#include #include #include #include "audio_receiver.h" #include "esp_heap_caps.h" #include "esp_log.h" #include "esp_timer.h" #include "audio_buffer.h" #include "audio_decoder.h" #include "audio_output.h" #include "audio_receiver_internal.h" #include "audio_stream.h" #include "audio_timing.h" #include "ptp_clock.h" #define DEFAULT_SAMPLE_RATE 44100 #define DEFAULT_CHANNELS 2 #define DEFAULT_BITS_PER_SAMPLE 16 #define DEFAULT_FRAME_SIZE 352 #define DECRYPT_BUFFER_SIZE 8192 static const char *TAG = "audio_recv"; static audio_receiver_state_t receiver = {0}; static void audio_receiver_reset_stats(void) { memset(&receiver.stats, 0, sizeof(receiver.stats)); } static void audio_receiver_reset_resend_state(void) { receiver.rtp_sequence_valid = false; receiver.resend_window_first = 0; receiver.resend_missing_mask = 0; receiver.resend_last_request_time_us = 0; receiver.last_resend_error_time_us = 0; } static void audio_receiver_reset_blocks(void) { receiver.blocks_read = 0; receiver.blocks_read_in_sequence = 0; } static void audio_receiver_copy_stream_state(audio_stream_t *dst, const audio_stream_t *src) { if (!dst || !src) { return; } dst->format = src->format; dst->encrypt = src->encrypt; } esp_err_t audio_receiver_init(void) { if (receiver.buffer.pool) { return ESP_OK; } receiver.realtime_stream = audio_stream_create_realtime(); if (receiver.realtime_stream) { receiver.realtime_stream->ctx = &receiver; } receiver.buffered_stream = audio_stream_create_buffered(); if (receiver.buffered_stream) { receiver.buffered_stream->ctx = &receiver; } if (!receiver.realtime_stream || !receiver.buffered_stream) { ESP_LOGE(TAG, "Failed to allocate audio streams"); audio_stream_destroy(receiver.realtime_stream); audio_stream_destroy(receiver.buffered_stream); receiver.realtime_stream = NULL; receiver.buffered_stream = NULL; return ESP_ERR_NO_MEM; } receiver.stream = receiver.realtime_stream; audio_format_t default_format = {0}; strcpy(default_format.codec, "AppleLossless"); default_format.sample_rate = DEFAULT_SAMPLE_RATE; default_format.channels = DEFAULT_CHANNELS; default_format.bits_per_sample = DEFAULT_BITS_PER_SAMPLE; default_format.frame_size = DEFAULT_FRAME_SIZE; receiver.realtime_stream->format = default_format; receiver.buffered_stream->format = default_format; esp_err_t err = audio_buffer_init(&receiver.buffer); if (err != ESP_OK) { audio_stream_destroy(receiver.realtime_stream); audio_stream_destroy(receiver.buffered_stream); receiver.realtime_stream = NULL; receiver.buffered_stream = NULL; return err; } receiver.decrypt_buffer_size = DECRYPT_BUFFER_SIZE; #ifdef CONFIG_SPIRAM receiver.decrypt_buffer = heap_caps_malloc( receiver.decrypt_buffer_size, MALLOC_CAP_SPIRAM | MALLOC_CAP_8BIT); #endif if (!receiver.decrypt_buffer) { receiver.decrypt_buffer = malloc(receiver.decrypt_buffer_size); } if (!receiver.decrypt_buffer) { ESP_LOGE(TAG, "Failed to allocate decrypt buffer"); audio_buffer_deinit(&receiver.buffer); audio_stream_destroy(receiver.realtime_stream); audio_stream_destroy(receiver.buffered_stream); receiver.realtime_stream = NULL; receiver.buffered_stream = NULL; return ESP_ERR_NO_MEM; } size_t pending_capacity = sizeof(audio_frame_header_t) + ((size_t)MAX_SAMPLES_PER_FRAME * AUDIO_MAX_CHANNELS * sizeof(int16_t)); audio_timing_init(&receiver.timing, pending_capacity); audio_timing_set_format(&receiver.timing, &receiver.stream->format); receiver.buffered_listen_socket = -1; receiver.buffered_client_socket = -1; audio_receiver_reset_blocks(); return ESP_OK; } void audio_receiver_set_format(const audio_format_t *format) { if (!format) { return; } if (!receiver.realtime_stream || !receiver.buffered_stream) { return; } receiver.realtime_stream->format = *format; receiver.buffered_stream->format = *format; audio_decoder_destroy(receiver.decoder); receiver.decoder = NULL; audio_decoder_config_t cfg = {.format = *format}; receiver.decoder = audio_decoder_create(&cfg); if (!receiver.decoder) { ESP_LOGW(TAG, "Decoder not initialized for codec: %s", format->codec); } audio_timing_set_format(&receiver.timing, format); audio_output_set_source_rate(format->sample_rate); } void audio_receiver_set_encryption(const audio_encrypt_t *encrypt) { if (!receiver.realtime_stream || !receiver.buffered_stream) { return; } if (encrypt) { receiver.realtime_stream->encrypt = *encrypt; receiver.buffered_stream->encrypt = *encrypt; } else { memset(&receiver.realtime_stream->encrypt, 0, sizeof(receiver.realtime_stream->encrypt)); memset(&receiver.buffered_stream->encrypt, 0, sizeof(receiver.buffered_stream->encrypt)); } } void audio_receiver_set_output_latency_us(uint32_t latency_us) { if (!receiver.stream) { return; } audio_timing_set_output_latency(&receiver.timing, &receiver.stream->format, latency_us); } uint32_t audio_receiver_get_output_latency_us(void) { return audio_timing_get_output_latency(&receiver.timing); } uint32_t audio_receiver_get_hardware_latency_us(void) { return audio_timing_get_hardware_latency(); } uint32_t audio_receiver_get_advertised_latency_us(void) { return audio_timing_get_advertised_latency(&receiver.timing); } void audio_receiver_set_anchor_time(uint64_t clock_id, uint64_t network_time_ns, uint32_t rtp_time) { if (!receiver.stream) { return; } int sample_rate = receiver.stream->format.sample_rate; if (sample_rate <= 0) { sample_rate = 44100; } // Window size for the upper RTP gate: 10 s of samples. Large enough that // a normal 2-4 s pre-buffer passes, but small enough to reject stale frames // left in the TCP socket buffer after a backward seek. const uint32_t gate_window = (uint32_t)(10 * sample_rate); const int32_t seek_threshold = 5 * sample_rate; // --- Phase 1: Arm RTP gates BEFORE opening the blanket gate ----------- // // The blanket gate (discard_all_until_anchor) blocks ALL frames from the // TCP task. The per-RTP gates filter by timestamp range. On single-core // ESP32-S2, ESP_LOGI can yield to the scheduler, so any gap between // clearing the blanket and arming the per-RTP gates lets the TCP task // queue stale frames. Arm first, then open. bool gates_armed = false; // Path A: seek_flush set arm_gate_on_next_anchor because the buffer was // already empty when the flush happened (forward-seek). if (receiver.arm_gate_on_next_anchor) { receiver.arm_gate_on_next_anchor = false; receiver.discard_before_rtp = rtp_time; receiver.discard_before_rtp_valid = true; receiver.discard_above_rtp = rtp_time + gate_window; receiver.discard_above_rtp_valid = true; gates_armed = true; ESP_LOGI(TAG, "RTP gates armed on anchor: discard_before=%lu discard_above=%lu", (unsigned long)rtp_time, (unsigned long)(rtp_time + gate_window)); } // Path B: Anchor-change detection — the phone changed track with a // PAUSE → RESUME cycle but no FLUSHBUFFERED. The buffer may already be // empty (consumed during playback), so the seek-detection heuristic // below (which needs oldest_rtp from the buffer) would miss it. // // Compare the new anchor against the EXPECTED current playback position // (old_anchor_rtp + elapsed_time × sample_rate), NOT against the raw old // anchor. The raw old anchor was set at the start of the previous play // segment; comparing against it gives a delta equal to (elapsed_play_time + // new_anchor_lead_time), which easily exceeds the 5-second threshold on a // normal pause/resume within the same track — causing a false flush that // empties valid pre-buffered frames and produces 6+ seconds of silence. // Using the expected position instead, normal resume gives a delta of only // the anchor's lead-time offset (< 2 s), while a real track-change or seek // gives a huge delta (many minutes). if (!gates_armed && receiver.timing.anchor_valid) { // Choose reference point for the seek-detection comparison: // - If we have a pause snapshot, use it. The snapshot was taken at the // exact moment the sender said PAUSE, so it reflects the true pause // position rather than a wall-clock estimate that keeps running during // the pause and overshoots by (pause_duration x sample_rate). // - Otherwise fall back to the elapsed-time estimate (covers the edge // case where a track changes without a prior PAUSE signal). uint32_t reference_rtp; if (receiver.paused_rtp_valid) { reference_rtp = receiver.paused_rtp; receiver.paused_rtp_valid = false; // one-shot: consume after use // Compute the pause duration from the RTP snapshot so we can notify // the PTP clock without tracking a separate wall-clock timestamp. // anchor_local_time_ns/1000 is the µs when the anchor was set; // adding the played-sample offset gives the µs when play paused. int32_t played = (int32_t)(reference_rtp - receiver.timing.anchor_rtp_time); int64_t pause_time_us = receiver.timing.anchor_local_time_ns / 1000LL + (int64_t)played * 1000000LL / sample_rate; int64_t pause_us = esp_timer_get_time() - pause_time_us; ptp_clock_notify_resume((pause_us > 0) ? (uint32_t)(pause_us / 1000LL) : 0); ESP_LOGD(TAG, "Path B: pause snapshot rtp=%lu pause=%.1f s", (unsigned long)reference_rtp, (float)pause_us / 1e6f); } else { int64_t elapsed_us = esp_timer_get_time() - (receiver.timing.anchor_local_time_ns / 1000LL); if (elapsed_us < 0) { elapsed_us = 0; } // Cap elapsed to prevent int64 overflow on very long pauses. if (elapsed_us > 600000000LL) { elapsed_us = 600000000LL; } int32_t elapsed_samples = (int32_t)((elapsed_us * (int64_t)sample_rate) / 1000000LL); reference_rtp = receiver.timing.anchor_rtp_time + (uint32_t)elapsed_samples; } int32_t delta = (int32_t)(rtp_time - reference_rtp); int32_t abs_delta = delta < 0 ? -delta : delta; if (abs_delta > seek_threshold) { ESP_LOGI(TAG, "Anchor change detected: ref_rtp=%lu new_rtp=%lu " "delta=%ld samples (%.1f s) - flushing & arming gates", (unsigned long)reference_rtp, (unsigned long)rtp_time, (long)delta, (float)delta / sample_rate); audio_buffer_flush(&receiver.buffer); receiver.timing.playout_started = false; receiver.timing.pending_valid = false; receiver.timing.pending_frame_len = 0; receiver.timing.ready_time_us = 0; receiver.timing.deferred_flush_pending = false; receiver.blocks_read_in_sequence = 0; receiver.discard_before_rtp = rtp_time; receiver.discard_before_rtp_valid = true; receiver.discard_above_rtp = rtp_time + gate_window; receiver.discard_above_rtp_valid = true; receiver.timing.quick_start = true; gates_armed = true; } else { ESP_LOGD(TAG, "Anchor resume OK: ref_rtp=%lu new_rtp=%lu " "delta=%ld samples (%.2f s) - same track, no flush", (unsigned long)reference_rtp, (unsigned long)rtp_time, (long)delta, (float)delta / (float)sample_rate); } } // NOW safe to clear the blanket gate — per-RTP gates are active. receiver.discard_all_until_anchor = false; // --- Phase 2: Seek detection from buffer content ---------------------- // // If stale data managed to enter the buffer (e.g. queued before // seek_flush was called), detect it by comparing the oldest buffered // RTP timestamp against the new anchor. uint32_t oldest_rtp = 0; if (audio_buffer_oldest_timestamp(&receiver.buffer, &oldest_rtp)) { int32_t rtp_ahead = (int32_t)(oldest_rtp - rtp_time); int32_t abs_ahead = rtp_ahead < 0 ? -rtp_ahead : rtp_ahead; if (abs_ahead > seek_threshold) { ESP_LOGI(TAG, "Seek detected: oldest_rtp=%lu, new anchor rtp=%lu, " "delta=%ld samples (%.1f s) — flushing stale buffer", (unsigned long)oldest_rtp, (unsigned long)rtp_time, (long)rtp_ahead, (float)rtp_ahead / sample_rate); audio_buffer_flush(&receiver.buffer); receiver.timing.playout_started = false; receiver.timing.pending_valid = false; receiver.timing.pending_frame_len = 0; receiver.timing.ready_time_us = 0; receiver.timing.deferred_flush_pending = false; receiver.blocks_read_in_sequence = 0; receiver.timing.quick_start = true; if (!gates_armed) { receiver.discard_before_rtp = rtp_time; receiver.discard_before_rtp_valid = true; receiver.discard_above_rtp = rtp_time + gate_window; receiver.discard_above_rtp_valid = true; } } } // Pin the PTP clock to the master announced by the anchor packet's // clock_id field. Without this, ptp_clock can lock to any PTP master // on the LAN (HomePods, AppleTVs, NTP-PTP gateways) and produce offsets // that have nothing to do with the AirPlay sender's clock domain. // clock_id == 0 happens on the AirPlay 1 NTP path; in that case leave // the filter as-is (set or cleared by a prior 0xD7 anchor / TEARDOWN). if (clock_id != 0) { ptp_clock_set_master_clock_id(clock_id); } audio_timing_set_anchor(&receiver.timing, &receiver.stream->format, clock_id, network_time_ns, rtp_time); } void audio_receiver_set_playing(bool playing) { audio_timing_set_playing(&receiver.timing, playing); if (!playing) { receiver.blocks_read_in_sequence = 0; // Snapshot the expected RTP position at the moment of pause so that // Path B in audio_receiver_set_anchor_time() can compare the next // resume anchor against the actual pause position. // // Without this, Path B uses (anchor_rtp + wall_clock_elapsed), which // overshoots by the pause duration and fires a false seek flush on any // pause >= seek_threshold (5 s) — causing up to 7+ s of silence when // pre-buffered frames end up far ahead of the unwanted new anchor. if (receiver.timing.anchor_valid && receiver.stream) { int sample_rate = receiver.stream->format.sample_rate; if (sample_rate <= 0) { sample_rate = 44100; } int64_t elapsed_us = esp_timer_get_time() - (receiver.timing.anchor_local_time_ns / 1000LL); if (elapsed_us < 0) { elapsed_us = 0; } if (elapsed_us > 600000000LL) { elapsed_us = 600000000LL; } int32_t elapsed_samples = (int32_t)((elapsed_us * (int64_t)sample_rate) / 1000000LL); receiver.paused_rtp = receiver.timing.anchor_rtp_time + (uint32_t)elapsed_samples; receiver.paused_rtp_valid = true; ESP_LOGD(TAG, "Pause: RTP snapshot=%lu (elapsed=%.2f s)", (unsigned long)receiver.paused_rtp, (float)elapsed_us / 1e6f); } } } void audio_receiver_reset_timing(void) { audio_timing_reset(&receiver.timing); } bool audio_receiver_is_playing(void) { return receiver.timing.playing; } void audio_receiver_set_stream_type(audio_stream_type_t type) { if (!receiver.realtime_stream || !receiver.buffered_stream) { return; } audio_stream_t *target = receiver.realtime_stream; if (type == AUDIO_STREAM_BUFFERED) { target = receiver.buffered_stream; } if (!target) { return; } if (receiver.stream != target) { if (receiver.stream) { audio_receiver_copy_stream_state(target, receiver.stream); if (receiver.stream->running && receiver.stream->ops && receiver.stream->ops->stop) { receiver.stream->ops->stop(receiver.stream); } } receiver.stream = target; } receiver.stream->type = type; } esp_err_t audio_receiver_start(uint16_t data_port, uint16_t control_port) { audio_receiver_set_stream_type(AUDIO_STREAM_REALTIME); if (!receiver.stream || !receiver.stream->ops || !receiver.stream->ops->start) { return ESP_FAIL; } // Always stop and restart fresh if (receiver.stream->running) { receiver.stream->ops->stop(receiver.stream); } receiver.data_port = data_port; receiver.control_port = control_port; // Starting a stream resets all timing state (including pause tracking) audio_receiver_reset_stats(); audio_buffer_flush(&receiver.buffer); audio_timing_reset(&receiver.timing); audio_receiver_reset_resend_state(); receiver.timing.ptp_locked = ptp_clock_is_locked(); audio_receiver_reset_blocks(); return receiver.stream->ops->start(receiver.stream, data_port); } esp_err_t audio_receiver_start_buffered(uint16_t tcp_port) { audio_receiver_set_stream_type(AUDIO_STREAM_BUFFERED); if (!receiver.stream || !receiver.stream->ops || !receiver.stream->ops->start) { return ESP_FAIL; } // Buffered streams use a fixed port, no need to restart if running if (receiver.stream->running) { return ESP_OK; } // Starting a stream resets all timing state (including pause tracking) audio_receiver_reset_stats(); audio_buffer_flush(&receiver.buffer); audio_timing_reset(&receiver.timing); audio_receiver_reset_resend_state(); receiver.timing.ptp_locked = ptp_clock_is_locked(); audio_receiver_reset_blocks(); return receiver.stream->ops->start(receiver.stream, tcp_port); } esp_err_t audio_receiver_start_stream(uint16_t data_port, uint16_t control_port, uint16_t tcp_port) { if (!receiver.stream) { return ESP_FAIL; } if (receiver.stream->type == AUDIO_STREAM_BUFFERED) { return audio_receiver_start_buffered(tcp_port); } return audio_receiver_start(data_port, control_port); } uint16_t audio_receiver_get_stream_port(void) { if (!receiver.stream || !receiver.stream->ops || !receiver.stream->ops->get_port) { return 0; } return receiver.stream->ops->get_port(receiver.stream); } void audio_receiver_set_client_control(uint32_t client_ip, uint16_t client_control_port) { if (client_ip == 0 || client_control_port == 0) { receiver.retransmit_enabled = false; audio_receiver_reset_resend_state(); return; } memset(&receiver.client_control_addr, 0, sizeof(receiver.client_control_addr)); receiver.client_control_addr.sin_family = AF_INET; receiver.client_control_addr.sin_addr.s_addr = client_ip; receiver.client_control_addr.sin_port = htons(client_control_port); receiver.retransmit_enabled = true; audio_receiver_reset_resend_state(); ESP_LOGI(TAG, "NACK retransmission enabled, client control port %u", client_control_port); } void audio_receiver_stop(void) { if (receiver.realtime_stream && receiver.realtime_stream->ops && receiver.realtime_stream->ops->stop) { receiver.realtime_stream->ops->stop(receiver.realtime_stream); } if (receiver.buffered_stream && receiver.buffered_stream->ops && receiver.buffered_stream->ops->stop) { receiver.buffered_stream->ops->stop(receiver.buffered_stream); } audio_decoder_destroy(receiver.decoder); receiver.decoder = NULL; if (receiver.realtime_stream) { memset(&receiver.realtime_stream->encrypt, 0, sizeof(receiver.realtime_stream->encrypt)); } if (receiver.buffered_stream) { memset(&receiver.buffered_stream->encrypt, 0, sizeof(receiver.buffered_stream->encrypt)); } receiver.retransmit_enabled = false; memset(&receiver.client_control_addr, 0, sizeof(receiver.client_control_addr)); audio_receiver_reset_resend_state(); audio_receiver_flush(); } void audio_receiver_stop_buffered_only(void) { if (receiver.buffered_stream && receiver.buffered_stream->ops && receiver.buffered_stream->ops->stop) { receiver.buffered_stream->ops->stop(receiver.buffered_stream); } } void audio_receiver_get_stats(audio_stats_t *stats) { if (!stats) { return; } memcpy(stats, &receiver.stats, sizeof(receiver.stats)); } size_t audio_receiver_read(int16_t *buffer, size_t samples) { if (!receiver.buffer.pool || !buffer || samples == 0) { return 0; } return audio_timing_read(&receiver.timing, &receiver.buffer, receiver.stream, &receiver.stats, buffer, samples); } bool audio_receiver_has_data(void) { int buffered_frames = audio_buffer_get_frame_count(&receiver.buffer); return buffered_frames > 0 || receiver.timing.pending_valid; } void audio_receiver_flush(void) { // Flush is an explicit reset — clear all timing state including pause // tracking. The sender will provide fresh anchor times after flush. // Also disarm any pending deferred flush so it does not fire on the // next track's frames. audio_buffer_flush(&receiver.buffer); audio_timing_reset(&receiver.timing); audio_receiver_reset_resend_state(); receiver.discard_before_rtp_valid = false; receiver.discard_above_rtp_valid = false; receiver.arm_gate_on_next_anchor = false; receiver.discard_all_until_anchor = false; receiver.paused_rtp_valid = false; receiver.blocks_read_in_sequence = 1; } void audio_receiver_seek_flush(void) { // Mid-stream seek flush (FLUSH / immediate FLUSHBUFFERED). Like // audio_receiver_flush() but sets timing.quick_start so audio_timing_read // starts as soon as 1 frame is available, with normal anchor-based timing. // Also disarms any pending deferred flush (audio_timing_reset clears it). audio_receiver_flush(); receiver.timing.quick_start = true; // Request that the RTP gate be armed as soon as the next anchor arrives. // This covers the forward-seek case where the buffer is already empty by // the time SETRATEANCHORTIME arrives, so the seek-detection heuristic // (which needs oldest_rtp from the buffer) would otherwise miss arming it. receiver.arm_gate_on_next_anchor = true; // Reject ALL incoming frames until the next anchor. Prevents stale TCP // data from filling the buffer between FLUSHBUFFERED and SETRATEANCHORTIME, // which would cause a second flush and double the startup delay. receiver.discard_all_until_anchor = true; } void audio_receiver_set_deferred_flush(uint32_t flush_until_ts) { if (!receiver.stream) { return; } // Write flush_until_ts before arming the flag so audio_timing_read never // sees deferred_flush_pending=true with a stale timestamp. receiver.timing.flush_until_ts = flush_until_ts; receiver.timing.deferred_flush_pending = true; ESP_LOGI(TAG, "Deferred flush armed: flush_until_ts=%" PRIu32, flush_until_ts); } void audio_receiver_pause(void) { // Stop the consumer. The receiver tasks keep running so the audio buffer // continues to fill with pre-buffered audio — TCP back-pressure naturally // throttles the sender. On resume the phone sends a fresh // SETRATEANCHORTIME anchor that re-aligns the buffered frames to the // correct wall-clock position; no flush or offset compensation is needed. audio_timing_set_playing(&receiver.timing, false); receiver.blocks_read_in_sequence = 0; } uint16_t audio_receiver_get_buffered_port(void) { return receiver.buffered_port; }