Add the source role and task with a PCM streaming pipeline - #114
Add the source role and task with a PCM streaming pipeline#114chrisuthe wants to merge 7 commits into
Conversation
68cfcc4 to
34c194a
Compare
Names the source config's default sample rate and makes the pass-through SourceRole::Impl::write_audio const, matching the other Impl methods that only write through pointers.
There was a problem hiding this comment.
🟡 Changes recommended
Trust enforcement, command concurrency, send ordering, lifecycle accuracy, documentation, and active-pipeline test coverage remain unresolved.
Once you've addressed the issues Copilot identified, you can request another Copilot review.
Pull request overview
Adds the source@v1 PCM capture pipeline, including public APIs, background streaming, protocol integration, build configuration, and tests.
Changes:
- Adds configurable PCM capture and timestamped source streaming.
- Integrates source lifecycle, commands, state, and Inbox events.
- Adds conditional builds and source-role tests.
File summaries
| File | Description |
|---|---|
CMakeLists.txt |
Adds source build option and sources. |
Kconfig |
Adds ESP source-role option. |
cmake/sources.cmake |
Defines source and shared audio sources. |
include/sendspin/client.h |
Exposes source-role registration and access. |
include/sendspin/config.h |
Adds source capture configuration. |
include/sendspin/source_role.h |
Defines the public source API. |
src/audio_ring_buffer.cpp |
Supports configurable memory placement. |
src/audio_ring_buffer.h |
Generalizes ring-buffer documentation and creation. |
src/client.cpp |
Integrates source lifecycle and message routing. |
src/constants.h |
Shares the time-sync wait interval. |
src/inbox.h |
Adds source command and lifecycle events. |
src/protocol_messages.h |
Uses the public source signal type. |
src/source_encoder.h |
Adds the PCM encoder abstraction. |
src/source_role.cpp |
Implements source configuration and lifecycle. |
src/source_role_impl.h |
Defines private source-role state. |
src/source_task.cpp |
Implements the capture-to-server pipeline. |
src/source_task.h |
Declares source task state and helpers. |
src/sync_task.cpp |
Removes the duplicated sync wait constant. |
tests/CMakeLists.txt |
Conditionally registers source tests. |
tests/test_source_role.cpp |
Tests source configuration and bookkeeping. |
Review details
- Files reviewed: 20/20 changed files
- Comments generated: 7
- Review effort level: Balanced
💡 Configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| if (process_server_command_source(root, &source_cmd)) { | ||
| // The instance id lets the main-loop latch discard stale-connection | ||
| // commands (per-connection permission; Sendspin spec, Source messages) | ||
| this->source_->impl_->handle_server_command(source_cmd, | ||
| conn->get_instance_id()); |
There was a problem hiding this comment.
The trust/activation gate cannot exist on this protocol revision: main predates the pairing/encryption transport entirely (there is no trust state to check), active_roles is parsed-but-unconsumed for every role, and the spec places unpaired-activation approval on the server side. This is the stack's documented sequencing caveat (see the PR body and the cover): client-side pairing enforcement lands with the encryption transport work, at which point the source role needs no changes. Part 6 adds an explicit authorization note to the integration guide and a startup warning to the example (842d16a).
| void SourceRole::Impl::handle_server_command(SourceCommand command, | ||
| uint64_t connection_instance_id) const { | ||
| this->event_state->command_slot.write(SourceCommandEnvelope{connection_instance_id, command}); |
There was a problem hiding this comment.
Fixed in 0b51d56: the dispatch site now forwards a source command only when its connection is the current one, so a concurrently delivering displaced connection can no longer overwrite the current connection's command; the drain-side instance-id check remains as the TOCTOU backstop.
| // STREAMING_STOPPED is pushed unconditionally: a second teardown in the same tick wipes the | ||
| // ring before it drains (see cleanup_connection_state()); streaming_active keeps the | ||
| // listener callbacks paired, so an extra STOPPED is harmless | ||
| this->pending_events.clear(); | ||
| this->enqueue_stream_event(SourceStreamCallbackType::STREAMING_STOPPED); |
There was a problem hiding this comment.
Fixed in 0b51d56: cleanup closes the write_audio gate synchronously before publishing its synthetic STOPPED, so the stopped callback and the gate can never disagree.
| if (conn->send_text_message(format_client_stream_start_message(&start_msg), nullptr) != | ||
| SsErr::OK) { | ||
| SS_LOGW(TAG, "Failed to send client-stream/start; stream not opened"); | ||
| return; |
There was a problem hiding this comment.
Fixed in 0b51d56: the task now passes its completion callback on the client-stream/start send and opens the gate only after a confirmed completion (bounded wait; timeout or failure is a failed open and retried), so queued-but-failed sends can no longer admit audio.
| SourceRole(SourceRoleConfig config, SendspinClient* client); | ||
| ~SourceRole(); |
There was a problem hiding this comment.
Documentation for the whole stack lands in part 6 (#117) per the cover's stack map — integration guide, internals (thread/Inbox/pipeline sections), CLAUDE.md, and README all cover the source role there; conventions' same-PR rule is satisfied at stack level.
| /// @brief Same budget as the sync task: the deepest paths are the stream start/end JSON build | ||
| /// and the transport send | ||
| static constexpr size_t SOURCE_TASK_STACK_SIZE = 6192; |
There was a problem hiding this comment.
Fixed in 0b51d56: the constant's comment now states its measured basis (host -O2 -fstack-usage, deepest task chain ~0.6 KB) and that the remainder is headroom pending an on-target high-water measurement.
| // Defends the accepting gate in SourceTask::write_audio(): with no open stream every write is | ||
| // rejected, whole frames or not. | ||
| TEST(SourceWriteAudio, RejectedWhenNotStreaming) { |
There was a problem hiding this comment.
The active-pipeline coverage is part 4 (#115) by design: loopback wire tests drive a real start/write/stop cycle and assert the emitted text/binary sequence, framing, and timestamps. Part 5 extends them for Opus.
Gate server/command source writes to the current connection so a concurrently delivering displaced connection cannot overwrite the current one's command before the drain, close the write_audio gate before cleanup's synthetic STOPPED so the stopped callback and the gate can never disagree, and wait for the client-stream/start send confirmation before accepting audio (queued is not sent on the async ESP-server transport). The default chunk drops to 20 ms so a codec-only switch to Opus keeps a valid config, and the stack-size and write_audio docs now state their measured basis and per-platform locking.
Part 3/6 of the source@v1 stack (tracker: #95).
What it adds: the source role itself —
SendspinClient::add_source(),SourceRoleConfig, and a dedicated task that assembles captured audio into timestamped PCM chunks and streams them to the server, gated byserver/commandstart/stop with per-connection permission.How it's used: the consumer adds the role with its capture format and feeds
write_audio()from its capture thread; the server starts and stops the stream. After this part the library is a complete PCM source. End-to-end use against Music Assistant additionally needs the encryption/pairing work (servers only activate source@v1 on paired connections) and aiosendspin's pendingclient-stream/*rename.Validation: exercised end-to-end against Music Assistant's
sendspin_sourceprovider on a dev MA server (with the two server-side interop shims aiosendspin needs anyway: the pairing-gate lift for legacy transport and theclient-stream/*rename aliases): mDNS discovery, source-typed registration, server-gated start/stop lifecycle, and decoded PCM ingest through MA's clock bridge to a live player.