From 8f208c26b6fa1a9f3372679c047cab559c06e26b Mon Sep 17 00:00:00 2001 From: Andrew Opalach Date: Mon, 14 Sep 2026 08:57:42 -0400 Subject: Server-side fixes from DIRECT_MODE testing Signed-off-by: Andrew Opalach --- src/libsink/common.h | 2 + src/libsink/desktop.c | 9 +- src/libsink/input_simulator.c | 7 +- src/libsink/sink.c | 563 ++++++++++++++++++++++++++++++------------ src/libsink/sink.h | 10 +- 5 files changed, 419 insertions(+), 172 deletions(-) (limited to 'src/libsink') diff --git a/src/libsink/common.h b/src/libsink/common.h index 254e246..475ef5f 100644 --- a/src/libsink/common.h +++ b/src/libsink/common.h @@ -2,6 +2,8 @@ enum { CAMU_SINK_SET = 0, + CAMU_SINK_UNSET, + CAMU_SINK_SEQUENCE, CAMU_SINK_PAUSE, CAMU_SINK_SEEK }; diff --git a/src/libsink/desktop.c b/src/libsink/desktop.c index 427a6f4..be3e797 100644 --- a/src/libsink/desktop.c +++ b/src/libsink/desktop.c @@ -3,7 +3,7 @@ #ifdef CAMU_NO_MINIAUDIO_BACKENDS // @TODO: audio_null is incomplete in that it never actually reads from the buffers. -// This is an issue in CAMU_DIRECT_MODE because eventually the server-side packet_pool +// This is an issue in DIRECT_MODE because eventually the server-side packet_pool // will be starved by pending audio packets. #define DESKTOP_NULL_AUDIO #endif @@ -188,10 +188,10 @@ static void screen_callback(void *userdata, u8 op, void *opaque) camu_sink_reseek(&c->sink); break; case CAMU_SCREEN_AUDIO_TRACK: - camu_sink_adjust_audio_track(&c->sink, *(s32 *)opaque); + camu_sink_change_audio_track(&c->sink, *(s32 *)opaque); break; case CAMU_SCREEN_SUBTITLE_TRACK: - camu_sink_adjust_subtitle_track(&c->sink, *(s32 *)opaque); + camu_sink_change_subtitle_track(&c->sink, *(s32 *)opaque); break; case CAMU_SCREEN_SHUFFLE: camu_sink_shuffle(&c->sink); @@ -219,9 +219,10 @@ static void screen_callback(void *userdata, u8 op, void *opaque) bool camu_desktop_open(struct camu_desktop *c, char *window_name) { + camu_screen_init(&c->scr); c->scr.callback = screen_callback; c->scr.userdata = c; - if (!(camu_screen_init(&c->scr) && camu_screen_create_window(&c->scr, window_name))) { + if (!camu_screen_create_window(&c->scr, window_name)) { log_error("Failed to create window."); return false; } diff --git a/src/libsink/input_simulator.c b/src/libsink/input_simulator.c index 5b9d6e6..883e631 100644 --- a/src/libsink/input_simulator.c +++ b/src/libsink/input_simulator.c @@ -7,7 +7,7 @@ #include "input_simulator.h" -// This is ignoring all thread-safety. +// Currently ignoring all thread-safety. static s32 quit = 1; static struct nn_thread thread; @@ -17,7 +17,8 @@ enum { TOGGLE_PAUSE, SEEK, SHUFFLE, - MARK, // count + COUNT, + RESEEK }; static nn_thread_result NNWT_THREADCALL input_simulation_thread(void *userdata) @@ -28,7 +29,7 @@ static nn_thread_result NNWT_THREADCALL input_simulation_thread(void *userdata) //nn_thread_sleep(NNWT_TS_FROM_USEC(2000000 + al_random_int(0, 1750000))); //nn_thread_sleep(NNWT_TS_FROM_USEC(60000)); nn_thread_sleep(NNWT_TS_FROM_USEC(30000)); - switch (al_random_int(0, MARK - 1)) { + switch (al_random_int(0, COUNT - 1)) { case SKIP: { s32 n = al_random_int(1, 5); log_info("SKIP (n: %d).", n); diff --git a/src/libsink/sink.c b/src/libsink/sink.c index 5f61bd4..bad51d9 100644 --- a/src/libsink/sink.c +++ b/src/libsink/sink.c @@ -12,8 +12,6 @@ #include "sink.h" #include "common.h" -//#define CAMU_SINK_ONESHOT - // Requested state of the sinks outputs. enum { SINK_PAUSED = 0, @@ -55,17 +53,25 @@ enum { TOGGLE_PAUSE, SEEK, RESEEK, + AUDIO_TRACK, + SUBTITLE_TRACK, SHUFFLE, END }; // Status reporting. enum { - NOTIFY_EMPTY = 1, - NOTIFY_NOT_EMPTY = 1 << 1, - NOTIFY_SEEK = 1 << 2 + NOTIFY_CONNECTED = 1, + NOTIFY_DISCONNECTED = 1 << 1, + NOTIFY_EMPTY = 1 << 2, + NOTIFY_NOT_EMPTY = 1 << 3, + NOTIFY_ENTRY_ADDED = 1 << 4, + NOTIFY_SEEK = 1 << 5 }; +// If the video output isn't running, the audio output might need to resolve a clock_pause(). +#define AUDIO_STOP_ON_CLOCK_PAUSE + // Number of entries to keep buffered at one time. #define ENTRY_MAX_AGE 4 #define SINK_LRU_MAX UINT16_MAX @@ -130,8 +136,8 @@ AL_STATIC_ASSERT(max_age_lt_lru, ENTRY_MAX_AGE, <, SINK_LRU_MAX); // video_buffer_callback() // CAMU_MIXER_THREADED or CAMU_SCREEN_THREADED: // clock_callback() -// client_callback()::LIANA_CLIENT_DATA -// client_callback()::LIANA_CLIENT_EOF/ERRORED +// client_callback(LIANA_CLIENT_DATA) +// client_callback(LIANA_CLIENT_EOF/ERRORED) static inline bool entry_audio_buffer_held(struct camu_sink_entry *entry) { @@ -276,6 +282,17 @@ static void maybe_disconnect_entry(struct camu_sink_entry *entry) } } +static struct lia_prefs *get_prefs_from_node_id(struct camu_sink *sink, u32 node_id) +{ + struct lia_prefs *prefs; + al_array_foreach_ptr(sink->node_prefs, i, prefs) { + if (prefs->node_id == node_id) { + return prefs; + } + } + return NULL; +} + // sink->current could be NULL. static inline struct camu_sink_entry *get_entry_for_command(struct camu_sink *sink) { @@ -385,7 +402,7 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) return; } case ADD: { - if (!sink->connected) return; + if (!sink->conn) return; struct nn_packet *packet = nn_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION); nn_packet_write_str(packet, &sink->default_list); nn_packet_write_u8(packet, CAMU_LIST_ADD); @@ -396,7 +413,7 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) break; } case SKIP: { - if (!sink->connected) return; + if (!sink->conn) return; struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque; struct nn_packet *packet = nn_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION); nn_packet_write_str(packet, &sink->default_list); @@ -407,7 +424,7 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) break; } case TOGGLE_PAUSE: { - if (!sink->connected) return; + if (!sink->conn) return; struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque; struct nn_packet *packet = nn_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION); nn_packet_write_str(packet, &sink->default_list); @@ -418,7 +435,7 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) break; } case SEEK: { - if (!sink->connected) return; + if (!sink->conn) return; struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque; struct nn_packet *packet = nn_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION); nn_packet_write_str(packet, &sink->default_list); @@ -434,8 +451,30 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) if (sink->conn) nn_rpc_conn_disconnect(sink->conn); break; } + case AUDIO_TRACK: { + if (!sink->conn) return; + struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque; + struct lia_prefs *prefs = get_prefs_from_node_id(sink, REMOTE_ENTRY_ID(entry->id)); + s32 i = prefs->index.audio + cmd->v.i; + if (i <= prefs->index.audio_max && i >= prefs->index.audio_min) { + prefs->index.audio = i; + nn_rpc_conn_disconnect(sink->conn); + } + break; + } + case SUBTITLE_TRACK: { + if (!sink->conn) return; + struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque; + struct lia_prefs *prefs = get_prefs_from_node_id(sink, REMOTE_ENTRY_ID(entry->id)); + s32 i = prefs->index.subtitles + cmd->v.i; + if (i <= prefs->index.subtitles_max && i >= prefs->index.subtitles_min) { + prefs->index.subtitles = i; + nn_rpc_conn_disconnect(sink->conn); + } + break; + } case SHUFFLE: { - if (!sink->connected) return; + if (!sink->conn) return; struct nn_packet *packet = nn_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION); nn_packet_write_str(packet, &sink->default_list); nn_packet_write_u8(packet, CAMU_LIST_SHUFFLE); @@ -443,7 +482,7 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) break; } case END: { - if (!sink->connected) return; + if (!sink->conn) return; struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque; struct nn_packet *packet = nn_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION); nn_packet_write_str(packet, &sink->default_list); @@ -485,21 +524,20 @@ static void mixer_callback(void *userdata, u8 op) } } -static s32 entry_lru_compare(const void *a, const void *b) +static s32 lru_compare_high_to_low(const void *a, const void *b) { - struct camu_sink_entry *aa = *((struct camu_sink_entry **)a); - struct camu_sink_entry *bb = *((struct camu_sink_entry **)b); - if (aa->lru > bb->lru) return -1; - else if (aa->lru < bb->lru) return 1; + struct camu_sink_entry *entry1 = *(struct camu_sink_entry **)a; + struct camu_sink_entry *entry2 = *(struct camu_sink_entry **)b; + if (entry1->lru > entry2->lru) return -1; + else if (entry1->lru < entry2->lru) return 1; return 0; } static void maybe_cleanup_old_entries(struct camu_sink *sink) { - al_array_sort(sink->entries, struct camu_sink_entry *, entry_lru_compare); // We check size <= MAX_AGE in the loops because sink->lru is not // indicative of the amount of entries we have loaded. - // The most obvious reason being it's incremented when moving back + // The most obvious reason being that it's incremented when moving back // and forth between two entries. As well as for buffer and queue operations. // We have to handle sink->lru wrapping in a step before the default case. // 0 65532 65533 65534 65535 @@ -507,26 +545,46 @@ static void maybe_cleanup_old_entries(struct camu_sink *sink) // 0 1 2 65534 65535 // 0 1 2 3 65535 // 0 1 2 3 4 + al_array_sort(sink->entries, struct camu_sink_entry *, lru_compare_high_to_low); + array(struct camu_sink_entry *) cleanup; + al_array_init(cleanup); struct camu_sink_entry *entry; al_array_foreach_rev(sink->entries, i, entry) { - if (sink->entries.count <= ENTRY_MAX_AGE) return; + if (sink->entries.count <= ENTRY_MAX_AGE) { + goto out; + } u16 entry_age = (SINK_LRU_MAX - entry->lru) + sink->lru; if (entry->lru > sink->lru && entry_age > ENTRY_MAX_AGE) { al_array_remove_at(sink->entries, i); - maybe_disconnect_entry(entry); + al_assert(!al_array_contains(sink->entries, entry)); + al_array_push(cleanup, entry); } } // Make sure we don't have to consider wrapping in the second loop. - if (sink->lru < ENTRY_MAX_AGE) return; - al_array_foreach_rev(sink->entries, i, entry) { - al_assert(sink->lru >= entry->lru); - if (sink->entries.count <= ENTRY_MAX_AGE) return; - u16 entry_age = sink->lru - entry->lru; - if (entry_age > ENTRY_MAX_AGE) { - al_array_remove_at(sink->entries, i); - maybe_disconnect_entry(entry); + if (sink->lru >= ENTRY_MAX_AGE) { + al_array_foreach_rev(sink->entries, i, entry) { + al_assert(sink->lru >= entry->lru); + if (sink->entries.count <= ENTRY_MAX_AGE) { + goto out; + } + u16 entry_age = sink->lru - entry->lru; + if (entry_age > ENTRY_MAX_AGE) { + al_array_remove_at(sink->entries, i); + al_assert(!al_array_contains(sink->entries, entry)); + al_array_push(cleanup, entry); + } } } +out: + // Why we can't disconnect an entry while iterating sink->entries: + // -> maybe_disconnect_entry() -> nn_packet_stream_disconnect() -> connection_closed_callback(). + // -> In either CLIENT_REMOVE_BUFFERS or CLIENT_CLOSED, BLOCKING_SLEEP() runs the event loop. + // -> An entry other than the disconnected entry gets removed from sink->entries. + // -> On the next loop `i` is invalid. + al_array_foreach(cleanup, i, entry) { + maybe_disconnect_entry(entry); + } + al_array_free(cleanup); } static void maybe_run_previous(struct camu_sink *sink) @@ -549,16 +607,16 @@ static void maybe_run_previous(struct camu_sink *sink) // being at the point it's freed. static void run_previous_if_contains(struct camu_sink *sink, struct camu_sink_entry *key) { - bool removed = false; + bool ran = false; struct camu_sink_entry *previous; al_array_foreach(sink->previous, i, previous) { if (previous == key) { maybe_run_previous(sink); - removed = true; + ran = true; break; } } - log_trace("run_previous_if_contains("ENTRY_FMT"), removed: %s.", ENTRY_ARG(key), BOOLSTR(removed)); + log_trace("run_previous_if_contains("ENTRY_FMT"), ran: %s.", ENTRY_ARG(key), BOOLSTR(ran)); } static bool maybe_remove_from_previous(struct camu_sink *sink, struct camu_sink_entry *entry) @@ -572,6 +630,7 @@ static bool maybe_remove_from_previous(struct camu_sink *sink, struct camu_sink_ break; } } + al_assert(!al_array_contains(sink->previous, entry)); log_trace("maybe_remove_from_previous("ENTRY_FMT"), removed: %s.", ENTRY_ARG(entry), BOOLSTR(removed)); return removed; } @@ -600,6 +659,7 @@ static void after_add_entry(struct camu_sink_entry *entry, bool skip_audio, bool ); // Clear the screen if skipping from a video to an audio-only entry. if (VIDEO_ENDED_OR_EMPTY(entry)) refresh_video_output(sink); + sink->notify_status |= NOTIFY_ENTRY_ADDED; } // Call this after setting state to ADDED because this entry might be in previous. @@ -688,7 +748,10 @@ static void ensure_static_video_removed(struct camu_sink_entry *entry) if (!VIDEO_EMPTY(entry) && VIDEO_IS_STATIC(entry)) { remove_entry_video_buffer(entry); struct camu_sink *sink = entry->sink; + // Unlock to wait, like in CLIENT_REMOVE_BUFFERS. + nn_mutex_unlock(&sink->lock); while (entry_video_buffer_held(entry)) { BLOCKING_SLEEP(sink, NNWT_TS_FROM_USEC(2000)); } + nn_mutex_lock(&sink->lock); } } @@ -705,7 +768,7 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target) if (suspended) { al_assert(suspended == current); // It should be impossible for a buffer to be QUEUED while it's entry is suspended. - // Even for a static video buffer because we wouldn't know if it was static yet. + // Even for a static video buffer because we wouldn't have known if it was static yet. al_assert(AUDIO_STATE(suspended) != BUFFER_QUEUED); al_assert(VIDEO_STATE(suspended) != BUFFER_QUEUED); ensure_static_video_removed(suspended); @@ -722,7 +785,7 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target) if (dangling_target) log_trace("Ignoring dangling target."); bool stop_video = dangling_target; if (!dangling_target) { - target->audio.ignore_paused = false; + target->audio.ignore_pause = false; if (!sink->local && !target->paused) { camu_audio_buffer_resync(&target->audio.buf); } @@ -761,9 +824,11 @@ static void pause_and_swap_to(struct camu_sink *sink, struct camu_sink_entry *ta al_assert(target != current); al_assert(!sink->target); sink->target = target; + // Checking !current here allows the list to send set(PAUSE_BOTH) from handle_add_sink(). bool immediate = !current || ENTRY_ENDED(current); if (current) { - current->audio.ignore_paused = true; + current->audio.ignore_pause = true; + camu_clock_set_pause_for_swap(¤t->clock); immediate |= camu_clock_pause(¤t->clock, at); } if (immediate) { @@ -772,11 +837,13 @@ static void pause_and_swap_to(struct camu_sink *sink, struct camu_sink_entry *ta } } +//#define SINK_ONESHOT + static bool end_entry_and_advance_queue(struct camu_sink *sink, struct camu_sink_entry *entry) { log_debug("Entry ("ENTRY_FMT") ended.", ENTRY_ARG(entry)); run_previous_if_contains(sink, entry); -#ifdef CAMU_SINK_ONESHOT +#ifdef SINK_ONESHOT sink->callback(sink->userdata, CAMU_SINK_MOCK_CLOSE, 0, NULL); return false; #endif @@ -815,22 +882,27 @@ static void audio_buffer_callback(void *userdata, u8 op) lia_vcr_uncork(entry->audio.track); break; case CAMU_BUFFER_PAUSED: // Comes from audio read() thread. +#ifndef AUDIO_STOP_ON_CLOCK_PAUSE nn_mutex_lock(&sink->lock); // entry->paused could plausibly be false here if the lock was held // by pause_command_callback() to resume. This can be simulated by // calling list_toggle_pause() twice in server/list_action_callback() // for each sink request. Spaced by an nn_event_loop_sleep(~15500us) // (no_video, MINIAUDIO_LOW_LATENCY mode). - if (!entry->audio.ignore_paused && entry->paused) { + if (!entry->audio.ignore_pause && entry->paused) { log_info("Audio buffer paused."); queue_cmd(sink, CMD(STOP, .v.u = CAMU_SINK_AUDIO)); } nn_mutex_unlock(&sink->lock); +#endif break; case CAMU_BUFFER_EOF: case CAMU_BUFFER_ERRORED: { bool error = op == CAMU_BUFFER_ERRORED; if (error) { + // There's two main considerations for whether to EJECT_ENTRY on error or not. + // 1. If the other buffer would have continued working, that's an unoptimal user experience. + // 2. If the server continues sending frames for the erorred buffer, that's wasteful. log_error("Audio buffer errored."); } else { log_debug("Audio EOF."); @@ -873,7 +945,7 @@ static void video_buffer_callback(void *userdata, u8 op) lia_vcr_uncork(entry->video.track); break; case CAMU_BUFFER_EOF: - case CAMU_BUFFER_ERRORED: { + case CAMU_BUFFER_ERRORED: { // See notes in audio_buffer_callback(). bool error = op == CAMU_BUFFER_ERRORED; if (error) { log_error("Video buffer errored."); @@ -889,7 +961,6 @@ static void video_buffer_callback(void *userdata, u8 op) return; } // Video state could be ADDED, SET_OR_BUFFERED, or CONFIGURED. - // See note about threaded outputs in audio_buffer_callback(EOF|ERRORED). if (VIDEO_STATE(entry) == BUFFER_ADDED) { remove_entry_video_buffer(entry); } @@ -920,6 +991,9 @@ static void clock_callback(void *userdata, u8 op) sink->target = NULL; } else if (entry->paused) { queue_cmd(sink, CMD(STOP, .v.u = CAMU_SINK_VIDEO)); +#ifdef AUDIO_STOP_ON_CLOCK_PAUSE + queue_cmd(sink, CMD(STOP, .v.u = CAMU_SINK_AUDIO)); +#endif } } nn_mutex_unlock(&sink->lock); @@ -943,7 +1017,7 @@ static void evaluate_and_set_buffer_params(struct camu_sink *sink, struct camu_s camu_audio_buffer_set_latency(&entry->audio.buf, audio); camu_audio_buffer_set_no_video(&entry->audio.buf, ignore_video); camu_video_buffer_set_latency(&entry->video.buf, video); - log_info("video_latency: %fs (%u frames), audio_latency: %fs.", video, frames, audio); + log_debug("video_latency: %fs (%u frames), audio_latency: %fs.", video, frames, audio); if (sink->local) { camu_audio_buffer_set_ignore_desync(&entry->audio.buf, ignore_video); // When the video buffer starts the clock, we have to consider the audio @@ -963,7 +1037,9 @@ static void run_queue_by_opaque(struct camu_sink *sink, void *opaque) camu_queue_lock(sink->queue); struct camu_sink_cmd *cmd; al_array_foreach_ptr(sink->queue.a, i, cmd) { - if (cmd->opaque == opaque) { + // Only run "Sink operations", otherwise in DIRECT_MODE we could + // end up with an insane call stack and likely deadlock. + if (cmd->op < ADD && cmd->opaque == opaque) { handle_sink_cmd(sink, cmd); al_array_remove_at_iter(sink->queue.a, i); } @@ -1043,6 +1119,9 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str // This is mainly to assert DETACHED handling. al_assert(!VIDEO_ENDED(entry) && !AUDIO_ENDED(entry)); evaluate_and_set_buffer_params(sink, entry); + struct lia_prefs *prefs = get_prefs_from_node_id(sink, REMOTE_ENTRY_ID(entry->id)); + al_assert(prefs); + *prefs = entry->client.prefs; nn_mutex_unlock(&sink->lock); break; } @@ -1091,7 +1170,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str } if (sink->target) { // This is necessary to avoid re-adding an entry with an in-between clock state. - // See note in clock.c::camu_clock_seek(). + // See note in clock.c:camu_clock_seek(). switch_to(sink, sink->target); sink->target = NULL; } else if (rec->reconnect) { @@ -1101,8 +1180,9 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str } // This entry might be in previous if it was added to previous then, - // 1. it's being cleaned up after ENTRY_MAX_AGE - 1 entries were added but none buffered. - // 2. it was seeked. + // 1. it was seeked. + // 2. it's being cleaned up after ENTRY_MAX_AGE - 1 entries were added but none buffered. + // 3. it got disconnected server-side. run_previous_if_contains(sink, entry); // AUDIO/VIDEO_STATE() could be INIT at this point, even if entry = current. @@ -1137,7 +1217,9 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str } // If MIXER_THREADED_START_STOP is not set, REMOVE_BUFFER is not thread-safe. // Meaning it must be run on the event loop. So, resolve any queued REMOVE_BUFFER - // requests so we can safely block the loop. + // requests so we can safely block the loop. As of now, remove_entry_video_buffer() + // never queues REMOVE_BUFFER. If that were to change, we may need to reevaluate + // ensure_static_video_removed() based on this. run_queue_by_opaque(sink, entry); // Unlock to wait. @@ -1202,7 +1284,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str bool ignore_video = VIDEO_EMPTY(entry) || VIDEO_IS_STATIC(entry); if (!AUDIO_EMPTY(entry)) { camu_audio_buffer_reset(&entry->audio.buf); - // no_video is set to true in video BUFFER_EOF as a fail-safe. Reset it here. + // no_video is set to true in video BUFFER_EOF as a fail-safe, reset it here. camu_audio_buffer_set_no_video(&entry->audio.buf, ignore_video); } if (!ignore_video) { @@ -1270,6 +1352,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str // If a client is closed after a failed reconnect, a static video buffer could still be added. if (rec->reconnect) ensure_static_video_removed(entry); + al_assert(AUDIO_STATE(entry) != BUFFER_ADDED); al_assert(VIDEO_STATE(entry) != BUFFER_ADDED); @@ -1293,10 +1376,13 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str queue_cmd(sink, CMD(STOP, .v.u = CAMU_SINK_VIDEO)); } } - // If current was never fully added we need to call this here. + // If current never got to after_add_entry(), run previous here. maybe_run_previous(sink); } + // If entry was in previous, it should have been removed in CLIENT_REMOVE_BUFFERS. + al_assert(!maybe_remove_from_previous(sink, entry)); + nn_mutex_unlock(&sink->lock); lia_client_free(&entry->client); @@ -1318,15 +1404,28 @@ static struct camu_sink_entry *create_entry(struct camu_sink *sink, u64 id) entry->disconnected = false; entry->ended = false; - camu_clock_init(&entry->clock, clock_callback, entry); - entry->client.callback = client_callback; entry->client.userdata = entry; entry->client.renderer = sink->video.renderer; - entry->client.prefs = sink->prefs; + struct lia_prefs *prefs = get_prefs_from_node_id(sink, REMOTE_ENTRY_ID(id)); + if (!prefs) { + prefs = &sink->prefs; + prefs->node_id = REMOTE_ENTRY_ID(id); + prefs->index.audio = -1; + prefs->index.audio_min = INT32_MAX; + prefs->index.audio_max = 0; + prefs->index.video = -1; + prefs->index.subtitles = -1; + prefs->index.subtitles_min = INT32_MAX; + prefs->index.subtitles_max = 0; + al_array_push(sink->node_prefs, *prefs); + } + entry->client.prefs = *prefs; + + camu_clock_init(&entry->clock, clock_callback, entry); AUDIO_STATE(entry) = BUFFER_INIT; - entry->audio.ignore_paused = false; + entry->audio.ignore_pause = false; camu_audio_buffer_init(&entry->audio.buf, &entry->clock); entry->audio.buf.callback = audio_buffer_callback; entry->audio.buf.userdata = entry; @@ -1342,8 +1441,6 @@ static struct camu_sink_entry *create_entry(struct camu_sink *sink, u64 id) union { f64 f; u64 u; } fv = { .u = id }; entry->video.buf.seek_pts = fv.f; - al_array_push(sink->entries, entry); - return entry; } @@ -1356,15 +1453,6 @@ static struct camu_sink_entry *get_entry_from_id(struct camu_sink *sink, u64 id) return NULL; } -static void unset_current(struct camu_sink *sink) -{ - struct camu_sink_entry *current = sink->current; - nn_mutex_unlock(&sink->lock); - if (current) { - maybe_disconnect_entry(current); - } -} - static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn, struct nn_packet *packet, struct nn_packet *rpacket) { @@ -1372,16 +1460,11 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn, (void)rpacket; u8 op = nn_packet_read_u8(packet); - if (op == LIANA_SINK_UNSET) { - nn_mutex_lock(&sink->lock); - unset_current(sink); - nn_mutex_lock(&sink->lock); - goto out; - } // Liana node info. - str addr; - nn_packet_read_str(packet, &addr); + str addr, paddr; + nn_packet_read_str(packet, &paddr); + al_str_clone(&addr, &paddr); u16 port = nn_packet_read_u16(packet); u32 node_id = nn_packet_read_u32(packet); @@ -1389,34 +1472,25 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn, u64 id = LOCAL_ENTRY_ID(sink, nn_packet_read_u32(packet)); s32 sequence = nn_packet_read_s32(packet); u64 at = nn_packet_read_u64(packet); + if (sink->local) at = 0; u64 pos = nn_packet_read_u64(packet); u8 pause = nn_packet_read_u8(packet); u32 reset_token = nn_packet_read_u32(packet); + nn_packet_stream_return_packet(conn->stream, packet); + struct camu_sink_entry *entry = get_entry_from_id(sink, id); bool create = !entry; - if (create) entry = create_entry(sink, id); - entry->sequence = sequence; - entry->lru = sink->lru; - sink->lru = al_u16_add_wrap(sink->lru, 1, SINK_LRU_MAX); - entry->reset_token = reset_token; if (create) { + entry = create_entry(sink, id); + al_array_push(sink->entries, entry); entry->paused = pause == LIANA_PAUSE_NONE || pause == LIANA_PAUSE_PAUSE; camu_clock_set(&entry->clock, pos / 1000000.0); - lia_client_connect(&entry->client, sink->loop, sink->type, &addr, port, node_id, pos); } - - // Don't lock before client_connect() or we could deadlock in CLIENT_CLOSED on a failed socket_connect(). - nn_mutex_lock(&sink->lock); - - // lia_client_connect() can fail and call connection_closed_callback() in-line. Meaning this entry - // could already be disconnected here. - if (!al_array_contains(sink->entries, entry)) { - unset_current(sink); - goto out; - } - - struct camu_sink_entry *current = sink->current; + entry->sequence = sequence; + entry->lru = sink->lru; + sink->lru = al_u16_add_wrap(sink->lru, 1, SINK_LRU_MAX); + entry->reset_token = reset_token; if (op == LIANA_SINK_BUFFER) { log_trace("buffered("ENTRY_FMT"), created: %s.", ENTRY_ARG(entry), BOOLSTR(create)); @@ -1426,12 +1500,32 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn, goto out; } - if (sink->local) at = 0; + nn_mutex_lock(&sink->lock); - // For target to be set that must mean current is set, armed to pause, and not ended. + // For target to be set that means: current is set, armed to pause, and not ended. + struct camu_sink_entry *current = sink->current; struct camu_sink_entry *prev_target = sink->target; + + // Go against the list and try to handle this case with a pause_and_swap_to() as it's much cleaner. + if (current && (CONNECTION_NUMBER(current->id) != CONNECTION_NUMBER(entry->id)) && + (REMOTE_ENTRY_ID(current->id) == REMOTE_ENTRY_ID(entry->id))) { + if (prev_target) { + sink->target = entry; + // Don't assume anything about the list-side pause state of current. + if (pause == LIANA_PAUSE_RESUME) { + // This will result in a slight jump due to differing `at`s. + camu_clock_resume(&entry->clock, at); + } + nn_mutex_unlock(&sink->lock); + goto out; + } else if (!current->paused) { + pause += LIANA_PAUSE_PAUSE; + } + } + log_trace("set("ENTRY_FMT"), %s, created: %s, target: "ENTRY_FMT".", ENTRY_ARG(entry), lia_pause_op_name(pause), BOOLSTR(create), ENTRY_ARG(prev_target)); + switch (pause) { case LIANA_PAUSE_NONE: if (prev_target) { @@ -1499,14 +1593,61 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn, break; } -out: nn_mutex_unlock(&sink->lock); + +out: + if (create) { + lia_client_connect(&entry->client, sink->loop, sink->type, &addr, port, node_id, pos); + } + + al_str_free(&addr); + if (op != LIANA_SINK_BUFFER) { maybe_cleanup_old_entries(sink); } + return false; +} + +static bool unset_command_callback(void *userdata, struct nn_rpc_connection *conn, + struct nn_packet *packet, struct nn_packet *rpacket) +{ + struct camu_sink *sink = (struct camu_sink *)userdata; + (void)rpacket; + + nn_packet_stream_return_packet(conn->stream, packet); + + nn_mutex_lock(&sink->lock); + struct camu_sink_entry *current = sink->current; + nn_mutex_unlock(&sink->lock); + + log_trace("unset("ENTRY_FMT")", ENTRY_ARG(current)); + + if (current) { + maybe_disconnect_entry(current); + } + + return false; +} + +static bool sequence_command_callback(void *userdata, struct nn_rpc_connection *conn, + struct nn_packet *packet, struct nn_packet *rpacket) +{ + struct camu_sink *sink = (struct camu_sink *)userdata; + (void)rpacket; + + u64 id = LOCAL_ENTRY_ID(sink, nn_packet_read_u32(packet)); + s32 sequence = nn_packet_read_s32(packet); + nn_packet_stream_return_packet(conn->stream, packet); + struct camu_sink_entry *entry = get_entry_from_id(sink, id); + if (!entry) goto out; + entry->sequence = sequence; + + log_trace("sequence("ENTRY_FMT"), sequence: %d", ENTRY_ARG(entry), sequence); + +out: return false; } @@ -1519,23 +1660,32 @@ static bool pause_command_callback(void *userdata, struct nn_rpc_connection *con u64 id = LOCAL_ENTRY_ID(sink, nn_packet_read_u32(packet)); s32 sequence = nn_packet_read_s32(packet); u64 at = nn_packet_read_u64(packet); + if (sink->local) at = 0; u8 pause = nn_packet_read_u8(packet); + nn_packet_stream_return_packet(conn->stream, packet); + struct camu_sink_entry *entry = get_entry_from_id(sink, id); if (!entry) goto out; - nn_mutex_lock(&sink->lock); - // As long as the list discards skips with a non-current sequence, this should hold true. + // List-side we discard skip/pause commands with a non-current sequence and send SINK_SEQUENCE + // when an erroring command is reduced to only a sequence change (in terms of what the sink + // cares about). That being said, as of now, there's no technical reason to enforce this. al_assert(entry->sequence == sequence); + + nn_mutex_lock(&sink->lock); + log_trace("pause("ENTRY_FMT"), %s, audio_state: %hhu, video_state: %hhu.", ENTRY_ARG(entry), lia_pause_op_name(pause), AUDIO_STATE(entry), VIDEO_STATE(entry)); - if (sink->local) at = 0; + switch (pause) { case LIANA_PAUSE_PAUSE: { entry->paused = true; - if (camu_clock_pause(&entry->clock, at)) { - if (entry == sink->current) { + bool immediate = camu_clock_pause(&entry->clock, at); + if (entry == sink->current) { + if (immediate) { queue_cmd(sink, CMD(STOP, .v.u = CAMU_SINK_VIDEO)); } + refresh_video_output(sink); } log_info("Clock paused."); // Audio will be stopped in a BUFFER_PAUSED callback. @@ -1560,11 +1710,10 @@ static bool pause_command_callback(void *userdata, struct nn_rpc_connection *con break; } } + nn_mutex_unlock(&sink->lock); out: - nn_packet_stream_return_packet(conn->stream, packet); - return false; } @@ -1580,25 +1729,34 @@ static bool seek_command_callback(void *userdata, struct nn_rpc_connection *conn u64 pos = nn_packet_read_u64(packet); u32 reset_token = nn_packet_read_u32(packet); + nn_packet_stream_return_packet(conn->stream, packet); + struct camu_sink_entry *entry = get_entry_from_id(sink, id); if (!entry) goto out; - nn_mutex_lock(&sink->lock); + // List-side seek() allows a non-current sequence. See comment in pause_command_callback(). entry->sequence = sequence; + + nn_mutex_lock(&sink->lock); + log_trace("seek("ENTRY_FMT", %.2f), reset_token: %u.", ENTRY_ARG(entry), pos / 1000000.0, reset_token); + entry->reset_token = reset_token; sink->notify_status |= NOTIFY_SEEK; + refresh_video_output(sink); + nn_mutex_unlock(&sink->lock); + // The rest of the seek is handled in CLIENT_REMOVE_BUFFERS/RESUME_AT/RECONNECTED. lia_client_seek(&entry->client, pos, at); out: - nn_packet_stream_return_packet(conn->stream, packet); - return false; } static struct nn_rpc_command commands[] = { { .op = CAMU_SINK_SET, .callback = set_command_callback, .userdata = NULL }, + { .op = CAMU_SINK_UNSET, .callback = unset_command_callback, .userdata = NULL }, + { .op = CAMU_SINK_SEQUENCE, .callback = sequence_command_callback, .userdata = NULL }, { .op = CAMU_SINK_PAUSE, .callback = pause_command_callback, .userdata = NULL }, { .op = CAMU_SINK_SEEK, .callback = seek_command_callback, .userdata = NULL } }; @@ -1606,17 +1764,17 @@ static struct nn_rpc_command commands[] = { static void identify_callback(void *userdata, struct nn_rpc_connection *conn, struct nn_packet *packet) { struct camu_sink *sink = (struct camu_sink *)userdata; + nn_packet_stream_return_packet(conn->stream, packet); #ifndef CAMU_DIRECT_MODE if (sink->type == NNWT_SOCKET_UNIX) { - log_info("Sink connected to %.*s.", al_str_x(&sink->addr)); + log_info("Connected to %.*s.", al_str_x(&sink->addr)); } else { - log_info("Sink connected to %.*s:%hu.", al_str_x(&sink->addr), sink->port); + log_info("Connected to %.*s:%hu.", al_str_x(&sink->addr), sink->port); } #else (void)sink; log_info("Sink directly bridged to server."); #endif - nn_packet_stream_return_packet(conn->stream, packet); } static void identify_on_connection(struct camu_sink *sink) @@ -1630,49 +1788,71 @@ static void identify_on_connection(struct camu_sink *sink) static void connection_callback(void *userdata, struct nn_rpc_connection *conn) { struct camu_sink *sink = (struct camu_sink *)userdata; - nn_timer_stop(&sink->reconnect_timer); - if (sink->conn) al_assert(sink->conn == conn); sink->conn = conn; - sink->connected = true; - refresh_video_output(sink); // For OSD. + sink->connecting = false; sink->connection_number = al_u16_inc_wrap(sink->connection_number); if (sink->connection_number == 0) sink->connection_number = 1; + nn_timer_stop(&sink->reconnect_timer); + sink->notify_status |= NOTIFY_CONNECTED; + refresh_video_output(sink); +} + +static void ready_callback(void *userdata, struct nn_rpc_connection *conn) +{ + struct camu_sink *sink = (struct camu_sink *)userdata; + (void)conn; identify_on_connection(sink); } +static inline void rpc_connect(struct camu_sink *sink) +{ +#ifdef CAMU_DIRECT_MODE + nn_multiplex_direct_connect(sink->client.conn->stream, CAMU_MULTIPLEX_RPC); +#else + sink->connecting = nn_rpc_connect(&sink->client, CAMU_MULTIPLEX_RPC, sink->type, &sink->addr, sink->port); +#endif +} + +static inline void rpc_reconnect(struct camu_sink *sink) +{ +#ifdef CAMU_DIRECT_MODE + nn_multiplex_direct_reconnect(sink->client.conn->stream); +#else + sink->connecting = nn_rpc_reconnect(&sink->client, &sink->addr, sink->port); +#endif +} + static void reconnect_timer_callback(void *userdata, struct nn_timer *timer) { struct camu_sink *sink = (struct camu_sink *)userdata; (void)timer; - if (sink->conn) { + if (sink->connecting) { // If the client was connecting, reconnect() will force a disconnect before reconnecting. log_warn("Forcing reconnect due to timeout."); } - sink->conn = nn_rpc_reconnect(&sink->client, &sink->addr, sink->port); + rpc_reconnect(sink); } static void connection_closed_callback(void *userdata, struct nn_rpc_connection *conn) { struct camu_sink *sink = (struct camu_sink *)userdata; - // @TODO: This is broken for an immediately failing reconnect(). - // If the nn_rpc_reconnect() in this function fails and recurses on connection_closed_callback(), - // sink->conn will be NULL and we will not attempt to reconnect. - bool reconnect = sink->conn != NULL; - bool disconnected = sink->conn && sink->connection_number > 0; + bool was_connected = sink->conn && sink->connection_number > 0; + sink->connecting = false; if (sink->conn) { al_assert(sink->conn == conn); sink->conn = NULL; - sink->connected = false; - } else { - al_assert(!reconnect || !sink->connected); + sink->notify_status |= NOTIFY_DISCONNECTED; + refresh_video_output(sink); } - if (reconnect) { - if (disconnected) { + if (!sink->closed) { + if (was_connected) { log_warn("Connection to server closed, attempting reconnect..."); - nn_rpc_reconnect(&sink->client, &sink->addr, sink->port); +#ifndef CAMU_DIRECT_MODE + nn_timer_again(&sink->reconnect_timer); +#endif + rpc_reconnect(sink); } else { log_warn("Failed to connect to server, trying again..."); - nn_timer_again(&sink->reconnect_timer); } } } @@ -1693,13 +1873,15 @@ bool camu_sink_init(struct camu_sink *sink, struct nn_event_loop *loop, struct camu_mixer *mixer, struct camu_renderer *renderer) { sink->loop = loop; - nn_rpc_init(&sink->client, sink->loop, connection_callback, connection_closed_callback, sink); + nn_rpc_init(&sink->client, sink->loop, connection_callback, ready_callback, connection_closed_callback, sink); sink->conn = NULL; + sink->closed = false; + sink->connecting = false; sink->connection_number = 0; nn_mutex_init(&sink->lock); nn_timer_init(&sink->reconnect_timer, sink->loop, reconnect_timer_callback, sink); - // This is also effectively a timeout for attempted reconnects. - nn_timer_set_repeat(&sink->reconnect_timer, NNWT_TS_FROM_USEC(1000000)); + // Also effectively a timeout. + nn_timer_set_repeat(&sink->reconnect_timer, NNWT_TS_FROM_USEC(2500000)); nn_signal_init(&sink->queue_signal, sink->loop, queue_signal_callback, sink); nn_signal_start(&sink->queue_signal); camu_queue_init(sink->queue); @@ -1709,7 +1891,7 @@ bool camu_sink_init(struct camu_sink *sink, struct nn_event_loop *loop, al_array_init(sink->previous); al_array_init(sink->entries); nn_timer_init(&sink->empty_timer, sink->loop, empty_timer_callback, sink); - nn_timer_set_repeat(&sink->empty_timer, NNWT_TS_FROM_USEC(100000)); + nn_timer_set_repeat(&sink->empty_timer, NNWT_TS_FROM_USEC(250000)); nn_timer_again(&sink->empty_timer); sink->notify_status = 0; // Start high to exercise the wrapping path. @@ -1720,6 +1902,7 @@ bool camu_sink_init(struct camu_sink *sink, struct nn_event_loop *loop, sink->audio.mixer = mixer; sink->video.state = SINK_PAUSED; sink->video.renderer = renderer; + al_array_init(sink->node_prefs); return true; } @@ -1729,18 +1912,16 @@ bool camu_sink_connect(struct camu_sink *sink, str *name, u8 type, str *addr, u1 sink->type = type; al_str_clone(&sink->addr, addr); sink->port = port; - sink->conn = nn_rpc_prepare_client(&sink->client); + nn_rpc_prepare_client(&sink->client); al_assert(sink->callback); for (u32 i = 0; i < ARRAY_SIZE(commands); i++) { commands[i].userdata = sink; nn_rpc_add_command(&sink->client, &commands[i]); } -#ifdef CAMU_DIRECT_MODE - // Note that direct_connect() runs connection_callback() directly. - nn_multiplex_direct_connect(sink->conn->stream, CAMU_MULTIPLEX_RPC); -#else - nn_rpc_connect(&sink->client, CAMU_MULTIPLEX_RPC, sink->type, &sink->addr, sink->port); +#ifndef CAMU_DIRECT_MODE + nn_timer_again(&sink->reconnect_timer); #endif + rpc_connect(sink); return true; } @@ -1765,40 +1946,46 @@ void camu_sink_add(struct camu_sink *sink, str *path) void camu_sink_skip(struct camu_sink *sink, s32 n) { nn_mutex_lock(&sink->lock); - struct camu_sink_entry *current = get_entry_for_command(sink); + struct camu_sink_entry *entry = get_entry_for_command(sink); nn_mutex_unlock(&sink->lock); - queue_cmd(sink, CMD(SKIP, .v.i = n, .opaque = current)); + queue_cmd(sink, CMD(SKIP, .v.i = n, .opaque = entry)); } void camu_sink_toggle_pause(struct camu_sink *sink) { nn_mutex_lock(&sink->lock); - struct camu_sink_entry *current = get_entry_for_command(sink); + struct camu_sink_entry *entry = get_entry_for_command(sink); + struct camu_sink_entry *current = (entry == sink->current) ? NULL : sink->current; + if (!entry) { + nn_mutex_unlock(&sink->lock); + return; + } + if (current) { + camu_clock_set_external_pause(¤t->clock); + } + camu_clock_set_external_pause(&entry->clock); + f64 pts = camu_clock_get_last_pts(&entry->clock); nn_mutex_unlock(&sink->lock); - if (!current) return; - bool armed_for_pause = false; - camu_clock_external_pause(¤t->clock); - f64 pts = camu_clock_get_pts(¤t->clock, 0.0, false, &armed_for_pause); - queue_cmd(sink, CMD(TOGGLE_PAUSE, .v.f = pts, .opaque = current)); + queue_cmd(sink, CMD(TOGGLE_PAUSE, .v.f = pts, .opaque = entry)); } void camu_sink_seek(struct camu_sink *sink, void *value, u8 mode) { nn_mutex_lock(&sink->lock); - struct camu_sink_entry *current = get_entry_for_command(sink); + struct camu_sink_entry *entry = get_entry_for_command(sink); u64 duration = 0; f64 pts = 0.0; - if (current) { - duration = current->client.duration; - pts = camu_clock_get_last_pts(¤t->clock); + if (entry) { + duration = entry->client.duration; + pts = camu_clock_get_last_pts(&entry->clock); } nn_mutex_unlock(&sink->lock); - if (!current || duration == 0) { + if (!entry || duration == 0) { return; } struct camu_sink_cmd cmd = { .op = SEEK, - .opaque = current + .opaque = entry }; switch (mode) { case CAMU_SEEK_POS: { @@ -1813,7 +2000,6 @@ void camu_sink_seek(struct camu_sink *sink, void *value, u8 mode) break; } case CAMU_SEEK_PERCENT: { - // There's probably a way to lose less precision here. f64 percent = *(f64 *)value; cmd.v.u = (u64)(duration * percent); break; @@ -1827,6 +2013,24 @@ void camu_sink_reseek(struct camu_sink *sink) queue_cmd(sink, CMD(RESEEK)); } +void camu_sink_change_audio_track(struct camu_sink *sink, s32 n) +{ + nn_mutex_lock(&sink->lock); + struct camu_sink_entry *entry = get_entry_for_command(sink); + nn_mutex_unlock(&sink->lock); + if (!entry) return; + queue_cmd(sink, CMD(AUDIO_TRACK, .v.i = n, .opaque = entry)); +} + +void camu_sink_change_subtitle_track(struct camu_sink *sink, s32 n) +{ + nn_mutex_lock(&sink->lock); + struct camu_sink_entry *entry = get_entry_for_command(sink); + nn_mutex_unlock(&sink->lock); + if (!entry) return; + queue_cmd(sink, CMD(SUBTITLE_TRACK, .v.i = n, .opaque = entry)); +} + void camu_sink_shuffle(struct camu_sink *sink) { queue_cmd(sink, CMD(SHUFFLE)); @@ -1836,7 +2040,21 @@ void camu_sink_status(struct camu_sink *sink, struct camu_osd *osd) { nn_mutex_lock(&sink->lock); struct camu_sink_entry *current = sink->current; - osd->connecting = !sink->connected; + osd->connecting = !sink->conn; + if (sink->notify_status & NOTIFY_DISCONNECTED) { + sink->notify_status &= ~NOTIFY_DISCONNECTED; + osd->show = true; + osd->shown_for |= CAMU_OSD_CONNECTING; + } + if (sink->notify_status & NOTIFY_CONNECTED) { + sink->notify_status &= ~NOTIFY_CONNECTED; + if (osd->shown_for & CAMU_OSD_CONNECTING) { + osd->shown_for &= ~CAMU_OSD_CONNECTING; + if (osd->shown_for == 0) { + osd->show = false; + } + } + } if (sink->notify_status & NOTIFY_EMPTY) { sink->notify_status &= ~NOTIFY_EMPTY; osd->show = true; @@ -1846,29 +2064,45 @@ void camu_sink_status(struct camu_sink *sink, struct camu_osd *osd) sink->notify_status &= ~NOTIFY_NOT_EMPTY; if (osd->shown_for & CAMU_OSD_EMPTY) { osd->shown_for &= ~CAMU_OSD_EMPTY; - osd->show = false; + if (osd->shown_for == 0) { + osd->show = false; + } } } if (sink->notify_status & NOTIFY_SEEK) { sink->notify_status &= ~NOTIFY_SEEK; if (!osd->show) { osd->flash = CAMU_OSD_FLASH_FOR(0.75); + osd->show = true; } } - nn_mutex_unlock(&sink->lock); + if (sink->notify_status & NOTIFY_ENTRY_ADDED) { + sink->notify_status &= ~NOTIFY_ENTRY_ADDED; + osd->no_force_render = false; + } + if (!osd->no_force_render) { + osd->no_force_render = sink->suspended; + } if (current) { - osd->paused = camu_clock_is_user_paused(¤t->clock); - bool armed_for_pause = false; - osd->pts = camu_clock_get_pts(¤t->clock, 0.0, false, &armed_for_pause); + u8 status = CAMU_CLOCK_NO_SIGNAL_PAUSE; + osd->pts = camu_clock_get_pts(¤t->clock, 0.0, false, &status); if (CAMU_PTS_CONSIDER_PAUSED(osd->pts)) { + osd->paused = !(status & CAMU_CLOCK_PAUSE_FOR_SWAP) && osd->pts != CAMU_PTS_UNSET; osd->pts = camu_clock_get_last_pts(¤t->clock); + } else { + osd->paused = !!(status & CAMU_CLOCK_EXTERNAL_PAUSE); + } + // Don't show paused on an image. + osd->paused &= VIDEO_EMPTY(current) || !(VIDEO_IS_STATIC(current) && AUDIO_EMPTY(current)); + if (current->client.duration != LIANA_TIMESTAMP_INVALID) { + osd->duration = current->client.duration; } - osd->duration = current->client.duration; } else { osd->paused = false; osd->pts = 0.0; osd->duration = 0; } + nn_mutex_unlock(&sink->lock); } void camu_sink_stop(struct camu_sink *sink) @@ -1883,25 +2117,30 @@ void camu_sink_stop(struct camu_sink *sink) void camu_sink_close(struct camu_sink *sink) { nn_timer_stop(&sink->reconnect_timer); - if (sink->conn) { - struct nn_rpc_connection *conn = sink->conn; - sink->conn = NULL; // Signal to connection_closed_callback() we're done. - nn_rpc_conn_disconnect(conn); + if (sink->conn || sink->connecting) { + sink->closed = true; // Signal to connection_closed_callback() we're done. + nn_rpc_disconnect(&sink->client); } nn_timer_stop(&sink->empty_timer); + array(struct camu_sink_entry *) cleanup; + al_array_init(cleanup); + al_array_copy(cleanup, sink->entries); + sink->entries.count = 0; struct camu_sink_entry *entry; - al_array_foreach_rev(sink->entries, i, entry) { - al_array_remove_at(sink->entries, i); + al_array_foreach(cleanup, i, entry) { maybe_disconnect_entry(entry); } + al_array_free(cleanup); } void camu_sink_free(struct camu_sink *sink) { + al_array_free(sink->node_prefs); al_assert(!sink->entries.count); al_array_free(sink->entries); - nn_rpc_free(&sink->client); + al_array_free(sink->previous); camu_queue_free(sink->queue); + nn_rpc_free(&sink->client); nn_mutex_destroy(&sink->lock); al_str_free(&sink->name); al_str_free(&sink->addr); diff --git a/src/libsink/sink.h b/src/libsink/sink.h index 6000b4f..5cc5a2a 100644 --- a/src/libsink/sink.h +++ b/src/libsink/sink.h @@ -57,7 +57,7 @@ struct camu_sink_entry { struct { u8 state; // Don't stop audio on a BUFFER_PAUSED from this entry. - bool ignore_paused; + bool ignore_pause; struct camu_audio_buffer buf; struct lia_vcr_track *track; } audio; @@ -81,11 +81,12 @@ struct camu_sink { u8 type; str addr; u16 port; + bool local; struct nn_rpc client; struct nn_rpc_connection *conn; - bool connected; + bool closed; + bool connecting; u16 connection_number; - bool local; struct nn_mutex lock; struct nn_timer reconnect_timer; struct nn_signal queue_signal; @@ -108,6 +109,7 @@ struct camu_sink { struct camu_renderer *renderer; } video; struct lia_prefs prefs; + array(struct lia_prefs) node_prefs; str default_list; u8 (*callback)(void *, u8, u8, void *); void *userdata; @@ -123,6 +125,8 @@ void camu_sink_skip(struct camu_sink *sink, s32 n); void camu_sink_toggle_pause(struct camu_sink *sink); void camu_sink_seek(struct camu_sink *sink, void *value, u8 mode); void camu_sink_reseek(struct camu_sink *sink); +void camu_sink_change_audio_track(struct camu_sink *sink, s32 n); +void camu_sink_change_subtitle_track(struct camu_sink *sink, s32 n); void camu_sink_shuffle(struct camu_sink *sink); void camu_sink_status(struct camu_sink *sink, struct camu_osd *osd); void camu_sink_stop(struct camu_sink *sink); -- cgit v1.2.3-101-g0448