Send SRT stream to WebRTC package

main.go

@@ -104,15 +104,16 @@ func main() {
 	remoteSdpChan := make(chan webrtc.SessionDescription)
 	localSdpChan := make(chan webrtc.SessionDescription)
 
-	// SRT channel for forwarding
+	// SRT channel for forwarding and webrtc
 	forwardingChannel := make(chan srt.Packet, 65536)
+	webrtcChannel := make(chan srt.Packet, 65536)
 
 	// Start stream, web and monitoring server, and stream forwarding
 	go forwarding.Serve(cfg.Forwarding, forwardingChannel)
 	go monitoring.Serve(&cfg.Monitoring)
-	go srt.Serve(&cfg.Srt, authBackend, forwardingChannel)
+	go srt.Serve(&cfg.Srt, authBackend, forwardingChannel, webrtcChannel)
 	go web.Serve(remoteSdpChan, localSdpChan, &cfg.Web)
-	go webrtc.Serve(remoteSdpChan, localSdpChan, &cfg.WebRTC)
+	go webrtc.Serve(remoteSdpChan, localSdpChan, webrtcChannel, &cfg.WebRTC)
 
 	// Wait for routines
 	select {}

stream/srt/handler.go

@@ -6,7 +6,7 @@ import (
 	"github.com/haivision/srtgo"
 )
 
-func handleStreamer(s *srtgo.SrtSocket, name string, clientDataChannels *[]chan Packet, forwardingChannel chan Packet) {
+func handleStreamer(s *srtgo.SrtSocket, name string, clientDataChannels *[]chan Packet, forwardingChannel, webrtcChannel chan Packet) {
 	log.Printf("New SRT streamer for stream %s", name)
 
 	// Create a new buffer
@@ -15,6 +15,7 @@ func handleStreamer(s *srtgo.SrtSocket, name string, clientDataChannels *[]chan
 
 	// Setup stream forwarding
 	forwardingChannel <- Packet{StreamName: name, PacketType: "register", Data: nil}
+	webrtcChannel <- Packet{StreamName: name, PacketType: "register", Data: nil}
 
 	// Read RTP packets forever and send them to the WebRTC Client
 	for {
@@ -36,12 +37,14 @@ func handleStreamer(s *srtgo.SrtSocket, name string, clientDataChannels *[]chan
 		data := make([]byte, n)
 		copy(data, buff[:n])
 		forwardingChannel <- Packet{StreamName: name, PacketType: "sendData", Data: data}
+		webrtcChannel <- Packet{StreamName: name, PacketType: "sendData", Data: data}
 		for _, dataChannel := range *clientDataChannels {
 			dataChannel <- Packet{StreamName: name, PacketType: "sendData", Data: data}
 		}
 	}
 
 	forwardingChannel <- Packet{StreamName: name, PacketType: "close", Data: nil}
+	webrtcChannel <- Packet{StreamName: name, PacketType: "close", Data: nil}
 }
 
 func handleViewer(s *srtgo.SrtSocket, name string, dataChannel chan Packet, dataChannels *[]chan Packet) {

stream/srt/srt.go

@@ -43,7 +43,7 @@ func splitHostPort(hostport string) (string, uint16) {
 }
 
 // Serve SRT server
-func Serve(cfg *Options, authBackend auth.Backend, forwardingChannel chan Packet) {
+func Serve(cfg *Options, authBackend auth.Backend, forwardingChannel, webrtcChannel chan Packet) {
 	// Start SRT in listening mode
 	log.Printf("SRT server listening on %s", cfg.ListenAddress)
 	host, port := splitHostPort(cfg.ListenAddress)
@@ -82,7 +82,7 @@ func Serve(cfg *Options, authBackend auth.Backend, forwardingChannel chan Packet
 				}
 			}
 
-			go handleStreamer(s, name, &clientDataChannels, forwardingChannel)
+			go handleStreamer(s, name, &clientDataChannels, forwardingChannel, webrtcChannel)
 		} else {
 			// password was not provided so it is a viewer
 			name := split[0]

stream/webrtc/webrtc.go

@@ -13,6 +13,7 @@ import (
 	"github.com/pion/webrtc/v3/pkg/media/ivfreader"
 	"github.com/pion/webrtc/v3/pkg/media/oggreader"
 	"gitlab.crans.org/nounous/ghostream/internal/monitoring"
+	"gitlab.crans.org/nounous/ghostream/stream/srt"
 )
 
 // Options holds web package configuration
@@ -242,10 +243,37 @@ func getPayloadType(m webrtc.MediaEngine, codecType webrtc.RTPCodecType, codecNa
 	panic(fmt.Sprintf("Remote peer does not support %s", codecName))
 }
 
+func waitForPackets(inputChannel chan srt.Packet) {
+	for {
+		var err error = nil
+		packet := <-inputChannel
+		switch packet.PacketType {
+		case "register":
+			log.Printf("WebRTC RegisterStream %s", packet.StreamName)
+			break
+		case "sendData":
+			log.Printf("WebRTC SendPacket %s", packet.StreamName)
+			// packet.Data
+			break
+		case "close":
+			log.Printf("WebRTC CloseConnection %s", packet.StreamName)
+			break
+		default:
+			log.Println("Unknown SRT packet type:", packet.PacketType)
+			break
+		}
+		if err != nil {
+			log.Printf("Error occured while receiving SRT packet of type %s: %s", packet.PacketType, err)
+		}
+	}
+}
+
 // Serve WebRTC media streaming server
-func Serve(remoteSdpChan, localSdpChan chan webrtc.SessionDescription, cfg *Options) {
+func Serve(remoteSdpChan, localSdpChan chan webrtc.SessionDescription, inputChannel chan srt.Packet, cfg *Options) {
 	log.Printf("WebRTC server using UDP from %d to %d", cfg.MinPortUDP, cfg.MaxPortUDP)
 
+	// FIXME: use data from inputChannel
+	go waitForPackets(inputChannel)
 	go playVideo()
 	go playAudio()