feat: gst compositor - work in progress

example/ion-cluster-simple/main.go

@@ -10,7 +10,7 @@ import (
 	"github.com/lucsky/cuid"
 	log "github.com/pion/ion-log"
 	sdk "github.com/pion/ion-sdk-go"
-	gst "github.com/pion/ion-sdk-go/pkg/gstreamer-sink"
+	gst "github.com/pion/ion-sdk-go/pkg/gst"
 	"github.com/pion/rtcp"
 	"github.com/pion/webrtc/v3"
 )

example/ion-sfu-gstreamer-receive/main.go

@@ -4,13 +4,13 @@ import (
 	"flag"
 	"fmt"
 	"runtime"
+	"strconv"
 	"strings"
 	"time"
 
 	log "github.com/pion/ion-log"
 	sdk "github.com/pion/ion-sdk-go"
-	gst "github.com/pion/ion-sdk-go/pkg/gstreamer-sink"
-	"github.com/pion/rtcp"
+	gst "github.com/pion/ion-sdk-go/pkg/gst"
 	"github.com/pion/webrtc/v3"
 )
 
@@ -49,46 +49,49 @@ func runClientLoop(addr, session string) {
 		return
 	}
 
-	// subscribe rtp from sessoin
-	// comment this if you don't need save to file
-	c.OnTrack = func(track *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) {
-		// Send a PLI on an interval so that the publisher is pushing a keyframe every rtcpPLIInterval
-		go func() {
-			ticker := time.NewTicker(time.Second * 3)
-			for range ticker.C {
-				rtcpSendErr := c.GetSubTransport().GetPeerConnection().WriteRTCP([]rtcp.Packet{&rtcp.PictureLossIndication{MediaSSRC: uint32(track.SSRC())}})
-				if rtcpSendErr != nil {
-					fmt.Println(rtcpSendErr)
-				}
-			}
-		}()
-
-		codecName := strings.Split(track.Codec().RTPCodecCapability.MimeType, "/")[1]
-		fmt.Printf("Track has started, of type %d: %s \n", track.PayloadType(), codecName)
-		pipeline := gst.CreatePipeline(strings.ToLower(codecName))
-		pipeline.Start()
-		defer pipeline.Stop()
-		buf := make([]byte, 1400)
-		for {
-			i, _, readErr := track.Read(buf)
-			if readErr != nil {
-				log.Errorf("%v", readErr)
-				return
-			}
-
-			pipeline.Push(buf[:i])
+	filename := strings.ReplaceAll(session, " ", "-") + "-" + strconv.FormatInt(time.Now().Unix(), 10) + ".avi"
+	destination := "filesink location=./" + filename
+	videoEncoder := "x264enc bframes=0 speed-preset=ultrafast key-int-max=60 ! video/x-h264, profile=baseline "
+	//destination="movie.avi"
+
+	compositorString := fmt.Sprintf(`
+		qtmux name=savemux ! queue ! %s sync=false async=false
+			vtee. ! queue ! savemux.
+			atee. ! queue ! savemux.
+
+	`, destination)
+
+	log.Infof("Beginning Recording Compositor[%s]: %s -> %s", addr, videoEncoder, filename)
+
+	pipelineID := addr + "|" + filename
+	log.Infof("connected pipeline[%s]!", pipelineID)
+	compositor := gst.NewCompositorPipeline(compositorString)
+
+	c.OnTrack = func(t *webrtc.TrackRemote, r *webrtc.RTPReceiver) {
+		log.Debugf("pipeline[%s] got track: %#v", pipelineID, t)
+		if t.Kind() == webrtc.RTPCodecTypeVideo && t.Codec().MimeType != webrtc.MimeTypeH264 {
+			log.Errorf("only h264 video is supported currently, please help me improve this example :) ")
+			panic("exiting")
 		}
+
+		compositor.AddInputTrack(t, c.GetSubTransport().GetPeerConnection())
+	}
+
+	compositor.OnRemoveTrack = func(t *webrtc.TrackRemote) {
+		log.Infof("REMOVED TRACK", t.Codec(), len(compositor.Tracks))
 	}
 
 	// client join a session
 	err = c.Join(session)
-
-	// publish file to session if needed
 	if err != nil {
-		log.Errorf("err=%v", err)
+		log.Errorf("error joining room:", err)
+		panic(err)
 	}
 
-	select {}
+	log.Infof("joined pipeline[%s]!", pipelineID)
+	compositor.Play()
+	log.Infof("compositing!")
+
 }
 
 func main() {
@@ -104,5 +107,6 @@ func main() {
 	flag.Parse()
 
 	go runClientLoop(addr, session)
-	gst.StartMainLoop()
+
+	gst.MainLoop()
 }

go.mod

@@ -15,6 +15,8 @@ require (
 	github.com/pion/rtp v1.6.2
 	github.com/pion/sdp/v3 v3.0.4
 	github.com/pion/webrtc/v3 v3.0.11
+	github.com/square/go-jose/v3 v3.0.0-20200630053402-0a67ce9b0693
 	google.golang.org/grpc v1.35.0
+	google.golang.org/protobuf v1.25.0
 	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
 )

pkg/gst/compositor.go

@@ -0,0 +1,161 @@
+package gst
+
+/*
+#cgo pkg-config: gstreamer-1.0 gstreamer-app-1.0 gstreamer-video-1.0
+
+#include "gst.h"
+
+*/
+import "C"
+import (
+	"fmt"
+	"runtime"
+	"strings"
+	"sync"
+	"unsafe"
+
+	log "github.com/pion/ion-log"
+	"github.com/pion/rtcp"
+	"github.com/pion/webrtc/v3"
+)
+
+// CompositorPipeline will decode incoming tracks in a single pipeline and compose the streams
+type CompositorPipeline struct {
+	mu       sync.Mutex
+	Pipeline *C.GstElement
+	OnRemoveTrack func(track *webrtc.TrackRemote)
+	Tracks              map[string]*webrtc.TrackRemote
+
+	trackBins              map[string]*C.GstElement
+	trackKeyframeCallbacks map[string]func()
+}
+
+// NewCompositorPipeline will create a pipeline controller for AV compositing.
+// It will include tee's (vtee,atee) for linking extra elements to the composited output using the extraPipelineStr
+// You can add more than one audio/video tracks to a pipeline, but having zero tracks will end the pipeline
+func NewCompositorPipeline(extraPipelineStr string) *CompositorPipeline {
+	pipelineStr := `
+		compositor name=vmix background=black ! video/x-raw,width=1920,height=1080,framerate=30/1,format=UYVY ! queue ! tee name=vtee 
+			vtee. ! queue ! glimagesink sync=false 
+		audiomixer name=amix ! queue ! tee name=atee 
+			atee. ! queue ! audioconvert ! autoaudiosink
+	` + extraPipelineStr
+
+	log.Infof("Creating gst compositor pipeline:\n%s", pipelineStr)
+	pipelineStrUnsafe := C.CString(pipelineStr)
+	defer C.free(unsafe.Pointer(pipelineStrUnsafe))
+
+	c := &CompositorPipeline{
+		Pipeline:  C.gstreamer_create_pipeline(pipelineStrUnsafe),
+		Tracks: make(map[string]*webrtc.TrackRemote),
+		trackBins: make(map[string]*C.GstElement),
+	}
+	runtime.SetFinalizer(c, func(c *CompositorPipeline) {
+		log.Infof("Destroying compositor pipeline...")
+		c.destroy()
+	})
+	C.gstreamer_start_pipeline(c.Pipeline)
+	return c
+}
+
+func (c *CompositorPipeline) AddInputTrack(t *webrtc.TrackRemote, pc *webrtc.PeerConnection) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	inputBin := fmt.Sprintf("appsrc format=time is-live=true do-timestamp=true name=%s ! application/x-rtp ", t.ID())
+
+	switch strings.ToLower(t.Codec().MimeType) {
+	case "audio/opus":
+		inputBin += ", encoding-name=OPUS, payload=96 ! rtpopusdepay ! queue ! opusdec "
+	case "audio/g722":
+		inputBin += " clock-rate=8000 ! rtpg722depay ! decodebin "
+	case "video/vp8":
+		inputBin += ", encoding-name=VP8-DRAFT-IETF-01 ! rtpvp8depay ! avdec_vp8 "
+	case "viode/vp9":
+		inputBin += " ! rtpvp9depay ! decodebin "
+	case "video/h264":
+		inputBin += fmt.Sprintf(", payload=%d ! rtph264depay ! queue ! %s !  videoconvert ! queue ", t.PayloadType(), getDecoderString())
+	default:
+		panic(fmt.Sprintf("couldn't build gst pipeline for codec: %s ", t.Codec().MimeType))
+	}
+
+	log.Debugf("adding input track with bin: %s", inputBin)
+	inputBinUnsafe := C.CString(inputBin)
+	// defer C.free(unsafe.Pointer(&inputBinUnsafe))
+	trackIdUnsafe := C.CString(t.ID())
+	// defer C.free(unsafe.Pointer(&trackIdUnsafe))
+
+	isVideo := t.Kind() == webrtc.RTPCodecTypeVideo
+	bin := C.gstreamer_compositor_add_input_track(c.Pipeline, inputBinUnsafe, trackIdUnsafe, C.bool(isVideo))
+	c.trackBins[t.ID()] = bin
+	c.Tracks[t.ID()] = t
+
+	if t.Kind() == webrtc.RTPCodecTypeVideo {
+		boundRemoteTrackKeyframeCallbacks[t.ID()] = func() {
+			log.Debugf("sending pli for track %s", t.ID())
+			rtcpSendErr := pc.WriteRTCP([]rtcp.Packet{&rtcp.PictureLossIndication{MediaSSRC: uint32(t.SSRC())}})
+			if rtcpSendErr != nil {
+				fmt.Println(rtcpSendErr)
+			}
+		}
+	}
+	go c.bindTrackToAppsrc(t)
+}
+
+func (c *CompositorPipeline) Play() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	C.gstreamer_play_pipeline(c.Pipeline)
+}
+
+func (c *CompositorPipeline) Stop() {
+	C.gstreamer_stop_pipeline(c.Pipeline)
+}
+
+func (c *CompositorPipeline) destroy() {
+	for _, b := range c.trackBins {
+		C.gst_object_unref(C.gpointer(b))
+	}
+}
+
+func (c *CompositorPipeline) bindTrackToAppsrc(t *webrtc.TrackRemote) {
+	buf := make([]byte, 1400)
+	for {
+		i, _, readErr := t.Read(buf)
+		if readErr != nil {
+			log.Debugf("end of track %v: cleaning up pipeline", t.ID())
+
+			trackBin := c.trackBins[t.ID()]
+			C.gstreamer_compositor_remove_input_track(c.Pipeline, trackBin, C.bool(t.Kind() == webrtc.RTPCodecTypeVideo))
+			delete(c.trackBins, t.ID())
+			delete(c.Tracks, t.ID())
+			delete(boundRemoteTrackKeyframeCallbacks, t.ID())
+
+			if c.OnRemoveTrack != nil {
+				c.OnRemoveTrack(t)
+			}
+			// panic(readErr)
+			return
+		}
+		c.pushAppsrc(buf[:i], t.ID())
+	}
+}
+
+// Push pushes a buffer on the appsrc of the GStreamer Pipeline
+func (c *CompositorPipeline) pushAppsrc(buffer []byte, appsrc string) {
+	b := C.CBytes(buffer)
+	defer C.free(b)
+	inputElementUnsafe := C.CString(appsrc)
+	// defer C.free(unsafe.Pointer(&inputElementUnsafe))
+	C.gstreamer_receive_push_buffer(c.Pipeline, b, C.int(len(buffer)), inputElementUnsafe)
+}
+
+//export goHandleAppsrcForceKeyUnit
+func goHandleAppsrcForceKeyUnit(remoteTrackID *C.char) {
+	id := C.GoString(remoteTrackID)
+	log.Debugf("go forceKeyUnit: %v", id)
+	if trackSendPLI, ok := boundRemoteTrackKeyframeCallbacks[id]; ok {
+		trackSendPLI()
+	}
+}