support recording LPCM tracks (#2475)

README.md

@@ -42,6 +42,12 @@ And can be read from the server with:
 |[RTMP](#rtmp)|RTMP, RTMPS, Enhanced RTMP|H264|MPEG-4 Audio (AAC), MPEG-1/2 Audio (MP3)|
 |[HLS](#hls)|Low-Latency HLS, MP4-based HLS, legacy HLS|AV1, VP9, H265, H264|Opus, MPEG-4 Audio (AAC)|
 
+And can be recorded with:
+
+|format|video codecs|audio codecs|
+|------|------------|------------|
+|[fMP4](#record-streams-to-disk)|AV1, VP9, H265, H264, MPEG-4 Video (H263, Xvid), MPEG-1/2 Video, M-JPEG|Opus, MPEG-4 Audio (AAC), MPEG-1/2 Audio (MP3), AC-3, LPCM|
+
 **Features**
 
 * Publish live streams to the server
@@ -1151,10 +1157,7 @@ recordPath: ./recordings/%path/%Y-%m-%d_%H-%M-%S-%f
 
 All available recording parameters are listed in the [sample configuration file](/mediamtx.yml).
 
-Currently the server supports recording tracks encoded with the following codecs:
-
-* Video: AV1, VP9, H265, H264, MPEG-4 Video (H263, Xvid), MPEG-1/2 Video, M-JPEG
-* Audio: Opus, MPEG-4 Audio (AAC), MPEG-1/2 Audio (MP3), AC-3
+Be aware that not all tracks can be saved. A compatibility matrix is available at the beginning of the README.
 
 To upload recordings to a remote location, you can use _MediaMTX_ together with [rclone](https://github.com/rclone/rclone), a command line tool that provides file synchronization capabilities with a huge variety of services (including S3, FTP, SMB, Google Drive):
 

go.mod

@@ -9,7 +9,7 @@ require (
 	github.com/aler9/writerseeker v1.1.0
 	github.com/bluenviron/gohlslib v1.0.3
 	github.com/bluenviron/gortsplib/v4 v4.2.0
-	github.com/bluenviron/mediacommon v1.4.1-0.20230924203439-7ac007e2ac2d
+	github.com/bluenviron/mediacommon v1.4.1-0.20231007133411-92ec4e147f89
 	github.com/datarhei/gosrt v0.5.4
 	github.com/fsnotify/fsnotify v1.6.0
 	github.com/gin-gonic/gin v1.9.1

go.sum

@@ -16,8 +16,8 @@ github.com/bluenviron/gohlslib v1.0.3 h1:FMHevlIrrZ67uzCXmlTSGflsfYREEtHb8L9BDyf
 github.com/bluenviron/gohlslib v1.0.3/go.mod h1:R/aIsSxLI61N0CVMjtcHqJouK6+Ddd5YIihcCr7IFIw=
 github.com/bluenviron/gortsplib/v4 v4.2.0 h1:EbIMqkFxFo/iG5Hkld+Flew9R8ORKnuxlgUyFdpd5Rk=
 github.com/bluenviron/gortsplib/v4 v4.2.0/go.mod h1:wz9d4Tn2qS/mexc+BnvNeWzlNOpyaHzNK6SXxtg4mfM=
-github.com/bluenviron/mediacommon v1.4.1-0.20230924203439-7ac007e2ac2d h1:VbzIg0t5HKfyLbuzWeNU64JdOtTUp981Fx9ljdMRGpM=
-github.com/bluenviron/mediacommon v1.4.1-0.20230924203439-7ac007e2ac2d/go.mod h1:/vlOVSebDwzdRtQONOKLua0fOSJg1tUDHpP+h9a0uqM=
+github.com/bluenviron/mediacommon v1.4.1-0.20231007133411-92ec4e147f89 h1:IEicF/CpzU78BJ95v0XwTWL5D2Kr2grlpe5jnuhbRN8=
+github.com/bluenviron/mediacommon v1.4.1-0.20231007133411-92ec4e147f89/go.mod h1:Ij/kE1LEucSjryNBVTyPL/gBI0d6/Css3f5PyrM957w=
 github.com/bytedance/sonic v1.5.0/go.mod h1:ED5hyg4y6t3/9Ku1R6dU/4KyJ48DZ4jPhfY1O2AihPM=
 github.com/bytedance/sonic v1.9.1 h1:6iJ6NqdoxCDr6mbY8h18oSO+cShGSMRGCEo7F2h0x8s=
 github.com/bytedance/sonic v1.9.1/go.mod h1:i736AoUSYt75HyZLoJW9ERYxcy6eaN6h4BZXU064P/U=

internal/formatprocessor/lpcm.go

@@ -0,0 +1,109 @@
+package formatprocessor //nolint:dupl
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/bluenviron/gortsplib/v4/pkg/format"
+	"github.com/bluenviron/gortsplib/v4/pkg/format/rtplpcm"
+	"github.com/pion/rtp"
+
+	"github.com/bluenviron/mediamtx/internal/unit"
+)
+
+type formatProcessorLPCM struct {
+	udpMaxPayloadSize int
+	format            *format.LPCM
+	encoder           *rtplpcm.Encoder
+	decoder           *rtplpcm.Decoder
+}
+
+func newLPCM(
+	udpMaxPayloadSize int,
+	forma *format.LPCM,
+	generateRTPPackets bool,
+) (*formatProcessorLPCM, error) {
+	t := &formatProcessorLPCM{
+		udpMaxPayloadSize: udpMaxPayloadSize,
+		format:            forma,
+	}
+
+	if generateRTPPackets {
+		err := t.createEncoder()
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return t, nil
+}
+
+func (t *formatProcessorLPCM) createEncoder() error {
+	t.encoder = &rtplpcm.Encoder{
+		PayloadMaxSize: t.udpMaxPayloadSize - 12,
+	}
+	return t.encoder.Init()
+}
+
+func (t *formatProcessorLPCM) ProcessUnit(uu unit.Unit) error { //nolint:dupl
+	u := uu.(*unit.LPCM)
+
+	pkts, err := t.encoder.Encode(u.Samples)
+	if err != nil {
+		return err
+	}
+
+	ts := uint32(multiplyAndDivide(u.PTS, time.Duration(t.format.ClockRate()), time.Second))
+	for _, pkt := range pkts {
+		pkt.Timestamp += ts
+	}
+
+	u.RTPPackets = pkts
+
+	return nil
+}
+
+func (t *formatProcessorLPCM) ProcessRTPPacket( //nolint:dupl
+	pkt *rtp.Packet,
+	ntp time.Time,
+	pts time.Duration,
+	hasNonRTSPReaders bool,
+) (Unit, error) {
+	u := &unit.LPCM{
+		Base: unit.Base{
+			RTPPackets: []*rtp.Packet{pkt},
+			NTP:        ntp,
+			PTS:        pts,
+		},
+	}
+
+	// remove padding
+	pkt.Header.Padding = false
+	pkt.PaddingSize = 0
+
+	if pkt.MarshalSize() > t.udpMaxPayloadSize {
+		return nil, fmt.Errorf("payload size (%d) is greater than maximum allowed (%d)",
+			pkt.MarshalSize(), t.udpMaxPayloadSize)
+	}
+
+	// decode from RTP
+	if hasNonRTSPReaders || t.decoder != nil {
+		if t.decoder == nil {
+			var err error
+			t.decoder, err = t.format.CreateDecoder()
+			if err != nil {
+				return nil, err
+			}
+		}
+
+		samples, err := t.decoder.Decode(pkt)
+		if err != nil {
+			return nil, err
+		}
+
+		u.Samples = samples
+	}
+
+	// route packet as is
+	return u, nil
+}

internal/formatprocessor/processor.go

@@ -75,6 +75,9 @@ func New(
 	case *format.AC3:
 		return newAC3(udpMaxPayloadSize, forma, generateRTPPackets)
 
+	case *format.LPCM:
+		return newLPCM(udpMaxPayloadSize, forma, generateRTPPackets)
+
 	default:
 		return newGeneric(udpMaxPayloadSize, forma, generateRTPPackets)
 	}

internal/record/agent.go

@@ -816,7 +816,27 @@ func NewAgent(
 				// TODO
 
 			case *format.LPCM:
-				// TODO
+				codec := &fmp4.CodecLPCM{
+					LittleEndian: false,
+					BitDepth:     forma.BitDepth,
+					SampleRate:   forma.SampleRate,
+					ChannelCount: forma.ChannelCount,
+				}
+				track := addTrack(codec)
+
+				stream.AddReader(r.writer, media, forma, func(u unit.Unit) error {
+					tunit := u.(*unit.LPCM)
+					if tunit.Samples == nil {
+						return nil
+					}
+
+					return track.record(&sample{
+						PartSample: &fmp4.PartSample{
+							Payload: tunit.Samples,
+						},
+						dts: tunit.PTS,
+					})
+				})
 			}
 		}
 	}

internal/unit/lpcm.go

@@ -0,0 +1,7 @@
+package unit
+
+// LPCM is a LPCM data unit.
+type LPCM struct {
+	Base
+	Samples []byte
+}