feat: renamed the module to avmuxer

Author: itzmanish

g711_stream.go

@@ -1,4 +1,4 @@
-package multiplexer
+package avmuxer
 
 import (
 	"bytes"

go.mod

@@ -1,4 +1,4 @@
-module github.com/itzmanish/multiplexer
+module github.com/itzmanish/avmuxer
 
 go 1.22.1
 
@@ -32,6 +32,7 @@ require (
 	github.com/pion/transport/v3 v3.0.6 // indirect
 	github.com/pion/turn/v3 v3.0.3 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/stretchr/objx v0.5.2 // indirect
 	github.com/wlynxg/anet v0.0.3 // indirect
 	golang.org/x/crypto v0.25.0 // indirect
 	golang.org/x/exp v0.0.0-20190731235908-ec7cb31e5a56 // indirect

go.sum

@@ -55,6 +55,7 @@ github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZN
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
 github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
+github.com/stretchr/objx v0.5.2 h1:xuMeJ0Sdp5ZMRXx/aWO6RZxdr3beISkG5/G/aIRr3pY=
 github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA=
 github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=

mutliplexer.go

@@ -1,4 +1,4 @@
-package multiplexer
+package avmuxer
 
 import (
 	"errors"

mutliplexer_test.go

@@ -1,4 +1,4 @@
-package multiplexer
+package avmuxer
 
 import (
 	"encoding/binary"
@@ -223,3 +223,13 @@ func TestMultiplexer_ReadPCM16(t *testing.T) {
 		time.Sleep(20 * time.Millisecond)
 	}
 }
+
+func generateTestPCM(sampleCount, repeatCount int) []int16 {
+	pcm := make([]int16, sampleCount*repeatCount)
+	for i := 0; i < repeatCount; i++ {
+		for j := 0; j < sampleCount; j++ {
+			pcm[i*sampleCount+j] = int16((j % 100) * 100)
+		}
+	}
+	return pcm
+}

opus_stream.go

@@ -1,11 +1,12 @@
-package multiplexer
+package avmuxer
 
 import (
 	"errors"
 	"io"
 	"log"
 )
 
+// OpusStream interface defines the methods for both encoding and decoding Opus streams
 type OpusStream interface {
 	Stream
 
@@ -17,8 +18,10 @@ type OpusStream interface {
 	ChannelCount() int
 	SampleDurationMs() int
 	SampleCount() int
+	ID() string
 }
 
+// opusEncodingStream implements OpusStream for encoding
 type opusEncodingStream struct {
 	id               string
 	sampleRate       int
@@ -31,6 +34,7 @@ type opusEncodingStream struct {
 	encoder *OpusEncoder
 }
 
+// opusDecodingStream implements OpusStream for decoding
 type opusDecodingStream struct {
 	id               string
 	sampleRate       int
@@ -43,12 +47,18 @@ type opusDecodingStream struct {
 	decoder *OpusDecoder
 }
 
+// NewDecodingOpusStream creates a new OpusStream for decoding
 func NewDecodingOpusStream(id string, sampleRate, sampleDuration, channel int) (OpusStream, error) {
+	// Calculate sample size based on duration and rate
 	sampleSize := sampleDuration * sampleRate / 1000
+
+	// Create a new OpusDecoder
 	dec, err := NewOpusDecoder(sampleRate, channel, sampleSize)
 	if err != nil {
 		return nil, err
 	}
+
+	// Return a new opusDecodingStream
 	return &opusDecodingStream{
 		id:               id,
 		sampleRate:       sampleRate,
@@ -60,7 +70,10 @@ func NewDecodingOpusStream(id string, sampleRate, sampleDuration, channel int) (
 	}, err
 }
 
+// NewEncodingOpusStream creates a new OpusStream for encoding
 func NewEncodingOpusStream(id string, sampleRate, sampleDuration, channel int) (OpusStream, error) {
+	// Similar to NewDecodingOpusStream, but for encoding
+	// ... existing code ...
 	sampleSize := sampleDuration * sampleRate / 1000
 	enc, err := NewOpusEncoder(sampleRate, channel, sampleSize)
 	if err != nil {
@@ -89,38 +102,46 @@ func (ods *opusDecodingStream) SampleRate() int {
 func (ods *opusDecodingStream) SampleDurationMs() int {
 	return ods.sampleDurationMs
 }
+func (ods *opusDecodingStream) ID() string {
+	return ods.id
+}
 
 func (ods *opusDecodingStream) Decode(src []byte, dst []int16) (int, error) {
 	return ods.decoder.Decode(src, dst)
 }
 
-// It takes encoded opus data and decode then store in the buffer
+// Write decodes Opus data and writes PCM to the sink
 func (ods *opusDecodingStream) Write(data []byte) (int, error) {
+	// Decode Opus data to PCM
 	pcm := make([]int16, ods.size*ods.channel)
 	n, err := ods.Decode(data, pcm)
 	if err != nil {
 		return 0, err
 	}
+
 	log.Printf("samples decoded: %v, os.size: %v, data size: %v\n", n, ods.size, len(data))
+
+	// Write decoded PCM to sink if available
 	if ods.sink != nil {
 		_, err := ods.sink.Write(int16ToByteSlice(pcm[:n*ods.channel]))
 		if err != nil {
 			return 0, err
 		}
 	}
-	// FIXME(itzmanish): do we need to store the pcm data?
+
+	// Write PCM data to buffer
 	return ods.decoder.buffer.Write(pcm[:n*ods.channel])
 }
 
-// This reads the raw PCM data from Decoding OPUS stream
+// ReadPCM reads raw PCM data from the decoding buffer
 func (ods *opusDecodingStream) ReadPCM(dst []int16) (int, error) {
 	if ods.decoder == nil {
 		return 0, errors.New("stream is not decoding supported")
 	}
 	return ods.decoder.buffer.Read(dst)
 }
 
-// This reads the raw PCM data from Decoding OPUS stream and convert them to byte array
+// Read reads raw PCM data and converts it to a byte array
 func (ods *opusDecodingStream) Read(dst []byte) (int, error) {
 	if ods.decoder == nil {
 		return 0, errors.New("stream is not decoding supported")
@@ -167,6 +188,10 @@ func (oes *opusEncodingStream) SampleDurationMs() int {
 	return oes.sampleDurationMs
 }
 
+func (oes *opusEncodingStream) ID() string {
+	return oes.id
+}
+
 func (*opusEncodingStream) ReadPCM([]int16) (int, error) {
 	return 0, errors.New("encoding stream doesn't support reading pcm")
 }
@@ -191,25 +216,21 @@ func (oes *opusEncodingStream) WritePCM(data []int16) (int, error) {
 	return oes.encoder.buffer.Write(byteData[:n])
 }
 
-// This reads encoded data after pulling and encoding from the reader
+// Read reads encoded Opus data from the encoder's buffer
 func (oes *opusEncodingStream) Read(dst []byte) (int, error) {
-	// if oes.encoder == nil {
-	// 	return 0, errors.New("stream is not encoding supported")
-	// }
-	// if oes.reader == nil {
-	// 	return 0, errors.New("reader is not connected")
-	// }
-	// byteArr := make([]byte, oes.size*oes.channel*2)
-	// n, err := oes.reader.Read(byteArr)
-	// if err != nil {
-	// 	return n, err
-	// }
-	// pcm := byteSliceToInt16(byteArr[:n])
-	// _, err = oes.WritePCM(pcm)
-	// if err != nil {
-	// 	return 0, err
-	// }
-	return oes.encoder.buffer.Read(dst)
+	if oes.encoder == nil {
+		return 0, errors.New("encoder is not initialized")
+	}
+
+	n, err := oes.encoder.buffer.Read(dst)
+	if err != nil {
+		if err == ErrEmptyBuffer {
+			return 0, io.EOF
+		}
+		return n, err
+	}
+
+	return n, nil
 }
 
 func (oes *opusEncodingStream) Encode(src []int16, dst []byte) (int, error) {

opus_stream_test.go

@@ -0,0 +1,73 @@
+package avmuxer
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNewDecodingOpusStream(t *testing.T) {
+	stream, err := NewDecodingOpusStream("stream1", 48000, 20, 2)
+	assert.NoError(t, err)
+	assert.NotNil(t, stream)
+
+	assert.Equal(t, 48000, stream.SampleRate())
+	assert.Equal(t, 2, stream.ChannelCount())
+	assert.Equal(t, 20, stream.SampleDurationMs())
+	assert.Equal(t, 960, stream.SampleCount()) // sampleDuration * sampleRate / 1000
+}
+
+func TestNewEncodingOpusStream(t *testing.T) {
+	stream, err := NewEncodingOpusStream("stream1", 48000, 20, 2)
+	assert.NoError(t, err)
+	assert.NotNil(t, stream)
+
+	assert.Equal(t, 48000, stream.SampleRate())
+	assert.Equal(t, 2, stream.ChannelCount())
+	assert.Equal(t, 20, stream.SampleDurationMs())
+	assert.Equal(t, 960, stream.SampleCount()) // sampleDuration * sampleRate / 1000
+}
+
+func TestDecodingStream_Write(t *testing.T) {
+	stream, err := NewDecodingOpusStream("stream1", 48000, 20, 2)
+	assert.NoError(t, err)
+	assert.NotNil(t, stream)
+
+	encodedData := make([]byte, 960)
+	n, err := stream.Write(encodedData)
+	assert.NoError(t, err)
+	assert.Equal(t, len(encodedData), n)
+}
+
+func TestEncodingStream_WritePCM(t *testing.T) {
+	stream, err := NewEncodingOpusStream("stream1", 48000, 20, 2)
+	assert.NoError(t, err)
+	assert.NotNil(t, stream)
+
+	pcmData := make([]int16, 960) // Example PCM data
+	n, err := stream.WritePCM(pcmData)
+	assert.NoError(t, err)
+	assert.Equal(t, 960, n)
+}
+
+func TestDecodingStream_ReadPCM(t *testing.T) {
+	stream, err := NewDecodingOpusStream("stream1", 48000, 20, 2)
+	assert.NoError(t, err)
+	assert.NotNil(t, stream)
+
+	dst := make([]int16, 960)
+	n, err := stream.ReadPCM(dst)
+	assert.NoError(t, err)
+	assert.Equal(t, 960, n)
+}
+
+func TestEncodingStream_Read(t *testing.T) {
+	stream, err := NewEncodingOpusStream("stream1", 48000, 20, 2)
+	assert.NoError(t, err)
+	assert.NotNil(t, stream)
+
+	dst := make([]byte, 4096)
+	n, err := stream.Read(dst)
+	assert.NoError(t, err)
+	assert.True(t, n > 0)
+}

ringbuffer.go

@@ -1,4 +1,4 @@
-package multiplexer
+package avmuxer
 
 import (
 	"errors"

ringbuffer_test.go

@@ -1 +1 @@
-package multiplexer
+package avmuxer

testdata/muxed.pcm

@@ -1,4 +1,4 @@
-package multiplexer
+package avmuxer
 
 import (
 	"errors"
@@ -16,15 +16,15 @@ func NewTranscoder() *Transcoder {
 	return &Transcoder{}
 }
 
-func (tc *Transcoder) AddSource(id string, stream Stream) error {
+func (tc *Transcoder) AddSource(stream Stream) error {
 	if tc.input != nil {
 		return errors.New("source is already present")
 	}
 	tc.input = stream
 	return nil
 }
 
-func (tc *Transcoder) AddEncoder(id string, enc Encoder) error {
+func (tc *Transcoder) AddEncoder(enc Encoder) error {
 	if tc.encoder != nil {
 		return errors.New("encoder is already present")
 	}