Try to mux into RTP with FFMPEG bindings

.gitignore

@@ -17,9 +17,3 @@ pkged.go
 # Profiler and test files
 *.prof
 *.test
-
-# Javascript tools
-.eslintrc.js
-node_modules
-package.json
-package-lock.json

.gitlab-ci.yml

@@ -2,18 +2,8 @@ stages:
   - test
   - quality-assurance
 
-.go-cache:
-  variables:
-    GOPATH: $CI_PROJECT_DIR/.go
-  before_script:
-    - mkdir -p .go
-  cache:
-    paths:
-      - .go/pkg/mod/
-
 unit_tests:
   image: golang:1.15-alpine
-  extends: .go-cache
   stage: test
   before_script:
     - apk add --no-cache -X http://dl-cdn.alpinelinux.org/alpine/edge/community build-base ffmpeg gcc libsrt-dev
@@ -28,7 +18,6 @@ unit_tests:
 
 linters:
   image: golang:1.15-alpine
-  extends: .go-cache
   stage: quality-assurance
   script:
     - go get -u golang.org/x/lint/golint

docs/ghostream.example.yml

@@ -38,17 +38,13 @@ auth:
 ## Stream forwarding ##
 # Forward an incoming stream to other servers
 # The URL can be anything FFMpeg can accept as an stream output
-# If a file is specified, the name may contains %Y, %m, %d, %H, %M or %S
-# that will be replaced by the current date information.
 forwarding:
   # By default nothing is forwarded.
   #
   # This example forwards a stream named "demo" to Twitch and YouTube,
-  # and save the record in a timestamped-file,
   #demo:
   #  - rtmp://live-cdg.twitch.tv/app/STREAM_KEY
   #  - rtmp://a.rtmp.youtube.com/live2/STREAM_KEY
-  #  - /home/ghostream/lives/%name/live-%Y-%m-%d-%H-%M-%S.flv
 
 ## Prometheus monitoring ##
 # Expose a monitoring endpoint for Prometheus

go.mod

@@ -3,8 +3,8 @@ module gitlab.crans.org/nounous/ghostream
 go 1.13
 
 require (
+	github.com/3d0c/gmf v0.0.0-20200614092945-e58d8d5a6035
 	github.com/go-ldap/ldap/v3 v3.2.3
-	github.com/gorilla/websocket v1.4.0
 	github.com/haivision/srtgo v0.0.0-20200731151239-e00427ae473a
 	github.com/markbates/pkger v0.17.1
 	github.com/pion/rtp v1.6.0

go.sum

@@ -7,6 +7,8 @@ dmitri.shuralyov.com/html/belt v0.0.0-20180602232347-f7d459c86be0/go.mod h1:JLBr
 dmitri.shuralyov.com/service/change v0.0.0-20181023043359-a85b471d5412/go.mod h1:a1inKt/atXimZ4Mv927x+r7UpyzRUf4emIoiiSC2TN4=
 dmitri.shuralyov.com/state v0.0.0-20180228185332-28bcc343414c/go.mod h1:0PRwlb0D6DFvNNtx+9ybjezNCa8XF0xaYcETyp6rHWU=
 git.apache.org/thrift.git v0.0.0-20180902110319-2566ecd5d999/go.mod h1:fPE2ZNJGynbRyZ4dJvy6G277gSllfV2HJqblrnkyeyg=
+github.com/3d0c/gmf v0.0.0-20200614092945-e58d8d5a6035 h1:QZb1aMKxiYdGGieyIDmXuw9I9YcGWGViTrpQ6vcZX7Q=
+github.com/3d0c/gmf v0.0.0-20200614092945-e58d8d5a6035/go.mod h1:0QMRcUq2JsDECeAq7bj4h79k7XbhtTsrPUQf6G7qfPs=
 github.com/Azure/go-ntlmssp v0.0.0-20200615164410-66371956d46c h1:/IBSNwUN8+eKzUzbJPqhK839ygXJ82sde8x3ogr6R28=
 github.com/Azure/go-ntlmssp v0.0.0-20200615164410-66371956d46c/go.mod h1:chxPXzSsl7ZWRAuOIE23GDNzjWuZquvFlgA8xmpunjU=
 github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ=
@@ -113,7 +115,6 @@ github.com/googleapis/gax-go v2.0.0+incompatible/go.mod h1:SFVmujtThgffbyetf+mdk
 github.com/googleapis/gax-go/v2 v2.0.3/go.mod h1:LLvjysVCY1JZeum8Z6l8qUty8fiNwE08qbEPm1M08qg=
 github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1 h1:EGx4pi6eqNxGaHF6qqu48+N2wcFQ5qg5FXgOdqsJ5d8=
 github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
-github.com/gorilla/websocket v1.4.0 h1:WDFjx/TMzVgy9VdMMQi2K2Emtwi2QcUQsztZ/zLaH/Q=
 github.com/gorilla/websocket v1.4.0/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ=
 github.com/gregjones/httpcache v0.0.0-20180305231024-9cad4c3443a7/go.mod h1:FecbI9+v66THATjSRHfNgh1IVFe/9kFxbXtjV0ctIMA=
 github.com/grpc-ecosystem/go-grpc-middleware v1.0.0/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs=

internal/config/config.go

@@ -2,6 +2,7 @@
 package config
 
 import (
+	"gitlab.crans.org/nounous/ghostream/transcoder/audio"
 	"net"
 
 	"github.com/sherifabdlnaby/configuro"
@@ -59,6 +60,10 @@ func New() *Config {
 			ListenAddress: ":8023",
 		},
 		Transcoder: transcoder.Options{
+			Audio: audio.Options{
+				Enabled: true,
+				Bitrate: 160,
+			},
 			Text: text.Options{
 				Enabled:   false,
 				Width:     80,

main.go

@@ -10,7 +10,7 @@ import (
 	"gitlab.crans.org/nounous/ghostream/auth"
 	"gitlab.crans.org/nounous/ghostream/internal/config"
 	"gitlab.crans.org/nounous/ghostream/internal/monitoring"
-	"gitlab.crans.org/nounous/ghostream/messaging"
+	"gitlab.crans.org/nounous/ghostream/stream"
 	"gitlab.crans.org/nounous/ghostream/stream/forwarding"
 	"gitlab.crans.org/nounous/ghostream/stream/srt"
 	"gitlab.crans.org/nounous/ghostream/stream/telnet"
@@ -38,8 +38,15 @@ func main() {
 		defer authBackend.Close()
 	}
 
+	// WebRTC session description channels
+	remoteSdpChan := make(chan struct {
+		StreamID          string
+		RemoteDescription webrtc.SessionDescription
+	})
+	localSdpChan := make(chan webrtc.SessionDescription)
+
 	// Init streams messaging
-	streams := messaging.New()
+	streams := make(map[string]*stream.Stream)
 
 	// Start routines
 	go transcoder.Init(streams, &cfg.Transcoder)
@@ -47,8 +54,8 @@ func main() {
 	go monitoring.Serve(&cfg.Monitoring)
 	go srt.Serve(streams, authBackend, &cfg.Srt)
 	go telnet.Serve(streams, &cfg.Telnet)
-	go web.Serve(streams, &cfg.Web)
-	go webrtc.Serve(streams, &cfg.WebRTC)
+	go web.Serve(streams, remoteSdpChan, localSdpChan, &cfg.Web)
+	go webrtc.Serve(streams, remoteSdpChan, localSdpChan, &cfg.WebRTC)
 
 	// Wait for routines
 	select {}

messaging/quality.go

@@ -1,96 +0,0 @@
-// Package messaging defines a structure to communication between inputs and outputs
-package messaging
-
-import (
-	"sync"
-
-	"github.com/pion/webrtc/v3"
-)
-
-// Quality holds a specific stream quality.
-// It makes packages able to subscribe to an incoming stream.
-type Quality struct {
-	// Type of the quality
-	Name string
-
-	// Source Stream
-	Stream *Stream
-
-	// Incoming data come from this channel
-	Broadcast chan<- []byte
-
-	// Incoming data will be outputted to all those outputs.
-	// Use a map to be able to delete an item.
-	outputs map[chan []byte]struct{}
-
-	// Mutex to lock outputs map
-	lockOutputs sync.Mutex
-
-	// WebRTC session descriptor exchange.
-	// When new client connects, a SDP arrives on WebRtcRemoteSdp,
-	// then webrtc package answers on WebRtcLocalSdp.
-	WebRtcLocalSdp  chan webrtc.SessionDescription
-	WebRtcRemoteSdp chan webrtc.SessionDescription
-}
-
-func newQuality(name string, stream *Stream) (q *Quality) {
-	q = &Quality{Name: name}
-	q.Stream = stream
-	broadcast := make(chan []byte, 1024)
-	q.Broadcast = broadcast
-	q.outputs = make(map[chan []byte]struct{})
-	q.WebRtcLocalSdp = make(chan webrtc.SessionDescription, 1)
-	q.WebRtcRemoteSdp = make(chan webrtc.SessionDescription, 1)
-	go q.run(broadcast)
-	return q
-}
-
-func (q *Quality) run(broadcast <-chan []byte) {
-	for msg := range broadcast {
-		q.lockOutputs.Lock()
-		for output := range q.outputs {
-			select {
-			case output <- msg:
-			default:
-				// If full, do a ring buffer
-				// Check that output is not of size zero
-				if len(output) > 1 {
-					<-output
-				}
-			}
-		}
-		q.lockOutputs.Unlock()
-	}
-
-	// Incoming chan has been closed, close all outputs
-	q.lockOutputs.Lock()
-	for ch := range q.outputs {
-		delete(q.outputs, ch)
-		close(ch)
-	}
-	q.lockOutputs.Unlock()
-}
-
-// Close the incoming chan, this will also delete all outputs.
-func (q *Quality) Close() {
-	close(q.Broadcast)
-}
-
-// Register a new output on a stream.
-func (q *Quality) Register(output chan []byte) {
-	q.lockOutputs.Lock()
-	q.outputs[output] = struct{}{}
-	q.lockOutputs.Unlock()
-}
-
-// Unregister removes an output.
-func (q *Quality) Unregister(output chan []byte) {
-	// Make sure we did not already close this output
-	q.lockOutputs.Lock()
-	_, ok := q.outputs[output]
-	if ok {
-		delete(q.outputs, output)
-		close(output)
-	}
-	defer q.lockOutputs.Unlock()
-}

messaging/stream.go

@@ -1,84 +0,0 @@
-// Package messaging defines a structure to communication between inputs and outputs
-package messaging
-
-import (
-	"errors"
-	"sync"
-)
-
-// Stream makes packages able to subscribe to an incoming stream
-type Stream struct {
-	// Different qualities of this stream
-	qualities map[string]*Quality
-
-	// Mutex to lock outputs map
-	lockQualities sync.Mutex
-
-	// Count clients for statistics
-	nbClients int
-}
-
-func newStream() (s *Stream) {
-	s = &Stream{}
-	s.qualities = make(map[string]*Quality)
-	s.nbClients = 0
-	return s
-}
-
-// Close stream.
-func (s *Stream) Close() {
-	for quality := range s.qualities {
-		s.DeleteQuality(quality)
-	}
-}
-
-// CreateQuality creates a new quality associated with this stream.
-func (s *Stream) CreateQuality(name string) (quality *Quality, err error) {
-	// If quality already exist, fail
-	if _, ok := s.qualities[name]; ok {
-		return nil, errors.New("quality already exists")
-	}
-
-	s.lockQualities.Lock()
-	quality = newQuality(name, s)
-	s.qualities[name] = quality
-	s.lockQualities.Unlock()
-	return quality, nil
-}
-
-// DeleteQuality removes a stream quality.
-func (s *Stream) DeleteQuality(name string) {
-	// Make sure we did not already close this output
-	s.lockQualities.Lock()
-	if _, ok := s.qualities[name]; ok {
-		s.qualities[name].Close()
-		delete(s.qualities, name)
-	}
-	s.lockQualities.Unlock()
-}
-
-// GetQuality gets a specific stream quality.
-func (s *Stream) GetQuality(name string) (quality *Quality, err error) {
-	s.lockQualities.Lock()
-	quality, ok := s.qualities[name]
-	s.lockQualities.Unlock()
-	if !ok {
-		return nil, errors.New("quality does not exist")
-	}
-	return quality, nil
-}
-
-// ClientCount returns the number of clients.
-func (s *Stream) ClientCount() int {
-	return s.nbClients
-}
-
-// IncrementClientCount increments the number of clients.
-func (s *Stream) IncrementClientCount() {
-	s.nbClients++
-}
-
-// DecrementClientCount decrements the number of clients.
-func (s *Stream) DecrementClientCount() {
-	s.nbClients--
-}

messaging/streams.go

@@ -1,98 +0,0 @@
-// Package messaging defines a structure to communication between inputs and outputs
-package messaging
-
-import (
-	"errors"
-	"log"
-	"sync"
-)
-
-// Streams hold all application streams.
-type Streams struct {
-	// Associate each stream name to the stream
-	streams map[string]*Stream
-
-	// Mutex to lock streams
-	lockStreams sync.Mutex
-
-	// Subscribers get notified when a new stream is created
-	// Use a map to be able to delete a subscriber
-	eventSubscribers map[chan string]struct{}
-
-	// Mutex to lock eventSubscribers
-	lockSubscribers sync.Mutex
-}
-
-// New creates a new stream list.
-func New() (l *Streams) {
-	l = &Streams{}
-	l.streams = make(map[string]*Stream)
-	l.eventSubscribers = make(map[chan string]struct{})
-	return l
-}
-
-// Subscribe to get notified on new stream.
-func (l *Streams) Subscribe(output chan string) {
-	l.lockSubscribers.Lock()
-	l.eventSubscribers[output] = struct{}{}
-	l.lockSubscribers.Unlock()
-}
-
-// Unsubscribe to no longer get notified on new stream.
-func (l *Streams) Unsubscribe(output chan string) {
-	// Make sure we did not already delete this subscriber
-	l.lockSubscribers.Lock()
-	if _, ok := l.eventSubscribers[output]; ok {
-		delete(l.eventSubscribers, output)
-	}
-	l.lockSubscribers.Unlock()
-}
-
-// Create a new stream.
-func (l *Streams) Create(name string) (s *Stream, err error) {
-	// If stream already exist, fail
-	if _, ok := l.streams[name]; ok {
-		return nil, errors.New("stream already exists")
-	}
-
-	// Create stream
-	s = newStream()
-	l.lockStreams.Lock()
-	l.streams[name] = s
-	l.lockStreams.Unlock()
-
-	// Notify
-	l.lockSubscribers.Lock()
-	for sub := range l.eventSubscribers {
-		select {
-		case sub <- name:
-		default:
-			log.Printf("Failed to announce stream '%s' to subscriber", name)
-		}
-	}
-	l.lockSubscribers.Unlock()
-	return s, nil
-}
-
-// Get a stream.
-func (l *Streams) Get(name string) (s *Stream, err error) {
-	// If stream does exist, return it
-	l.lockStreams.Lock()
-	s, ok := l.streams[name]
-	l.lockStreams.Unlock()
-	if !ok {
-		return nil, errors.New("stream does not exist")
-	}
-	return s, nil
-}
-
-// Delete a stream.
-func (l *Streams) Delete(name string) {
-	// Make sure we did not already delete this stream
-	l.lockStreams.Lock()
-	if _, ok := l.streams[name]; ok {
-		l.streams[name].Close()
-		delete(l.streams, name)
-	}
-	l.lockStreams.Unlock()
-}

messaging/streams_test.go

@@ -1,55 +0,0 @@
-package messaging
-
-import "testing"
-
-func TestWithOneStream(t *testing.T) {
-	streams := New()
-
-	// Subscribe to new streams
-	event := make(chan string, 8)
-	streams.Subscribe(event)
-
-	// Create a stream
-	stream, err := streams.Create("demo")
-	if err != nil {
-		t.Errorf("Failed to create stream")
-	}
-
-	// Check that we receive the creation event
-	e := <-event
-	if e != "demo" {
-		t.Errorf("Message has wrong content: %s != demo", e)
-	}
-
-	// Create a quality
-	quality, err := stream.CreateQuality("source")
-	if err != nil {
-		t.Errorf("Failed to create quality")
-	}
-
-	// Register one output
-	output := make(chan []byte, 64)
-	quality.Register(output)
-	stream.IncrementClientCount()
-
-	// Try to pass one message
-	quality.Broadcast <- []byte("hello world")
-	msg := <-output
-	if string(msg) != "hello world" {
-		t.Errorf("Message has wrong content: %s != hello world", msg)
-	}
-
-	// Check client count
-	if count := stream.ClientCount(); count != 1 {
-		t.Errorf("Client counter returned %d, expected 1", count)
-	}
-
-	// Unregister
-	quality.Unregister(output)
-	stream.DecrementClientCount()
-
-	// Check client count
-	if count := stream.ClientCount(); count != 0 {
-		t.Errorf("Client counter returned %d, expected 0", count)
-	}
-}

stream/forwarding/forwarding.go

@@ -3,13 +3,11 @@ package forwarding
 
 import (
 	"bufio"
-	"fmt"
 	"log"
 	"os/exec"
-	"strings"
 	"time"
 
-	"gitlab.crans.org/nounous/ghostream/messaging"
+	"gitlab.crans.org/nounous/ghostream/stream"
 )
 
 // Options to configure the stream forwarding.
@@ -17,65 +15,43 @@ import (
 type Options map[string][]string
 
 // Serve handles incoming packets from SRT and forward them to other external services
-func Serve(streams *messaging.Streams, cfg Options) {
+func Serve(streams map[string]*stream.Stream, cfg Options) {
 	if len(cfg) < 1 {
 		// No forwarding, ignore
 		return
 	}
 
-	// Subscribe to new stream event
-	event := make(chan string, 8)
-	streams.Subscribe(event)
 	log.Printf("Stream forwarding initialized")
+	for {
+		for name, st := range streams {
+			fwdCfg, ok := cfg[name]
+			if !ok {
+				// Not configured
+				continue
+			}
 
-	// For each new stream
-	for name := range event {
-		streamCfg, ok := cfg[name]
-		if !ok {
-			// Not configured
-			continue
+			// Start forwarding
+			log.Printf("Starting forwarding for '%s'", name)
+			go forward(st, fwdCfg)
 		}
 
-		// Get stream
-		stream, err := streams.Get(name)
-		if err != nil {
-			log.Printf("Failed to get stream '%s'", name)
-		}
-
-		// Get specific quality
-		// FIXME: make it possible to forward other qualities
-		qualityName := "source"
-		quality, err := stream.GetQuality(qualityName)
-		if err != nil {
-			log.Printf("Failed to get quality '%s'", qualityName)
-		}
-
-		// Start forwarding
-		log.Printf("Starting forwarding for '%s' quality '%s'", name, qualityName)
-		go forward(name, quality, streamCfg)
+		// Regulary pull stream list,
+		// it may be better to tweak the messaging system
+		// to get an event on a new stream.
+		time.Sleep(time.Second)
 	}
 }
 
 // Start a FFMPEG instance and redirect stream output to forwarded streams
-func forward(streamName string, q *messaging.Quality, fwdCfg []string) {
+func forward(st *stream.Stream, fwdCfg []string) {
 	output := make(chan []byte, 1024)
-	q.Register(output)
+	st.Register(output)
 
 	// Launch FFMPEG instance
-	params := []string{"-hide_banner", "-loglevel", "error", "-i", "pipe:0"}
+	params := []string{"-hide_banner", "-loglevel", "error", "-re", "-i", "pipe:0"}
 	for _, url := range fwdCfg {
-		// If the url should be date-formatted, replace special characters with the current time information
-		now := time.Now()
-		formattedURL := strings.ReplaceAll(url, "%Y", fmt.Sprintf("%04d", now.Year()))
-		formattedURL = strings.ReplaceAll(formattedURL, "%m", fmt.Sprintf("%02d", now.Month()))
-		formattedURL = strings.ReplaceAll(formattedURL, "%d", fmt.Sprintf("%02d", now.Day()))
-		formattedURL = strings.ReplaceAll(formattedURL, "%H", fmt.Sprintf("%02d", now.Hour()))
-		formattedURL = strings.ReplaceAll(formattedURL, "%M", fmt.Sprintf("%02d", now.Minute()))
-		formattedURL = strings.ReplaceAll(formattedURL, "%S", fmt.Sprintf("%02d", now.Second()))
-		formattedURL = strings.ReplaceAll(formattedURL, "%name", streamName)
-
 		params = append(params, "-f", "flv", "-preset", "ultrafast", "-tune", "zerolatency",
-			"-c", "copy", formattedURL)
+			"-c", "copy", url)
 	}
 	ffmpeg := exec.Command("ffmpeg", params...)
 
@@ -101,14 +77,14 @@ func forward(streamName string, q *messaging.Quality, fwdCfg []string) {
 		_ = input.Close()
 		_ = errOutput.Close()
 		_ = ffmpeg.Process.Kill()
-		q.Unregister(output)
+		st.Unregister(output)
 	}()
 
 	// Log standard error output
 	go func() {
 		scanner := bufio.NewScanner(errOutput)
 		for scanner.Scan() {
-			log.Printf("[FORWARDING FFMPEG %s] %s", streamName, scanner.Text())
+			log.Printf("[FORWARDING FFMPEG] %s", scanner.Text())
 		}
 	}()
 

stream/forwarding/forwarding_test.go

@@ -6,7 +6,7 @@ import (
 	"testing"
 	"time"
 
-	"gitlab.crans.org/nounous/ghostream/messaging"
+	"gitlab.crans.org/nounous/ghostream/stream"
 	"gitlab.crans.org/nounous/ghostream/stream/srt"
 )
 
@@ -35,7 +35,7 @@ func TestForwardStream(t *testing.T) {
 	cfg["demo"] = []string{"rtmp://127.0.0.1:1936/live/app"}
 
 	// Register forwarding stream list
-	streams := messaging.New()
+	streams := make(map[string]*stream.Stream)
 	go Serve(streams, cfg)
 
 	// Serve SRT Server without authentification backend

stream/messaging.go

@@ -0,0 +1,99 @@
+// Package stream defines a structure to communication between inputs and outputs
+package stream
+
+import (
+	"sync"
+)
+
+// Stream makes packages able to subscribe to an incoming stream
+type Stream struct {
+	// Incoming data come from this channel
+	Broadcast chan<- []byte
+
+	// Use a map to be able to delete an item
+	outputs map[chan []byte]struct{}
+
+	// Count clients for statistics
+	nbClients int
+
+	// Mutex to lock outputs map
+	lock sync.Mutex
+}
+
+// New creates a new stream.
+func New() *Stream {
+	s := &Stream{}
+	broadcast := make(chan []byte, 1024)
+	s.Broadcast = broadcast
+	s.outputs = make(map[chan []byte]struct{})
+	s.nbClients = 0
+	go s.run(broadcast)
+	return s
+}
+
+func (s *Stream) run(broadcast <-chan []byte) {
+	for msg := range broadcast {
+		s.lock.Lock()
+		for output := range s.outputs {
+			select {
+			case output <- msg:
+			default:
+				// If full, do a ring buffer
+				// Check that output is not of size zero
+				if len(output) > 1 {
+					<-output
+				}
+			}
+		}
+		s.lock.Unlock()
+	}
+
+	// Incoming chan has been closed, close all outputs
+	s.lock.Lock()
+	for ch := range s.outputs {
+		delete(s.outputs, ch)
+		close(ch)
+	}
+	s.lock.Unlock()
+}
+
+// Close the incoming chan, this will also delete all outputs
+func (s *Stream) Close() {
+	close(s.Broadcast)
+}
+
+// Register a new output on a stream.
+func (s *Stream) Register(output chan []byte) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	s.outputs[output] = struct{}{}
+}
+
+// Unregister removes an output.
+// If hidden in true, then do not count this client.
+func (s *Stream) Unregister(output chan []byte) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	// Make sure we did not already close this output
+	_, ok := s.outputs[output]
+	if ok {
+		delete(s.outputs, output)
+		close(output)
+	}
+}
+
+// ClientCount returns the number of clients
+func (s *Stream) ClientCount() int {
+	return s.nbClients
+}
+
+// IncrementClientCount increments the number of clients
+func (s *Stream) IncrementClientCount() {
+	s.nbClients++
+}
+
+// DecrementClientCount decrements the number of clients
+func (s *Stream) DecrementClientCount() {
+	s.nbClients--
+}

stream/messaging_test.go

@@ -0,0 +1,42 @@
+package stream
+
+import (
+	"testing"
+)
+
+func TestWithoutOutputs(t *testing.T) {
+	stream := New()
+	defer stream.Close()
+	stream.Broadcast <- []byte("hello world")
+}
+
+func TestWithOneOutput(t *testing.T) {
+	stream := New()
+	defer stream.Close()
+
+	// Register one output
+	output := make(chan []byte, 64)
+	stream.Register(output)
+	stream.IncrementClientCount()
+
+	// Try to pass one message
+	stream.Broadcast <- []byte("hello world")
+	msg := <-output
+	if string(msg) != "hello world" {
+		t.Errorf("Message has wrong content: %s != hello world", msg)
+	}
+
+	// Check client count
+	if count := stream.ClientCount(); count != 1 {
+		t.Errorf("Client counter returned %d, expected 1", count)
+	}
+
+	// Unregister
+	stream.Unregister(output)
+	stream.DecrementClientCount()
+
+	// Check client count
+	if count := stream.ClientCount(); count != 0 {
+		t.Errorf("Client counter returned %d, expected 0", count)
+	}
+}

stream/srt/handler.go

@@ -5,37 +5,21 @@ import (
 	"log"
 
 	"github.com/haivision/srtgo"
-	"gitlab.crans.org/nounous/ghostream/messaging"
+	"gitlab.crans.org/nounous/ghostream/stream"
 )
 
-func handleStreamer(socket *srtgo.SrtSocket, streams *messaging.Streams, name string) {
+func handleStreamer(socket *srtgo.SrtSocket, streams map[string]*stream.Stream, name string) {
+	// Check stream does not exist
+	if _, ok := streams[name]; ok {
+		log.Print("Stream already exists, refusing new streamer")
+		socket.Close()
+		return
+	}
+
 	// Create stream
-	stream, err := streams.Create(name)
-	if err != nil {
-		log.Printf("Error on stream creating: %s", err)
-		socket.Close()
-		return
-	}
-
-	// Create source quality
-	q, err := stream.CreateQuality("source")
-	if err != nil {
-		log.Printf("Error on quality creating: %s", err)
-		socket.Close()
-		return
-	}
-
-	// Create sub-qualities
-	for _, qualityName := range []string{"audio", "480p", "360p", "240p"} {
-		_, err := stream.CreateQuality(qualityName)
-		if err != nil {
-			log.Printf("Error on quality creating: %s", err)
-			socket.Close()
-			return
-		}
-	}
-
-	log.Printf("New SRT streamer for stream '%s' quality 'source'", name)
+	log.Printf("New SRT streamer for stream %s", name)
+	st := stream.New()
+	streams[name] = st
 
 	// Read RTP packets forever and send them to the WebRTC Client
 	for {
@@ -58,38 +42,29 @@ func handleStreamer(socket *srtgo.SrtSocket, streams *messaging.Streams, name st
 
 		// Send raw data to other streams
 		buff = buff[:n]
-		q.Broadcast <- buff
+		st.Broadcast <- buff
 	}
 
 	// Close stream
-	streams.Delete(name)
+	st.Close()
 	socket.Close()
+	delete(streams, name)
 }
 
-func handleViewer(socket *srtgo.SrtSocket, streams *messaging.Streams, name string) {
-	// Get requested stream
-	stream, err := streams.Get(name)
-	if err != nil {
-		log.Printf("Failed to get stream: %s", err)
-		socket.Close()
-		return
-	}
+func handleViewer(s *srtgo.SrtSocket, streams map[string]*stream.Stream, name string) {
+	log.Printf("New SRT viewer for stream %s", name)
 
-	// Get requested quality
-	// FIXME: make qualities available
-	qualityName := "source"
-	q, err := stream.GetQuality(qualityName)
-	if err != nil {
-		log.Printf("Failed to get quality: %s", err)
-		socket.Close()
+	// Get requested stream
+	st, ok := streams[name]
+	if !ok {
+		log.Println("Stream does not exist, refusing new viewer")
 		return
 	}
-	log.Printf("New SRT viewer for stream %s quality %s", name, qualityName)
 
 	// Register new output
 	c := make(chan []byte, 1024)
-	q.Register(c)
-	stream.IncrementClientCount()
+	st.Register(c)
+	st.IncrementClientCount()
 
 	// Receive data and send them
 	for data := range c {
@@ -99,7 +74,7 @@ func handleViewer(socket *srtgo.SrtSocket, streams *messaging.Streams, name stri
 		}
 
 		// Send data
-		_, err := socket.Write(data, 1000)
+		_, err := s.Write(data, 1000)
 		if err != nil {
 			log.Printf("Remove SRT viewer because of sending error, %s", err)
 			break
@@ -107,7 +82,7 @@ func handleViewer(socket *srtgo.SrtSocket, streams *messaging.Streams, name stri
 	}
 
 	// Close output
-	q.Unregister(c)
-	stream.DecrementClientCount()
-	socket.Close()
+	st.Unregister(c)
+	st.DecrementClientCount()
+	s.Close()
 }

stream/srt/srt.go

@@ -12,7 +12,7 @@ import (
 
 	"github.com/haivision/srtgo"
 	"gitlab.crans.org/nounous/ghostream/auth"
-	"gitlab.crans.org/nounous/ghostream/messaging"
+	"gitlab.crans.org/nounous/ghostream/stream"
 )
 
 // Options holds web package configuration
@@ -39,7 +39,7 @@ func splitHostPort(hostport string) (string, uint16, error) {
 }
 
 // Serve SRT server
-func Serve(streams *messaging.Streams, authBackend auth.Backend, cfg *Options) {
+func Serve(streams map[string]*stream.Stream, authBackend auth.Backend, cfg *Options) {
 	if !cfg.Enabled {
 		// SRT is not enabled, ignore
 		return

stream/srt/srt_test.go

@@ -6,7 +6,7 @@ import (
 	"testing"
 	"time"
 
-	"gitlab.crans.org/nounous/ghostream/messaging"
+	"gitlab.crans.org/nounous/ghostream/stream"
 )
 
 // TestSplitHostPort Try to split a host like 127.0.0.1:1234 in host, port (127.0.0.1, 1234à
@@ -58,7 +58,7 @@ func TestServeSRT(t *testing.T) {
 	}
 
 	// Init streams messaging and SRT server
-	streams := messaging.New()
+	streams := make(map[string]*stream.Stream)
 	go Serve(streams, nil, &Options{Enabled: true, ListenAddress: ":9711", MaxClients: 2})
 
 	ffmpeg := exec.Command("ffmpeg", "-hide_banner", "-loglevel", "error",

stream/telnet/telnet.go

@@ -7,7 +7,7 @@ import (
 	"strings"
 	"time"
 
-	"gitlab.crans.org/nounous/ghostream/messaging"
+	"gitlab.crans.org/nounous/ghostream/stream"
 )
 
 // Options holds telnet package configuration
@@ -17,7 +17,7 @@ type Options struct {
 }
 
 // Serve Telnet server
-func Serve(streams *messaging.Streams, cfg *Options) {
+func Serve(streams map[string]*stream.Stream, cfg *Options) {
 	if !cfg.Enabled {
 		// Telnet is not enabled, ignore
 		return
@@ -32,17 +32,17 @@ func Serve(streams *messaging.Streams, cfg *Options) {
 
 	// Handle each new client
 	for {
-		socket, err := listener.Accept()
+		s, err := listener.Accept()
 		if err != nil {
-			log.Printf("Error while accepting TCP socket: %s", err)
+			log.Printf("Error while accepting TCP socket: %s", s)
 			continue
 		}
 
-		go handleViewer(socket, streams, cfg)
+		go handleViewer(s, streams, cfg)
 	}
 }
 
-func handleViewer(s net.Conn, streams *messaging.Streams, cfg *Options) {
+func handleViewer(s net.Conn, streams map[string]*stream.Stream, cfg *Options) {
 	// Prompt user about stream name
 	if _, err := s.Write([]byte("[GHOSTREAM]\nEnter stream name: ")); err != nil {
 		log.Printf("Error while writing to TCP socket: %s", err)
@@ -56,7 +56,7 @@ func handleViewer(s net.Conn, streams *messaging.Streams, cfg *Options) {
 		s.Close()
 		return
 	}
-	name := strings.TrimSpace(string(buff[:n]))
+	name := strings.TrimSpace(string(buff[:n])) + "@text"
 	if len(name) < 1 {
 		// Too short, exit
 		s.Close()
@@ -67,9 +67,9 @@ func handleViewer(s net.Conn, streams *messaging.Streams, cfg *Options) {
 	time.Sleep(time.Second)
 
 	// Get requested stream
-	stream, err := streams.Get(name)
-	if err != nil {
-		log.Printf("Kicking new Telnet viewer: %s", err)
+	st, ok := streams[name]
+	if !ok {
+		log.Println("Stream does not exist, kicking new Telnet viewer")
 		if _, err := s.Write([]byte("This stream is inactive.\n")); err != nil {
 			log.Printf("Error while writing to TCP socket: %s", err)
 		}
@@ -77,23 +77,11 @@ func handleViewer(s net.Conn, streams *messaging.Streams, cfg *Options) {
 		return
 	}
 
-	// Get requested quality
-	qualityName := "text"
-	q, err := stream.GetQuality(qualityName)
-	if err != nil {
-		log.Printf("Kicking new Telnet viewer: %s", err)
-		if _, err := s.Write([]byte("This stream is not converted to text.\n")); err != nil {
-			log.Printf("Error while writing to TCP socket: %s", err)
-		}
-		s.Close()
-		return
-	}
-	log.Printf("New Telnet viewer for stream %s quality %s", name, qualityName)
-
 	// Register new client
+	log.Printf("New Telnet viewer for stream '%s'", name)
 	c := make(chan []byte, 128)
-	q.Register(c)
-	stream.IncrementClientCount()
+	st.Register(c)
+	st.IncrementClientCount()
 
 	// Hide terminal cursor
 	if _, err = s.Write([]byte("\033[?25l")); err != nil {
@@ -118,7 +106,7 @@ func handleViewer(s net.Conn, streams *messaging.Streams, cfg *Options) {
 	}
 
 	// Close output
-	q.Unregister(c)
-	stream.DecrementClientCount()
+	st.Unregister(c)
+	st.DecrementClientCount()
 	s.Close()
 }

stream/telnet/telnet_test.go

@@ -3,13 +3,13 @@ package telnet
 import (
 	"testing"
 
-	"gitlab.crans.org/nounous/ghostream/messaging"
+	"gitlab.crans.org/nounous/ghostream/stream"
 )
 
 // TestTelnetOutput creates a TCP client that connects to the server and get one image.
 func TestTelnetOutput(t *testing.T) {
 	// Try to start Telnet server while it is disabled
-	streams := messaging.New()
+	streams := make(map[string]*stream.Stream)
 	go Serve(streams, &Options{Enabled: false})
 
 	// FIXME test connect

stream/webrtc/ingest.go

@@ -2,142 +2,258 @@
 package webrtc
 
 import (
-	"bufio"
-	"log"
-	"net"
-	"os/exec"
-
+	"github.com/3d0c/gmf"
 	"github.com/pion/rtp"
 	"github.com/pion/webrtc/v3"
-	"gitlab.crans.org/nounous/ghostream/messaging"
+	"gitlab.crans.org/nounous/ghostream/stream"
+	"log"
+	"net"
+	"strings"
+	"time"
 )
 
-func ingest(name string, q *messaging.Quality) {
-	// Register to get stream
-	input := make(chan []byte, 1024)
-	// FIXME Stream data should already be transcoded
-	source, _ := q.Stream.GetQuality("source")
-	source.Register(input)
+var (
+	activeStream map[string]struct{}
+)
 
-	// FIXME Bad code
-	port := 5000
-	var tracks map[string][]*webrtc.Track
-	qualityName := ""
-	switch q.Name {
-	case "audio":
-		port = 5004
-		tracks = audioTracks
-		break
-	case "source":
-		port = 5005
-		tracks = videoTracks
-		qualityName = "@source"
-		break
-	case "480p":
-		port = 5006
-		tracks = videoTracks
-		qualityName = "@480p"
-		break
-	case "360p":
-		port = 5007
-		tracks = videoTracks
-		qualityName = "@360p"
-		break
-	case "240p":
-		port = 5008
-		tracks = videoTracks
-		qualityName = "@240p"
-		break
-	}
-
-	// Open a UDP Listener for RTP Packets
-	listener, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: port})
-	if err != nil {
-		log.Printf("Faited to open UDP listener %s", err)
-		return
-	}
-
-	// Start ffmpag to convert input to video and audio UDP
-	ffmpeg, err := startFFmpeg(q, input)
-	if err != nil {
-		log.Printf("Error while starting ffmpeg: %s", err)
-		return
-	}
-
-	// Receive stream
-	go func() {
-		inboundRTPPacket := make([]byte, 1500) // UDP MTU
-		for {
-			n, _, err := listener.ReadFromUDP(inboundRTPPacket)
-			if err != nil {
-				log.Printf("Failed to read from UDP: %s", err)
-				break
+func autoIngest(streams map[string]*stream.Stream) {
+	// Regulary check existing streams
+	activeStream = make(map[string]struct{})
+	for {
+		for name, st := range streams {
+			if strings.Contains(name, "@") {
+				// Not a source stream, pass
+				continue
 			}
+
+			if _, ok := activeStream[name]; ok {
+				// Stream is already ingested
+				continue
+			}
+
+			// Start ingestion
+			log.Printf("Starting webrtc for '%s'", name)
+			// FIXME Ensure that the audio stream exist, but that's poop code
+			time.Sleep(time.Second)
+			go ingest(name, st, streams[name+"@audio"])
+		}
+
+		// Regulary pull stream list,
+		// it may be better to tweak the messaging system
+		// to get an event on a new stream.
+		time.Sleep(time.Second)
+	}
+}
+
+func ingest(name string, input *stream.Stream, audio *stream.Stream) {
+	// Register to get stream
+	videoInput := make(chan []byte, 1024)
+	input.Register(videoInput)
+	audioInput := make(chan []byte, 1024)
+	audio.Register(audioInput)
+	activeStream[name] = struct{}{}
+
+	inputCtx := gmf.NewCtx()
+	avioInputCtx, _ := gmf.NewAVIOContext(inputCtx, &gmf.AVIOHandlers{ReadPacket: func() ([]byte, int) {
+		data := <-audioInput
+		return data, len(data)
+	}})
+	log.Println("Open input")
+	inputCtx.SetPb(avioInputCtx).OpenInput("")
+	log.Println("Opened")
+	defer inputCtx.CloseInput()
+	defer avioInputCtx.Release()
+
+	if audioTracks[name] == nil {
+		audioTracks[name] = make([]*webrtc.Track, 0)
+	}
+
+	udpListener, _ := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 1234})
+
+	outputCtx, _ := gmf.NewOutputCtxWithFormatName("rtp://127.0.0.1:1234", "rtp")
+	avioOutputCtx, _ := gmf.NewAVIOContext(outputCtx, &gmf.AVIOHandlers{WritePacket: func(data []byte) int {
+		n := len(data)
+		log.Printf("Read %d bytes", n)
+
+		return n
+	}})
+	// FIXME DON'T RAN AN UDP LISTENER, PLIZ GET DIRECTLY UDP PACKETS, WHY IS IT SO COMPLICATED????
+	// outputCtx.SetPb(avioOutputCtx)
+	defer outputCtx.CloseOutput()
+	defer avioOutputCtx.Release()
+
+	log.Printf("%d streams", inputCtx.StreamsCnt())
+
+	for i := 0; i < inputCtx.StreamsCnt(); i++ {
+		srcStream, err := inputCtx.GetStream(i)
+		if err != nil {
+			log.Println("GetStream error")
+		}
+
+		outputCtx.AddStreamWithCodeCtx(srcStream.CodecCtx())
+	}
+	outputCtx.Dump()
+
+	if err := outputCtx.WriteHeader(); err != nil {
+		log.Printf("Unable to write RTP header: %s", err)
+	}
+
+	// Receive audio data
+	go func() {
+		buff := make([]byte, 1500)
+		for {
+			n, _ := udpListener.Read(buff)
+
+			if n == 0 {
+				return
+			}
+
 			packet := &rtp.Packet{}
-			if err := packet.Unmarshal(inboundRTPPacket[:n]); err != nil {
+			if err := packet.Unmarshal(buff[:n]); err != nil {
 				log.Printf("Failed to unmarshal RTP srtPacket: %s", err)
 				continue
 			}
 
-			// Write RTP srtPacket to all tracks
+			if audioTracks[name] == nil {
+				audioTracks[name] = make([]*webrtc.Track, 0)
+			}
+
+			// Write RTP srtPacket to all audio tracks
 			// Adapt payload and SSRC to match destination
-			for _, track := range tracks[name+qualityName] {
-				packet.Header.PayloadType = track.PayloadType()
-				packet.Header.SSRC = track.SSRC()
-				if writeErr := track.WriteRTP(packet); writeErr != nil {
-					log.Printf("Failed to write to track: %s", writeErr)
+			for _, audioTrack := range audioTracks[name] {
+				packet.Header.PayloadType = audioTrack.PayloadType()
+				packet.Header.SSRC = audioTrack.SSRC()
+				if writeErr := audioTrack.WriteRTP(packet); writeErr != nil {
+					log.Printf("Failed to write to audio track: %s", writeErr)
 					continue
 				}
 			}
 		}
 	}()
 
-	// Wait for stopped ffmpeg
-	if err = ffmpeg.Wait(); err != nil {
-		log.Printf("Faited to wait for ffmpeg: %s", err)
+	first := false
+	for packet := range inputCtx.GetNewPackets() {
+		if first { //if read from rtsp ,the first packets is wrong.
+			if err := outputCtx.WritePacketNoBuffer(packet); err != nil {
+				log.Printf("Error while writing packet: %s", err)
+			}
+		}
+
+		first = true
+
+		packet.Free()
 	}
 
-	// Close UDP listener
-	if err = listener.Close(); err != nil {
-		log.Printf("Faited to close UDP listener: %s", err)
-	}
-	q.Unregister(input)
+	select {}
+
+	// TODO Register to all substreams and make RTP packets. Don't transcode in this package.
+
+	/*	// Open a UDP Listener for RTP Packets on port 5004
+		videoListener, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 5004})
+		if err != nil {
+			log.Printf("Faited to open UDP listener %s", err)
+			return
+		}
+		audioListener, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 5005})
+		if err != nil {
+			log.Printf("Faited to open UDP listener %s", err)
+			return
+		}
+
+		// Start ffmpag to convert videoInput to video and audio UDP
+		ffmpeg, err := startFFmpeg(videoInput)
+		if err != nil {
+			log.Printf("Error while starting ffmpeg: %s", err)
+			return
+		}
+
+		// Receive video
+		go func() {
+			inboundRTPPacket := make([]byte, 1500) // UDP MTU
+			for {
+				n, _, err := videoListener.ReadFromUDP(inboundRTPPacket)
+				if err != nil {
+					log.Printf("Failed to read from UDP: %s", err)
+					break
+				}
+				packet := &rtp.Packet{}
+				if err := packet.Unmarshal(inboundRTPPacket[:n]); err != nil {
+					log.Printf("Failed to unmarshal RTP srtPacket: %s", err)
+					continue
+				}
+
+				if videoTracks[name] == nil {
+					videoTracks[name] = make([]*webrtc.Track, 0)
+				}
+
+				// Write RTP srtPacket to all video tracks
+				// Adapt payload and SSRC to match destination
+				for _, videoTrack := range videoTracks[name] {
+					packet.Header.PayloadType = videoTrack.PayloadType()
+					packet.Header.SSRC = videoTrack.SSRC()
+					if writeErr := videoTrack.WriteRTP(packet); writeErr != nil {
+						log.Printf("Failed to write to video track: %s", err)
+						continue
+					}
+				}
+			}
+		}()
+
+		// Receive audio
+		go func() {
+			inboundRTPPacket := make([]byte, 1500) // UDP MTU
+			for {
+				n, _, err := audioListener.ReadFromUDP(inboundRTPPacket)
+				if err != nil {
+					log.Printf("Failed to read from UDP: %s", err)
+					break
+				}
+				packet := &rtp.Packet{}
+				if err := packet.Unmarshal(inboundRTPPacket[:n]); err != nil {
+					log.Printf("Failed to unmarshal RTP srtPacket: %s", err)
+					continue
+				}
+
+				if audioTracks[name] == nil {
+					audioTracks[name] = make([]*webrtc.Track, 0)
+				}
+
+				// Write RTP srtPacket to all audio tracks
+				// Adapt payload and SSRC to match destination
+				for _, audioTrack := range audioTracks[name] {
+					packet.Header.PayloadType = audioTrack.PayloadType()
+					packet.Header.SSRC = audioTrack.SSRC()
+					if writeErr := audioTrack.WriteRTP(packet); writeErr != nil {
+						log.Printf("Failed to write to audio track: %s", err)
+						continue
+					}
+				}
+			}
+		}()
+
+		// Wait for stopped ffmpeg
+		if err = ffmpeg.Wait(); err != nil {
+			log.Printf("Faited to wait for ffmpeg: %s", err)
+		}
+
+		// Close UDP listeners
+		if err = videoListener.Close(); err != nil {
+			log.Printf("Faited to close UDP listener: %s", err)
+		}
+		if err = audioListener.Close(); err != nil {
+			log.Printf("Faited to close UDP listener: %s", err)
+		}*/
+	delete(activeStream, name)
 }
 
-func startFFmpeg(q *messaging.Quality, in <-chan []byte) (ffmpeg *exec.Cmd, err error) {
-	// FIXME Use transcoders to downscale, then remux in RTP
-	ffmpegArgs := []string{"-hide_banner", "-loglevel", "error", "-i", "pipe:0"}
-	switch q.Name {
-	case "audio":
-		ffmpegArgs = append(ffmpegArgs, "-vn", "-c:a", "libopus", "-b:a", "160k",
-			"-f", "rtp", "rtp://127.0.0.1:5004")
-		break
-	case "source":
-		ffmpegArgs = append(ffmpegArgs, "-an", "-c:v", "copy",
-			"-f", "rtp", "rtp://127.0.0.1:5005")
-		break
-	case "480p":
-		ffmpegArgs = append(ffmpegArgs,
-			"-an", "-c:v", "libx264", "-b:v", "1200k", "-maxrate", "2000k", "-bufsize", "3000k",
-			"-preset", "ultrafast", "-profile", "main", "-tune", "zerolatency",
-			"-vf", "scale=854:480",
-			"-f", "rtp", "rtp://127.0.0.1:5006")
-		break
-	case "360p":
-		ffmpegArgs = append(ffmpegArgs,
-			"-an", "-c:v", "libx264", "-b:v", "800k", "-maxrate", "1200k", "-bufsize", "1500k",
-			"-preset", "ultrafast", "-profile", "main", "-tune", "zerolatency",
-			"-vf", "scale=480:360",
-			"-f", "rtp", "rtp://127.0.0.1:5007")
-		break
-	case "240p":
-		ffmpegArgs = append(ffmpegArgs,
-			"-an", "-c:v", "libx264", "-b:v", "500k", "-maxrate", "800k", "-bufsize", "1000k",
-			"-preset", "ultrafast", "-profile", "main", "-tune", "zerolatency",
-			"-vf", "scale=360:240",
-			"-f", "rtp", "rtp://127.0.0.1:5008")
-		break
-	}
+/* func startFFmpeg(in <-chan []byte) (ffmpeg *exec.Cmd, err error) {
+	ffmpegArgs := []string{"-hide_banner", "-loglevel", "error", "-i", "pipe:0",
+		"-an", "-vcodec", "libvpx", "-crf", "10", "-cpu-used", "5", "-b:v", "6000k", "-maxrate", "8000k", "-bufsize", "12000k", // TODO Change bitrate when changing quality
+		"-qmin", "10", "-qmax", "42", "-threads", "4", "-deadline", "1", "-error-resilient", "1",
+		"-auto-alt-ref", "1",
+		"-f", "rtp", "rtp://127.0.0.1:5004",
+		"-vn", "-acodec", "libopus", "-cpu-used", "5", "-deadline", "1", "-qmin", "10", "-qmax", "42", "-error-resilient", "1", "-auto-alt-ref", "1",
+		"-f", "rtp", "rtp://127.0.0.1:5005"}
 	ffmpeg = exec.Command("ffmpeg", ffmpegArgs...)
 
 	// Handle errors output
@@ -171,4 +287,4 @@ func startFFmpeg(q *messaging.Quality, in <-chan []byte) (ffmpeg *exec.Cmd, err
 	// Start process
 	err = ffmpeg.Start()
 	return ffmpeg, err
-}
+} */

stream/webrtc/webrtc.go

@@ -8,7 +8,7 @@ import (
 
 	"github.com/pion/webrtc/v3"
 	"gitlab.crans.org/nounous/ghostream/internal/monitoring"
-	"gitlab.crans.org/nounous/ghostream/messaging"
+	"gitlab.crans.org/nounous/ghostream/stream"
 )
 
 // Options holds web package configuration
@@ -40,15 +40,18 @@ func removeTrack(tracks []*webrtc.Track, track *webrtc.Track) []*webrtc.Track {
 
 // GetNumberConnectedSessions get the number of currently connected clients
 func GetNumberConnectedSessions(streamID string) int {
-	return len(audioTracks[streamID])
+	return len(videoTracks[streamID])
 }
 
 // newPeerHandler is called when server receive a new session description
 // this initiates a WebRTC connection and return server description
-func newPeerHandler(name string, localSdpChan chan webrtc.SessionDescription, remoteSdp webrtc.SessionDescription, cfg *Options) {
+func newPeerHandler(localSdpChan chan webrtc.SessionDescription, remoteSdp struct {
+	StreamID          string
+	RemoteDescription webrtc.SessionDescription
+}, cfg *Options) {
 	// Create media engine using client SDP
 	mediaEngine := webrtc.MediaEngine{}
-	if err := mediaEngine.PopulateFromSDP(remoteSdp); err != nil {
+	if err := mediaEngine.PopulateFromSDP(remoteSdp.RemoteDescription); err != nil {
 		log.Println("Failed to create new media engine", err)
 		localSdpChan <- webrtc.SessionDescription{}
 		return
@@ -75,7 +78,7 @@ func newPeerHandler(name string, localSdpChan chan webrtc.SessionDescription, re
 	}
 
 	// Create video track
-	codec, payloadType := getPayloadType(mediaEngine, webrtc.RTPCodecTypeVideo, "H264")
+	codec, payloadType := getPayloadType(mediaEngine, webrtc.RTPCodecTypeVideo, "VP8")
 	videoTrack, err := webrtc.NewTrack(payloadType, rand.Uint32(), "video", "pion", codec)
 	if err != nil {
 		log.Println("Failed to create new video track", err)
@@ -103,13 +106,13 @@ func newPeerHandler(name string, localSdpChan chan webrtc.SessionDescription, re
 	}
 
 	// Set the remote SessionDescription
-	if err = peerConnection.SetRemoteDescription(remoteSdp); err != nil {
+	if err = peerConnection.SetRemoteDescription(remoteSdp.RemoteDescription); err != nil {
 		log.Println("Failed to set remote description", err)
 		localSdpChan <- webrtc.SessionDescription{}
 		return
 	}
 
-	streamID := name
+	streamID := remoteSdp.StreamID
 	split := strings.SplitN(streamID, "@", 2)
 	streamID = split[0]
 	quality := "source"
@@ -117,20 +120,21 @@ func newPeerHandler(name string, localSdpChan chan webrtc.SessionDescription, re
 		quality = split[1]
 	}
 	log.Printf("New WebRTC session for stream %s, quality %s", streamID, quality)
+	// TODO Consider the quality
 
 	// Set the handler for ICE connection state
 	// This will notify you when the peer has connected/disconnected
 	peerConnection.OnICEConnectionStateChange(func(connectionState webrtc.ICEConnectionState) {
 		log.Printf("Connection State has changed %s \n", connectionState.String())
-		if videoTracks[streamID+"@"+quality] == nil {
-			videoTracks[streamID+"@"+quality] = make([]*webrtc.Track, 0, 1)
+		if videoTracks[streamID] == nil {
+			videoTracks[streamID] = make([]*webrtc.Track, 0, 1)
 		}
 		if audioTracks[streamID] == nil {
 			audioTracks[streamID] = make([]*webrtc.Track, 0, 1)
 		}
 		if connectionState == webrtc.ICEConnectionStateConnected {
 			// Register tracks
-			videoTracks[streamID+"@"+quality] = append(videoTracks[streamID+"@"+quality], videoTrack)
+			videoTracks[streamID] = append(videoTracks[streamID], videoTrack)
 			audioTracks[streamID] = append(audioTracks[streamID], audioTrack)
 			monitoring.WebRTCConnectedSessions.Inc()
 		} else if connectionState == webrtc.ICEConnectionStateDisconnected {
@@ -178,7 +182,10 @@ func getPayloadType(m webrtc.MediaEngine, codecType webrtc.RTPCodecType, codecNa
 }
 
 // Serve WebRTC media streaming server
-func Serve(streams *messaging.Streams, cfg *Options) {
+func Serve(streams map[string]*stream.Stream, remoteSdpChan chan struct {
+	StreamID          string
+	RemoteDescription webrtc.SessionDescription
+}, localSdpChan chan webrtc.SessionDescription, cfg *Options) {
 	if !cfg.Enabled {
 		// WebRTC is not enabled, ignore
 		return
@@ -186,43 +193,17 @@ func Serve(streams *messaging.Streams, cfg *Options) {
 
 	log.Printf("WebRTC server using UDP from port %d to %d", cfg.MinPortUDP, cfg.MaxPortUDP)
 
-	// WebRTC ingested tracks
+	// Allocate memory
 	videoTracks = make(map[string][]*webrtc.Track)
 	audioTracks = make(map[string][]*webrtc.Track)
 
-	// Subscribe to new stream event
-	event := make(chan string, 8)
-	streams.Subscribe(event)
+	// Ingest data
+	go autoIngest(streams)
 
-	// For each new stream
-	for name := range event {
-		// Get stream
-		stream, err := streams.Get(name)
-		if err != nil {
-			log.Printf("Failed to get stream '%s'", name)
-		}
-
-		// Get specific quality
-		// FIXME: make it possible to forward other qualities
-		for _, qualityName := range []string{"source", "audio", "480p", "360p", "240p"} {
-			quality, err := stream.GetQuality(qualityName)
-			if err != nil {
-				log.Printf("Failed to get quality '%s'", qualityName)
-			}
-
-			// Start forwarding
-			log.Printf("Starting webrtc for '%s' quality '%s'", name, qualityName)
-			go ingest(name, quality)
-			go listenSdp(name, quality.WebRtcLocalSdp, quality.WebRtcRemoteSdp, cfg)
-		}
-	}
-}
-
-func listenSdp(name string, localSdp, remoteSdp chan webrtc.SessionDescription, cfg *Options) {
 	// Handle new connections
 	for {
 		// Wait for incoming session description
 		// then send the local description to browser
-		newPeerHandler(name, localSdp, <-remoteSdp, cfg)
+		newPeerHandler(localSdpChan, <-remoteSdpChan, cfg)
 	}
 }

stream/webrtc/webrtc_test.go

@@ -5,19 +5,24 @@ import (
 	"testing"
 
 	"github.com/pion/webrtc/v3"
-	"gitlab.crans.org/nounous/ghostream/messaging"
+	"gitlab.crans.org/nounous/ghostream/stream"
 )
 
 func TestServe(t *testing.T) {
 	// Init streams messaging and WebRTC server
-	streams := messaging.New()
+	streams := make(map[string]*stream.Stream)
+	remoteSdpChan := make(chan struct {
+		StreamID          string
+		RemoteDescription webrtc.SessionDescription
+	})
+	localSdpChan := make(chan webrtc.SessionDescription)
 	cfg := Options{
 		Enabled:     true,
 		MinPortUDP:  10000,
 		MaxPortUDP:  10005,
 		STUNServers: []string{"stun:stun.l.google.com:19302"},
 	}
-	go Serve(streams, &cfg)
+	go Serve(streams, remoteSdpChan, localSdpChan, &cfg)
 
 	// New client connection
 	mediaEngine := webrtc.MediaEngine{}
@@ -26,7 +31,7 @@ func TestServe(t *testing.T) {
 	peerConnection, _ := api.NewPeerConnection(webrtc.Configuration{})
 
 	// Create video track
-	codec, payloadType := getPayloadType(mediaEngine, webrtc.RTPCodecTypeVideo, "H264")
+	codec, payloadType := getPayloadType(mediaEngine, webrtc.RTPCodecTypeVideo, "VP8")
 	videoTrack, err := webrtc.NewTrack(payloadType, rand.Uint32(), "video", "pion", codec)
 	if err != nil {
 		t.Error("Failed to create new video track", err)
@@ -53,6 +58,12 @@ func TestServe(t *testing.T) {
 	peerConnection.SetLocalDescription(offer)
 	<-gatherComplete
 
-	// FIXME: Send offer to server
+	// Send offer to server
+	remoteSdpChan <- struct {
+		StreamID          string
+		RemoteDescription webrtc.SessionDescription
+	}{"demo", *peerConnection.LocalDescription()}
+	_ = <-localSdpChan
+
 	// FIXME: verify connection did work
 }

transcoder/audio/audio.go

@@ -0,0 +1,153 @@
+// Package audio transcode a stream to filter the audio
+package audio
+
+import (
+	"bufio"
+	"fmt"
+	"github.com/3d0c/gmf"
+	"io"
+	"log"
+	"os/exec"
+	"strings"
+	"time"
+
+	"gitlab.crans.org/nounous/ghostream/stream"
+)
+
+// Options holds audio package configuration
+type Options struct {
+	Enabled bool
+	Bitrate int
+}
+
+// Init text transcoder
+func Init(streams map[string]*stream.Stream, cfg *Options) {
+	if !cfg.Enabled {
+		// Audio transcode is not enabled, ignore
+		return
+	}
+
+	// Regulary check existing streams
+	for {
+		for sourceName, sourceStream := range streams {
+			if strings.Contains(sourceName, "@") {
+				// Not a source stream, pass
+				continue
+			}
+
+			// Check that the transcoded stream does not already exist
+			name := sourceName + "@audio"
+			_, ok := streams[name]
+			if ok {
+				// Stream is already transcoded
+				continue
+			}
+
+			// Start conversion
+			log.Printf("Starting audio transcode '%s'", name)
+			st := stream.New()
+			streams[name] = st
+
+			go transcode(sourceStream, st, cfg)
+		}
+
+		// Regulary pull stream list,
+		// it may be better to tweak the messaging system
+		// to get an event on a new stream.
+		time.Sleep(time.Second)
+	}
+}
+
+// Extract audio from stream
+func transcode(input, output *stream.Stream, cfg *Options) {
+	// Start ffmpeg to transcode video to audio
+	videoInput := make(chan []byte, 1024)
+	input.Register(videoInput)
+	ffmpeg, audio, err := startFFmpeg(videoInput, cfg)
+	if err != nil {
+		log.Printf("Error while starting ffmpeg: %s", err)
+		return
+	}
+
+	dataBuff := make([]byte, gmf.IO_BUFFER_SIZE) // UDP MTU
+	for {
+		n, err := (*audio).Read(dataBuff)
+		if err != nil {
+			log.Printf("An error occurred while reading input: %s", err)
+			break
+		}
+		if n == 0 {
+			// Stream is finished
+			break
+		}
+
+		output.Broadcast <- dataBuff[:n]
+	}
+
+	// Stop transcode
+	_ = ffmpeg.Process.Kill()
+	_ = (*audio).Close()
+}
+
+// Start a ffmpeg instance to convert stream into audio
+func startFFmpeg(in <-chan []byte, cfg *Options) (*exec.Cmd, *io.ReadCloser, error) {
+	// TODO in a future release: remove FFMPEG dependency and transcode directly using the libopus API
+
+	// FIXME It seems impossible to get a RTP Packet from standard output.
+	// We need to find a clean solution, without waiting on UDP listeners.
+	// FIXME We should also not build RTP packets here.
+
+	/*	port := 0
+		var udpListener *net.UDPConn
+		var err error
+		for {
+			port = rand.Intn(65535)
+			udpListener, err = net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: port})
+			if err != nil {
+				if strings.Contains(fmt.Sprintf("%s", err), "address already in use") {
+					continue
+				}
+				return nil, nil, err
+			}
+			break
+		}*/
+
+	bitrate := fmt.Sprintf("%dk", cfg.Bitrate)
+	// Use copy audio codec, assume for now that libopus is used by the streamer
+	ffmpegArgs := []string{"-hide_banner", "-loglevel", "error", "-i", "pipe:0",
+		"-vn", "-c:a", "copy", "-b:a", bitrate, "-f", "mpegts", "pipe:1"}
+	ffmpeg := exec.Command("ffmpeg", ffmpegArgs...)
+
+	// Handle errors output
+	errOutput, err := ffmpeg.StderrPipe()
+	if err != nil {
+		return nil, nil, err
+	}
+	go func() {
+		scanner := bufio.NewScanner(errOutput)
+		for scanner.Scan() {
+			log.Printf("[AUDIO FFMPEG %s] %s", "demo", scanner.Text())
+		}
+	}()
+
+	// Handle text output
+	output, err := ffmpeg.StdoutPipe()
+	if err != nil {
+		return nil, nil, err
+	}
+
+	// Handle stream input
+	input, err := ffmpeg.StdinPipe()
+	if err != nil {
+		return nil, nil, err
+	}
+	go func() {
+		for data := range in {
+			_, _ = input.Write(data)
+		}
+	}()
+
+	// Start process
+	err = ffmpeg.Start()
+	return ffmpeg, &output, err
+}

transcoder/audio/audio_test.go

@@ -0,0 +1 @@
+package audio

transcoder/text/text.go

@@ -8,8 +8,10 @@ import (
 	"io"
 	"log"
 	"os/exec"
+	"strings"
+	"time"
 
-	"gitlab.crans.org/nounous/ghostream/messaging"
+	"gitlab.crans.org/nounous/ghostream/stream"
 )
 
 // Options holds text package configuration
@@ -21,46 +23,45 @@ type Options struct {
 }
 
 // Init text transcoder
-func Init(streams *messaging.Streams, cfg *Options) {
+func Init(streams map[string]*stream.Stream, cfg *Options) {
 	if !cfg.Enabled {
 		// Text transcode is not enabled, ignore
 		return
 	}
 
-	// Subscribe to new stream event
-	event := make(chan string, 8)
-	streams.Subscribe(event)
+	// Regulary check existing streams
+	for {
+		for sourceName, sourceStream := range streams {
+			if strings.Contains(sourceName, "@") {
+				// Not a source stream, pass
+				continue
+			}
 
-	// For each new stream
-	for name := range event {
-		// Get stream
-		stream, err := streams.Get(name)
-		if err != nil {
-			log.Printf("Failed to get stream '%s'", name)
+			// Check that the transcoded stream does not already exist
+			name := sourceName + "@text"
+			_, ok := streams[name]
+			if ok {
+				// Stream is already transcoded
+				continue
+			}
+
+			// Start conversion
+			log.Printf("Starting text transcode '%s'", name)
+			st := stream.New()
+			streams[name] = st
+
+			go transcode(sourceStream, st, cfg)
 		}
 
-		// Get specific quality
-		// FIXME: make it possible to forward other qualities
-		qualityName := "source"
-		quality, err := stream.GetQuality(qualityName)
-		if err != nil {
-			log.Printf("Failed to get quality '%s'", qualityName)
-		}
-
-		// Create new text quality
-		outputQuality, err := stream.CreateQuality("text")
-		if err != nil {
-			log.Printf("Failed to create quality 'text': %s", err)
-		}
-
-		// Start forwarding
-		log.Printf("Starting text transcoder for '%s' quality '%s'", name, qualityName)
-		go transcode(quality, outputQuality, cfg)
+		// Regulary pull stream list,
+		// it may be better to tweak the messaging system
+		// to get an event on a new stream.
+		time.Sleep(time.Second)
 	}
 }
 
 // Convert video to ANSI text
-func transcode(input, output *messaging.Quality, cfg *Options) {
+func transcode(input, output *stream.Stream, cfg *Options) {
 	// Start ffmpeg to transcode video to rawvideo
 	videoInput := make(chan []byte, 1024)
 	input.Register(videoInput)

transcoder/transcoder.go

@@ -2,16 +2,19 @@
 package transcoder
 
 import (
-	"gitlab.crans.org/nounous/ghostream/messaging"
+	"gitlab.crans.org/nounous/ghostream/stream"
+	"gitlab.crans.org/nounous/ghostream/transcoder/audio"
 	"gitlab.crans.org/nounous/ghostream/transcoder/text"
 )
 
 // Options holds text package configuration
 type Options struct {
-	Text text.Options
+	Text  text.Options
+	Audio audio.Options
 }
 
 // Init all transcoders
-func Init(streams *messaging.Streams, cfg *Options) {
+func Init(streams map[string]*stream.Stream, cfg *Options) {
 	go text.Init(streams, &cfg.Text)
+	go audio.Init(streams, &cfg.Audio)
 }

web/handler.go

@@ -21,20 +21,61 @@ var (
 	validPath = regexp.MustCompile("^/[a-z0-9@_-]*$")
 )
 
-// Handle site index and viewer pages
-func viewerHandler(w http.ResponseWriter, r *http.Request) {
-	// Validation on path
-	if validPath.FindStringSubmatch(r.URL.Path) == nil {
-		http.NotFound(w, r)
-		log.Printf("Replied not found on %s", r.URL.Path)
+// Handle WebRTC session description exchange via POST
+func viewerPostHandler(w http.ResponseWriter, r *http.Request) {
+	// Limit response body to 128KB
+	r.Body = http.MaxBytesReader(w, r.Body, 131072)
+
+	// Get stream ID from URL, or from domain name
+	path := r.URL.Path[1:]
+	host := r.Host
+	if strings.Contains(host, ":") {
+		realHost, _, err := net.SplitHostPort(r.Host)
+		if err != nil {
+			log.Printf("Failed to split host and port from %s", r.Host)
+			return
+		}
+		host = realHost
+	}
+	host = strings.Replace(host, ".", "-", -1)
+	if streamID, ok := cfg.MapDomainToStream[host]; ok {
+		path = streamID
+	}
+
+	// Decode client description
+	dec := json.NewDecoder(r.Body)
+	dec.DisallowUnknownFields()
+	remoteDescription := webrtc.SessionDescription{}
+	if err := dec.Decode(&remoteDescription); err != nil {
+		http.Error(w, "The JSON WebRTC offer is malformed", http.StatusBadRequest)
 		return
 	}
 
-	// Check method
-	if r.Method != http.MethodGet {
-		http.Error(w, "Method not allowed.", http.StatusMethodNotAllowed)
+	// Exchange session descriptions with WebRTC stream server
+	remoteSdpChan <- struct {
+		StreamID          string
+		RemoteDescription webrtc.SessionDescription
+	}{StreamID: path, RemoteDescription: remoteDescription}
+	localDescription := <-localSdpChan
+
+	// Send server description as JSON
+	jsonDesc, err := json.Marshal(localDescription)
+	if err != nil {
+		http.Error(w, "An error occurred while formating response", http.StatusInternalServerError)
+		log.Println("An error occurred while sending session description", err)
+		return
+	}
+	w.Header().Set("Content-Type", "application/json")
+	_, err = w.Write(jsonDesc)
+	if err != nil {
+		log.Println("An error occurred while sending session description", err)
 	}
 
+	// Increment monitoring
+	monitoring.WebSessions.Inc()
+}
+
+func viewerGetHandler(w http.ResponseWriter, r *http.Request) {
 	// Get stream ID from URL, or from domain name
 	path := r.URL.Path[1:]
 	host := r.Host
@@ -81,6 +122,27 @@ func viewerHandler(w http.ResponseWriter, r *http.Request) {
 	monitoring.WebViewerServed.Inc()
 }
 
+// Handle site index and viewer pages
+// POST requests are used to exchange WebRTC session descriptions
+func viewerHandler(w http.ResponseWriter, r *http.Request) {
+	// Validation on path
+	if validPath.FindStringSubmatch(r.URL.Path) == nil {
+		http.NotFound(w, r)
+		log.Printf("Replied not found on %s", r.URL.Path)
+		return
+	}
+
+	// Route depending on HTTP method
+	switch r.Method {
+	case http.MethodGet:
+		viewerGetHandler(w, r)
+	case http.MethodPost:
+		viewerPostHandler(w, r)
+	default:
+		http.Error(w, "Sorry, only GET and POST methods are supported.", http.StatusBadRequest)
+	}
+}
+
 func staticHandler() http.Handler {
 	// Set up static files server
 	staticFs := http.FileServer(pkger.Dir("/web/static"))
@@ -91,16 +153,19 @@ func statisticsHandler(w http.ResponseWriter, r *http.Request) {
 	name := strings.SplitN(strings.Replace(r.URL.Path[7:], "/", "", -1), "@", 2)[0]
 	userCount := 0
 
-	// Get requested stream
-	stream, err := streams.Get(name)
-	if err == nil {
-		userCount = stream.ClientCount()
-		userCount += webrtc.GetNumberConnectedSessions(name)
+	// Get all substreams
+	for _, outputType := range []string{"", "@720p", "@480p", "@360p", "@240p", "@text"} {
+		// Get requested stream
+		stream, ok := streams[name+outputType]
+		if ok {
+			// Get number of output channels
+			userCount += stream.ClientCount()
+		}
 	}
 
 	// Display connected users statistics
 	enc := json.NewEncoder(w)
-	err = enc.Encode(struct{ ConnectedViewers int }{userCount})
+	err := enc.Encode(struct{ ConnectedViewers int }{userCount})
 	if err != nil {
 		http.Error(w, "Failed to generate JSON.", http.StatusInternalServerError)
 		log.Printf("Failed to generate JSON: %s", err)

web/handler_test.go

@@ -4,8 +4,6 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"testing"
-
-	"gitlab.crans.org/nounous/ghostream/messaging"
 )
 
 func TestViewerPageGET(t *testing.T) {
@@ -14,9 +12,6 @@ func TestViewerPageGET(t *testing.T) {
 		t.Errorf("Failed to load templates: %v", err)
 	}
 
-	// Init streams messaging
-	streams = messaging.New()
-
 	cfg = &Options{}
 
 	// Test GET request

web/static/css/player.css

@@ -2,7 +2,7 @@
 video {
   display: block;
   flex-grow: 1;
-  max-width: 100%;
+  width: 100%;
 
   /* Black borders around video */
   background-color: #000;

web/static/js/modules/viewerCounter.js

@@ -1,29 +0,0 @@
-/**
- * ViewerCounter show the number of active viewers
- */
-export class ViewerCounter {
-    /**
-     * @param {HTMLElement} element 
-     * @param {String} streamName 
-     */
-    constructor(element, streamName) {
-        this.element = element;
-        this.url = "/_stats/" + streamName;
-    }
-
-    /**
-     * Regulary update counter
-     * 
-     * @param {Number} updatePeriod 
-     */
-    regularUpdate(updatePeriod) {
-        setInterval(() => this.refreshViewersCounter(), updatePeriod);
-    }
-
-    refreshViewersCounter() {
-        fetch(this.url)
-            .then(response => response.json())
-            .then((data) => this.element.innerText = data.ConnectedViewers)
-            .catch(console.log);
-    }
-}

web/static/js/modules/webrtc.js

@@ -1,99 +0,0 @@
-/**
- * GsWebRTC to connect to Ghostream
- */
-export class GsWebRTC {
-    /**
-     * @param {list} stunServers STUN servers
-     * @param {HTMLElement} viewer Video HTML element
-     * @param {HTMLElement} connectionIndicator Connection indicator element
-     */
-    constructor(stunServers, viewer, connectionIndicator) {
-        this.viewer = viewer;
-        this.connectionIndicator = connectionIndicator;
-        this.pc = new RTCPeerConnection({
-            iceServers: [{ urls: stunServers }]
-        });
-
-        // We want to receive audio and video
-        this.pc.addTransceiver("video", { "direction": "sendrecv" });
-        this.pc.addTransceiver("audio", { "direction": "sendrecv" });
-
-        // Configure events
-        this.pc.oniceconnectionstatechange = () => this._onConnectionStateChange();
-        this.pc.ontrack = (e) => this._onTrack(e);
-    }
-
-    /**
-     * On connection change, log it and change indicator.
-     * If connection closed or failed, try to reconnect.
-     */
-    _onConnectionStateChange() {
-        console.log("[WebRTC] ICE connection state changed to " + this.pc.iceConnectionState);
-        switch (this.pc.iceConnectionState) {
-        case "disconnected":
-            this.connectionIndicator.style.fill = "#dc3545";
-            break;
-        case "checking":
-            this.connectionIndicator.style.fill = "#ffc107";
-            break;
-        case "connected":
-            this.connectionIndicator.style.fill = "#28a745";
-            break;
-        case "closed":
-        case "failed":
-            console.log("[WebRTC] Connection closed, restarting...");
-            /*peerConnection.close();
-                    peerConnection = null;
-                    setTimeout(startPeerConnection, 1000);*/
-            break;
-        }
-    }
-
-    /**
-     * On new track, add it to the player
-     * @param {Event} event 
-     */
-    _onTrack(event) {
-        console.log(`[WebRTC] New ${event.track.kind} track`);
-        if (event.track.kind === "video") {
-            this.viewer.srcObject = event.streams[0];
-        }
-    }
-
-    /**
-     * Create an offer and set local description.
-     * After that the browser will fire onicecandidate events.
-     */
-    createOffer() {
-        this.pc.createOffer().then(offer => {
-            this.pc.setLocalDescription(offer);
-            console.log("[WebRTC] WebRTC offer created");
-        }).catch(console.log);
-    }
-
-    /**
-     * Register a function to call to send local descriptions
-     * @param {Function} sendFunction Called with a local description to send.
-     */
-    onICECandidate(sendFunction) {
-        // When candidate is null, ICE layer has run out of potential configurations to suggest
-        // so let's send the offer to the server.
-        // FIXME: Send offers progressively to do Trickle ICE
-        this.pc.onicecandidate = event => {
-            if (event.candidate === null) {
-                // Send offer to server
-                console.log("[WebRTC] Sending session description to server");
-                sendFunction(this.pc.localDescription);
-            }
-        };
-    }
-
-    /**
-     * Set WebRTC remote description
-     * After that, the connection will be established and ontrack will be fired.
-     * @param {RTCSessionDescription} sdp Session description data
-     */
-    setRemoteDescription(sdp) {
-        this.pc.setRemoteDescription(sdp);
-    }
-}

web/static/js/modules/websocket.js

@@ -1,62 +0,0 @@
-/**
- * GsWebSocket to do Ghostream signalling
- */
-export class GsWebSocket {
-    constructor() {
-        const protocol = (window.location.protocol === "https:") ? "wss://" : "ws://";
-        this.url = protocol + window.location.host + "/_ws/";
-
-        // Open WebSocket
-        this._open();
-
-        // Configure events
-        this.socket.addEventListener("open", () => {
-            console.log("[WebSocket] Connection established");
-        });
-        this.socket.addEventListener("close", () => {
-            console.log("[WebSocket] Connection closed, retrying connection in 1s...");
-            setTimeout(() => this._open(), 1000);
-        });
-        this.socket.addEventListener("error", () => {
-            console.log("[WebSocket] Connection errored, retrying connection in 1s...");
-            setTimeout(() => this._open(), 1000);
-        });
-    }
-
-    _open() {
-        console.log(`[WebSocket] Connecting to ${this.url}...`);
-        this.socket = new WebSocket(this.url);
-    }
-
-    /**
-     * Send local WebRTC session description to remote.
-     * @param {SessionDescription} localDescription WebRTC local SDP
-     * @param {string} stream Name of the stream
-     * @param {string} quality Requested quality 
-     */
-    sendLocalDescription(localDescription, stream, quality) {
-        if (this.socket.readyState !== 1) {
-            console.log("[WebSocket] Waiting for connection to send data...");
-            setTimeout(() => this.sendLocalDescription(localDescription, stream, quality), 100);
-            return;
-        }
-        console.log(`[WebSocket] Sending WebRTC local session description for stream ${stream} quality ${quality}`);
-        this.socket.send(JSON.stringify({
-            "webRtcSdp": localDescription,
-            "stream": stream,
-            "quality": quality
-        }));
-    }
-
-    /**
-     * Set callback function on new remote session description.
-     * @param {Function} callback Function called when data is received
-     */
-    onRemoteDescription(callback) {
-        this.socket.addEventListener("message", (event) => {
-            console.log("[WebSocket] Received WebRTC remote session description");
-            const sdp = new RTCSessionDescription(JSON.parse(event.data));
-            callback(sdp);
-        });
-    }
-}

web/static/js/sideWidget.js

@@ -0,0 +1,12 @@
+// Side widget toggler
+const sideWidgetToggle = document.getElementById("sideWidgetToggle")
+sideWidgetToggle.addEventListener("click", function () {
+    const sideWidget = document.getElementById("sideWidget")
+    if (sideWidget.style.display === "none") {
+        sideWidget.style.display = "block"
+        sideWidgetToggle.textContent = "»"
+    } else {
+        sideWidget.style.display = "none"
+        sideWidgetToggle.textContent = "«"
+    }
+})

web/static/js/videoQuality.js

@@ -0,0 +1,9 @@
+document.getElementById("quality").addEventListener("change", (event) => {
+    console.log(`Stream quality changed to ${event.target.value}`)
+
+    // Restart the connection with a new quality
+    peerConnection.close()
+    peerConnection = null
+    streamPath = window.location.href + event.target.value
+    startPeerConnection()
+})

web/static/js/viewer.js

@@ -1,87 +1,97 @@
-import { GsWebSocket } from "./modules/websocket.js";
-import { ViewerCounter } from "./modules/viewerCounter.js";
-import { GsWebRTC } from "./modules/webrtc.js";
+let peerConnection
+let streamPath = window.location.href
 
-/**
- * Initialize viewer page
- * 
- * @param {String} stream 
- * @param {List} stunServers 
- * @param {Number} viewersCounterRefreshPeriod 
- */
-export function initViewerPage(stream, stunServers, viewersCounterRefreshPeriod) {
-    // Viewer element
-    const viewer = document.getElementById("viewer");
+startPeerConnection = () => {
+    // Init peer connection
+    peerConnection = new RTCPeerConnection({
+        iceServers: [{ urls: stunServers }]
+    })
 
-    // Default quality
-    let quality = "240p";
-
-    // Create WebSocket and WebRTC
-    const websocket = new GsWebSocket();
-    const webrtc = new GsWebRTC(
-        stunServers,
-        viewer,
-        document.getElementById("connectionIndicator"),
-    );
-    webrtc.createOffer();
-    webrtc.onICECandidate(localDescription => {
-        websocket.sendLocalDescription(localDescription, stream, quality);
-    });
-    websocket.onRemoteDescription(sdp => {
-        webrtc.setRemoteDescription(sdp);
-    });
-
-    // Register keyboard events
-    window.addEventListener("keydown", (event) => {
-        switch (event.key) {
-        case "f":
-            // F key put player in fullscreen
-            if (document.fullscreenElement !== null) {
-                document.exitFullscreen();
-            } else {
-                viewer.requestFullscreen();
-            }
-            break;
-        case "m":
-        case " ":
-            // M and space key mute player
-            viewer.muted = !viewer.muted;
-            event.preventDefault();
-            viewer.play();
-            break;
+    // On connection change, change indicator color
+    // if connection failed, restart peer connection
+    peerConnection.oniceconnectionstatechange = e => {
+        console.log("ICE connection state changed, " + peerConnection.iceConnectionState)
+        switch (peerConnection.iceConnectionState) {
+            case "disconnected":
+                document.getElementById("connectionIndicator").style.fill = "#dc3545"
+                break
+            case "checking":
+                document.getElementById("connectionIndicator").style.fill = "#ffc107"
+                break
+            case "connected":
+                document.getElementById("connectionIndicator").style.fill = "#28a745"
+                break
+            case "closed":
+            case "failed":
+                console.log("Connection failed, restarting...")
+                peerConnection.close()
+                peerConnection = null
+                setTimeout(startPeerConnection, 1000)
+                break
         }
-    });
-
-    // Create viewer counter
-    const viewerCounter = new ViewerCounter(
-        document.getElementById("connected-people"),
-        stream,
-    );
-    viewerCounter.regularUpdate(viewersCounterRefreshPeriod);
-    viewerCounter.refreshViewersCounter();
-
-    // Side widget toggler
-    const sideWidgetToggle = document.getElementById("sideWidgetToggle");
-    const sideWidget = document.getElementById("sideWidget");
-    if (sideWidgetToggle !== null && sideWidget !== null) {
-        // On click, toggle side widget visibility
-        sideWidgetToggle.addEventListener("click", function () {
-            if (sideWidget.style.display === "none") {
-                sideWidget.style.display = "block";
-                sideWidgetToggle.textContent = "»";
-            } else {
-                sideWidget.style.display = "none";
-                sideWidgetToggle.textContent = "«";
-            }
-        });
     }
 
-    // Video quality toggler
-    document.getElementById("quality").addEventListener("change", (event) => {
-        quality = event.target.value;
-        console.log(`Stream quality changed to ${quality}`);
+    // We want to receive audio and video
+    peerConnection.addTransceiver('video', { 'direction': 'sendrecv' })
+    peerConnection.addTransceiver('audio', { 'direction': 'sendrecv' })
 
-        // Restart WebRTC negociation
-        webrtc.createOffer();
-    });
+    // Create offer and set local description
+    peerConnection.createOffer().then(offer => {
+        // After setLocalDescription, the browser will fire onicecandidate events
+        peerConnection.setLocalDescription(offer)
+    }).catch(console.log)
+
+    // When candidate is null, ICE layer has run out of potential configurations to suggest
+    // so let's send the offer to the server
+    peerConnection.onicecandidate = event => {
+        if (event.candidate === null) {
+            // Send offer to server
+            // The server know the stream name from the url
+            // The server replies with its description
+            // After setRemoteDescription, the browser will fire ontrack events
+            console.log("Sending session description to server")
+            fetch(streamPath, {
+                method: 'POST',
+                headers: {
+                    'Accept': 'application/json',
+                    'Content-Type': 'application/json'
+                },
+                body: JSON.stringify(peerConnection.localDescription)
+            })
+                .then(response => response.json())
+                .then((data) => peerConnection.setRemoteDescription(new RTCSessionDescription(data)))
+                .catch(console.log)
+        }
+    }
+
+    // When video track is received, configure player
+    peerConnection.ontrack = function (event) {
+        console.log(`New ${event.track.kind} track`)
+        if (event.track.kind === "video") {
+            const viewer = document.getElementById('viewer')
+            viewer.srcObject = event.streams[0]
+        }
+    }
 }
+
+// Register keyboard events
+let viewer = document.getElementById("viewer")
+window.addEventListener("keydown", (event) => {
+    switch (event.key) {
+        case 'f':
+            // F key put player in fullscreen
+            if (document.fullscreenElement !== null) {
+                document.exitFullscreen()
+            } else {
+                viewer.requestFullscreen()
+            }
+            break
+        case 'm':
+        case ' ':
+            // M and space key mute player
+            viewer.muted = !viewer.muted
+            event.preventDefault()
+            viewer.play()
+            break
+    }
+})

web/static/js/viewersCounter.js

@@ -0,0 +1,12 @@
+// Refresh viewer count by pulling metric from server
+function refreshViewersCounter(streamID, period) {
+    // Distinguish oneDomainPerStream mode
+    fetch("/_stats/" + streamID)
+        .then(response => response.json())
+        .then((data) => document.getElementById("connected-people").innerText = data.ConnectedViewers)
+        .catch(console.log)
+
+    setTimeout(() => {
+        refreshViewersCounter(streamID, period)
+    }, period)
+}

web/template/_base.html

@@ -4,7 +4,6 @@
 
 <head>
   <meta charset="UTF-8">
-  <meta name="viewport" content="width=device-width, initial-scale=1, shrink-to-fit=no">
   <title>{{if .Path}}{{.Path}} - {{end}}{{.Cfg.Name}}</title>
   <link rel="stylesheet" href="static/css/style.css">
   <link rel="stylesheet" href="static/css/player.css">

web/template/player.html

@@ -8,10 +8,10 @@
     <div class="controls">
       <span class="control-quality">
         <select id="quality">
-          <option value="240p">Source</option>
-          <option value="480p">480p</option>
-          <option value="360p">360p</option>
-          <option value="240p">240p</option>
+          <option value="">Source</option>
+          <option value="@720p">720p</option>
+          <option value="@480p">480p</option>
+          <option value="@240p">240p</option>
         </select>  
       </span>
       <code class="control-srt-link">srt://{{.Cfg.Hostname}}:{{.Cfg.SRTServerPort}}?streamid={{.Path}}</code>
@@ -34,17 +34,21 @@
   {{end}}
 </div>
 
-<script type="module">
-  import { initViewerPage } from "/static/js/viewer.js";
-
-  // Some variables that need to be fixed by web page
-  const viewersCounterRefreshPeriod = Number("{{.Cfg.ViewersCounterRefreshPeriod}}");
-  const stream = "{{.Path}}";
+{{if .WidgetURL}}<script src="/static/js/sideWidget.js"></script>{{end}}
+<script src="/static/js/videoQuality.js"></script>
+<script src="/static/js/viewer.js"></script>
+<script src="/static/js/viewersCounter.js"></script>
+<script>
   const stunServers = [
     {{range $id, $value := .Cfg.STUNServers}}
-    "{{$value}}",
+    '{{$value}}',
     {{end}}
   ]
-  initViewerPage(stream, stunServers, viewersCounterRefreshPeriod)
+  startPeerConnection()
+
+  // Wait a bit before pulling viewers counter for the first time
+  setTimeout(() => {
+      refreshViewersCounter("{{.Path}}", {{.Cfg.ViewersCounterRefreshPeriod}})
+  }, 1000)
 </script>
-{{end}}
+{{end}}

web/web.go

@@ -10,7 +10,8 @@ import (
 	"strings"
 
 	"github.com/markbates/pkger"
-	"gitlab.crans.org/nounous/ghostream/messaging"
+	"github.com/pion/webrtc/v3"
+	"gitlab.crans.org/nounous/ghostream/stream"
 )
 
 // Options holds web package configuration
@@ -32,11 +33,18 @@ type Options struct {
 var (
 	cfg *Options
 
+	// WebRTC session description channels
+	remoteSdpChan chan struct {
+		StreamID          string
+		RemoteDescription webrtc.SessionDescription
+	}
+	localSdpChan chan webrtc.SessionDescription
+
 	// Preload templates
 	templates *template.Template
 
 	// Streams to get statistics
-	streams *messaging.Streams
+	streams map[string]*stream.Stream
 )
 
 // Load templates with pkger
@@ -70,8 +78,13 @@ func loadTemplates() error {
 }
 
 // Serve HTTP server
-func Serve(s *messaging.Streams, c *Options) {
+func Serve(s map[string]*stream.Stream, rSdpChan chan struct {
+	StreamID          string
+	RemoteDescription webrtc.SessionDescription
+}, lSdpChan chan webrtc.SessionDescription, c *Options) {
 	streams = s
+	remoteSdpChan = rSdpChan
+	localSdpChan = lSdpChan
 	cfg = c
 
 	if !cfg.Enabled {
@@ -88,7 +101,6 @@ func Serve(s *messaging.Streams, c *Options) {
 	mux := http.NewServeMux()
 	mux.HandleFunc("/", viewerHandler)
 	mux.Handle("/static/", staticHandler())
-	mux.HandleFunc("/_ws/", websocketHandler)
 	mux.HandleFunc("/_stats/", statisticsHandler)
 	log.Printf("HTTP server listening on %s", cfg.ListenAddress)
 	log.Fatal(http.ListenAndServe(cfg.ListenAddress, mux))

web/web_test.go

@@ -5,16 +5,16 @@ import (
 	"testing"
 	"time"
 
-	"gitlab.crans.org/nounous/ghostream/messaging"
+	"gitlab.crans.org/nounous/ghostream/stream"
 )
 
 // TestHTTPServe tries to serve a real HTTP server and load some pages
 func TestHTTPServe(t *testing.T) {
 	// Init streams messaging
-	streams := messaging.New()
+	streams := make(map[string]*stream.Stream)
 
 	// Create a disabled web server
-	go Serve(streams, &Options{Enabled: false, ListenAddress: "127.0.0.1:8081"})
+	go Serve(streams, nil, nil, &Options{Enabled: false, ListenAddress: "127.0.0.1:8081"})
 
 	// Sleep 500ms to ensure that the web server is running, to avoid fails because the request came too early
 	time.Sleep(500 * time.Millisecond)
@@ -26,7 +26,7 @@ func TestHTTPServe(t *testing.T) {
 	}
 
 	// Now let's really start the web server
-	go Serve(streams, &Options{Enabled: true, ListenAddress: "127.0.0.1:8081"})
+	go Serve(streams, nil, nil, &Options{Enabled: true, ListenAddress: "127.0.0.1:8081"})
 
 	// Sleep 500ms to ensure that the web server is running, to avoid fails because the request came too early
 	time.Sleep(500 * time.Millisecond)

web/websocket_handler.go

@@ -1,67 +0,0 @@
-// Package web serves the JavaScript player and WebRTC negotiation
-package web
-
-import (
-	"log"
-	"net/http"
-
-	"github.com/gorilla/websocket"
-	"gitlab.crans.org/nounous/ghostream/stream/webrtc"
-)
-
-var upgrader = websocket.Upgrader{
-	ReadBufferSize:  1024,
-	WriteBufferSize: 1024,
-}
-
-// clientDescription is sent by new client
-type clientDescription struct {
-	WebRtcSdp webrtc.SessionDescription
-	Stream    string
-	Quality   string
-}
-
-// websocketHandler exchanges WebRTC SDP and viewer count
-func websocketHandler(w http.ResponseWriter, r *http.Request) {
-	// Upgrade client connection to WebSocket
-	conn, err := upgrader.Upgrade(w, r, nil)
-	if err != nil {
-		log.Printf("Failed to upgrade client to websocket: %s", err)
-		return
-	}
-
-	for {
-		// Get client description
-		c := &clientDescription{}
-		err = conn.ReadJSON(c)
-		if err != nil {
-			log.Printf("Failed to receive client description: %s", err)
-			continue
-		}
-
-		// Get requested stream
-		stream, err := streams.Get(c.Stream)
-		if err != nil {
-			log.Printf("Stream not found: %s", c.Stream)
-			continue
-		}
-
-		// Get requested quality
-		q, err := stream.GetQuality(c.Quality)
-		if err != nil {
-			log.Printf("Quality not found: %s", c.Quality)
-			continue
-		}
-
-		// Exchange session descriptions with WebRTC stream server
-		// FIXME: Add trickle ICE support
-		q.WebRtcRemoteSdp <- c.WebRtcSdp
-		localDescription := <-q.WebRtcLocalSdp
-
-		// Send new local description
-		if err := conn.WriteJSON(localDescription); err != nil {
-			log.Println(err)
-			continue
-		}
-	}
-}