diff options
Diffstat (limited to 'src/libsink/sink.c')
| -rw-r--r-- | src/libsink/sink.c | 148 |
1 files changed, 131 insertions, 17 deletions
diff --git a/src/libsink/sink.c b/src/libsink/sink.c index 783f242..5f61bd4 100644 --- a/src/libsink/sink.c +++ b/src/libsink/sink.c @@ -3,6 +3,7 @@ #include <al/log.h> #include <al/random.h> #include <nnwt/multiplex.h> +#include <nnwt/time.h> #include "../server/common.h" #include "../buffer/common.h" @@ -49,6 +50,7 @@ enum { EJECT_ENTRY, CLOSE, // List actions. + ADD, SKIP, TOGGLE_PAUSE, SEEK, @@ -57,6 +59,13 @@ enum { END }; +// Status reporting. +enum { + NOTIFY_EMPTY = 1, + NOTIFY_NOT_EMPTY = 1 << 1, + NOTIFY_SEEK = 1 << 2 +}; + // Number of entries to keep buffered at one time. #define ENTRY_MAX_AGE 4 #define SINK_LRU_MAX UINT16_MAX @@ -375,6 +384,17 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) sink->callback(sink->userdata, CAMU_SINK_EXIT, 0, NULL); return; } + case ADD: { + if (!sink->connected) 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); + nn_packet_write_str(packet, (str *)cmd->v.p); + al_str_free((str *)cmd->v.p); + al_free(cmd->v.p); + nn_rpc_connection_command(sink->conn, packet, NULL, NULL); + break; + } case SKIP: { if (!sink->connected) return; struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque; @@ -719,12 +739,18 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target) if (stop_video) { queue_cmd(sink, CMD(STOP, .v.u = CAMU_SINK_VIDEO)); - if (dangling_target || VIDEO_ENDED_OR_EMPTY(target)) { + // Refresh on VIDEO_ENDED_OR_EMPTY(target) should be covered by after_add_entry(). + if (dangling_target) { refresh_video_output(sink); } } sink->current = dangling_target ? NULL : target; + + if (sink->current) { + sink->notify_status |= NOTIFY_NOT_EMPTY; + nn_timer_again(&sink->empty_timer); + } } static void pause_and_swap_to(struct camu_sink *sink, struct camu_sink_entry *target, u64 at) @@ -1330,6 +1356,15 @@ 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) { @@ -1339,11 +1374,7 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn, u8 op = nn_packet_read_u8(packet); if (op == LIANA_SINK_UNSET) { nn_mutex_lock(&sink->lock); - struct camu_sink_entry *current = sink->current; - nn_mutex_unlock(&sink->lock); - if (current) { - maybe_disconnect_entry(current); - } + unset_current(sink); nn_mutex_lock(&sink->lock); goto out; } @@ -1378,6 +1409,13 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn, // 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; if (op == LIANA_SINK_BUFFER) { @@ -1494,24 +1532,30 @@ static bool pause_command_callback(void *userdata, struct nn_rpc_connection *con switch (pause) { case LIANA_PAUSE_PAUSE: { entry->paused = true; - camu_clock_pause(&entry->clock, at); + if (camu_clock_pause(&entry->clock, at)) { + if (entry == sink->current) { + queue_cmd(sink, CMD(STOP, .v.u = CAMU_SINK_VIDEO)); + } + } log_info("Clock paused."); // Audio will be stopped in a BUFFER_PAUSED callback. - // Video will be stopped in a CLOCK_PAUSED callback. + // Video will be stopped above or in a CLOCK_PAUSED callback. break; } case LIANA_PAUSE_RESUME: { entry->paused = false; camu_clock_resume(&entry->clock, at); log_info("Clock resumed."); - if (!VIDEO_ENDED_OR_EMPTY(entry) && !VIDEO_IS_STATIC(entry)) { - queue_cmd(sink, CMD(START, .v.u = CAMU_SINK_VIDEO)); - } - if (!AUDIO_ENDED_OR_EMPTY(entry)) { - if (!sink->local) { - camu_audio_buffer_resync(&entry->audio.buf); + if (entry == sink->current) { + if (!VIDEO_ENDED_OR_EMPTY(entry) && !VIDEO_IS_STATIC(entry)) { + queue_cmd(sink, CMD(START, .v.u = CAMU_SINK_VIDEO)); + } + if (!AUDIO_ENDED_OR_EMPTY(entry)) { + if (!sink->local) { + camu_audio_buffer_resync(&entry->audio.buf); + } + queue_cmd(sink, CMD(START, .v.u = CAMU_SINK_AUDIO)); } - queue_cmd(sink, CMD(START, .v.u = CAMU_SINK_AUDIO)); } break; } @@ -1542,6 +1586,7 @@ static bool seek_command_callback(void *userdata, struct nn_rpc_connection *conn entry->sequence = sequence; 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; 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); @@ -1589,6 +1634,7 @@ static void connection_callback(void *userdata, struct nn_rpc_connection *conn) if (sink->conn) al_assert(sink->conn == conn); sink->conn = conn; sink->connected = true; + refresh_video_output(sink); // For OSD. sink->connection_number = al_u16_inc_wrap(sink->connection_number); if (sink->connection_number == 0) sink->connection_number = 1; identify_on_connection(sink); @@ -1608,6 +1654,9 @@ static void reconnect_timer_callback(void *userdata, struct nn_timer *timer) 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; if (sink->conn) { @@ -1628,6 +1677,18 @@ static void connection_closed_callback(void *userdata, struct nn_rpc_connection } } +static void empty_timer_callback(void *userdata, struct nn_timer *timer) +{ + struct camu_sink *sink = (struct camu_sink *)userdata; + (void)timer; + if (!sink->current) { + sink->notify_status = NOTIFY_EMPTY; + refresh_video_output(sink); + nn_timer_stop(&sink->empty_timer); + } + nn_timer_set_repeat(&sink->empty_timer, NNWT_TS_FROM_USEC(1000000)); +} + bool camu_sink_init(struct camu_sink *sink, struct nn_event_loop *loop, struct camu_mixer *mixer, struct camu_renderer *renderer) { @@ -1647,6 +1708,10 @@ bool camu_sink_init(struct camu_sink *sink, struct nn_event_loop *loop, sink->suspended = NULL; 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_again(&sink->empty_timer); + sink->notify_status = 0; // Start high to exercise the wrapping path. sink->lru = SINK_LRU_MAX - al_random_int(0, ENTRY_MAX_AGE); mixer->callback = mixer_callback; @@ -1690,6 +1755,13 @@ void camu_sink_return_current(struct camu_sink *sink) nn_mutex_unlock(&sink->lock); } +void camu_sink_add(struct camu_sink *sink, str *path) +{ + str *copy = al_alloc_object(str); + al_str_clone(copy, path); + queue_cmd(sink, CMD(ADD, .v.p = copy)); +} + void camu_sink_skip(struct camu_sink *sink, s32 n) { nn_mutex_lock(&sink->lock); @@ -1721,7 +1793,9 @@ void camu_sink_seek(struct camu_sink *sink, void *value, u8 mode) pts = camu_clock_get_last_pts(¤t->clock); } nn_mutex_unlock(&sink->lock); - if (!current || duration == 0) return; + if (!current || duration == 0) { + return; + } struct camu_sink_cmd cmd = { .op = SEEK, .opaque = current @@ -1735,7 +1809,7 @@ void camu_sink_seek(struct camu_sink *sink, void *value, u8 mode) case CAMU_SEEK_RELATIVE: { f64 offset = *(f64 *)value; pts = MAX(pts + offset, 0.0); - cmd.v.u = (u64)(pts * 1000000); + cmd.v.u = (u64)(pts * 1000000.0); break; } case CAMU_SEEK_PERCENT: { @@ -1758,6 +1832,45 @@ void camu_sink_shuffle(struct camu_sink *sink) queue_cmd(sink, CMD(SHUFFLE)); } +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; + if (sink->notify_status & NOTIFY_EMPTY) { + sink->notify_status &= ~NOTIFY_EMPTY; + osd->show = true; + osd->shown_for |= CAMU_OSD_EMPTY; + } + if (sink->notify_status & NOTIFY_NOT_EMPTY) { + sink->notify_status &= ~NOTIFY_NOT_EMPTY; + if (osd->shown_for & CAMU_OSD_EMPTY) { + osd->shown_for &= ~CAMU_OSD_EMPTY; + 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); + } + } + nn_mutex_unlock(&sink->lock); + 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); + if (CAMU_PTS_CONSIDER_PAUSED(osd->pts)) { + osd->pts = camu_clock_get_last_pts(¤t->clock); + } + osd->duration = current->client.duration; + } else { + osd->paused = false; + osd->pts = 0.0; + osd->duration = 0; + } +} + void camu_sink_stop(struct camu_sink *sink) { queue_cmds(sink, 3, @@ -1775,6 +1888,7 @@ void camu_sink_close(struct camu_sink *sink) sink->conn = NULL; // Signal to connection_closed_callback() we're done. nn_rpc_conn_disconnect(conn); } + nn_timer_stop(&sink->empty_timer); struct camu_sink_entry *entry; al_array_foreach_rev(sink->entries, i, entry) { al_array_remove_at(sink->entries, i); |