add opus audio initial support

Dockerfile-dev

@@ -9,9 +9,10 @@ ENV LD_LIBRARY_PATH="/usr/local/lib:/usr/lib:/usr/lib/x86_64-linux-gnu/"
 ENV CGO_CFLAGS="-I/usr/local/include/"
 ENV CGO_LDFLAGS="-L/usr/local/lib"
 
+RUN apt-get clean && apt-get update && \
+    DEBIAN_FRONTEND=noninteractive apt-get install -y \
+      tclsh pkg-config cmake libssl-dev build-essential git \
+    && apt-get clean
+
 ENV WD=/usr/src/app
 WORKDIR ${WD}
-
-ENV GOPROXY=direct
-
-RUN curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(go env GOPATH)/bin v1.55.2

Makefile

@@ -1,16 +1,20 @@
 run:
-	docker compose stop && docker compose up origin srt app
+	docker compose stop && docker compose up app
 
 run-dev:
-	docker compose stop && docker compose down && docker compose build app && docker compose up origin srt app
+	docker compose stop && docker compose down && docker compose build app && docker compose up app
 
 run-dev-total-rebuild:
-	docker compose stop && docker compose down && docker compose build && docker compose up origin srt app
+	docker compose stop && docker compose down && docker compose build && docker compose up app
 
 clean-docker:
 	docker-compose down -v --rmi all --remove-orphans &&  docker volume prune -a -f && docker system prune -a -f && docker builder prune -a -f
 
+run-docker-dev:
+	docker compose run --rm --service-ports dev
+
+run-server-inside-docker:
+	go run main.go -- --enable-ice-mux=true
+
 lint:
 	docker compose stop lint && docker compose down lint && docker compose run --rm lint	
-
-.PHONY: run lint test run-srt mac-run-local mac-test-local html-local-coverage install-ffmpeg

docker-compose.yaml

@@ -12,11 +12,86 @@ services:
       - "8081:8081/udp"
       - "6060:6060"
     depends_on:
-      - srt
+      - haivision_srt
-      - rtmp
+      - nginx_rtmp
     links:
-      - srt
+      - haivision_srt
-      - rtmp
+      - nginx_rtmp
+
+  dev:
+    build:
+      context: .
+      dockerfile: Dockerfile-dev
+    working_dir: "/app"
+    platform: "linux/amd64"
+    volumes:
+      - "./:/app/"
+    command: "bash"
+    ports:
+      - "8080:8080"
+      - "8081:8081"
+      - "8081:8081/udp"
+      - "6060:6060"
+    depends_on:
+      - haivision_srt
+      - nginx_rtmp
+    links:
+      - haivision_srt
+      - nginx_rtmp
+
+  nginx_rtmp:
+    image: alfg/nginx-rtmp
+    ports:
+      - "1935:1935"
+    volumes:
+      - ./nginx.conf:/etc/nginx/nginx.conf.template
+    depends_on:
+      - origin_rtmp
+    links:
+      - origin_rtmp
+
+  origin_rtmp: # simulating an RTMP flv (h264/aac) live transmission
+    image: jrottenberg/ffmpeg:4.4-alpine
+    entrypoint: sh
+    command: "/scripts/ffmpeg_rtmp.sh"
+    volumes:
+      - "./scripts:/scripts"
+      - "./fonts/0xProto:/usr/share/fonts"
+    environment:
+      - RTMP_HOST=nginx_rtmp
+      - RTMP_PORT=1935
+
+  haivision_srt:
+    build:
+      context: .
+      dockerfile: Dockerfile-srt-live
+    entrypoint: sh
+    command: "./srt.sh"
+    working_dir: "/scripts"
+    volumes:
+      - "./scripts:/scripts"
+    environment:
+      - SRT_LISTENING_PORT=40052
+      - SRT_UDP_TS_INPUT_HOST=0.0.0.0
+      - SRT_UDP_TS_INPUT_PORT=1234
+    ports:
+      - "40052:40052/udp"
+    depends_on:
+      - origin_srt
+    links:
+      - origin_srt
+
+  origin_srt: # simulating an (h264/aac) mpeg-ts upd origin live transmission
+    image: jrottenberg/ffmpeg:4.4-alpine
+    entrypoint: sh
+    command: "/scripts/ffmpeg_mpegts_udp.sh"
+    volumes:
+      - "./scripts:/scripts"
+      - "./fonts/0xProto:/usr/share/fonts"      
+    environment:
+      - SRT_INPUT_HOST=haivision_srt
+      - SRT_INPUT_PORT=1234
+      - PKT_SIZE=1316
 
   test:
     build:
@@ -37,48 +112,3 @@ services:
     volumes:
       - "./:/app/"
     command: "golangci-lint run -v"
-
-  rtmp: # simulating an RTMP live transmission
-    image: jrottenberg/ffmpeg:4.4-alpine
-    entrypoint: sh
-    command: "/scripts/ffmpeg_rtmp.sh"
-    volumes:
-      - "./scripts:/scripts"
-      - "./fonts/0xProto:/usr/share/fonts"
-    environment:
-      - RTMP_HOST=0.0.0.0
-      - RTMP_PORT=1935
-    ports:
-      - "1935:1935"
-
-  srt:
-    build:
-      context: .
-      dockerfile: Dockerfile-srt-live
-    entrypoint: sh
-    command: "./srt.sh"
-    working_dir: "/scripts"
-    volumes:
-      - "./scripts:/scripts"
-    environment:
-      - SRT_LISTENING_PORT=40052
-      - SRT_UDP_TS_INPUT_HOST=0.0.0.0
-      - SRT_UDP_TS_INPUT_PORT=1234
-    ports:
-      - "40052:40052/udp"
-    depends_on:
-      - origin
-    links:
-      - origin
-
-  origin: # simulating an mpeg-ts upd origin live transmission
-    image: jrottenberg/ffmpeg:4.4-alpine
-    entrypoint: sh
-    command: "/scripts/ffmpeg_mpegts_udp.sh"
-    volumes:
-      - "./scripts:/scripts"
-      - "./fonts/0xProto:/usr/share/fonts"      
-    environment:
-      - SRT_INPUT_HOST=srt
-      - SRT_INPUT_PORT=1234
-      - PKT_SIZE=1316

internal/controllers/engine/donut_engine_controller.go

@@ -123,12 +123,8 @@ func (d *donutEngine) RecipeFor(server, client *entities.StreamInfo) (*entities.
 			Codec:  entities.Opus,
 			// TODO: create method list options per Codec
 			CodecContextOptions: []entities.LibAVOptionsCodecContext{
-				// opus specifically works under 48000 Hz
 				entities.SetSampleRate(48000),
-				// once we changed the sample rate we need to update the time base
+				entities.SetSampleFormat("fltp"),
-				entities.SetTimeBase(1, 48000),
-				// for some reason it's setting "s16"
-				// entities.SetSampleFormat("fltp"),
 			},
 		},
 	}

internal/controllers/streamers/libav_ffmpeg.go

@@ -4,8 +4,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"reflect"
-	"runtime"
 	"strconv"
 	"strings"
 	"time"
@@ -95,7 +93,8 @@ func (c *LibAVFFmpegStreamer) Stream(donut *entities.DonutParameters) {
 	}
 
 	// it's useful for debugging
-	astiav.SetLogLevel(astiav.LogLevelDebug)
+	// astiav.SetLogLevel(astiav.LogLevelDebug)
+	astiav.SetLogLevel(astiav.LogLevelInfo)
 	astiav.SetLogCallback(func(_ astiav.Classer, l astiav.LogLevel, fmt, msg string) {
 		c.l.Infof("ffmpeg %s: - %s", c.libAVLogToString(l), strings.TrimSpace(msg))
 	})
@@ -147,17 +146,17 @@ func (c *LibAVFFmpegStreamer) Stream(donut *entities.DonutParameters) {
 
 			s, ok := p.streams[inPkt.StreamIndex()]
 			if !ok {
-				c.l.Warnf("cannot find stream id=%d", inPkt.StreamIndex())
+				c.l.Warnf("skipping to process stream id=%d", inPkt.StreamIndex())
 				continue
 			}
 
 			if s.bsfContext != nil {
-				if err := c.applyBitStreamFilter(inPkt, s, donut); err != nil {
+				if err := c.applyBitStreamFilter(p, inPkt, s, donut); err != nil {
 					c.onError(err, donut)
 					return
 				}
 			} else {
-				if err := c.processPacket(inPkt, s, donut); err != nil {
+				if err := c.processPacket(p, inPkt, s, donut); err != nil {
 					c.onError(err, donut)
 					return
 				}
@@ -215,6 +214,9 @@ func (c *LibAVFFmpegStreamer) prepareInput(p *libAVParams, closer *astikit.Close
 			return fmt.Errorf("ffmpeg/libav: updating codec context failed %w", err)
 		}
 
+		//FFMPEG_NEW
+		s.decCodecContext.SetTimeBase(s.inputStream.TimeBase())
+
 		if is.CodecParameters().MediaType() == astiav.MediaTypeVideo {
 			s.decCodecContext.SetFramerate(p.inputFormatContext.GuessFrameRate(is, nil))
 		}
@@ -239,17 +241,11 @@ func (c *LibAVFFmpegStreamer) prepareInput(p *libAVParams, closer *astikit.Close
 	return nil
 }
 
-func functionNameFor(i interface{}) string {
-	fullName := runtime.FuncForPC(reflect.ValueOf(i).Pointer()).Name()
-	components := strings.Split(fullName, ".")
-	return components[len(components)-2]
-}
-
 func (c *LibAVFFmpegStreamer) prepareOutput(p *libAVParams, closer *astikit.Closer, donut *entities.DonutParameters) error {
 	for _, is := range p.inputFormatContext.Streams() {
 		s, ok := p.streams[is.Index()]
 		if !ok {
-			c.l.Infof("skipping stream index = %d", is.Index())
+			c.l.Infof("skipping absent stream index = %d", is.Index())
 			continue
 		}
 
@@ -308,10 +304,9 @@ func (c *LibAVFFmpegStreamer) prepareOutput(p *libAVParams, closer *astikit.Clos
 			}
 			s.encCodecContext.SetTimeBase(s.decCodecContext.TimeBase())
 
-			// supplying custom config
+			// overriding with user provide config
 			if len(donut.Recipe.Audio.CodecContextOptions) > 0 {
 				for _, opt := range donut.Recipe.Audio.CodecContextOptions {
-					c.l.Infof("overriding av codec context %s", functionNameFor(opt))
 					opt(s.encCodecContext)
 				}
 			}
@@ -327,12 +322,11 @@ func (c *LibAVFFmpegStreamer) prepareOutput(p *libAVParams, closer *astikit.Clos
 			s.encCodecContext.SetTimeBase(s.decCodecContext.TimeBase())
 			s.encCodecContext.SetHeight(s.decCodecContext.Height())
 			s.encCodecContext.SetWidth(s.decCodecContext.Width())
-			s.encCodecContext.SetFramerate(s.inputStream.AvgFrameRate())
+			// s.encCodecContext.SetFramerate(s.inputStream.AvgFrameRate())
 
-			// supplying custom config
+			// overriding with user provide config
 			if len(donut.Recipe.Video.CodecContextOptions) > 0 {
 				for _, opt := range donut.Recipe.Video.CodecContextOptions {
-					c.l.Infof("overriding av codec context %s", functionNameFor(opt))
 					opt(s.encCodecContext)
 				}
 			}
@@ -388,7 +382,7 @@ func (c *LibAVFFmpegStreamer) prepareFilters(p *libAVParams, closer *astikit.Clo
 		}
 		closer.Add(inputs.Free)
 
-		if s.decCodecContext.MediaType() == astiav.MediaTypeAudio {
+		if isAudio {
 			args = astiav.FilterArgs{
 				"channel_layout": s.decCodecContext.ChannelLayout().String(),
 				"sample_fmt":     s.decCodecContext.SampleFormat().Name(),
@@ -403,7 +397,7 @@ func (c *LibAVFFmpegStreamer) prepareFilters(p *libAVParams, closer *astikit.Clo
 			)
 		}
 
-		if s.decCodecContext.MediaType() == astiav.MediaTypeVideo {
+		if isVideo {
 			args = astiav.FilterArgs{
 				"pix_fmt":      strconv.Itoa(int(s.decCodecContext.PixelFormat())),
 				"pixel_aspect": s.decCodecContext.SampleAspectRatio().String(),
@@ -502,7 +496,7 @@ func (c *LibAVFFmpegStreamer) prepareBitStreamFilters(p *libAVParams, closer *as
 	return nil
 }
 
-func (c *LibAVFFmpegStreamer) processPacket(pkt *astiav.Packet, s *streamContext, donut *entities.DonutParameters) error {
+func (c *LibAVFFmpegStreamer) processPacket(p *libAVParams, pkt *astiav.Packet, s *streamContext, donut *entities.DonutParameters) error {
 	isVideo := s.decCodecContext.MediaType() == astiav.MediaTypeVideo
 	isAudio := s.decCodecContext.MediaType() == astiav.MediaTypeAudio
 	var currentMedia *entities.DonutMediaTask
@@ -516,11 +510,10 @@ func (c *LibAVFFmpegStreamer) processPacket(pkt *astiav.Packet, s *streamContext
 		return nil
 	}
 
-	pkt.RescaleTs(s.inputStream.TimeBase(), s.decCodecContext.TimeBase())
-
 	byPass := currentMedia.Action == entities.DonutBypass
 	if isVideo && byPass {
 		if donut.OnVideoFrame != nil {
+			pkt.RescaleTs(s.inputStream.TimeBase(), s.decCodecContext.TimeBase())
 			if err := donut.OnVideoFrame(pkt.Data(), entities.MediaFrameContext{
 				PTS:      int(pkt.Pts()),
 				DTS:      int(pkt.Dts()),
@@ -533,6 +526,7 @@ func (c *LibAVFFmpegStreamer) processPacket(pkt *astiav.Packet, s *streamContext
 	}
 	if isAudio && byPass {
 		if donut.OnAudioFrame != nil {
+			pkt.RescaleTs(s.inputStream.TimeBase(), s.decCodecContext.TimeBase())
 			if err := donut.OnAudioFrame(pkt.Data(), entities.MediaFrameContext{
 				PTS:      int(pkt.Pts()),
 				DTS:      int(pkt.Dts()),
@@ -559,14 +553,14 @@ func (c *LibAVFFmpegStreamer) processPacket(pkt *astiav.Packet, s *streamContext
 			}
 			return err
 		}
-		if err := c.filterAndEncode(s.decFrame, s, donut); err != nil {
+		if err := c.filterAndEncode(p, s.decFrame, s, donut); err != nil {
 			return err
 		}
 	}
 	return nil
 }
 
-func (c *LibAVFFmpegStreamer) applyBitStreamFilter(pkt *astiav.Packet, s *streamContext, donut *entities.DonutParameters) error {
+func (c *LibAVFFmpegStreamer) applyBitStreamFilter(p *libAVParams, pkt *astiav.Packet, s *streamContext, donut *entities.DonutParameters) error {
 	if err := s.bsfContext.SendPacket(pkt); err != nil && !errors.Is(err, astiav.ErrEagain) {
 		return fmt.Errorf("sending bit stream packet failed: %w", err)
 	}
@@ -579,13 +573,13 @@ func (c *LibAVFFmpegStreamer) applyBitStreamFilter(pkt *astiav.Packet, s *stream
 			return fmt.Errorf("receiving bit stream packet failed: %w", err)
 		}
 
-		c.processPacket(s.bsfPacket, s, donut)
+		c.processPacket(p, s.bsfPacket, s, donut)
 		s.bsfPacket.Unref()
 	}
 	return nil
 }
 
-func (c *LibAVFFmpegStreamer) filterAndEncode(f *astiav.Frame, s *streamContext, donut *entities.DonutParameters) (err error) {
+func (c *LibAVFFmpegStreamer) filterAndEncode(p *libAVParams, f *astiav.Frame, s *streamContext, donut *entities.DonutParameters) (err error) {
 	if err = s.buffersrcContext.BuffersrcAddFrame(f, astiav.NewBuffersrcFlags(astiav.BuffersrcFlagKeepRef)); err != nil {
 		return fmt.Errorf("adding frame failed: %w", err)
 	}
@@ -601,7 +595,7 @@ func (c *LibAVFFmpegStreamer) filterAndEncode(f *astiav.Frame, s *streamContext,
 		}
 		// TODO: should we avoid setting the picture type for audio?
 		s.filterFrame.SetPictureType(astiav.PictureTypeNone)
-		if err = c.encodeFrame(s.filterFrame, s, donut); err != nil {
+		if err = c.encodeFrame(p, s.filterFrame, s, donut); err != nil {
 			err = fmt.Errorf("main: encoding and writing frame failed: %w", err)
 			return
 		}
@@ -609,12 +603,15 @@ func (c *LibAVFFmpegStreamer) filterAndEncode(f *astiav.Frame, s *streamContext,
 	return nil
 }
 
-func (c *LibAVFFmpegStreamer) encodeFrame(f *astiav.Frame, s *streamContext, donut *entities.DonutParameters) (err error) {
+func (c *LibAVFFmpegStreamer) encodeFrame(p *libAVParams, f *astiav.Frame, s *streamContext, donut *entities.DonutParameters) (err error) {
 	s.encPkt.Unref()
 
-	// when converting from aac to opus using filters, the np samples are bigger than the frame size
+	// when converting from aac to opus using filters,
+	// the np samples are bigger than the frame size
 	// to fix the error "more samples than frame size"
+	if f != nil {
 		f.SetNbSamples(s.encCodecContext.FrameSize())
+	}
 
 	if err = s.encCodecContext.SendFrame(f); err != nil {
 		return fmt.Errorf("sending frame failed: %w", err)

internal/web/handlers/signaling.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/json"
 	"net/http"
-	"time"
 
 	"github.com/flavioribeiro/donut/internal/controllers"
 	"github.com/flavioribeiro/donut/internal/controllers/engine"
@@ -79,11 +78,6 @@ func (h *SignalingHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) err
 	}
 	h.l.Infof("WebRTCResponse %#v", webRTCResponse)
 
-	//TODO: remove the sleeping
-	// The simulated RTMP stream (/scripts/ffmpeg_rtmp.sh) goes down every time a client disconnects.
-	// The prober is forcing the first restart therefore it waits for 4 seconds.
-	time.Sleep(4 * time.Second)
-
 	go donutEngine.Serve(&entities.DonutParameters{
 		Cancel: cancel,
 		Ctx:    ctx,

nginx.conf

@@ -0,0 +1,19 @@
+daemon off;
+
+error_log /dev/stdout info;
+
+events {
+    worker_connections 1024;
+}
+
+rtmp {
+    server {
+        listen 1935;
+        chunk_size 4000;
+
+        application live {
+            live on;
+            record off;
+        }
+    }
+}

scripts/ffmpeg_mpegts_udp.sh

@@ -1,4 +1,4 @@
-ffmpeg -hide_banner -loglevel verbose \
+ffmpeg -hide_banner -loglevel info \
     -re -f lavfi -i testsrc2=size=768x432:rate=30,format=yuv420p \
     -f lavfi -i sine=frequency=1000:sample_rate=44100 \
     -c:v libx264 -preset veryfast -tune zerolatency -profile:v baseline \

scripts/ffmpeg_rtmp.sh

@@ -1,12 +1,9 @@
 #!/bin/bash
-while true
+    ffmpeg -hide_banner -loglevel info \
-do
-    ffmpeg -hide_banner -loglevel debug \
         -re -f lavfi -i testsrc2=size=768x432:rate=30,format=yuv420p \
         -f lavfi -i sine=frequency=1000:sample_rate=44100 \
         -c:v libx264 -preset veryfast -tune zerolatency -profile:v baseline \
         -vf "drawtext=text='RTMP streaming':box=1:boxborderw=10:x=(w-text_w)/2:y=(h-text_h)/2:fontsize=64:fontcolor=black" \
         -b:v 1000k -bufsize 2000k -x264opts keyint=30:min-keyint=30:scenecut=-1 \
         -c:a aac -b:a 128k \
-        -f flv -listen 1 -rtmp_live live "rtmp://${RTMP_HOST}:${RTMP_PORT}/live/app"
+        -f flv -rtmp_live live "rtmp://${RTMP_HOST}:${RTMP_PORT}/live/app"
-done

static/demo.css

@@ -2,7 +2,70 @@
     SPDX-FileCopyrightText: 2023 The Pion community <https://pion.ly>
     SPDX-License-Identifier: MIT
 */
-textarea {
+
-    width: 500px;
+* {
-    min-height: 75px;
+  font-family: "Open Sans", sans-serif;
+}
+input {
+    width: 100%;
+    padding: 12px 20px;
+    margin: 8px 0;
+    box-sizing: border-box;
+}
+
+legend {
+  background-color: #000;
+  color: #fff;
+  padding: 3px 6px;
+}
+
+.hint {
+  color:darkgray;
+}
+
+button {
+  align-items: center;
+  background-color: #FFFFFF;
+  border: 1px solid rgba(0, 0, 0, 0.1);
+  border-radius: .25rem;
+  box-shadow: rgba(0, 0, 0, 0.02) 0 1px 3px 0;
+  box-sizing: border-box;
+  color: rgba(0, 0, 0, 0.85);
+  cursor: pointer;
+  display: inline-flex;
+  font-family: system-ui,-apple-system,system-ui,"Helvetica Neue",Helvetica,Arial,sans-serif;
+  font-size: 16px;
+  font-weight: 600;
+  justify-content: center;
+  line-height: 1.25;
+  margin: 0;
+  min-height: 3rem;
+  padding: calc(.875rem - 1px) calc(1.5rem - 1px);
+  position: relative;
+  text-decoration: none;
+  transition: all 250ms;
+  user-select: none;
+  -webkit-user-select: none;
+  touch-action: manipulation;
+  vertical-align: baseline;
+  width: auto;
+}
+
+button:hover,
+button:focus {
+  border-color: rgba(0, 0, 0, 0.15);
+  box-shadow: rgba(0, 0, 0, 0.1) 0 4px 12px;
+  color: rgba(0, 0, 0, 0.65);
+}
+
+button {
+  transform: translateY(-1px);
+}
+
+button {
+  background-color: #F0F0F1;
+  border-color: rgba(0, 0, 0, 0.15);
+  box-shadow: rgba(0, 0, 0, 0.06) 0 2px 4px;
+  color: rgba(0, 0, 0, 0.65);
+  transform: translateY(0);
 }

static/index.html

@@ -3,29 +3,48 @@
 <head>
 	<title>donut</title>
 	<script src="demo.js"></script>
-	<style src="demo.css"></style>
+	<link rel="stylesheet" href="demo.css">
+	</link>
+	<link rel="preconnect" href="https://fonts.googleapis.com">
+	<link rel="preconnect" href="https://fonts.gstatic.com" crossorigin>
+	<link href="https://fonts.googleapis.com/css2?family=Open+Sans:ital,wght@0,300..800;1,300..800&display=swap"
+		rel="stylesheet">
 </head>
 
 <body>
 
-	<h1>Remote streaming</h1>
+	<fieldset>
-	<b> URL </b>
+		<legend>Remote streaming</legend>
-	<input type="text" id="stream-url" value="srt://srt:40052"> <br />
+		<p>
-
+			<label for="stream-url">URL: <span aria-label="required">*</span></label>
-	<b> ID </b>
+			<input id="stream-url" type="text" name="stream-url" required value="srt://haivision_srt:40052" />
-	<input type="text" id="stream-id" value="stream-id" /> <br />
+			<label class="hint">rtmp://nginx_rtmp/live</label>
-
+		</p>
+		<p>
+			<label for="stream-id">ID: <span aria-label="required">*</span></label>
+			<input id="stream-id" type="text" name="stream-id" required value="stream-id" />
+			<label class="hint">app</label>
+		</p>
+		<p>
 			<button onclick="onConnect()"> Connect </button>
+		</p>
+	</fieldset>
 
-	<h1>Video</h1>
+	<fieldset>
+		<legend>Video</legend>
 		<div id="remoteVideos"></div>
+	</fieldset>
 
-	<h1>Metadata</h1>
+
+	<fieldset>
+		<legend>Metadata</legend>
 		<div id="metadata"></div>
+	</fieldset>
 
-	<h1>Logs</h1>
+	<fieldset>
+		<legend>Logs</legend>
 		<div id="log"></div>
-
+	</fieldset>
 </body>
 
 <script>