diff --git a/machinery/src/components/backchannel.go b/machinery/src/components/backchannel.go index 0769854..62f115d 100644 --- a/machinery/src/components/backchannel.go +++ b/machinery/src/components/backchannel.go @@ -2,7 +2,9 @@ package components import ( "bufio" + "context" "fmt" + "math/rand" "os" "time" @@ -15,6 +17,62 @@ import ( "github.com/zaf/g711" ) +const ( + backchannelSampleRate = 8000 + backchannelTalkspurtGap = 500 * time.Millisecond + backchannelReconnectInitial = time.Second + backchannelReconnectMax = 30 * time.Second +) + +type backchannelClient interface { + ConnectBackChannel(ctx context.Context, otelContext context.Context) error + StartBackChannel(ctx context.Context, otelContext context.Context) error + WritePacket(pkt packets.Packet) error + Close(otelContext context.Context) error +} + +type backchannelPacketizer struct { + sequenceNumber uint16 + timestamp uint32 + ssrc uint32 + lastPacketAt time.Time +} + +func newBackchannelPacketizer() backchannelPacketizer { + return backchannelPacketizer{ + sequenceNumber: uint16(rand.Uint32()), + timestamp: rand.Uint32(), + ssrc: rand.Uint32(), + } +} + +func (p *backchannelPacketizer) packet(audio models.AudioDataPartial, now time.Time) packets.Packet { + bufferUlaw := make([]byte, len(audio.Data)) + for index, sample := range audio.Data { + bufferUlaw[index] = g711.EncodeUlawFrame(sample) + } + + pkt := packets.Packet{ + Packet: &rtp.Packet{ + Header: rtp.Header{ + Version: 2, + Marker: p.lastPacketAt.IsZero() || now.Sub(p.lastPacketAt) >= backchannelTalkspurtGap, + PayloadType: 0, + SequenceNumber: p.sequenceNumber, + Timestamp: p.timestamp, + SSRC: p.ssrc, + }, + Payload: bufferUlaw, + }, + } + + p.timestamp += uint32(len(bufferUlaw)) + p.sequenceNumber++ + p.lastPacketAt = now + + return pkt +} + func GetBackChannelAudioCodec(streams []av.CodecData, communication *models.Communication) av.AudioCodecData { for _, stream := range streams { if stream.Type().IsAudio() { @@ -31,41 +89,115 @@ func GetBackChannelAudioCodec(streams []av.CodecData, communication *models.Comm } func WriteAudioToBackchannel(communication *models.Communication, rtspClient capture.RTSPClient) { - log.Log.Info("Audio.WriteAudioToBackchannel(): writing to backchannel audio codec") - length := uint32(0) - sequenceNumber := uint16(0) - for audio := range communication.HandleAudio { - // Encode PCM to MULAW - var bufferUlaw []byte - for _, v := range audio.Data { - b := g711.EncodeUlawFrame(v) - bufferUlaw = append(bufferUlaw, b) - } - - pkt := packets.Packet{ - Packet: &rtp.Packet{ - Header: rtp.Header{ - Version: 2, - Marker: true, // should be true - PayloadType: 0, //packet.PayloadType, // will be owerwriten - SequenceNumber: sequenceNumber, - Timestamp: uint32(length), - SSRC: 1293847657, - }, - Payload: bufferUlaw, - }, - } - err := rtspClient.WritePacket(pkt) - if err != nil { - log.Log.Error("Audio.WriteAudioToBackchannel(): error writing packet to backchannel") - } - - length = (length + uint32(len(bufferUlaw))) % 65536 - sequenceNumber = (sequenceNumber + 1) % 65535 - time.Sleep(128 * time.Millisecond) + ctx := context.Background() + if communication.Context != nil { + ctx = *communication.Context } - log.Log.Info("Audio.WriteAudioToBackchannel(): finished") + writeAudioToBackchannel(ctx, ctx, communication.HandleAudio, rtspClient) +} + +func writeAudioToBackchannel(ctx context.Context, otelContext context.Context, audioChannel <-chan models.AudioDataPartial, rtspClient backchannelClient) { + log.Log.Info("Audio.WriteAudioToBackchannel(): writing to backchannel audio codec") + + if err := rtspClient.StartBackChannel(ctx, otelContext); err != nil { + log.Log.Error("Audio.WriteAudioToBackchannel(): error starting backchannel: " + err.Error()) + if !reconnectBackchannel(ctx, otelContext, rtspClient) { + log.Log.Info("Audio.WriteAudioToBackchannel(): stopped while reconnecting") + return + } + } + + packetizer := newBackchannelPacketizer() + for { + select { + case <-ctx.Done(): + log.Log.Info("Audio.WriteAudioToBackchannel(): stopped") + return + case audio, ok := <-audioChannel: + if !ok { + log.Log.Info("Audio.WriteAudioToBackchannel(): finished") + return + } + + audio = latestBackchannelAudio(audio, audioChannel) + if len(audio.Data) == 0 { + continue + } + + pkt := packetizer.packet(audio, time.Now()) + if err := rtspClient.WritePacket(pkt); err != nil { + log.Log.Error("Audio.WriteAudioToBackchannel(): error writing packet to backchannel: " + err.Error()) + if !reconnectBackchannel(ctx, otelContext, rtspClient) { + log.Log.Info("Audio.WriteAudioToBackchannel(): stopped while reconnecting") + return + } + packetizer = newBackchannelPacketizer() + continue + } + + if !waitForBackchannel(ctx, time.Duration(len(audio.Data))*time.Second/backchannelSampleRate) { + log.Log.Info("Audio.WriteAudioToBackchannel(): stopped") + return + } + } + } +} + +func latestBackchannelAudio(audio models.AudioDataPartial, audioChannel <-chan models.AudioDataPartial) models.AudioDataPartial { + for { + select { + case next, ok := <-audioChannel: + if !ok { + return audio + } + audio = next + default: + return audio + } + } +} + +func reconnectBackchannel(ctx context.Context, otelContext context.Context, rtspClient backchannelClient) bool { + backoff := backchannelReconnectInitial + for { + if err := rtspClient.Close(otelContext); err != nil { + log.Log.Error("Audio.WriteAudioToBackchannel(): error closing failed backchannel: " + err.Error()) + } + if ctx.Err() != nil { + return false + } + + err := rtspClient.ConnectBackChannel(ctx, otelContext) + if err == nil { + err = rtspClient.StartBackChannel(ctx, otelContext) + } + if err == nil { + log.Log.Info("Audio.WriteAudioToBackchannel(): reconnected backchannel") + return true + } + + log.Log.Error("Audio.WriteAudioToBackchannel(): error reconnecting backchannel: " + err.Error()) + if !waitForBackchannel(ctx, backoff) { + return false + } + backoff *= 2 + if backoff > backchannelReconnectMax { + backoff = backchannelReconnectMax + } + } +} + +func waitForBackchannel(ctx context.Context, duration time.Duration) bool { + timer := time.NewTimer(duration) + defer timer.Stop() + + select { + case <-ctx.Done(): + return false + case <-timer.C: + return true + } } func WriteFileToBackChannel(infile av.DemuxCloser) { diff --git a/machinery/src/components/backchannel_test.go b/machinery/src/components/backchannel_test.go new file mode 100644 index 0000000..9c5d58f --- /dev/null +++ b/machinery/src/components/backchannel_test.go @@ -0,0 +1,60 @@ +package components + +import ( + "testing" + "time" + + "github.com/kerberos-io/agent/machinery/src/models" +) + +func TestBackchannelPacketizerUsesFullRTPClock(t *testing.T) { + packetizer := backchannelPacketizer{ssrc: 1} + audio := models.AudioDataPartial{Data: make([]int16, 1024)} + startedAt := time.Unix(1, 0) + + var timestamp uint32 + for index := 0; index <= 64; index++ { + pkt := packetizer.packet(audio, startedAt.Add(time.Duration(index)*128*time.Millisecond)) + timestamp = pkt.Packet.Timestamp + } + + if timestamp != 65536 { + t.Fatalf("timestamp after 64 frames = %d, want 65536", timestamp) + } +} + +func TestBackchannelPacketizerUsesNaturalSequenceRollover(t *testing.T) { + packetizer := backchannelPacketizer{sequenceNumber: ^uint16(0), ssrc: 1} + audio := models.AudioDataPartial{Data: []int16{0}} + startedAt := time.Unix(1, 0) + + last := packetizer.packet(audio, startedAt) + firstAfterRollover := packetizer.packet(audio, startedAt.Add(time.Millisecond)) + + if last.Packet.SequenceNumber != ^uint16(0) { + t.Fatalf("last sequence number = %d, want %d", last.Packet.SequenceNumber, ^uint16(0)) + } + if firstAfterRollover.Packet.SequenceNumber != 0 { + t.Fatalf("first sequence number after rollover = %d, want 0", firstAfterRollover.Packet.SequenceNumber) + } +} + +func TestBackchannelPacketizerMarksTalkspurtStart(t *testing.T) { + packetizer := backchannelPacketizer{ssrc: 1} + audio := models.AudioDataPartial{Data: []int16{0}} + startedAt := time.Unix(1, 0) + + first := packetizer.packet(audio, startedAt) + continuous := packetizer.packet(audio, startedAt.Add(128*time.Millisecond)) + afterGap := packetizer.packet(audio, startedAt.Add(backchannelTalkspurtGap+128*time.Millisecond)) + + if !first.Packet.Marker { + t.Fatal("first packet must mark the start of a talkspurt") + } + if continuous.Packet.Marker { + t.Fatal("continuous packet must not carry the marker bit") + } + if !afterGap.Packet.Marker { + t.Fatal("packet after an audio gap must mark a new talkspurt") + } +} \ No newline at end of file diff --git a/machinery/src/components/kerberos.go b/machinery/src/components/kerberos.go index 68544e2..a372772 100644 --- a/machinery/src/components/kerberos.go +++ b/machinery/src/components/kerberos.go @@ -74,6 +74,7 @@ func Bootstrap(ctx context.Context, configDirectory string, configuration *model communication.HandleLiveHDKeepalive = make(chan string, 1) communication.HandleLiveHDPeers = make(chan string, 1) communication.HandleLiveHLS = make(chan string, 1) + communication.HandleAudio = make(chan models.AudioDataPartial, 10) communication.IsConfiguring = abool.New() communication.IsRecordingManual = abool.New() communication.RecordingManualHeartbeat = &atomic.Int64{} @@ -269,11 +270,11 @@ func RunAgent(configDirectory string, configuration *models.Configuration, commu communication.MainStreamConnected = true // Try to create backchannel + communication.HasBackChannel = false rtspBackChannelClient := captureDevice.SetBackChannelClient(rtspUrl) err = rtspBackChannelClient.ConnectBackChannel(ctx, ctxRunAgent) if err == nil { log.Log.Info("components.Kerberos.RunAgent(): opened RTSP backchannel stream: " + rtspUrl) - go rtspBackChannelClient.StartBackChannel(ctx, ctxRunAgent) } rtspSubClient := captureDevice.RTSPSubClient @@ -350,7 +351,6 @@ func RunAgent(configDirectory string, configuration *models.Configuration, commu // is a no-op if ONVIFMotion is not enabled. go onvif.HandleONVIFEventStream(*communication.Context, configuration, communication) - communication.HandleAudio = make(chan models.AudioDataPartial, 10) if rtspBackChannelClient.HasBackChannel { communication.HasBackChannel = true go WriteAudioToBackchannel(communication, rtspBackChannelClient) @@ -441,9 +441,6 @@ func RunAgent(configDirectory string, configuration *models.Configuration, commu close(communication.HandleMotion) communication.HandleMotion = nil - close(communication.HandleAudio) - communication.HandleAudio = nil - close(communication.HandleONVIF) communication.HandleONVIF = nil