Enable LK CLI to publish multiple H264 streams as a simulcast track (#604)

- add simulcast support for multiple H.264 streams.

Author: chenosaurus

README.md

@@ -203,6 +203,47 @@ lk room join --identity bot \
   <room_name>
 ```
 
+### Publish H.264 simulcast track from TCP
+
+You can publish multiple H.264 video tracks from different TCP ports as a single [Simulcast](https://docs.livekit.io/home/client/tracks/advanced/#video-simulcast) track.  This is done by using multiple `--publish` flags.
+
+The track will be published in simulcast mode if multiple `--publish` flags with the syntax `h264://<host>:<port>/<width>x<height>` are passed in as arguments.
+
+Example:
+
+Use Gstreamer to scale a video input to 3 resolutions (1920x1080, 1280x720, 640x360), encode each as a H.264 stream and output each H.264 stream on a different port using `tcpserversink`.
+
+```shell
+# Note: this is just an example of a Gstreamer pipeline structure
+# It uses a `tee` element to split the raw frame input to 3 pipelines for 
+# scaling to a specific resolution then encoding to H.264 byte stream.
+gst-launch-1.0 -e -v \
+  v4l2src device=<device> \
+  tee name=t  \
+  t. ! <scale to 1920x1080, H.264 encode elements> ! \
+      tcpserversink host=0.0.0.0 port=5005 sync=false async=false \
+  t. ! <scale to 1280x720, H.264 encode elements> ! \
+      tcpserversink host=0.0.0.0 port=5006 sync=false async=false \
+  t. ! <scale to 640x480, H.264 encode elements> ! \
+      tcpserversink host=0.0.0.0 port=5007 sync=false async=false
+```
+
+Use `livekit-cli` to publish the 3 resolution streams to a single Simulcast track.
+
+```shell
+lk room join --identity <name> --url "<url>" --api-key "<key>" --api-secret "<secret>" \
+--publish h264://127.0.0.1:5005/1920x1080 \
+--publish h264://127.0.0.1:5006/1280x720 \
+--publish h264://127.0.0.1:5007/640x480 <room>
+```
+
+Notes:
+- LiveKit CLI can only publish simulcast tracks using H.264 codec.
+- You can only use multiple `--publish` flags to create a simulcast track.
+- Using more than 1 `--publish` flag for other types of streams will not work.
+- Tracks will automatically be set to HIGH/MED/LOW resolution based on the order of their width.
+- If only 2 tracks are published, they will be published as HIGH and LOW resolution layers. 
+
 ### Publish streams from your application
 
 Using unix sockets, it's also possible to publish streams from your application. The tracks need to be encoded into

cmd/lk/join.go

@@ -22,7 +22,10 @@ import (
 	"os"
 	"os/signal"
 	"path/filepath"
+	"sort"
+	"strconv"
 	"strings"
+	"sync"
 	"syscall"
 	"time"
 
@@ -368,3 +371,176 @@ func publishReader(room *lksdk.Room,
 	}
 	return nil
 }
+
+// simulcastURLParts represents the parsed components of a simulcast URL
+type simulcastURLParts struct {
+	network string // "tcp" or "unix"
+	address string
+	width   uint32
+	height  uint32
+}
+
+// parseSimulcastURL validates and parses a simulcast URL in the format h264://<host:port>/<width>x<height> or h264://<socket_path>/<width>x<height>
+func parseSimulcastURL(url string) (*simulcastURLParts, error) {
+	matches := simulcastURLRegex.FindStringSubmatch(url)
+	if matches == nil {
+		return nil, fmt.Errorf("simulcast URL must be in format h264://<host:port>/<width>x<height> or h264://<socket_path>/<width>x<height>, got: %s", url)
+	}
+
+	address, widthStr, heightStr := matches[1], matches[2], matches[3]
+
+	// Parse dimensions
+	width, err := strconv.ParseUint(widthStr, 10, 32)
+	if err != nil || width == 0 {
+		return nil, fmt.Errorf("invalid width in URL %s: must be > 0", url)
+	}
+
+	height, err := strconv.ParseUint(heightStr, 10, 32)
+	if err != nil || height == 0 {
+		return nil, fmt.Errorf("invalid height in URL %s: must be > 0", url)
+	}
+
+	network := "unix"
+	if strings.Contains(address, ":") {
+		network = "tcp"
+	}
+
+	return &simulcastURLParts{
+		network: network,
+		address: address,
+		width:   uint32(width),
+		height:  uint32(height),
+	}, nil
+}
+
+// createSimulcastVideoTrack creates a simulcast video track from a TCP or Unix socket H.264 streams
+func createSimulcastVideoTrack(urlParts *simulcastURLParts, quality livekit.VideoQuality, fps float64, onComplete func()) (*lksdk.LocalTrack, error) {
+	conn, err := net.Dial(urlParts.network, urlParts.address)
+	if err != nil {
+		return nil, fmt.Errorf("failed to connect to %s://%s: %w", urlParts.network, urlParts.address, err)
+	}
+
+	var opts []lksdk.ReaderSampleProviderOption
+
+	// Add completion handler if provided
+	if onComplete != nil {
+		opts = append(opts, lksdk.ReaderTrackWithOnWriteComplete(onComplete))
+	}
+
+	// Set frame rate if FPS is set
+	if fps != 0 {
+		frameDuration := time.Second / time.Duration(fps)
+		opts = append(opts, lksdk.ReaderTrackWithFrameDuration(frameDuration))
+	}
+
+	// Configure simulcast layer
+	opts = append(opts, lksdk.ReaderTrackWithSampleOptions(lksdk.WithSimulcast("simulcast", &livekit.VideoLayer{
+		Quality: quality,
+		Width:   urlParts.width,
+		Height:  urlParts.height,
+	})))
+
+	return lksdk.NewLocalReaderTrack(conn, webrtc.MimeTypeH264, opts...)
+}
+
+// simulcastLayer represents a parsed H.264 stream with quality info
+type simulcastLayer struct {
+	url     string
+	parts   *simulcastURLParts
+	quality livekit.VideoQuality
+	name    string
+}
+
+// handleSimulcastPublish handles publishing multiple H.264 streams as a simulcast track
+func handleSimulcastPublish(room *lksdk.Room, urls []string, fps float64, onPublishComplete func(*lksdk.LocalTrackPublication)) error {
+	// Parse all URLs
+	var layers []simulcastLayer
+	for _, url := range urls {
+		parts, err := parseSimulcastURL(url)
+		if err != nil {
+			return fmt.Errorf("invalid simulcast URL %s: %w", url, err)
+		}
+		if parts != nil {
+			layers = append(layers, simulcastLayer{
+				url:   url,
+				parts: parts,
+			})
+		}
+	}
+
+	if len(layers) == 0 {
+		return fmt.Errorf("no valid simulcast URLs provided")
+	}
+
+	// Sort streams by width to determine quality levels
+	sort.Slice(layers, func(i, j int) bool {
+		return layers[i].parts.width < layers[j].parts.width
+	})
+
+	// Assign quality levels based on stream count and order
+	if len(layers) == 2 {
+		// 2 streams: low and high quality
+		layers[0].quality = livekit.VideoQuality_LOW
+		layers[0].name = "low"
+		layers[1].quality = livekit.VideoQuality_HIGH
+		layers[1].name = "high"
+	} else if len(layers) == 3 {
+		// 3 streams: low, medium, high quality
+		layers[0].quality = livekit.VideoQuality_LOW
+		layers[0].name = "low"
+		layers[1].quality = livekit.VideoQuality_MEDIUM
+		layers[1].name = "medium"
+		layers[2].quality = livekit.VideoQuality_HIGH
+		layers[2].name = "high"
+	} else {
+		return fmt.Errorf("simulcast requires 2 or 3 streams, got %d", len(layers))
+	}
+
+	// Create tracks for each stream
+	var tracks []*lksdk.LocalTrack
+	var trackNames []string
+
+	// Track completion - if any stream ends, signal completion
+	var pub *lksdk.LocalTrackPublication
+	completionSignaled := false
+	var completionMutex sync.Mutex
+
+	signalCompletion := func() {
+		completionMutex.Lock()
+		defer completionMutex.Unlock()
+		if !completionSignaled && onPublishComplete != nil {
+			completionSignaled = true
+			onPublishComplete(pub)
+		}
+	}
+
+	for _, layer := range layers {
+		track, err := createSimulcastVideoTrack(layer.parts, layer.quality, fps, signalCompletion)
+		if err != nil {
+			// Clean up any tracks we've already created
+			for _, t := range tracks {
+				t.Close()
+			}
+			return fmt.Errorf("failed to create %s quality track (%dx%d): %w",
+				layer.name, layer.parts.width, layer.parts.height, err)
+		}
+		tracks = append(tracks, track)
+		trackNames = append(trackNames, fmt.Sprintf("%s(%dx%d)", layer.name, layer.parts.width, layer.parts.height))
+	}
+
+	// Publish simulcast track
+	var err error
+	pub, err = room.LocalParticipant.PublishSimulcastTrack(tracks, &lksdk.TrackPublicationOptions{
+		Name: "simulcast",
+	})
+	if err != nil {
+		// Clean up tracks on publish failure
+		for _, track := range tracks {
+			track.Close()
+		}
+		return fmt.Errorf("failed to publish simulcast track: %w", err)
+	}
+
+	fmt.Printf("Successfully published H.264 simulcast track with qualities: %v\n", trackNames)
+	return nil
+}

cmd/lk/join_test.go

@@ -83,3 +83,47 @@ func TestParseSocketString(t *testing.T) {
 	assert.Equal(t, address, "foobar.com:1234")
 	assert.Equal(t, err, nil, "Expected no error for valid vp8 TCP socket")
 }
+
+func TestParseSimulcastURL(t *testing.T) {
+	// Test TCP format
+	parts, err := parseSimulcastURL("h264://localhost:8080/640x480")
+	assert.NoError(t, err, "Expected no error for valid TCP simulcast URL")
+	assert.Equal(t, "tcp", parts.network)
+	assert.Equal(t, "localhost:8080", parts.address)
+	assert.Equal(t, uint32(640), parts.width)
+	assert.Equal(t, uint32(480), parts.height)
+
+	// Test Unix socket format with multiple slashes
+	parts, err = parseSimulcastURL("h264:///tmp/my.socket/1280x720")
+	assert.NoError(t, err, "Expected no error for valid Unix socket simulcast URL")
+	assert.Equal(t, "unix", parts.network)
+	assert.Equal(t, "/tmp/my.socket", parts.address)
+	assert.Equal(t, uint32(1280), parts.width)
+	assert.Equal(t, uint32(720), parts.height)
+
+	// Test Unix socket format with nested paths
+	parts, err = parseSimulcastURL("h264:///tmp/deep/nested/path/my.socket/1920x1080")
+	assert.NoError(t, err, "Expected no error for valid nested path Unix socket simulcast URL")
+	assert.Equal(t, "unix", parts.network)
+	assert.Equal(t, "/tmp/deep/nested/path/my.socket", parts.address)
+	assert.Equal(t, uint32(1920), parts.width)
+	assert.Equal(t, uint32(1080), parts.height)
+
+	// Test simple socket name without path
+	parts, err = parseSimulcastURL("h264://mysocket/640x480")
+	assert.NoError(t, err, "Expected no error for simple socket name")
+	assert.Equal(t, "unix", parts.network)
+	assert.Equal(t, "mysocket", parts.address)
+	assert.Equal(t, uint32(640), parts.width)
+	assert.Equal(t, uint32(480), parts.height)
+
+	// Test invalid format
+	_, err = parseSimulcastURL("h264://localhost:8080")
+	assert.Error(t, err, "Expected error for URL without dimensions")
+
+	_, err = parseSimulcastURL("opus:///tmp/socket/640x480")
+	assert.Error(t, err, "Expected error for non-h264 protocol")
+
+	_, err = parseSimulcastURL("h264:///tmp/socket/invalidxinvalid")
+	assert.Error(t, err, "Expected error for invalid dimensions")
+}