Skip to content

feat(capture): dynamic source-level mic capture #30

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
odsod merged 1 commit into from
Aug 6, 2026
Merged

feat(capture): dynamic source-level mic capture #30

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion internal/capture/capture.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,6 @@ type Source interface {
// Narrowed for testability; *parec.Client satisfies it structurally.
type sinkClient interface {
ListSinks(ctx context.Context, req parec.ListSinksRequest) (parec.ListSinksResponse, error)
GetDefaultSource(ctx context.Context, req parec.GetDefaultSourceRequest) (parec.GetDefaultSourceResponse, error)
ListSources(ctx context.Context, req parec.ListSourcesRequest) (parec.ListSourcesResponse, error)
StartCapture(ctx context.Context, req parec.StartCaptureRequest) (*parec.CaptureStream, error)
}
29 changes: 13 additions & 16 deletions internal/capture/parec.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,14 +15,13 @@ import (
var tickInterval = time.Second

// Parec implements Source by dynamically monitoring every PulseAudio sink
// plus the default microphone, mixing simultaneous sink audio together.
// plus all input sources (microphones), mixing simultaneous audio together.
type Parec struct {
client sinkClient

mu sync.Mutex
sinks map[string]*reader // keyed by sink name
mic *reader
micName string
mu sync.Mutex
sinks map[string]*reader // keyed by sink name
mics map[string]*reader // keyed by source name

sinkBackoff backoff
sinkLastTry time.Time
Expand All @@ -36,7 +35,7 @@ type Parec struct {

// NewParec creates a Parec source using the given parec protocol client.
func NewParec(client *parec.Client) *Parec {
return &Parec{client: client, sinks: make(map[string]*reader)}
return &Parec{client: client, sinks: make(map[string]*reader), mics: make(map[string]*reader)}
}

// Start begins dynamic sink and microphone capture, returning a channel of
Expand Down Expand Up @@ -83,17 +82,15 @@ func (c *Parec) mixTick() frame.Dual {
sinkFrames = append(sinkFrames, data)
}
}
micReader := c.mic
c.mu.Unlock()

mic := frame.Silent(pcm.FrameBytes)
if micReader != nil {
if data, ok := micReader.take(); ok {
mic = data
micFrames := make([][]byte, 0, len(c.mics))
for _, r := range c.mics {
if data, ok := r.take(); ok {
micFrames = append(micFrames, data)
}
}
c.mu.Unlock()

return frame.Dual{Sys: pcm.Mix(sinkFrames...), Mic: mic}
return frame.Dual{Sys: pcm.Mix(sinkFrames...), Mic: pcm.Mix(micFrames...)}
}

// Stop terminates all capture streams. Idempotent.
Expand All @@ -108,8 +105,8 @@ func (c *Parec) Stop() error {
for _, r := range c.sinks {
_ = r.stop()
}
if c.mic != nil {
_ = c.mic.stop()
for _, r := range c.mics {
_ = r.stop()
}
c.mu.Unlock()
})
Expand Down
95 changes: 47 additions & 48 deletions internal/capture/parec_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -96,8 +96,8 @@ type fakeSinkClient struct {
client *parec.Client
sinks []parec.Sink
sinksErr error
source string
sourceErr error
sources []parec.Source
sourcesErr error
startErrFn func(device string) error
}

Expand All @@ -115,16 +115,16 @@ func (f *fakeSinkClient) ListSinks(ctx context.Context, _ parec.ListSinksRequest
return parec.ListSinksResponse{Sinks: f.sinks}, nil
}

func (f *fakeSinkClient) GetDefaultSource(
func (f *fakeSinkClient) ListSources(
ctx context.Context,
_ parec.GetDefaultSourceRequest,
) (parec.GetDefaultSourceResponse, error) {
_ parec.ListSourcesRequest,
) (parec.ListSourcesResponse, error) {
f.mu.Lock()
defer f.mu.Unlock()
if f.sourceErr != nil {
return parec.GetDefaultSourceResponse{}, f.sourceErr
if f.sourcesErr != nil {
return parec.ListSourcesResponse{}, f.sourcesErr
}
return parec.GetDefaultSourceResponse{Source: f.source}, nil
return parec.ListSourcesResponse{Sources: f.sources}, nil
}

func (f *fakeSinkClient) StartCapture(
Expand Down Expand Up @@ -154,16 +154,16 @@ func (f *fakeSinkClient) setSinksErr(err error) {
f.sinksErr = err
}

func (f *fakeSinkClient) setSource(name string) {
func (f *fakeSinkClient) setSources(sources ...parec.Source) {
f.mu.Lock()
defer f.mu.Unlock()
f.source = name
f.sources = sources
}

func (f *fakeSinkClient) setSourceErr(err error) {
func (f *fakeSinkClient) setSourcesErr(err error) {
f.mu.Lock()
defer f.mu.Unlock()
f.sourceErr = err
f.sourcesErr = err
}

// waitFor polls cond until it returns true or the timeout elapses.
Expand Down Expand Up @@ -206,7 +206,7 @@ func TestParec_MultiSinkMixing(t *testing.T) {
parec.Sink{Name: "b", MonitorSource: "b.monitor"},
)

src := &Parec{client: client, sinks: make(map[string]*reader)}
src := &Parec{client: client, sinks: make(map[string]*reader), mics: make(map[string]*reader)}
frames, err := src.Start(context.Background())
if err != nil {
t.Fatal(err)
Expand All @@ -227,7 +227,7 @@ func TestParec_SinkAppears(t *testing.T) {
client := newFakeSinkClient()
client.runner.setStream("a.monitor", bytes.Repeat([]byte{0x05, 0x00}, pcm.FrameBytes/2))

src := &Parec{client: client, sinks: make(map[string]*reader)}
src := &Parec{client: client, sinks: make(map[string]*reader), mics: make(map[string]*reader)}
frames, err := src.Start(context.Background())
if err != nil {
t.Fatal(err)
Expand Down Expand Up @@ -259,7 +259,7 @@ func TestParec_SinkDisappears(t *testing.T) {
parec.Sink{Name: "b", MonitorSource: "b.monitor"},
)

src := &Parec{client: client, sinks: make(map[string]*reader)}
src := &Parec{client: client, sinks: make(map[string]*reader), mics: make(map[string]*reader)}
if _, err := src.Start(context.Background()); err != nil {
t.Fatal(err)
}
Expand Down Expand Up @@ -288,7 +288,7 @@ func TestParec_SinkReappearsNewIndex(t *testing.T) {
client.runner.setStream("a.monitor", bytes.Repeat([]byte{0x05, 0x00}, pcm.FrameBytes/2))
client.setSinks(parec.Sink{Name: "a", MonitorSource: "a.monitor"})

src := &Parec{client: client, sinks: make(map[string]*reader)}
src := &Parec{client: client, sinks: make(map[string]*reader), mics: make(map[string]*reader)}
_, err := src.Start(context.Background())
if err != nil {
t.Fatal(err)
Expand Down Expand Up @@ -319,14 +319,14 @@ func TestParec_SinkReappearsNewIndex(t *testing.T) {
})
}

func TestParec_MicHotSwapSuccess(t *testing.T) {
func TestParec_MicSourceAppears(t *testing.T) {
withFastPolling(t)
client := newFakeSinkClient()
client.runner.setStream("mic1", bytes.Repeat([]byte{0x07, 0x00}, pcm.FrameBytes/2))
client.runner.setStream("mic2", bytes.Repeat([]byte{0x08, 0x00}, pcm.FrameBytes/2))
client.setSource("mic1")
client.setSources(parec.Source{Name: "mic1"})

src := &Parec{client: client, sinks: make(map[string]*reader)}
src := &Parec{client: client, sinks: make(map[string]*reader), mics: make(map[string]*reader)}
frames, err := src.Start(context.Background())
if err != nil {
t.Fatal(err)
Expand All @@ -336,36 +336,38 @@ func TestParec_MicHotSwapSuccess(t *testing.T) {
waitFor(t, time.Second, func() bool {
src.mu.Lock()
defer src.mu.Unlock()
return src.micName == "mic1"
_, ok := src.mics["mic1"]
return ok
})

client.setSource("mic2")
client.setSources(parec.Source{Name: "mic1"}, parec.Source{Name: "mic2"})
waitFor(t, time.Second, func() bool {
src.mu.Lock()
defer src.mu.Unlock()
return src.micName == "mic2"
_, ok := src.mics["mic2"]
return ok
})

found := false
for range 200 {
f := <-frames
if f.Mic[0] == 0x08 {
if f.Mic[0] == 0x08 || f.Mic[0] == 0x0f {
found = true
break
}
}
if !found {
t.Error("expected mic audio from new source after hot-swap")
t.Error("expected mic audio from new source after discovery")
}
}

func TestParec_MicHotSwapFailureRetainsOld(t *testing.T) {
func TestParec_MicCaptureFailureDoesNotAffectOthers(t *testing.T) {
withFastPolling(t)
client := newFakeSinkClient()
client.runner.setStream("mic1", bytes.Repeat([]byte{0x07, 0x00}, pcm.FrameBytes/2))
client.setSource("mic1")
client.setSources(parec.Source{Name: "mic1"})

src := &Parec{client: client, sinks: make(map[string]*reader)}
src := &Parec{client: client, sinks: make(map[string]*reader), mics: make(map[string]*reader)}
_, err := src.Start(context.Background())
if err != nil {
t.Fatal(err)
Expand All @@ -375,7 +377,8 @@ func TestParec_MicHotSwapFailureRetainsOld(t *testing.T) {
waitFor(t, time.Second, func() bool {
src.mu.Lock()
defer src.mu.Unlock()
return src.micName == "mic1"
_, ok := src.mics["mic1"]
return ok
})

client.mu.Lock()
Expand All @@ -386,24 +389,19 @@ func TestParec_MicHotSwapFailureRetainsOld(t *testing.T) {
return nil
}
client.mu.Unlock()
client.setSource("mic2")
client.setSources(parec.Source{Name: "mic1"}, parec.Source{Name: "mic2"})

time.Sleep(100 * time.Millisecond)

src.mu.Lock()
name := src.micName
mic := src.mic
_, hasMic1 := src.mics["mic1"]
_, hasMic2 := src.mics["mic2"]
src.mu.Unlock()
if name != "mic1" {
t.Errorf("expected mic to remain mic1 after failed swap, got %q", name)
}
if mic == nil {
t.Fatal("expected mic reader to still be present")
if !hasMic1 {
t.Error("expected mic1 to still be active")
}
select {
case <-mic.dead():
t.Error("old mic reader should still be alive after a failed swap")
default:
if hasMic2 {
t.Error("expected mic2 to not be active after failed start")
}
}

Expand All @@ -417,7 +415,7 @@ func TestParec_ReaderExitTriggersRestartWithoutKillingOthers(t *testing.T) {
parec.Sink{Name: "b", MonitorSource: "b.monitor"},
)

src := &Parec{client: client, sinks: make(map[string]*reader)}
src := &Parec{client: client, sinks: make(map[string]*reader), mics: make(map[string]*reader)}
_, err := src.Start(context.Background())
if err != nil {
t.Fatal(err)
Expand Down Expand Up @@ -465,9 +463,9 @@ func TestParec_PulseAudioUnreachableThenRecovers(t *testing.T) {
withFastPolling(t)
client := newFakeSinkClient()
client.setSinksErr(io.ErrClosedPipe)
client.setSourceErr(io.ErrClosedPipe)
client.setSourcesErr(io.ErrClosedPipe)

src := &Parec{client: client, sinks: make(map[string]*reader)}
src := &Parec{client: client, sinks: make(map[string]*reader), mics: make(map[string]*reader)}
frames, err := src.Start(context.Background())
if err != nil {
t.Fatal(err)
Expand All @@ -484,15 +482,16 @@ func TestParec_PulseAudioUnreachableThenRecovers(t *testing.T) {
client.runner.setStream("a.monitor", bytes.Repeat([]byte{0x09, 0x00}, pcm.FrameBytes/2))
client.setSinksErr(nil)
client.setSinks(parec.Sink{Name: "a", MonitorSource: "a.monitor"})
client.setSourceErr(nil)
client.setSource("mic1")
client.setSourcesErr(nil)
client.setSources(parec.Source{Name: "mic1"})
client.runner.setStream("mic1", bytes.Repeat([]byte{0x0a, 0x00}, pcm.FrameBytes/2))

waitFor(t, 2*time.Second, func() bool {
src.mu.Lock()
defer src.mu.Unlock()
_, hasSink := src.sinks["a"]
return hasSink && src.micName == "mic1"
_, hasMic := src.mics["mic1"]
return hasSink && hasMic
})
}

Expand All @@ -508,7 +507,7 @@ func isSilent(data []byte) bool {
func TestParec_Stop_Idempotent(t *testing.T) {
withFastPolling(t)
client := newFakeSinkClient()
src := &Parec{client: client, sinks: make(map[string]*reader)}
src := &Parec{client: client, sinks: make(map[string]*reader), mics: make(map[string]*reader)}
if _, err := src.Start(context.Background()); err != nil {
t.Fatal(err)
}
Expand All @@ -523,7 +522,7 @@ func TestParec_Stop_Idempotent(t *testing.T) {
func TestParec_Backpressure_SlowConsumerDoesNotDropFrames(t *testing.T) {
withFastPolling(t)
client := newFakeSinkClient()
src := &Parec{client: client, sinks: make(map[string]*reader)}
src := &Parec{client: client, sinks: make(map[string]*reader), mics: make(map[string]*reader)}
frames, err := src.Start(context.Background())
if err != nil {
t.Fatal(err)
Expand Down
Loading
Loading