From 8269a36d16743045efdb91691d4021afe04f3f24 Mon Sep 17 00:00:00 2001
From: Josh Allmann <joshua.allmann@gmail.com>
Date: Fri, 17 Jan 2025 06:53:45 +0000
Subject: [PATCH] rtmp: Support Opus writes from internal stream.

---
 internal/protocols/rtmp/from_stream.go | 20 ++++++++++++++-
 internal/protocols/rtmp/writer.go      | 34 ++++++++++++++++++++++++++
 2 files changed, 53 insertions(+), 1 deletion(-)

diff --git a/internal/protocols/rtmp/from_stream.go b/internal/protocols/rtmp/from_stream.go
index 724dd1c1e97..e89d1f6d538 100644
--- a/internal/protocols/rtmp/from_stream.go
+++ b/internal/protocols/rtmp/from_stream.go
@@ -16,7 +16,7 @@ import (
 )
 
 var errNoSupportedCodecsFrom = errors.New(
-	"the stream doesn't contain any supported codec, which are currently H264, MPEG-4 Audio, MPEG-1/2 Audio")
+	"the stream doesn't contain any supported codec, which are currently H264, MPEG-4 Audio, MPEG-1/2 Audio, Opus")
 
 func multiplyAndDivide2(v, m, d time.Duration) time.Duration {
 	secs := v / d
@@ -179,6 +179,24 @@ func setupAudio(
 		return audioFormatMPEG1
 	}
 
+	var audioFormatOpus *format.Opus
+	audioMedia = strea.Desc().FindFormat(&audioFormatOpus)
+	if audioMedia != nil {
+		strea.AddReader(
+			reader,
+			audioMedia,
+			audioFormatOpus,
+			func(u unit.Unit) error {
+				tunit := u.(*unit.Opus)
+				if tunit.Packets == nil {
+					return nil
+				}
+				nconn.SetWriteDeadline(time.Now().Add(writeTimeout))
+				return (*w).WriteOpus(timestampToDuration(tunit.PTS, audioFormatOpus.ClockRate()), tunit.Packets)
+			})
+		return audioFormatOpus
+	}
+
 	return nil
 }
 
diff --git a/internal/protocols/rtmp/writer.go b/internal/protocols/rtmp/writer.go
index 575d0d11de2..583175f9c88 100644
--- a/internal/protocols/rtmp/writer.go
+++ b/internal/protocols/rtmp/writer.go
@@ -100,6 +100,9 @@ func (w *Writer) writeTracks(videoTrack format.Format, audioTrack format.Format)
 						case *format.MPEG4Audio:
 							return message.CodecMPEG4Audio
 
+						case *format.Opus:
+							return float64(message.FourCCOpus)
+
 						default:
 							return 0
 						}
@@ -162,6 +165,21 @@ func (w *Writer) writeTracks(videoTrack format.Format, audioTrack format.Format)
 		}
 	}
 
+	if track, ok := audioTrack.(*format.Opus); ok {
+		err = w.conn.Write(&message.AudioExSequenceStart{
+			ChunkStreamID:   message.AudioChunkStreamID,
+			MessageStreamID: 0x1000000,
+			FourCC:          message.FourCCOpus,
+			OpusHeader: &message.OpusIDHeader{
+				ChannelCount: uint8(track.ChannelCount),
+				PreSkip:      3840,
+			},
+		})
+		if err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 
@@ -212,3 +230,19 @@ func (w *Writer) WriteMPEG1Audio(pts time.Duration, h *mpeg1audio.FrameHeader, f
 		DTS:             pts,
 	})
 }
+
+func (w *Writer) WriteOpus(pts time.Duration, packets [][]byte) error {
+	// only write one packet for now
+	// additional packets will require self-delimiting framing
+	// see Appendix B of RFC 6716
+	for _, packet := range packets {
+		return w.conn.Write(&message.AudioExCodedFrames{
+			ChunkStreamID:   message.AudioChunkStreamID,
+			MessageStreamID: 0x1000000,
+			DTS:             pts,
+			FourCC:          message.FourCCOpus,
+			Payload:         packet,
+		})
+	}
+	return nil
+}