Lock counter map

docs/Server-docker.xml

@@ -36,7 +36,7 @@
             </Domain>
             <Applications>
                 <Application>
-                    <Name>app</Name>
+                    <Name>play</Name>
                     <Type>live</Type>
                     <Encodes>
                         <Encode>

docs/docker-compose.yml

@@ -51,7 +51,7 @@ services:
     labels:
       - "traefik.http.middlewares.sslheader.headers.customrequestheaders.X-Forwarded-Proto=https"
 
-      - "traefik.http.routers.ovenmediaengine.rule=Host(`stream.example.com`) && PathPrefix(`/app/`)"
+      - "traefik.http.routers.ovenmediaengine.rule=Host(`stream.example.com`) && PathPrefix(`/play/`)"
       - "traefik.http.routers.ovenmediaengine.priority=101"
       - "traefik.http.routers.ovenmediaengine.entrypoints=websecure"
       - "traefik.http.routers.ovenmediaengine.tls.certresolver=mytlschallenge"
@@ -59,7 +59,7 @@ services:
       - "traefik.http.routers.ovenmediaengine.service=ovenmediaengine"
       - "traefik.http.routers.ovenmediaengine.middlewares=sslheader"
 
-      - "traefik.http.routers.ovenmediaengine-hls.rule=Host(`stream.example.com`) && Path(`/app/{app_name:.*}/{filename:.*}.{ext:(m3u8|mpd|ts)}`)"
+      - "traefik.http.routers.ovenmediaengine-hls.rule=Host(`stream.example.com`) && Path(`/play/{app_name:.*}/{filename:.*}.{ext:(m3u8|mpd|ts)}`)"
       - "traefik.http.routers.ovenmediaengine-hls.priority=102"
       - "traefik.http.routers.ovenmediaengine-hls.entrypoints=websecure"
       - "traefik.http.routers.ovenmediaengine-hls.tls.certresolver=mytlschallenge"

docs/ghostream.example.yml

@@ -49,7 +49,6 @@ forwarding:
   #  - 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
-  #  - rtmp://ovenmediaengine:1915/app/demo    # For player
 
 ## Prometheus monitoring ##
 # Expose a monitoring endpoint for Prometheus
@@ -62,6 +61,19 @@ monitoring:
   # To limit access to only localhost, use 127.0.0.1:2112
   #listenAddress: :2112
 
+## OvenMediaEngine ##
+# Send the stream data to OvenMediaEngine to handle properly the web client
+ome:
+  # If you disable OME module, the laggy webrtc client will be used.
+  #
+  #enabled: true
+  #
+  # The URL where OME listens RTMP, without the prefix.
+  #url: ovenmediaengine:1915
+  #
+  # The OME app where OME is waiting for the data of Ghostream.
+  #app: play
+
 ## SRT server ##
 # The SRT server receive incoming stream and can also serve video to clients.
 srt:

internal/config/config.go

@@ -2,6 +2,7 @@
 package config
 
 import (
+	"gitlab.crans.org/nounous/ghostream/stream/ovenmediaengine"
 	"net"
 
 	"github.com/sherifabdlnaby/configuro"
@@ -23,6 +24,7 @@ type Config struct {
 	Auth       auth.Options
 	Forwarding forwarding.Options
 	Monitoring monitoring.Options
+	OME        ovenmediaengine.Options
 	Srt        srt.Options
 	Telnet     telnet.Options
 	Transcoder transcoder.Options
@@ -49,6 +51,11 @@ func New() *Config {
 			Enabled:       true,
 			ListenAddress: ":2112",
 		},
+		OME: ovenmediaengine.Options{
+			Enabled: true,
+			URL:     "ovenmediaengine:1915",
+			App:     "play",
+		},
 		Srt: srt.Options{
 			Enabled:       true,
 			ListenAddress: ":9710",

main.go

@@ -5,6 +5,7 @@
 package main
 
 import (
+	"gitlab.crans.org/nounous/ghostream/stream/ovenmediaengine"
 	"log"
 
 	"github.com/pkg/profile"
@@ -49,9 +50,10 @@ func main() {
 	go transcoder.Init(streams, &cfg.Transcoder)
 	go forwarding.Serve(streams, cfg.Forwarding)
 	go monitoring.Serve(&cfg.Monitoring)
+	go ovenmediaengine.Serve(streams, &cfg.OME)
 	go srt.Serve(streams, authBackend, &cfg.Srt)
 	go telnet.Serve(streams, &cfg.Telnet)
-	go web.Serve(streams, &cfg.Web)
+	go web.Serve(streams, &cfg.Web, &cfg.OME)
 	go webrtc.Serve(streams, &cfg.WebRTC)
 
 	// Wait for routines

stream/forwarding/forwarding.go

@@ -40,6 +40,7 @@ func Serve(streams *messaging.Streams, cfg Options) {
 		stream, err := streams.Get(name)
 		if err != nil {
 			log.Printf("Failed to get stream '%s'", name)
+			return
 		}
 
 		// Get specific quality

stream/ovenmediaengine/ovenmediaengine.go

@@ -0,0 +1,112 @@
+// Package ovenmediaengine provides the forwarding to an ovenmediaengine server to handle the web client
+package ovenmediaengine
+
+import (
+	"bufio"
+	"fmt"
+	"log"
+	"os/exec"
+
+	"gitlab.crans.org/nounous/ghostream/messaging"
+)
+
+// Options holds ovenmediaengine package configuration
+type Options struct {
+	Enabled bool
+	URL     string
+	App     string
+}
+
+var (
+	cfg *Options
+)
+
+// Serve handles incoming packets from SRT and forward them to OME
+func Serve(streams *messaging.Streams, c *Options) {
+	cfg = c
+
+	if !c.Enabled {
+		return
+	}
+
+	// Subscribe to new stream event
+	event := make(chan string, 8)
+	streams.Subscribe(event)
+	log.Printf("Stream forwarding to OME initialized")
+
+	// 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)
+			return
+		}
+
+		qualityName := "source"
+		quality, err := stream.GetQuality(qualityName)
+		if err != nil {
+			log.Printf("Failed to get quality '%s'", qualityName)
+		}
+
+		// Start forwarding
+		log.Printf("Starting forwarding to OME for '%s'", name)
+		go forward(name, quality)
+	}
+}
+
+// Start a FFMPEG instance and redirect stream output to OME
+func forward(name string, q *messaging.Quality) {
+	output := make(chan []byte, 1024)
+	q.Register(output)
+
+	// TODO When a new OME version got released with SRT support, directly forward SRT packets, without using unwanted RTMP transport
+	// Launch FFMPEG instance
+	params := []string{"-hide_banner", "-loglevel", "error", "-i", "pipe:0", "-f", "flv", "-c:v", "copy",
+		"-c:a", "aac", "-b:a", "160k", "-ar", "44100",
+		fmt.Sprintf("rtmp://%s/%s/%s", cfg.URL, cfg.App, name)}
+	ffmpeg := exec.Command("ffmpeg", params...)
+
+	// Open pipes
+	input, err := ffmpeg.StdinPipe()
+	if err != nil {
+		log.Printf("Error while opening forwarding ffmpeg input pipe: %s", err)
+		return
+	}
+	errOutput, err := ffmpeg.StderrPipe()
+	if err != nil {
+		log.Printf("Error while opening forwarding ffmpeg output pipe: %s", err)
+		return
+	}
+
+	// Start FFMpeg
+	if err := ffmpeg.Start(); err != nil {
+		log.Printf("Error while starting forwarding ffmpeg instance: %s", err)
+	}
+
+	// Kill FFMPEG when stream is ended
+	defer func() {
+		_ = input.Close()
+		_ = errOutput.Close()
+		_ = ffmpeg.Process.Kill()
+		q.Unregister(output)
+	}()
+
+	// Log standard error output
+	go func() {
+		scanner := bufio.NewScanner(errOutput)
+		for scanner.Scan() {
+			log.Printf("[FORWARDING OME FFMPEG %s] %s", name, scanner.Text())
+		}
+	}()
+
+	// Read stream output and redirect immediately to ffmpeg
+	for data := range output {
+		_, err := input.Write(data)
+
+		if err != nil {
+			log.Printf("Error while writing to forwarded stream: %s", err)
+			break
+		}
+	}
+}

web/handler.go

@@ -10,15 +10,21 @@ import (
 	"net/http"
 	"regexp"
 	"strings"
+	"sync"
+	"time"
 
 	"github.com/markbates/pkger"
 	"gitlab.crans.org/nounous/ghostream/internal/monitoring"
+	"gitlab.crans.org/nounous/ghostream/stream/ovenmediaengine"
 	"gitlab.crans.org/nounous/ghostream/stream/webrtc"
 )
 
 var (
 	// Precompile regex
 	validPath = regexp.MustCompile("^/[a-z0-9@_-]*$")
+
+	counterMutex     = new(sync.Mutex)
+	connectedClients = make(map[string]map[string]int64)
 )
 
 // Handle site index and viewer pages
@@ -61,7 +67,8 @@ func viewerHandler(w http.ResponseWriter, r *http.Request) {
 		Cfg       *Options
 		Path      string
 		WidgetURL string
-	}{Path: path, Cfg: cfg, WidgetURL: ""}
+		OMECfg    *ovenmediaengine.Options
+	}{Path: path, Cfg: cfg, WidgetURL: "", OMECfg: omeCfg}
 
 	// Load widget is user does not disable it with ?nowidget
 	if _, ok := r.URL.Query()["nowidget"]; !ok {
@@ -88,14 +95,43 @@ func staticHandler() http.Handler {
 }
 
 func statisticsHandler(w http.ResponseWriter, r *http.Request) {
+	// Retrieve stream name from URL
 	name := strings.SplitN(strings.Replace(r.URL.Path[7:], "/", "", -1), "@", 2)[0]
 	userCount := 0
 
+	// Clients have a unique generated identifier per session, that expires in 40 seconds.
+	// Each time the client connects to this page, the identifier is renewed.
+	// Yeah, that's not a good way to have stats, but it works...
+	if connectedClients[name] == nil {
+		counterMutex.Lock()
+		connectedClients[name] = make(map[string]int64)
+		counterMutex.Unlock()
+	}
+	currentTime := time.Now().Unix()
+	if _, ok := r.URL.Query()["uid"]; ok {
+		uid := r.URL.Query()["uid"][0]
+		counterMutex.Lock()
+		connectedClients[name][uid] = currentTime
+		counterMutex.Unlock()
+	}
+	toDelete := make([]string, 0)
+	counterMutex.Lock()
+	for uid, oldTime := range connectedClients[name] {
+		if currentTime-oldTime > 40 {
+			toDelete = append(toDelete, uid)
+		}
+	}
+	for _, uid := range toDelete {
+		delete(connectedClients[name], uid)
+	}
+	counterMutex.Unlock()
+
 	// Get requested stream
 	stream, err := streams.Get(name)
 	if err == nil {
 		userCount = stream.ClientCount()
 		userCount += webrtc.GetNumberConnectedSessions(name)
+		userCount += len(connectedClients[name])
 	}
 
 	// Display connected users statistics

web/static/js/modules/viewerCounter.js

@@ -9,6 +9,7 @@ export class ViewerCounter {
     constructor(element, streamName) {
         this.element = element;
         this.url = "/_stats/" + streamName;
+        this.uid = Math.floor(1e19 * Math.random()).toString(16);
     }
 
     /**
@@ -21,7 +22,7 @@ export class ViewerCounter {
     }
 
     refreshViewersCounter() {
-        fetch(this.url)
+        fetch(this.url + "?uid=" + this.uid)
             .then(response => response.json())
             .then((data) => this.element.innerText = data.ConnectedViewers)
             .catch(console.log);

web/static/js/ovenplayer.js

@@ -4,10 +4,11 @@ import { ViewerCounter } from "./modules/viewerCounter.js";
  * Initialize viewer page
  *
  * @param {String} stream
+ * @param {String} omeApp
  * @param {Number} viewersCounterRefreshPeriod
  * @param {String} posterUrl
  */
-export function initViewerPage(stream, viewersCounterRefreshPeriod, posterUrl) {
+export function initViewerPage(stream, omeApp, viewersCounterRefreshPeriod, posterUrl) {
     // Create viewer counter
     const viewerCounter = new ViewerCounter(
         document.getElementById("connected-people"),
@@ -41,13 +42,13 @@ export function initViewerPage(stream, viewersCounterRefreshPeriod, posterUrl) {
         expandFullScreenUI: true,
         sources: [
             {
-                "file": "wss://" + window.location.host + "/app/" + stream,
+                "file": "wss://" + window.location.host + "/" + omeApp + "/" + stream,
                 "type": "webrtc",
                 "label": " WebRTC - Source"
             },
             {
                 "type": "hls",
-                "file": "https://" + window.location.host + "/app/" + stream + "_bypass/playlist.m3u8",
+                "file": "https://" + window.location.host + "/" + omeApp + "/" + stream + "_bypass/playlist.m3u8",
                 "label": " HLS"
             }
         ]

web/template/player.html

@@ -34,13 +34,17 @@
   {{end}}
 </div>
 
-
+{{if .OMECfg.Enabled}}
   <script src="https://cdn.jsdelivr.net/npm/hls.js@latest"></script>
   <script src="/static/ovenplayer/ovenplayer.js"></script>
   <script src="/static/js/ovenplayer.js"></script>
+{{end}}
 <script type="module">
-  // import { initViewerPage } from "/static/js/viewer.js";
+  {{if .OMECfg.Enabled}}
     import { initViewerPage } from "/static/js/ovenplayer.js";
+  {{else}}
+    import { initViewerPage } from "/static/js/viewer.js";
+  {{end}}
 
   // Some variables that need to be fixed by web page
   const viewersCounterRefreshPeriod = Number("{{.Cfg.ViewersCounterRefreshPeriod}}");
@@ -50,6 +54,10 @@
     "{{$value}}",
     {{end}}
   ]
-  initViewerPage(stream, /*stunServers,*/ viewersCounterRefreshPeriod, {{.Cfg.PlayerPoster}})
+  {{if .OMECfg.Enabled}}
+    initViewerPage(stream, {{.OMECfg.App}}, viewersCounterRefreshPeriod, {{.Cfg.PlayerPoster}})
+  {{else}}
+    initViewerPage(stream, stunServers, viewersCounterRefreshPeriod)
+  {{end}}
 </script>
 {{end}}

web/web.go

@@ -2,6 +2,7 @@
 package web
 
 import (
+	"gitlab.crans.org/nounous/ghostream/stream/ovenmediaengine"
 	"html/template"
 	"io/ioutil"
 	"log"
@@ -32,6 +33,8 @@ type Options struct {
 var (
 	cfg *Options
 
+	omeCfg *ovenmediaengine.Options
+
 	// Preload templates
 	templates *template.Template
 
@@ -70,9 +73,10 @@ func loadTemplates() error {
 }
 
 // Serve HTTP server
-func Serve(s *messaging.Streams, c *Options) {
+func Serve(s *messaging.Streams, c *Options, ome *ovenmediaengine.Options) {
 	streams = s
 	cfg = c
+	omeCfg = ome
 
 	if !cfg.Enabled {
 		// Web server is not enabled, ignore