feat: drain process for HLS, webm, mp4, WHEP

media/streamer/egress/hls/handler.go

@@ -100,7 +100,29 @@ func (h *HLS) Start(ctx context.Context, source hub.Source) error {
 				h.onVideo(ctx, data.H264Video)
 			}
 		}
+		if audioTranscodingProcess != nil {
+			log.Info(ctx, "draining audio transcoding process for HLS")
+			packets, err := audioTranscodingProcess.Drain()
+			if err != nil {
+				log.Error(ctx, err, "failed to drain audio transcoder for HLS")
+			}
+			for _, packet := range packets {
+				h.onAudio(ctx, source, &hub.AACAudio{
+					Data:                  packet.Data,
+					SequenceHeader:        false,
+					MPEG4AudioConfigBytes: h.mpeg4AudioConfigBytes,
+					MPEG4AudioConfig:      h.mpeg4AudioConfig,
+					PTS:                   packet.PTS,
+					DTS:                   packet.DTS,
+					AudioClockRate:        uint32(packet.SampleRate),
+				})
+			}
+			log.Info(ctx, "audio transcoding process for HLS drained")
+		}
 		log.Info(ctx, "[HLS] end of streamID: ", source.StreamID())
+		if h.muxer != nil {
+			h.muxer.Close()
+		}
 	}()
 	return nil
 }

media/streamer/egress/record/mp4/handler.go

@@ -118,6 +118,29 @@ func (m *MP4) Start(ctx context.Context, source hub.Source) error {
 				}
 			}
 		}
+
+		if audioTranscodingProcess != nil {
+			log.Info(ctx, "draining audio transcoding process")
+			// Drain the remaining frames from the transcoder
+			packets, err := audioTranscodingProcess.Drain()
+			if err != nil {
+				log.Error(ctx, err, "failed to drain audio transcoder")
+			}
+			for _, packet := range packets {
+				m.onAudio(ctx, &hub.AACAudio{
+					Data:                  packet.Data,
+					SequenceHeader:        false,
+					MPEG4AudioConfigBytes: m.mpeg4AudioConfigBytes,
+					MPEG4AudioConfig:      m.mpeg4AudioConfig,
+					PTS:                   packet.PTS,
+					DTS:                   packet.DTS,
+					AudioClockRate:        uint32(packet.SampleRate),
+				})
+			}
+			log.Info(ctx, "audio transcoding process drained")
+		} else {
+			log.Info(ctx, "no audio transcoding process to drain")
+		}
 	}()
 	return nil
 }

media/streamer/egress/record/webm/handler.go

@@ -99,6 +99,24 @@ func (w *WebM) Start(ctx context.Context, source hub.Source) error {
 				w.onAudio(ctx, data.OPUSAudio)
 			}
 		}
+
+		if w.audioTranscodingProcess != nil {
+			log.Info(ctx, "draining audio transcoding process for WebM")
+			packets, err := w.audioTranscodingProcess.Drain()
+			if err != nil {
+				log.Error(ctx, err, "failed to drain audio transcoder for WebM")
+			}
+			for _, packet := range packets {
+				w.onAudio(ctx, &hub.OPUSAudio{
+					Data:           packet.Data,
+					PTS:            packet.PTS,
+					DTS:            packet.DTS,
+					AudioClockRate: uint32(packet.SampleRate),
+				})
+			}
+			log.Info(ctx, "audio transcoding process for WebM drained")
+		}
+
 		// Ensure the muxer is finalized
 		w.closeMuxer(ctx)
 	}()

media/streamer/egress/whep/whep.go

@@ -105,7 +105,24 @@ func (w *WHEP) Start(ctx context.Context, source hub.Source) error {
 		}
 
 		if audioTranscodingProcess != nil {
+			log.Info(ctx, "draining audio transcoding process for WHEP")
+			packets, err := audioTranscodingProcess.Drain()
+			if err != nil {
+				log.Error(ctx, err, "failed to drain audio transcoder for WHEP")
+			}
+			for _, packet := range packets {
+				err := w.onAudio(source, &hub.OPUSAudio{
+					Data:           packet.Data,
+					PTS:            packet.PTS,
+					DTS:            packet.DTS,
+					AudioClockRate: uint32(packet.SampleRate),
+				})
+				if err != nil {
+					log.Error(ctx, err, "failed to process drained OPUS audio for WHEP")
+				}
+			}
 			audioTranscodingProcess.Close()
+			log.Info(ctx, "audio transcoding process for WHEP drained and closed")
 		}
 		log.Info(ctx, "end whep")
 		//C.__lsan_do_leak_check()

media/streamer/processes/transcoder.go

@@ -194,3 +194,63 @@ func (t *AudioTranscodingProcess) Process(data *MediaPacket) ([]*MediaPacket, er
 
 	return opusAudio, nil
 }
+
+func (t *AudioTranscodingProcess) Drain() ([]*MediaPacket, error) {
+	var packets []*MediaPacket
+	ctx := context.Background()
+
+	// 1. Flush Decoder
+	if err := t.decCodecContext.SendPacket(nil); err != nil {
+		log.Error(ctx, err, "failed to send nil packet to decoder for draining")
+		// Continue to encoder flushing
+	}
+
+	frame := astiav.AllocFrame()
+	defer frame.Free()
+
+	for {
+		err := t.decCodecContext.ReceiveFrame(frame)
+		if errors.Is(err, astiav.ErrEof) || errors.Is(err, astiav.ErrEagain) {
+			break
+		}
+		if err != nil {
+			log.Error(ctx, err, "failed to receive frame from decoder during draining")
+			continue
+		}
+
+		if err := t.encCodecContext.SendFrame(frame); err != nil {
+			log.Error(ctx, err, "failed to send flushed frame to encoder")
+		}
+		frame.Unref()
+	}
+
+	// 2. Flush Encoder
+	if err := t.encCodecContext.SendFrame(nil); err != nil {
+		log.Error(ctx, err, "failed to send nil frame to encoder for draining")
+		return packets, err
+	}
+
+	pkt := astiav.AllocPacket()
+	defer pkt.Free()
+
+	for {
+		err := t.encCodecContext.ReceivePacket(pkt)
+		if errors.Is(err, astiav.ErrEof) || errors.Is(err, astiav.ErrEagain) {
+			break
+		}
+		if err != nil {
+			log.Error(ctx, err, "failed to receive packet from encoder during draining")
+			continue
+		}
+
+		packets = append(packets, &MediaPacket{
+			Data:       append([]byte{}, pkt.Data()...),
+			PTS:        pkt.Pts(),
+			DTS:        pkt.Dts(),
+			SampleRate: t.encCodecContext.SampleRate(),
+		})
+		pkt.Unref()
+	}
+
+	return packets, nil
+}