summaryrefslogtreecommitdiff
path: root/src/shrub
diff options
context:
space:
mode:
Diffstat (limited to 'src/shrub')
-rw-r--r--src/shrub/client.c172
-rw-r--r--src/shrub/client.h3
-rw-r--r--src/shrub/common.h1
-rw-r--r--src/shrub/handler.h7
-rw-r--r--src/shrub/server.c192
-rw-r--r--src/shrub/server.h34
-rw-r--r--src/shrub/vcr.c4
7 files changed, 189 insertions, 224 deletions
diff --git a/src/shrub/client.c b/src/shrub/client.c
index 015f47f..abe2d23 100644
--- a/src/shrub/client.c
+++ b/src/shrub/client.c
@@ -1,6 +1,7 @@
#include <al/log.h>
#include "../codec/ffmpeg/packet_ext.h"
+#include "../server/common.h"
#include "handlers/codec.h"
@@ -8,18 +9,6 @@
#include "handler.h"
#include "common.h"
-static void data_connection_callback(void *userdata, struct aki_packet_stream *stream)
-{
- struct shrb_client *client = (struct shrb_client *)userdata;
- (void)stream;
- struct aki_packet *packet = aki_packet_create();
- aki_packet_write_u8(packet, SHRUB_CONNECTION_DATA);
- aki_packet_write_u16(packet, client->connection_id);
- aki_packet_write_u16(packet, client->node_id);
- aki_packet_write_u64(packet, client->seek_pos);
- aki_packet_stream_send_packet(&client->data, packet);
-}
-
static void send_reconnect_packet(struct shrb_client *client)
{
struct aki_packet *packet = aki_packet_create();
@@ -33,86 +22,55 @@ static void cleanup_connection_internal(struct shrb_client *client)
client->callback(client->userdata, SHRUB_CLIENT_CLOSED, NULL, NULL);
}
-static void data_packet_callback(void *userdata, struct aki_packet_stream *stream,
- struct aki_packet *packet)
-{
- struct shrb_client *client = (struct shrb_client *)userdata;
- (void)stream;
- shrb_vcr_push_packet(&client->vcr, packet);
-}
-
static void packet_sent_callback(void *userdata, struct aki_packet *packet)
{
(void)userdata;
aki_packet_free(packet);
}
-static void data_connection_closed_callback(void *userdata, struct aki_packet_stream *stream);
-
-void connect_data(struct shrb_client *client)
-{
- if (!client->reconnect) {
- client->reconnect = true;
- } else {
- aki_packet_stream_free(&client->data);
- }
- aki_packet_stream_init(&client->data, AKI_SOCKET_TCP, data_connection_callback,
- data_connection_closed_callback, data_packet_callback, packet_sent_callback, client);
- aki_packet_stream_connect(&client->data, client->loop, &client->addr, client->port);
-}
-
-static void parse_pause_packet(struct shrb_client *client, struct aki_packet *packet)
-{
- u64 delay = aki_packet_read_u64(packet);
- u64 seek_pos = aki_packet_read_u64(packet);
- struct shrb_seek_req req = { .base = seek_pos, .ts = UINT64_MAX, .delay = delay };
- // TODO: separare function.
- client->callback(client->userdata, SHRUB_CLIENT_SET, NULL, &req);
- client->seek_pos = req.base;
-}
-
-void data_connection_closed_callback(void *userdata, struct aki_packet_stream *stream)
+static void data_packet_callback(void *userdata, struct aki_packet_stream *stream,
+ struct aki_packet *packet)
{
struct shrb_client *client = (struct shrb_client *)userdata;
(void)stream;
- if (client->closed) {
- if (!client->control.connected) cleanup_connection_internal(client);
- return;
- }
- client->callback(client->userdata, SHRUB_CLIENT_REMOVE_BUFFERS, NULL, NULL);
- shrb_vcr_flush(&client->vcr);
- if (!client->paused) {
- send_reconnect_packet(client);
- } else {
- parse_pause_packet(client, client->paused);
- aki_packet_free(client->paused);
- client->paused = NULL;
- client->callback(client->userdata, SHRUB_CLIENT_PAUSE, NULL, NULL);
- connect_data(client);
- }
+ shrb_vcr_push_packet(&client->vcr, packet);
}
-static void control_connection_callback(void *userdata, struct aki_packet_stream *stream)
+static void data_connection_callback(void *userdata, struct aki_packet_stream *stream)
{
struct shrb_client *client = (struct shrb_client *)userdata;
(void)stream;
+ client->data.packet_callback = data_packet_callback;
+ client->data.packet_sent_callback = packet_sent_callback;
struct aki_packet *packet = aki_packet_create();
- aki_packet_write_u8(packet, SHRUB_CONNECTION_CONTROL);
- aki_packet_write_u16(packet, 0);
+ aki_packet_write_u8(packet, SHRUB_CONNECTION_DATA);
+ aki_packet_write_u16(packet, client->connection_id);
aki_packet_write_u16(packet, client->node_id);
- aki_packet_stream_send_packet(&client->control, packet);
+ aki_packet_write_u64(packet, client->seek_pos);
+ aki_packet_stream_send_packet(&client->data, packet);
}
-static void control_connection_closed_callback(void *userdata, struct aki_packet_stream *stream)
+static void data_connection_closed_callback(void *userdata, struct aki_packet_stream *stream)
{
struct shrb_client *client = (struct shrb_client *)userdata;
(void)stream;
if (client->closed) {
- if (!client->data.connected) cleanup_connection_internal(client);
+ if (!client->control.connected) cleanup_connection_internal(client);
return;
}
- aki_packet_stream_free(&client->control);
- // TODO: Reconnect.
+ client->callback(client->userdata, SHRUB_CLIENT_REMOVE_BUFFERS, NULL, NULL);
+ shrb_vcr_flush(&client->vcr);
+ send_reconnect_packet(client);
+}
+
+void connect_data(struct shrb_client *client)
+{
+ if (!client->reconnect) client->reconnect = true;
+ else aki_packet_stream_free(&client->data);
+ aki_packet_stream_init(&client->data, AKI_SOCKET_TCP, data_connection_callback,
+ data_connection_closed_callback, client);
+ aki_packet_stream_set_multiplex(&client->data, CAMU_MULTIPLEX_SHRUB);
+ aki_packet_stream_connect(&client->data, client->loop, &client->addr, client->port);
}
static void parse_info_packet(struct shrb_client *client, struct aki_packet *packet)
@@ -188,22 +146,18 @@ static void parse_start_packet(struct shrb_client *client, struct aki_packet *pa
u64 delay = aki_packet_read_u64(packet);
u64 seek_pos = aki_packet_read_u64(packet);
u8 paused = aki_packet_read_u8(packet);
- (void)paused;
- struct shrb_seek_req req = { .base = seek_pos, .ts = ts, .delay = delay };
+ u64 paused_at = aki_packet_read_u64(packet);
+ struct shrb_seek_req req = {
+ .base = seek_pos,
+ .ts = ts,
+ .delay = delay,
+ .paused = paused,
+ .paused_at = paused_at
+ };
client->callback(client->userdata, SHRUB_CLIENT_SET, NULL, &req);
client->seek_pos = req.base;
}
-static void parse_resume_packet(struct shrb_client *client, struct aki_packet *packet)
-{
- u64 ts = aki_packet_read_u64(packet);
- s64 delay = aki_packet_read_s64(packet);
- u64 seek_pos = aki_packet_read_u64(packet);
- struct shrb_seek_req req = { .base = seek_pos, .ts = ts, .delay = delay };
- client->callback(client->userdata, SHRUB_CLIENT_RESUME, NULL, &req);
- client->seek_pos = req.base;
-}
-
static void control_packet_callback(void *userdata, struct aki_packet_stream *stream,
struct aki_packet *packet)
{
@@ -221,17 +175,18 @@ static void control_packet_callback(void *userdata, struct aki_packet_stream *st
connect_data(client);
break;
case SHRUB_CONTROL_PAUSE: {
- al_assert(!client->paused);
f64 pts = aki_packet_read_u64(packet) / 1000000.f;
client->callback(client->userdata, SHRUB_CLIENT_PAUSE, NULL, &pts);
- //client->paused = packet;
- //disconnect_data_internal(client);
break;
}
case SHRUB_CONTROL_RESUME: {
- u64 ts = aki_packet_read_u64(packet);
- client->callback(client->userdata, SHRUB_CLIENT_RESUME, NULL, &ts);
- //parse_resume_packet(client, packet);
+ u64 start = aki_packet_read_u64(packet);
+ u64 offset = aki_packet_read_u64(packet);
+ struct shrb_resume_req req = {
+ .start = start,
+ .offset = offset
+ };
+ client->callback(client->userdata, SHRUB_CLIENT_RESUME, NULL, &req);
break;
}
case SHRUB_CONTROL_SEEK:
@@ -245,7 +200,31 @@ static void control_packet_callback(void *userdata, struct aki_packet_stream *st
aki_packet_free(packet);
}
-void shrb_client_connect(struct shrb_client *client, struct aki_event_loop *loop, str *addr, s32 port, u16 node_id)
+static void control_connection_callback(void *userdata, struct aki_packet_stream *stream)
+{
+ struct shrb_client *client = (struct shrb_client *)userdata;
+ (void)stream;
+ client->control.packet_callback = control_packet_callback;
+ client->control.packet_sent_callback = packet_sent_callback;
+ struct aki_packet *packet = aki_packet_create();
+ aki_packet_write_u8(packet, SHRUB_CONNECTION_CONTROL);
+ aki_packet_write_u16(packet, 0);
+ aki_packet_write_u16(packet, client->node_id);
+ aki_packet_stream_send_packet(&client->control, packet);
+}
+
+static void control_connection_closed_callback(void *userdata, struct aki_packet_stream *stream)
+{
+ struct shrb_client *client = (struct shrb_client *)userdata;
+ (void)stream;
+ if (client->closed) {
+ if (!client->data.connected) cleanup_connection_internal(client);
+ return;
+ }
+ // TODO: Reconnect.
+}
+
+bool shrb_client_connect(struct shrb_client *client, struct aki_event_loop *loop, str *addr, s32 port, u16 node_id)
{
client->loop = loop;
client->node_id = node_id;
@@ -255,21 +234,16 @@ void shrb_client_connect(struct shrb_client *client, struct aki_event_loop *loop
client->reconnect = false;
client->level = 0;
client->corked = false;
- client->paused = NULL;
al_array_init(client->subscribed);
shrb_vcr_init(&client->vcr, client->loop, &client->data);
- aki_packet_stream_init(&client->control, AKI_SOCKET_TCP, control_connection_callback,
- control_connection_closed_callback, control_packet_callback, packet_sent_callback, client);
- aki_packet_stream_set_no_delay(&client->control, 1);
+ if (!aki_packet_stream_init(&client->control, AKI_SOCKET_TCP, control_connection_callback,
+ control_connection_closed_callback, client)) {
+ return false;
+ }
+ aki_packet_stream_set_nodelay(&client->control, 1);
+ aki_packet_stream_set_multiplex(&client->control, CAMU_MULTIPLEX_SHRUB);
aki_packet_stream_connect(&client->control, client->loop, &client->addr, client->port);
-}
-
-void shrb_client_toggle_pause(struct shrb_client *client, f64 pts)
-{
- struct aki_packet *packet = aki_packet_create();
- aki_packet_write_u8(packet, SHRUB_CONTROL_TOGGLE_PAUSE);
- aki_packet_write_u64(packet, pts * 1000000L);
- aki_packet_stream_send_packet(&client->control, packet);
+ return true;
}
void shrb_client_reseek(struct shrb_client *client)
diff --git a/src/shrub/client.h b/src/shrub/client.h
index cb3250d..14f7da2 100644
--- a/src/shrub/client.h
+++ b/src/shrub/client.h
@@ -19,14 +19,13 @@ struct shrb_client {
bool corked;
u64 duration;
u64 seek_pos;
- struct aki_packet *paused;
array(s32) subscribed;
struct shrb_vcr vcr;
void (*callback)(void *, u8, struct shrb_vcr_stream *stream, void *);
void *userdata;
};
-void shrb_client_connect(struct shrb_client *client, struct aki_event_loop *loop, str *addr, s32 port, u16 node_id);
+bool shrb_client_connect(struct shrb_client *client, struct aki_event_loop *loop, str *addr, s32 port, u16 node_id);
void shrb_client_toggle_pause(struct shrb_client *client, f64 pts);
void shrb_client_reseek(struct shrb_client *client);
void shrb_client_seek(struct shrb_client *client, f64 percent);
diff --git a/src/shrub/common.h b/src/shrub/common.h
index 0be00e4..882eb49 100644
--- a/src/shrub/common.h
+++ b/src/shrub/common.h
@@ -16,7 +16,6 @@ enum {
SHRUB_CONTROL_SUBSCRIBE,
SHRUB_CONTROL_RECONNECT,
SHRUB_CONTROL_START,
- SHRUB_CONTROL_TOGGLE_PAUSE,
SHRUB_CONTROL_PAUSE,
SHRUB_CONTROL_RESUME,
SHRUB_CONTROL_SEEK
diff --git a/src/shrub/handler.h b/src/shrub/handler.h
index 4ff2f2d..7242130 100644
--- a/src/shrub/handler.h
+++ b/src/shrub/handler.h
@@ -34,10 +34,17 @@ enum {
SHRUB_CLIENT_CLOSED
};
+struct shrb_resume_req {
+ u64 start;
+ u64 offset;
+};
+
struct shrb_seek_req {
u64 base;
u64 ts;
u64 delay;
+ bool paused;
+ u64 paused_at;
};
struct shrb_client_stream;
diff --git a/src/shrub/server.c b/src/shrub/server.c
index ba7259c..31d8787 100644
--- a/src/shrub/server.c
+++ b/src/shrub/server.c
@@ -65,11 +65,12 @@ static u8 packet_pool_callback(void *userdata, struct aki_packet *packet)
return AKI_PACKET_POOL_RETURN;
}
-static void send_resume_packet(struct shrb_node_connection *connection, u64 ts)
+static void send_resume_packet(struct shrb_node_connection *connection, u64 start, u64 offset)
{
struct aki_packet *packet = aki_packet_create();
aki_packet_write_u8(packet, SHRUB_CONTROL_RESUME);
- aki_packet_write_u64(packet, ts);
+ aki_packet_write_u64(packet, start);
+ aki_packet_write_u64(packet, offset);
aki_packet_stream_send_packet(connection->control, packet);
}
@@ -107,16 +108,21 @@ static void send_start_packet(struct shrb_node_connection *connection)
u64 seek_pos = node->seek_pos;
u64 delay = SHRUB_BASE_DELAY * powl(3, connection->level);
u64 start = node->start;
- s64 diff = aki_get_timestamp() - start;
- if (diff > (s64)SHRUB_BASE_DELAY) {
- start += diff;
- seek_pos += diff;
+ bool paused = node->paused != SHRUB_NOT_PAUSED;
+ if (!paused) {
+ s64 diff = aki_get_timestamp() - start;
+ if (diff > (s64)SHRUB_BASE_DELAY) {
+ start += diff;
+ seek_pos += diff;
+ }
}
- seek_pos += delay - SHRUB_BASE_DELAY;
+ // Start is not used if the node is paused.
aki_packet_write_u64(packet, start);
+ seek_pos += delay - SHRUB_BASE_DELAY;
aki_packet_write_u64(packet, delay);
aki_packet_write_u64(packet, seek_pos);
- aki_packet_write_u8(packet, node->paused ? 1 : 0);
+ aki_packet_write_u8(packet, paused);
+ aki_packet_write_u64(packet, node->paused_at);
aki_packet_stream_send_packet(connection->control, packet);
}
@@ -147,46 +153,6 @@ static void control_packet_callback(void *userdata, struct aki_packet_stream *st
send_start_packet(connection);
break;
}
- case SHRUB_CONTROL_TOGGLE_PAUSE: {
- struct shrb_node *node = connection->node;
- /*
- u64 pos = aki_packet_read_u64(packet);
- u8 paused = connection->paused;
- if (paused == SHRUB_PLAYING) {
- node->seek_pos = pos;
- node->flags |= SHRUB_NODE_PAUSED;
- // node->start won't be touched while SHRUB_NODE_PAUSED is set.
- } else if (paused == SHRUB_PAUSED_AND_RUNNING) {
- node->flags &= ~SHRUB_NODE_PAUSED;
- // TODO:
- node->start = aki_get_timestamp() + DELAY;
- }
- struct shrb_node_connection *rconnection;
- al_array_foreach(node->connections, i, rconnection) {
- switch (rconnection->paused) {
- case SHRUB_PLAYING:
- al_assert(paused == SHRUB_PLAYING || paused == SHRUB_NOT_PAUSED);
- rconnection->paused = SHRUB_PAUSED;
- send_pause_packet(rconnection, node->seek_pos);
- break;
- case SHRUB_NOT_PAUSED:
- al_assert(paused != SHRUB_PAUSED);
- rconnection->paused = SHRUB_PAUSED;
- break;
- case SHRUB_PAUSED:
- al_assert(paused != SHRUB_NOT_PAUSED);
- rconnection->paused = SHRUB_NOT_PAUSED;
- break;
- case SHRUB_PAUSED_AND_RUNNING:
- al_assert(paused != SHRUB_NOT_PAUSED);
- send_resume_packet(rconnection, node->seek_pos, node->start);
- rconnection->paused = SHRUB_PLAYING;
- break;
- }
- }
- */
- break;
- }
case SHRUB_CONTROL_SEEK: {
u64 pos = aki_packet_read_u64(packet);
connection->node->seek_pos = pos;
@@ -207,17 +173,18 @@ static struct shrb_node_connection *get_connection_by_id(struct shrb_node *node,
return NULL;
}
-static void send_control_info_packet(struct shrb_node_connection *connection)
+static bool send_control_info_packet(struct shrb_node_connection *connection)
{
- struct aki_packet *packet = aki_packet_create();
- aki_packet_write_u8(packet, SHRUB_CONTROL_INFO);
- aki_packet_write_u16(packet, connection->id);
connection->handler = shrb_codec_server_create();
if (!connection->handler->init(connection->handler, &connection->handle, &connection->pool)) {
- al_assert(false);
+ return false;
}
+ struct aki_packet *packet = aki_packet_create();
+ aki_packet_write_u8(packet, SHRUB_CONTROL_INFO);
+ aki_packet_write_u16(packet, connection->id);
connection->handler->write_info(connection->handler, packet);
aki_packet_stream_send_packet(connection->control, packet);
+ return true;
}
static aki_thread_result AKI_THREADCALL handler_thread(void *userdata)
@@ -230,19 +197,12 @@ static aki_thread_result AKI_THREADCALL handler_thread(void *userdata)
static void maybe_start_handler(struct shrb_node_connection *connection)
{
al_assert(!connection->running);
- // Don't seek to 0 unless NODE_SEEKED is set.
+ // Don't seek to 0 unless node->seeked is set.
if (connection->seek_pos > 0 || connection->node->seeked) {
connection->handler->seek(connection->handler, connection->seek_pos);
}
connection->running = 1;
aki_thread_create(&connection->thread, handler_thread, connection);
- //struct shrb_node *node = connection->node;
- //if (connection->paused == SHRUB_NOT_PAUSED) {
- // send_resume_packet(connection, connection->seek_pos, node->start);
- // connection->paused = SHRUB_PLAYING;
- //} else if (connection->paused == SHRUB_PAUSED) {
- // connection->paused = SHRUB_PAUSED_AND_RUNNING;
- //}
}
static void identify_packet_callback(void *userdata, struct aki_packet_stream *stream, struct aki_packet *packet)
@@ -254,6 +214,12 @@ static void identify_packet_callback(void *userdata, struct aki_packet_stream *s
u16 connection_id = aki_packet_read_u16(packet);
u16 node_id = aki_packet_read_u16(packet);
+ // Before this point:
+ // - We get socket from accept
+ // - Create packet_stream
+ // - connection_closed_callback could have been called before this point
+ // - the socket exists only to wait for an identify packet
+
// Disconnecting a stream at any point before setting the new callbacks should
// be valid and handled properly by the client.
// Generalize erronious connection checking/handling.
@@ -276,26 +242,28 @@ static void identify_packet_callback(void *userdata, struct aki_packet_stream *s
connection->data = NULL;
aki_packet_pool_init(&connection->pool, 16, server->loop, packet_pool_callback, connection);
cch_entry_get_handle(connection->node->entry, &connection->handle);
- //connection->paused = SHRUB_PLAYING;
connection->level = 0;
connection->running = 0;
}
switch (type) {
case SHRUB_CONNECTION_CONTROL:
- aki_packet_stream_set_no_delay(stream, 1);
+ aki_packet_stream_set_nodelay(stream, 1);
connection->control = stream;
stream->packet_callback = control_packet_callback;
stream->packet_sent_callback = default_packet_sent_callback;
stream->connection_closed_callback = control_connection_closed_callback;
stream->userdata = connection;
- send_control_info_packet(connection);
+ if (!send_control_info_packet(connection)) {
+ aki_packet_stream_disconnect(stream);
+ }
break;
case SHRUB_CONNECTION_DATA:
// TODO: Account for control connection being in a reconnect state.
// TODO: is it possible for data->data to be non-NULL during normal operation
// (disconnect and reconnect very fast?).
// - Call maybe_cleanup_data in this function?
+ // ^^^^ Literally any connection could call itself data, this needs to be handled better.
al_assert(!connection->data);
connection->seek_pos = aki_packet_read_u64(packet);
connection->data = stream;
@@ -310,32 +278,34 @@ static void identify_packet_callback(void *userdata, struct aki_packet_stream *s
aki_packet_free(packet);
}
-static void connection_closed_callback(void *userdata, struct aki_packet_stream *stream)
-{
- (void)userdata;
- (void)stream;
-}
-
static void connection_callback(void *userdata, struct aki_packet_stream *stream)
{
struct shrb_server *server = (struct shrb_server *)userdata;
- stream->connection_closed_callback = connection_closed_callback;
stream->packet_callback = identify_packet_callback;
stream->packet_sent_callback = default_packet_sent_callback;
- stream->flushed_callback = NULL;
stream->userdata = server;
}
-bool shrb_server_init(struct shrb_server *server)
+bool shrb_server_init(struct shrb_server *server, struct aki_event_loop *loop)
{
+ server->loop = loop;
al_array_init(server->nodes);
- return aki_packet_stream_init(&server->server, AKI_SOCKET_TCP, connection_callback, NULL, NULL, NULL, server);
+ return true;
}
-void shrb_server_listen(struct shrb_server *server, struct aki_event_loop *loop, str *addr, s32 port)
+static void connection_closed_callback(void *userdata, struct aki_packet_stream *stream)
{
- server->loop = loop;
- aki_packet_stream_listen(&server->server, loop, addr, port);
+ (void)userdata;
+ (void)stream;
+}
+
+void shrb_server_add_socket(struct shrb_server *server, struct aki_socket *s)
+{
+ struct aki_packet_stream *stream = al_alloc_object(struct aki_packet_stream);
+ stream->connection_callback = connection_callback;
+ stream->connection_closed_callback = connection_closed_callback;
+ stream->userdata = server;
+ aki_packet_stream_from_socket(stream, server->loop, s);
}
struct shrb_node *shrb_server_create_node(struct shrb_server *server, u16 prev, struct cch_entry *entry)
@@ -344,14 +314,6 @@ struct shrb_node *shrb_server_create_node(struct shrb_server *server, u16 prev,
node->id = al_rand_u16();
node->prev = prev;
node->entry = entry;
- node->paused = false;
- node->seeked = false;
- node->start = 0;
- node->seek_pos = 0;
- node->offset = 0;
- node->connection_id = 1;
- al_array_init(node->connections);
- node->server = server;
struct cch_handle handle;
if (!cch_entry_get_handle(node->entry, &handle)) {
al_free(node);
@@ -365,47 +327,67 @@ struct shrb_node *shrb_server_create_node(struct shrb_server *server, u16 prev,
node->duration = handler->get_duration(handler);
handler->free(&handler);
cch_entry_return_handle(node->entry, &handle);
+ node->seeked = false;
+ node->start = 0;
+ node->paused = SHRUB_NOT_PAUSED;
+ node->paused_at = 0;
+ node->seek_pos = 0;
+ node->connection_id = 1;
+ al_array_init(node->connections);
+ node->server = server;
al_array_push(server->nodes, node);
return node;
}
void shrb_node_set_start(struct shrb_node *node, u64 start)
{
+ if (node->paused == SHRUB_PAUSED) {
+ node->paused = SHRUB_NOT_PAUSED;
+ }
node->start = start;
}
-void shrb_node_set_offset(struct shrb_node *node, u64 offset)
+void shrb_node_user_pause(struct shrb_node *node, u64 pos)
{
- node->offset = offset;
+ al_assert(node->paused != SHRUB_USER_PAUSED);
+ pos += SHRUB_PAUSE_DELAY;
+ node->seek_pos = pos;
+ node->paused_at = pos;
+ struct shrb_node_connection *connection;
+ al_array_foreach(node->connections, i, connection) {
+ send_pause_packet(connection, node->seek_pos);
+ }
+ node->paused = SHRUB_USER_PAUSED;
}
-void shrb_node_toggle_pause(struct shrb_node *node, u64 pos)
+void shrb_node_user_resume(struct shrb_node *node, struct shrb_resume_req *req)
{
- if (!node->paused) {
- node->seek_pos = pos;
- node->seek_pos += SHRUB_PAUSE_DELAY;
- } else {
- node->start = aki_get_timestamp() + SHRUB_PAUSE_DELAY;
- }
+ al_assert(node->paused == SHRUB_USER_PAUSED);
+ node->start = req->start + SHRUB_PAUSE_DELAY;
struct shrb_node_connection *connection;
al_array_foreach(node->connections, i, connection) {
- if (!node->paused) {
- send_pause_packet(connection, node->seek_pos);
- } else {
- send_resume_packet(connection, node->start);
- }
+ send_resume_packet(connection, node->start, req->offset);
}
- if (node->paused) {
- node->start -= SHRUB_BASE_DELAY;
- node->offset = 0;
- }
- node->paused = !node->paused;
+ node->start -= SHRUB_BASE_DELAY;
+ node->paused = SHRUB_NOT_PAUSED;
+}
+
+void shrb_node_pause(struct shrb_node *node)
+{
+ return;
+ if (node->paused != SHRUB_NOT_PAUSED) return;
+ u64 start = node->start;
+ s64 pos = node->seek_pos + (aki_get_timestamp() - (start + SHRUB_BASE_DELAY));
+ pos += SHRUB_PAUSE_DELAY;
+ al_assert(pos >= 0);
+ node->seek_pos = (u64)pos;
+ node->paused_at = pos;
+ node->paused = SHRUB_PAUSED;
}
void shrb_node_seek(struct shrb_node *node, u64 pos)
{
node->seek_pos = pos;
- node->start = aki_get_timestamp();
node->seeked = true;
struct shrb_node_connection *connection;
al_array_foreach(node->connections, i, connection) {
diff --git a/src/shrub/server.h b/src/shrub/server.h
index 2bce369..cb901c5 100644
--- a/src/shrub/server.h
+++ b/src/shrub/server.h
@@ -1,6 +1,7 @@
#pragma once
#include <al/array.h>
+#include <aki/multiplex.h>
#include <aki/packet_stream.h>
#include <aki/packet_pool.h>
@@ -9,13 +10,6 @@
#include "handler.h"
-enum {
- SHRUB_PLAYING = 0,
- SHRUB_NOT_PAUSED,
- SHRUB_PAUSED,
- SHRUB_PAUSED_AND_RUNNING
-};
-
struct shrb_node_connection {
u16 id;
struct shrb_node *node;
@@ -23,7 +17,6 @@ struct shrb_node_connection {
struct aki_packet_stream *data;
struct aki_packet_pool pool;
s32 running;
- //u8 paused;
u32 level;
u64 seek_pos;
struct cch_handle handle;
@@ -31,16 +24,22 @@ struct shrb_node_connection {
struct aki_thread thread;
};
+enum {
+ SHRUB_NOT_PAUSED = 0,
+ SHRUB_PAUSED,
+ SHRUB_USER_PAUSED
+};
+
struct shrb_node {
u16 id;
u16 prev;
struct cch_entry *entry;
- bool paused;
+ u64 duration;
bool seeked;
- u64 start;
+ u64 start; // Exact start of the stream minus BASE_DELAY.
+ u8 paused;
+ u64 paused_at;
u64 seek_pos;
- u64 offset;
- u64 duration;
u16 connection_id;
array(struct shrb_node_connection *) connections;
struct shrb_server *server;
@@ -48,15 +47,16 @@ struct shrb_node {
struct shrb_server {
struct aki_event_loop *loop;
- struct aki_packet_stream server;
array(struct shrb_node *) nodes;
};
-bool shrb_server_init(struct shrb_server *server);
-void shrb_server_listen(struct shrb_server *server, struct aki_event_loop *loop, str *addr, s32 port);
+bool shrb_server_init(struct shrb_server *server, struct aki_event_loop *loop);
+void shrb_server_add_socket(struct shrb_server *server, struct aki_socket *s);
struct shrb_node *shrb_server_create_node(struct shrb_server *server, u16 prev, struct cch_entry *entry);
void shrb_node_set_start(struct shrb_node *node, u64 start);
-void shrb_node_set_offset(struct shrb_node *node, u64 offset);
-void shrb_node_toggle_pause(struct shrb_node *node, u64 pos);
+void shrb_node_user_pause(struct shrb_node *node, u64 pos);
+void shrb_node_user_resume(struct shrb_node *node, struct shrb_resume_req *req);
+void shrb_node_pause(struct shrb_node *node);
+void shrb_node_resume(struct shrb_node *node);
void shrb_node_seek(struct shrb_node *node, u64 pos);
void shrb_server_close(struct shrb_server *server);
diff --git a/src/shrub/vcr.c b/src/shrub/vcr.c
index 1816142..7f2b058 100644
--- a/src/shrub/vcr.c
+++ b/src/shrub/vcr.c
@@ -149,6 +149,10 @@ void shrb_vcr_stream_cork(struct shrb_vcr_stream *stream)
void shrb_vcr_stream_uncork(struct shrb_vcr_stream *stream)
{
+ if (al_atomic_s32_load(&stream->state, AL_ATOMIC_RELAXED) == SHRUB_STREAM_RUNNING) {
+ // DEBUG: This will be reached during normal operation.
+ return;
+ }
al_atomic_s32_store(&stream->state, SHRUB_STREAM_RUNNING, AL_ATOMIC_RELAXED);
aki_mutex_lock(&stream->mutex);
if (aki_cond_is_waiting(&stream->cond)) {
generated by cgit v1.2.3-101-g0448 (git 2.54.0) at 2026-07-25 21:36:08 +0000