diff options
Diffstat (limited to 'src/liana')
| -rw-r--r-- | src/liana/client.c | 39 | ||||
| -rw-r--r-- | src/liana/client.h | 6 | ||||
| -rw-r--r-- | src/liana/handler.h | 1 | ||||
| -rw-r--r-- | src/liana/list.c | 138 | ||||
| -rw-r--r-- | src/liana/list.h | 2 | ||||
| -rw-r--r-- | src/liana/server.c | 16 | ||||
| -rw-r--r-- | src/liana/server.h | 6 |
7 files changed, 144 insertions, 64 deletions
diff --git a/src/liana/client.c b/src/liana/client.c index 1454ba0..a841f7a 100644 --- a/src/liana/client.c +++ b/src/liana/client.c @@ -1,3 +1,5 @@ +#include <al/log.h> + #include "../server/common.h" #ifdef CAMU_HAVE_FFMPEG #include "../codec/ffmpeg/packet_ext.h" @@ -121,7 +123,7 @@ static void parse_info_packet(struct lia_client *client, struct aki_packet *pack static void info_packet_callback(void *userdata, struct aki_packet_stream *stream, struct aki_packet *packet) { struct lia_client *client = (struct lia_client *)userdata; - client->connection_id = aki_packet_read_u16(packet); + client->connection_id = aki_packet_read_u32(packet); parse_info_packet(client, packet); aki_packet_free(packet); if (client->mask == 0 || lia_vcr_is_empty(&client->vcr)) { @@ -144,11 +146,27 @@ static void packet_sent_callback(void *userdata, struct aki_packet *packet) static void connection_callback(void *userdata, struct aki_packet_stream *stream) { struct lia_client *client = (struct lia_client *)userdata; + if (client->reconnect) { + // Even if seek() was called before the initial connection_callback(), + // we still want to call RESUME_AT here. + struct lia_timing time = { + .at = client->at, + .seek_pos = client->pos, + .pause = LIANA_PAUSE_NONE + }; + client->callback(client->userdata, LIANA_CLIENT_RESUME_AT, NULL, &time); + if (client->mask == 0) { + al_log_warn("liana", "Handling reconnect on unconfigured client."); + } else { + client->callback(client->userdata, LIANA_CLIENT_RECONNECTED, NULL, NULL); + } + client->reconnect = false; + } lia_vcr_start(&client->vcr); stream->packet_sent_callback = packet_sent_callback; struct aki_packet *packet = aki_packet_create(); - aki_packet_write_u16(packet, client->id); - aki_packet_write_u16(packet, 0); + aki_packet_write_u32(packet, client->id); + aki_packet_write_u32(packet, 0); aki_packet_write_s32(packet, client->mask); aki_packet_write_u64(packet, client->pos); if (client->mask == 0) { @@ -172,13 +190,6 @@ static void connection_closed_callback(void *userdata, struct aki_packet_stream // any unexpected behavior. client->callback(client->userdata, LIANA_CLIENT_REMOVE_BUFFERS, NULL, &client->reconnect); if (client->reconnect) { - client->reconnect = false; - struct lia_timing time = { - .at = client->at, - .seek_pos = client->pos, - .pause = LIANA_PAUSE_NONE - }; - client->callback(client->userdata, LIANA_CLIENT_RESUME_AT, NULL, &time); aki_packet_stream_reconnect(stream, &client->addr, client->port); } else { client->callback(client->userdata, LIANA_CLIENT_CLOSED, NULL, NULL); @@ -186,7 +197,7 @@ static void connection_closed_callback(void *userdata, struct aki_packet_stream } void lia_client_connect(struct lia_client *client, struct aki_event_loop *loop, u8 type, - str *addr, u16 port, u16 id, u64 pos, struct camu_renderer *renderer) + str *addr, u16 port, u32 id, u64 pos, struct camu_renderer *renderer) { client->loop = loop; client->id = id; @@ -211,10 +222,12 @@ void lia_client_set_renderer(struct lia_client *client, struct camu_renderer *re void lia_client_seek(struct lia_client *client, u64 pos, u64 at) { - client->reconnect = true; client->pos = pos; client->at = at; - aki_packet_stream_disconnect(&client->data); + if (!client->reconnect) { + client->reconnect = true; + aki_packet_stream_disconnect(&client->data); + } } void lia_client_reseek(struct lia_client *client) diff --git a/src/liana/client.h b/src/liana/client.h index 24526af..cd1297e 100644 --- a/src/liana/client.h +++ b/src/liana/client.h @@ -8,14 +8,14 @@ struct lia_client { struct aki_event_loop *loop; - u16 id; + u32 id; s32 mask; u64 pos; u64 at; bool reconnect; str addr; u16 port; - u16 connection_id; + u32 connection_id; struct aki_packet_stream data; u64 duration; struct lia_vcr vcr; @@ -25,7 +25,7 @@ struct lia_client { }; void lia_client_connect(struct lia_client *client, struct aki_event_loop *loop, u8 type, - str *addr, u16 port, u16 id, u64 pos, struct camu_renderer *renderer); + str *addr, u16 port, u32 id, u64 pos, struct camu_renderer *renderer); void lia_client_seek(struct lia_client *client, u64 pos, u64 at); void lia_client_reseek(struct lia_client *client); void lia_client_disconnect(struct lia_client *client); diff --git a/src/liana/handler.h b/src/liana/handler.h index 84aa1de..21437c4 100644 --- a/src/liana/handler.h +++ b/src/liana/handler.h @@ -29,6 +29,7 @@ enum { LIANA_CLIENT_SUBTITLE, LIANA_CLIENT_REMOVE_BUFFERS, LIANA_CLIENT_RESUME_AT, + LIANA_CLIENT_RECONNECTED, LIANA_CLIENT_EOF, LIANA_CLIENT_CLOSED }; diff --git a/src/liana/list.c b/src/liana/list.c index 059cd07..23eabc3 100644 --- a/src/liana/list.c +++ b/src/liana/list.c @@ -10,13 +10,19 @@ static void buffer_ahead(struct lia_list *list) { s32 size = (s32)list->entries.size; - if (list->queued >= 0 && list->queued + 1 < size) { - s32 ahead = list->queued + 1; + if (list->current >= 0 && list->current + 1 < size) { + s32 ahead = list->current + 1; for (s32 i = ahead; i < MIN(ahead + LIANA_BUFFER_AHEAD, size); i++) { - struct lia_list_entry *buffered = al_array_at(list->entries, i); + struct lia_list_entry *entry = al_array_at(list->entries, i); + struct lia_timing time = { + .at = LIANA_TIMESTAMP_INVALID, + .seek_pos = entry->offset, + .pause = LIANA_PAUSE_NONE, + .ended = false + }; struct lia_list_sink *sink; al_array_foreach(list->sinks, i, sink) { - sink->callback(sink->userdata, LIANA_SINK_BUFFER, buffered, i, LIANA_TIMESTAMP_INVALID); + sink->callback(sink->userdata, LIANA_SINK_BUFFER, entry, i, &time); } } } @@ -46,6 +52,7 @@ void lia_list_init(struct lia_list *list, str *name) list->previous = -1; list->queued = -1; list->idle = true; + list->increment = 0; al_array_init(list->entries); al_array_init(list->sinks); al_array_init(list->queue); @@ -56,6 +63,7 @@ static void pump_queue(struct lia_list *list); static bool assume_ended(struct lia_list_entry *entry, u64 at) { + if (entry->duration == LIANA_TIMESTAMP_INVALID) return false; if (entry->ended || entry->duration == 0) return true; if (entry->paused_at == LIANA_TIMESTAMP_INVALID && entry->start != LIANA_TIMESTAMP_INVALID) { if (entry->offset > entry->duration) { @@ -92,7 +100,13 @@ static bool handle_add_sink(struct lia_list *list, struct lia_list_sink *sink) u8 pause; u64 at = LIANA_TIMESTAMP_INVALID; u64 seek_pos = current->offset; - if (current->paused_at == LIANA_TIMESTAMP_INVALID) { + bool ended = assume_ended(current, now); + if (ended && current->duration != LIANA_TIMESTAMP_INVALID) { + seek_pos = current->duration; + } + if (ended || current->paused_at != LIANA_TIMESTAMP_INVALID) { + pause = LIANA_PAUSE_NONE; + } else { at = now + LIANA_BASE_DELAY; if (at > current->start && at - current->start > LIANA_BASE_DELAY) { seek_pos += at - current->start; @@ -100,14 +114,12 @@ static bool handle_add_sink(struct lia_list *list, struct lia_list_sink *sink) at = current->start; } pause = LIANA_PAUSE_RESUME; - } else { - pause = LIANA_PAUSE_NONE; } struct lia_timing time = { .at = at, .seek_pos = seek_pos, .pause = pause, - .ended = assume_ended(current, now) + .ended = ended }; sink->callback(sink->userdata, LIANA_SINK_SET, current, list->current, &time); } else { @@ -236,6 +248,7 @@ static bool handle_skipto(struct lia_list *list, s32 sequence, s32 index) { if (index == list->current) return true; if (sequence == LIANA_SEQUENCE_ANY) sequence = list->current; + if (sequence < 0) return true; if (sequence != list->current) { // Skipping from an entry other than current is not handled and will cause very // confusing errors. On top of likely resulting in unexpected behavior. @@ -244,7 +257,8 @@ static bool handle_skipto(struct lia_list *list, s32 sequence, s32 index) struct lia_list_entry *current = get_entry_from_sequence(list, sequence); struct lia_list_entry *target = get_entry_from_sequence(list, index); - if (!current || !target) return true; + al_assert(current && !current->held); + if (!target) return true; al_assert(current != target); if (!entry_load_and_get_duration(list, target)) { return false; @@ -258,23 +272,30 @@ static bool handle_skipto(struct lia_list *list, s32 sequence, s32 index) if (target->held) { al_assert(target->paused_at != LIANA_TIMESTAMP_INVALID); target->paused_at = LIANA_TIMESTAMP_INVALID; + target->start = at; target->held = false; + } else if (target->start == LIANA_TIMESTAMP_INVALID && target->paused_at == LIANA_TIMESTAMP_INVALID) { + // Target was never started before. + target->start = at; + } else { + // An ended entry may never have been paused, but a non-ended entry that wasn't set + // cannot be unpaused. Checking assume_ended(target) should be safe here as long + // as it can't go from true to false (consideration for seek?). + al_assert(assume_ended(target, now) || target->paused_at != LIANA_TIMESTAMP_INVALID); } - al_assert(!current->held); - // These are not equivalent to current/target->ended. bool current_ended = assume_ended(current, now); bool target_ended = assume_ended(target, now); - // Current is ended or paused. + // If current is ended or paused. if (current_ended || current->paused_at != LIANA_TIMESTAMP_INVALID) { if (target_ended || target->paused_at != LIANA_TIMESTAMP_INVALID) { // Swap entries and ignore their clocks. pause = LIANA_PAUSE_NONE; } else { // Swap entries and resume target. - target->start = at; + al_assert(target->start != LIANA_TIMESTAMP_INVALID); pause = LIANA_PAUSE_RESUME; } } else { // Current is playing. @@ -283,7 +304,7 @@ static bool handle_skipto(struct lia_list *list, s32 sequence, s32 index) pause = LIANA_PAUSE_PAUSE; } else { // Pause-swap current and resume target. - target->start = at; + al_assert(target->start != LIANA_TIMESTAMP_INVALID); pause = LIANA_PAUSE_BOTH; } } @@ -292,12 +313,16 @@ static bool handle_skipto(struct lia_list *list, s32 sequence, s32 index) // if it becomes the target of a skip. if (!current_ended && current->paused_at == LIANA_TIMESTAMP_INVALID) { current->paused_at = at; + if (current->paused_at < current->start) { + current->paused_at = current->start; + } current->offset += current->paused_at - current->start; current->start = LIANA_TIMESTAMP_INVALID; current->held = true; } - al_log_info("list", "pause: %u, held: %s.", pause, BOOLSTR(current->held)); + al_log_debug("list", "skipto [%u-%u]: pause: %hhu, held: %s, current_ended: %s, target_ended: %s.", + current->id, target->id, pause, BOOLSTR(current->held), BOOLSTR(current_ended), BOOLSTR(target_ended)); list->current = index; list->previous = sequence; @@ -334,30 +359,42 @@ static bool handle_skip(struct lia_list *list, s32 sequence, s32 n) static void handle_toggle_pause(struct lia_list *list, s32 sequence, f64 pts) { + // pts should be treated as a hint. if (sequence == LIANA_SEQUENCE_ANY) sequence = list->current; - struct lia_list_entry *current = get_entry_from_sequence(list, sequence); - if (!current) return; + if (sequence < 0) return; + if (sequence != list->current) { + // A non-current entry should always be paused. + return; + } + struct lia_list_entry *entry = get_entry_from_sequence(list, sequence); + al_assert(entry && !entry->held); u64 now = aki_get_timestamp(); - u8 pause = current->paused_at == LIANA_TIMESTAMP_INVALID ? LIANA_PAUSE_PAUSE : LIANA_PAUSE_RESUME; + u8 pause = entry->paused_at == LIANA_TIMESTAMP_INVALID ? LIANA_PAUSE_PAUSE : LIANA_PAUSE_RESUME; u64 at; + bool ended = assume_ended(entry, now); + if (ended) return; switch (pause) { case LIANA_PAUSE_PAUSE: - // This assert should exist but the correct behavior for this is unfinished. - //al_assert(pts != -1.0); - (void)pts; - current->paused_at = now + LIANA_PAUSE_DELAY; - current->offset += current->paused_at - current->start; - current->start = LIANA_TIMESTAMP_INVALID; - at = current->paused_at; + al_assert(entry->start != LIANA_TIMESTAMP_INVALID); + entry->paused_at = now + LIANA_PAUSE_DELAY; + if (entry->paused_at < entry->start) { + entry->paused_at = entry->start; + } + entry->offset += entry->paused_at - entry->start; + entry->start = LIANA_TIMESTAMP_INVALID; + at = entry->paused_at; break; case LIANA_PAUSE_RESUME: - current->paused_at = LIANA_TIMESTAMP_INVALID; - current->start = now + LIANA_PAUSE_DELAY; - at = current->start; + al_assert(entry->start == LIANA_TIMESTAMP_INVALID); + entry->paused_at = LIANA_TIMESTAMP_INVALID; + entry->start = now + LIANA_PAUSE_DELAY; + at = entry->start; break; } + al_log_debug("list", "toggle_pause [%u]: pts: %f, pause: %hhu.", entry->id, pts, pause); + struct lia_timing time = { .at = at, .seek_pos = LIANA_TIMESTAMP_INVALID, @@ -367,30 +404,49 @@ static void handle_toggle_pause(struct lia_list *list, s32 sequence, f64 pts) struct lia_list_sink *sink; al_array_foreach(list->sinks, i, sink) { - sink->callback(sink->userdata, LIANA_SINK_PAUSE, current, sequence, &time); + sink->callback(sink->userdata, LIANA_SINK_PAUSE, entry, sequence, &time); } } +// TODO: Take entry id instead of sequence for seek() and end()? + static void handle_seek(struct lia_list *list, s32 sequence, f64 percent) { if (sequence == LIANA_SEQUENCE_ANY) sequence = list->current; if (sequence < 0) return; - list->idle = false; - struct lia_list_entry *current = get_entry_from_sequence(list, sequence); - u64 pos = (u64)(current->duration * percent); + if (sequence != list->current) { + return; + } + struct lia_list_entry *entry = get_entry_from_sequence(list, sequence); + al_assert(entry); + if (entry->duration == LIANA_TIMESTAMP_INVALID) { + // Until any kind of live resource buffering. + return; + } + entry->ended = false; + u64 now = aki_get_timestamp(); - current->offset = pos; - current->start = now + LIANA_BASE_DELAY; - current->ended = false; + u64 pos = (u64)(entry->duration * percent); + u64 at = now + LIANA_BASE_DELAY; + entry->offset = pos; + if (entry->paused_at == LIANA_TIMESTAMP_INVALID) { + entry->start = at; + } + + list->idle = false; + + al_log_debug("list", "seek [%u]: pos: %f.", entry->id, pos / 1000000.0); + struct lia_timing time = { - .at = current->start, - .seek_pos = current->offset, + .at = at, + .seek_pos = pos, .pause = LIANA_PAUSE_NONE, .ended = false }; + struct lia_list_sink *sink; al_array_foreach(list->sinks, i, sink) { - sink->callback(sink->userdata, LIANA_SINK_SEEK, current, sequence, &time); + sink->callback(sink->userdata, LIANA_SINK_SEEK, entry, sequence, &time); } } @@ -400,11 +456,13 @@ static bool handle_end(struct lia_list *list, s32 sequence) s32 size = (s32)list->entries.size; struct lia_list_entry *entry = al_array_at(list->entries, sequence); + al_assert(entry); if (entry->ended) { al_log_warn("list", "Got end() from an already ended resource, ignoring."); return true; } entry->ended = true; + // TODO: Calculate duration for live resources? entry->offset = entry->duration; if (sequence == list->current) { @@ -510,11 +568,13 @@ void pump_queue(struct lia_list *list) break; case SKIPTO: if (!handle_skipto(list, cmd->sequence, cmd->i)) { + // Target entry not loaded. return; } break; case SKIP: if (!handle_skip(list, cmd->sequence, cmd->i)) { + // Converted to skipto and entry not loaded. return; } break; @@ -576,6 +636,8 @@ void lia_list_add(struct lia_list *list, void *opaque, u64 duration, wstr *name) { struct lia_list_entry *entry = al_alloc_object(struct lia_list_entry); entry->opaque = opaque; + entry->id = list->increment; + list->increment = al_u32_inc_wrap(list->increment); entry->paused_at = LIANA_TIMESTAMP_INVALID; entry->held = false; entry->offset = 0; diff --git a/src/liana/list.h b/src/liana/list.h index 6443db6..90f8c02 100644 --- a/src/liana/list.h +++ b/src/liana/list.h @@ -51,6 +51,7 @@ struct lia_timing { struct lia_list_entry { void *opaque; + u32 id; u64 start; u64 paused_at; bool held; @@ -84,6 +85,7 @@ struct lia_list { s32 previous; s32 queued; bool idle; + u32 increment; array(struct lia_list_entry *) entries; array(struct lia_list_sink *) sinks; array(struct lia_list_cmd *) queue; diff --git a/src/liana/server.c b/src/liana/server.c index e03b8ef..44b3f12 100644 --- a/src/liana/server.c +++ b/src/liana/server.c @@ -3,6 +3,7 @@ #include "server.h" #include "handler.h" #include "handlers.h" +#include "list.h" bool lia_server_init(struct lia_server *server, struct aki_event_loop *loop) { @@ -124,7 +125,7 @@ static void handle_connection(struct lia_node_connection *conn, struct aki_packe if (mask == 0) { struct aki_packet *rpacket = aki_packet_create(); - aki_packet_write_u16(rpacket, conn->id); + aki_packet_write_u32(rpacket, conn->id); aki_packet_write_str(rpacket, cch_entry_get_liana(conn->node->entry)); conn->handler->write_info(conn->handler, rpacket); stream->packet_callback = subscribe_packet_callback; @@ -164,7 +165,7 @@ static void signal_callback(void *userdata) } if (!conn->errored) { conn->id = server->increment; - server->increment = al_u16_inc_wrap(server->increment); + server->increment = al_u32_inc_wrap(server->increment); aki_packet_pool_init(&conn->pool, 96, server->loop, packet_pool_callback, conn); al_array_push(node->connections, conn); handle_connection(conn, packet); @@ -192,7 +193,7 @@ static aki_thread_result AKI_THREADCALL init_thread(void *userdata) return 0; } -static struct lia_node_connection *get_connection_from_id(struct lia_node *node, u16 id) +static struct lia_node_connection *get_connection_from_id(struct lia_node *node, u32 id) { struct lia_node_connection *conn = NULL; al_array_foreach(node->connections, i, conn) { @@ -201,7 +202,7 @@ static struct lia_node_connection *get_connection_from_id(struct lia_node *node, return NULL; } -static struct lia_node *get_node_from_id(struct lia_server *server, u16 id) +static struct lia_node *get_node_from_id(struct lia_server *server, u32 id) { struct lia_node *node; al_array_foreach(server->nodes, i, node) { @@ -226,8 +227,8 @@ static void packet_callback(void *userdata, struct aki_packet_stream *stream, st // We got a packet, this connection is no longer a zombie. remove_zombie(server, stream); - u16 node_id = aki_packet_read_u16(packet); - u16 connection_id = aki_packet_read_u16(packet); + u32 node_id = aki_packet_read_u32(packet); + u32 connection_id = aki_packet_read_u32(packet); struct lia_node *node = get_node_from_id(server, node_id); struct lia_node_connection *conn = NULL; @@ -283,7 +284,7 @@ struct lia_node *lia_server_create_node(struct lia_server *server, struct cch_en { struct lia_node *node = al_alloc_object(struct lia_node); node->id = server->increment; - server->increment = al_u16_inc_wrap(server->increment); + server->increment = al_u32_inc_wrap(server->increment); node->entry = entry; al_array_init(node->connections); node->server = server; @@ -296,6 +297,7 @@ static aki_thread_result AKI_THREADCALL init_duration_thread(void *userdata) struct lia_node *node = (struct lia_node *)userdata; if (!node->handler->init(node->handler, &node->handle)) { node->errored = true; + node->duration = LIANA_TIMESTAMP_INVALID; } else { node->duration = node->handler->get_duration(node->handler); } diff --git a/src/liana/server.h b/src/liana/server.h index 11fdfa9..c62eb29 100644 --- a/src/liana/server.h +++ b/src/liana/server.h @@ -9,7 +9,7 @@ #include "../cache/entry.h" struct lia_node_connection { - u16 id; + u32 id; struct aki_packet *packet; struct aki_packet_stream *stream; struct lia_server_handler *handler; @@ -26,7 +26,7 @@ enum { }; struct lia_node { - u16 id; + u32 id; struct cch_entry *entry; array(struct lia_node_connection *) connections; struct lia_server *server; @@ -44,7 +44,7 @@ struct lia_node { struct lia_server { struct aki_event_loop *loop; - u16 increment; + u32 increment; array(struct lia_node *) nodes; array(struct aki_packet_stream *) zombies; }; |