Utilize audio property latency in microphone AudioRecord (#439)

Also add test to other codecs

pkg/codec/internal/codectest/codectest.go

@@ -119,3 +119,41 @@ func VideoEncoderCloseTwiceTest(t *testing.T, c codec.VideoEncoderBuilder, p pro
 		t.Fatal(err)
 	}
 }
+
+func AudioEncoderReadAfterCloseTest(t *testing.T, c codec.AudioEncoderBuilder, p prop.Media, w wave.Audio) {
+	enc, err := c.BuildAudioEncoder(audio.ReaderFunc(func() (wave.Audio, func(), error) {
+		return w, nil, nil
+	}), p)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if err := assertNoPanic(t, enc.Close, "on Close()"); err != nil {
+		t.Fatal(err)
+	}
+	if err := assertNoPanic(t, func() error {
+		_, _, err := enc.Read()
+		return err
+	}, "on Read()"); err != io.EOF {
+		t.Fatalf("Expected: %v, got: %v", io.EOF, err)
+	}
+}
+
+func VideoEncoderReadAfterCloseTest(t *testing.T, c codec.VideoEncoderBuilder, p prop.Media, img image.Image) {
+	enc, err := c.BuildVideoEncoder(video.ReaderFunc(func() (image.Image, func(), error) {
+		return img, nil, nil
+	}), p)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if err := assertNoPanic(t, enc.Close, "on Close()"); err != nil {
+		t.Fatal(err)
+	}
+	if err := assertNoPanic(t, func() error {
+		_, _, err := enc.Read()
+		return err
+	}, "on Read()"); err != io.EOF {
+		t.Fatalf("Expected: %v, got: %v", io.EOF, err)
+	}
+}

pkg/codec/mmal/mmal_test.go

@@ -44,6 +44,25 @@ func TestEncoder(t *testing.T) {
 			},
 		})
 	})
+	t.Run("ReadAfterClose", func(t *testing.T) {
+		p, err := NewParams()
+		if err != nil {
+			t.Fatal(err)
+		}
+		codectest.VideoEncoderReadAfterCloseTest(t, &p,
+			prop.Media{
+				Video: prop.Video{
+					Width:       256,
+					Height:      144,
+					FrameFormat: frame.FormatI420,
+				},
+			},
+			image.NewYCbCr(
+				image.Rect(0, 0, 256, 144),
+				image.YCbCrSubsampleRatio420,
+			),
+		)
+	})
 }
 
 func TestShouldImplementBitRateControl(t *testing.T) {

pkg/codec/openh264/openh264_test.go

@@ -60,4 +60,23 @@ func TestEncoder(t *testing.T) {
 			},
 		})
 	})
+	t.Run("ReadAfterClose", func(t *testing.T) {
+		p, err := NewParams()
+		if err != nil {
+			t.Fatal(err)
+		}
+		codectest.VideoEncoderReadAfterCloseTest(t, &p,
+			prop.Media{
+				Video: prop.Video{
+					Width:       256,
+					Height:      144,
+					FrameFormat: frame.FormatI420,
+				},
+			},
+			image.NewYCbCr(
+				image.Rect(0, 0, 256, 144),
+				image.YCbCrSubsampleRatio420,
+			),
+		)
+	})
 }

pkg/codec/opus/opus.go

@@ -3,6 +3,8 @@ package opus
 import (
 	"errors"
 	"fmt"
+	"io"
+	"sync"
 
 	"github.com/pion/mediadevices/pkg/codec"
 	"github.com/pion/mediadevices/pkg/io/audio"
@@ -25,6 +27,8 @@ type encoder struct {
 	inBuff wave.Audio
 	reader audio.Reader
 	engine *C.OpusEncoder
+
+	mu sync.Mutex
 }
 
 func newEncoder(r audio.Reader, p prop.Media, params Params) (codec.ReadCloser, error) {
@@ -79,6 +83,12 @@ func (e *encoder) Read() ([]byte, func(), error) {
 		return nil, func() {}, err
 	}
 
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	if e.engine == nil {
+		return nil, nil, io.EOF
+	}
+
 	encoded := make([]byte, 1024)
 	var n C.opus_int32
 	switch b := buff.(type) {
@@ -126,6 +136,8 @@ func (e *encoder) Controller() codec.EncoderController {
 }
 
 func (e *encoder) Close() error {
+	e.mu.Lock()
+	defer e.mu.Unlock()
 	if e.engine == nil {
 		return nil
 	}

pkg/codec/opus/opus_test.go

@@ -57,4 +57,23 @@ func TestEncoder(t *testing.T) {
 			},
 		})
 	})
+	t.Run("ReadAfterClose", func(t *testing.T) {
+		p, err := NewParams()
+		if err != nil {
+			t.Fatal(err)
+		}
+		codectest.AudioEncoderReadAfterCloseTest(t, &p,
+			prop.Media{
+				Audio: prop.Audio{
+					SampleRate:   48000,
+					ChannelCount: 2,
+				},
+			},
+			wave.NewInt16Interleaved(wave.ChunkInfo{
+				Len:          960,
+				SamplingRate: 48000,
+				Channels:     2,
+			}),
+		)
+	})
 }

pkg/codec/vaapi/vaapi_test.go

@@ -65,6 +65,25 @@ func TestEncoder(t *testing.T) {
 					},
 				})
 			})
+			t.Run("ReadAfterClose", func(t *testing.T) {
+				p, err := factory()
+				if err != nil {
+					t.Fatal(err)
+				}
+				codectest.VideoEncoderReadAfterCloseTest(t, p,
+					prop.Media{
+						Video: prop.Video{
+							Width:       256,
+							Height:      144,
+							FrameFormat: frame.FormatI420,
+						},
+					},
+					image.NewYCbCr(
+						image.Rect(0, 0, 256, 144),
+						image.YCbCrSubsampleRatio420,
+					),
+				)
+			})
 		})
 	}
 }

pkg/codec/vpx/vpx_test.go

@@ -60,6 +60,25 @@ func TestEncoder(t *testing.T) {
 					},
 				})
 			})
+			t.Run("ReadAfterClose", func(t *testing.T) {
+				p, err := factory()
+				if err != nil {
+					t.Fatal(err)
+				}
+				codectest.VideoEncoderReadAfterCloseTest(t, p,
+					prop.Media{
+						Video: prop.Video{
+							Width:       256,
+							Height:      144,
+							FrameFormat: frame.FormatI420,
+						},
+					},
+					image.NewYCbCr(
+						image.Rect(0, 0, 256, 144),
+						image.YCbCrSubsampleRatio420,
+					),
+				)
+			})
 		})
 	}
 }

pkg/codec/x264/x264_test.go

@@ -46,6 +46,26 @@ func TestEncoder(t *testing.T) {
 			},
 		})
 	})
+	t.Run("ReadAfterClose", func(t *testing.T) {
+		p, err := NewParams()
+		if err != nil {
+			t.Fatal(err)
+		}
+		p.BitRate = 200000
+		codectest.VideoEncoderReadAfterCloseTest(t, &p,
+			prop.Media{
+				Video: prop.Video{
+					Width:       256,
+					Height:      144,
+					FrameFormat: frame.FormatI420,
+				},
+			},
+			image.NewYCbCr(
+				image.Rect(0, 0, 256, 144),
+				image.YCbCrSubsampleRatio420,
+			),
+		)
+	})
 }
 
 func TestShouldImplementKeyFrameControl(t *testing.T) {

pkg/driver/microphone/microphone.go

@@ -113,6 +113,7 @@ func (m *microphone) AudioRecord(inputProp prop.Media) (audio.Reader, error) {
 	config.PerformanceProfile = malgo.LowLatency
 	config.Capture.Channels = uint32(inputProp.ChannelCount)
 	config.SampleRate = uint32(inputProp.SampleRate)
+	config.PeriodSizeInMilliseconds = uint32(inputProp.Latency.Milliseconds())
 	//FIX: Turn on the microphone with the current device id
 	config.Capture.DeviceID = m.ID.Pointer()
 	if inputProp.SampleSize == 4 && inputProp.IsFloat {