Rough seek implementation

Signed-off-by: Andrew Opalach <andrew@akon.city>

7 files changed, 117 insertions, 43 deletions

diff --git a/src/bimu/client.c b/src/bimu/client.c
index b05d950..8a72753 100644
--- a/src/bimu/client.c
+++ b/src/bimu/client.c
@@ -1,3 +1,5 @@
+#include <al/log.h>
+
 #include "../codec/libav/packet_ext.h"
 
 #include "client.h"
@@ -13,10 +15,29 @@ static void data_connection_callback(void *userdata, struct aki_packet_stream *s
     aki_packet_stream_send_packet(&client->data, packet);
 }
 
+static void send_reconnect_packet(struct bmu_client *client)
+{
+    struct aki_packet *packet = aki_packet_create();
+    aki_packet_write_u8(packet, BIMU_CONTROL_RECONNECT);
+    aki_packet_stream_send_packet(&client->control, packet);
+}
+
+static void flush_streams(struct bmu_client *client)
+{
+    struct bmu_client_stream *stream;
+    al_array_foreach(client->streams, i, stream) {
+        stream->client->flush(stream->client);
+    }
+}
+
 static void data_connection_closed_callback(void *userdata, struct aki_packet_stream *stream)
 {
-    (void)userdata;
+    struct bmu_client *client = (struct bmu_client *)userdata;
     (void)stream;
+    aki_packet_stream_free(&client->data);
+    client->callback(client->userdata, BIMU_CLIENT_REMOVE_BUFFERS, NULL, NULL);
+    flush_streams(client);
+    send_reconnect_packet(client);
 }
 
 static struct bmu_client_stream *stream_from_index(struct bmu_client *client, s32 index)
@@ -32,16 +53,17 @@ static void handle_data_packet_internal(struct bmu_client *client, struct aki_pa
 {
     struct bmu_client_stream *stream;
     switch (aki_packet_read_u8(packet)) {
-    case 0:
+    case BIMU_PACKET_DATA:
         stream = stream_from_index(client, aki_packet_read_s32(packet));
         stream->client->handle_data_packet(stream->client, packet);
         break;
-    case 1:
+    case BIMU_PACKET_EOF:
         al_array_foreach(client->streams, i, stream) {
             stream->client->handle_eof(stream->client);
         }
         break;
-    case 2:
+    case BIMU_PACKET_ERROR:
+        al_log_warn("bimu", "Unhandled error packet.");
         break;
     }
 }
@@ -61,7 +83,7 @@ static void packet_sent_callback(void *userdata, struct aki_packet *packet)
     aki_packet_free(packet);
 }
 
-static void connect_data(struct bmu_client *client)
+void connect_data(struct bmu_client *client)
 {
     aki_packet_stream_init(&client->data, AKI_SOCKET_TCP, data_connection_callback,
         data_connection_closed_callback, data_packet_callback, packet_sent_callback, client);
@@ -87,6 +109,7 @@ static void control_connection_closed_callback(void *userdata, struct aki_packet
 
 static void parse_info_packet(struct bmu_client *client, struct aki_packet *packet)
 {
+    client->duration = aki_packet_read_s64(packet);
     u32 count = aki_packet_read_u32(packet);
     for (u32 i = 0; i < count; i++) {
         struct bmu_client_stream *stream = al_alloc_object(struct bmu_client_stream);
@@ -147,12 +170,23 @@ static void send_subscribe_packet(struct bmu_client *client)
 static void parse_start_packet(struct bmu_client *client, struct aki_packet *packet)
 {
     u64 seek_pos = aki_packet_read_u64(packet);
-    u64 ats = aki_packet_read_u64(packet);
+    u64 ts = aki_packet_read_u64(packet);
     u8 paused = aki_packet_read_u8(packet);
     (void)paused;
     struct bmu_seek_req req = {
         .base = seek_pos,
-        .start = ats
+        .start = ts
+    };
+    client->callback(client->userdata, BIMU_CLIENT_SET, NULL, &req);
+}
+
+static void parse_resume_packet(struct bmu_client *client, struct aki_packet *packet)
+{
+    u64 seek_pos = aki_packet_read_u64(packet);
+    u64 ts = aki_packet_read_u64(packet);
+    struct bmu_seek_req req = {
+        .base = seek_pos,
+        .start = ts
     };
     client->callback(client->userdata, BIMU_CLIENT_SET, NULL, &req);
 }
@@ -176,6 +210,11 @@ static void control_packet_callback(void *userdata, struct aki_packet_stream *st
     case BIMU_CONTROL_PAUSE:
         break;
     case BIMU_CONTROL_RESUME:
+        //parse_resume_packet(client, packet);
+        break;
+    case BIMU_CONTROL_SEEK:
+        client->corked = false;
+        aki_packet_stream_disconnect(&client->data);
         break;
     default:
         break;
@@ -205,6 +244,15 @@ void bmu_client_cork(struct bmu_client *client, bool cork)
     client->corked = cork;
 }
 
+void bmu_client_seek(struct bmu_client *client, f64 percent)
+{
+    u64 pos = client->duration * percent;
+    struct aki_packet *packet = aki_packet_create();
+    aki_packet_write_u8(packet, BIMU_CONTROL_SEEK);
+    aki_packet_write_u64(packet, pos);
+    aki_packet_stream_send_packet(&client->control, packet);
+}
+
 void bmu_client_free(struct bmu_client *client)
 {
     aki_packet_stream_free(&client->control);
diff --git a/src/bimu/client.h b/src/bimu/client.h
index 464ee03..0ebf282 100644
--- a/src/bimu/client.h
+++ b/src/bimu/client.h
@@ -23,6 +23,7 @@ struct bmu_client {
     struct aki_packet_stream control;
     struct aki_packet_stream data;
     bool corked;
+    s64 duration;
     array(s32) subscribed;
     array(struct bmu_client_stream *) streams;
     void (*callback)(void *, u8, struct bmu_client_stream *stream, void *);
@@ -31,4 +32,5 @@ struct bmu_client {
 
 void bmu_client_connect(struct bmu_client *client, struct aki_event_loop *loop, str *addr, s32 port, u16 node_id);
 void bmu_client_cork(struct bmu_client *client, bool cork);
+void bmu_client_seek(struct bmu_client *client, f64 percent);
 void bmu_client_free(struct bmu_client *client);
diff --git a/src/bimu/common.h b/src/bimu/common.h
index fe8e0af..485d454 100644
--- a/src/bimu/common.h
+++ b/src/bimu/common.h
@@ -14,6 +14,7 @@ enum {
 enum {
     BIMU_CONTROL_INFO = 0,
     BIMU_CONTROL_SUBSCRIBE,
+    BIMU_CONTROL_RECONNECT,
     BIMU_CONTROL_START,
     BIMU_CONTROL_TOGGLE_PAUSE,
     BIMU_CONTROL_PAUSE,
diff --git a/src/bimu/handler.h b/src/bimu/handler.h
index f9e8440..9b9d348 100644
--- a/src/bimu/handler.h
+++ b/src/bimu/handler.h
@@ -6,7 +6,7 @@
 #include "../cache/handle.h"
 
 struct bmu_server_handler {
-    bool (*init)(struct bmu_server_handler *, struct cch_handle *, struct aki_packet_pool *, void *);
+    bool (*init)(struct bmu_server_handler *, struct cch_handle *, struct aki_packet_pool *);
     void (*write_info)(struct bmu_server_handler *, struct aki_packet *);
     void (*subscribe)(struct bmu_server_handler *, s32, bool);
     s64 (*get_duration)(struct bmu_server_handler *);
@@ -17,10 +17,16 @@ struct bmu_server_handler {
 };
 
 enum {
+    BIMU_PACKET_DATA = 0,
+    BIMU_PACKET_EOF,
+    BIMU_PACKET_ERROR
+};
+
+enum {
     BIMU_CLIENT_SET = 0,
     BIMU_CLIENT_CONFIGURE,
     BIMU_CLIENT_DATA,
-    BIMU_CLIENT_SEEK,
+    BIMU_CLIENT_REMOVE_BUFFERS,
     BIMU_CLIENT_EOF,
     BIMU_CLIENT_CLOSED
 };
diff --git a/src/bimu/handlers/codec_server.c b/src/bimu/handlers/codec_server.c
index 3c27179..99c4e71 100644
--- a/src/bimu/handlers/codec_server.c
+++ b/src/bimu/handlers/codec_server.c
@@ -10,7 +10,7 @@
 #include "codec.h"
 
 static bool codec_server_init(struct bmu_server_handler *handler, struct cch_handle *handle,
-    struct aki_packet_pool *pool, void *userdata)
+    struct aki_packet_pool *pool)
 {
     struct bmu_codec_server *codec = (struct bmu_codec_server *)handler;
     codec->demux = camu_lav_demuxer_create();
@@ -27,13 +27,13 @@ static bool codec_server_init(struct bmu_server_handler *handler, struct cch_han
 #endif
     }
     codec->pool = pool;
-    codec->handler.userdata = userdata;
     return true;
 }
 
 static void codec_server_write_info(struct bmu_server_handler *handler, struct aki_packet *packet)
 {
     struct bmu_codec_server *codec = (struct bmu_codec_server *)handler;
+    aki_packet_write_s64(packet, codec->demux->get_duration(codec->demux));
     aki_packet_write_u32(packet, codec->demux->streams.size);
     struct camu_stream *stream;
     al_array_foreach_ptr(codec->demux->streams, i, stream) {
@@ -78,13 +78,12 @@ static bool codec_server_step(struct bmu_server_handler *handler)
     struct bmu_codec_server *codec = (struct bmu_codec_server *)handler;
     struct aki_packet *packet = aki_packet_pool_get(codec->pool);
     if (!packet) return false;
-    packet->userdata = codec->handler.userdata;
     s32 ret = codec->demux->get_packet(codec->demux, &codec->packet);
     if (ret == CAMU_OK) {
         aki_packet_write_u8(packet, 0);
         switch (codec->packet.type) {
         case CAMU_NORMAL: {
-            aki_packet_write_s32(packet, 0);
+            aki_packet_write_s32(packet, BIMU_PACKET_DATA);
             aki_packet_write_u8(packet, codec->packet.type);
             aki_packet_write_buffer(packet, codec->packet.buffer);
             break;
@@ -102,11 +101,11 @@ static bool codec_server_step(struct bmu_server_handler *handler)
         }
         aki_packet_pool_submit(codec->pool, packet);
     } else if (ret == CAMU_ERR_EOF) {
-        aki_packet_write_u8(packet, 1);
+        aki_packet_write_u8(packet, BIMU_PACKET_EOF);
         aki_packet_pool_submit(codec->pool, packet);
         return false;
     } else {
-        aki_packet_write_u8(packet, 2);
+        aki_packet_write_u8(packet, BIMU_PACKET_ERROR);
         aki_packet_pool_submit(codec->pool, packet);
         return false;
     }
diff --git a/src/bimu/server.c b/src/bimu/server.c
index 565f0b1..b823827 100644
--- a/src/bimu/server.c
+++ b/src/bimu/server.c
@@ -5,21 +5,34 @@
 
 #include "server.h"
 
-#define START_DELAY 700000 // 700ms
+#define START_DELAY 3000000 // 3s
 #define DELAY_GIVE 200000 // 200ms
 
 static void control_connection_closed_callback(void *userdata, struct aki_packet_stream *stream)
 {
     struct bmu_node_connection *connection = (struct bmu_node_connection *)userdata;
     (void)stream;
+    struct bmu_node_connection *rconnection;
+    al_array_foreach(connection->node->connections, i, rconnection) {
+        if (rconnection == connection) {
+            al_array_remove_at_iter(connection->node->connections, i);
+            break;
+        }
+    }
     aki_packet_stream_free(connection->control);
+    connection->control = NULL;
 }
 
 static void data_connection_closed_callback(void *userdata, struct aki_packet_stream *stream)
 {
     struct bmu_node_connection *connection = (struct bmu_node_connection *)userdata;
     (void)stream;
+    connection->running = 0;
+    aki_packet_pool_disable(&connection->pool);
+    aki_thread_join(&connection->thread);
+    aki_packet_pool_enable(&connection->pool);
     aki_packet_stream_free(connection->data);
+    connection->data = NULL;
 }
 
 static void default_packet_sent_callback(void *userdata, struct aki_packet *packet)
@@ -31,7 +44,7 @@ static void default_packet_sent_callback(void *userdata, struct aki_packet *pack
 static void packet_sent_callback(void *userdata, struct aki_packet *packet)
 {
     struct bmu_node_connection *connection = (struct bmu_node_connection *)userdata;
-    aki_packet_pool_return(&connection->node->pool, packet);
+    aki_packet_pool_return(&connection->pool, packet);
 }
 
 static void data_packet_callback(void *userdata, struct aki_packet_stream *stream, struct aki_packet *packet)
@@ -43,8 +56,7 @@ static void data_packet_callback(void *userdata, struct aki_packet_stream *strea
 
 static u8 packet_pool_callback(void *userdata, struct aki_packet *packet)
 {
-    (void)userdata;
-    struct bmu_node_connection *connection = (struct bmu_node_connection *)packet->userdata;
+    struct bmu_node_connection *connection = (struct bmu_node_connection *)userdata;
     if (connection->data->connected) {
         aki_packet_stream_send_packet(connection->data, packet);
         return AKI_PACKET_POOL_KEEP;
@@ -73,18 +85,17 @@ static void send_start_packet(struct bmu_node_connection *connection)
 {
     struct aki_packet *packet = aki_packet_create();
     aki_packet_write_u8(packet, BIMU_CONTROL_START);
-    u64 ts = aki_get_timestamp();
-    u64 ats = ts - START_DELAY;
+    u64 ts = aki_get_timestamp() - START_DELAY;
     s64 seek_pos = connection->node->seek_pos;
     if (!connection->node->paused) {
         if (connection->node->start >= 0) {
-            if ((ats - DELAY_GIVE) < (u64)connection->node->start) {
-                ats = connection->node->start;
+            if ((ts - DELAY_GIVE) < (u64)connection->node->start) {
+                ts = connection->node->start;
             } else {
-                seek_pos += ats - connection->node->start;
+                seek_pos += ts - connection->node->start;
             }
         } else {
-            connection->node->start = ats;
+            connection->node->start = ts;
         }
     }
     s64 duration = connection->handler->get_duration(connection->handler);
@@ -93,16 +104,20 @@ static void send_start_packet(struct bmu_node_connection *connection)
     // and the start of a stream.
     connection->seek_pos = seek_pos > 0 ? seek_pos : -1;
     aki_packet_write_u64(packet, seek_pos);
-    aki_packet_write_u64(packet, ats);
+    aki_packet_write_u64(packet, ts);
     aki_packet_write_u8(packet, connection->node->paused);
     aki_packet_stream_send_packet(connection->control, packet);
 }
 
-static void send_seek_packet(struct bmu_node_connection *connection)
+static void send_seek_packet(struct bmu_node *node)
 {
-    struct aki_packet *packet = aki_packet_create();
-    aki_packet_write_u8(packet, BIMU_CONTROL_SEEK);
-    aki_packet_stream_send_packet(connection->control, packet);
+    struct aki_packet *packet;
+    struct bmu_node_connection *connection;
+    al_array_foreach(node->connections, i, connection) {
+        packet = aki_packet_create();
+        aki_packet_write_u8(packet, BIMU_CONTROL_SEEK);
+        aki_packet_stream_send_packet(connection->control, packet);
+    }
 }
 
 static void control_packet_callback(void *userdata, struct aki_packet_stream *stream, struct aki_packet *packet)
@@ -119,6 +134,10 @@ static void control_packet_callback(void *userdata, struct aki_packet_stream *st
         send_start_packet(connection);
         break;
     }
+    case BIMU_CONTROL_RECONNECT: {
+        send_start_packet(connection);
+        break;
+    }
     case BIMU_CONTROL_TOGGLE_PAUSE: {
         struct bmu_node *node = connection->node;
         u64 pos = aki_packet_read_u64(packet);
@@ -160,8 +179,10 @@ static void control_packet_callback(void *userdata, struct aki_packet_stream *st
     }
     case BIMU_CONTROL_SEEK: {
         u64 pos = aki_packet_read_u64(packet);
+        connection->node->seek_pos = pos;
+        connection->node->start = -1;
         connection->seek_pos = pos;
-        send_seek_packet(connection);
+        send_seek_packet(connection->node);
         break;
     }
     }
@@ -192,7 +213,7 @@ static void send_control_info_packet(struct bmu_node_connection *connection)
     aki_packet_write_u8(packet, BIMU_CONTROL_INFO);
     aki_packet_write_u16(packet, connection->id);
     connection->handler = bmu_codec_server_create();
-    connection->handler->init(connection->handler, &connection->handle, &connection->node->pool, connection);
+    connection->handler->init(connection->handler, &connection->handle, &connection->pool);
     connection->handler->write_info(connection->handler, packet);
     aki_packet_stream_send_packet(connection->control, packet);
 }
@@ -200,7 +221,7 @@ static void send_control_info_packet(struct bmu_node_connection *connection)
 static aki_thread_result AKI_THREADCALL handler_thread(void *userdata)
 {
     struct bmu_node_connection *connection = (struct bmu_node_connection *)userdata;
-    while (connection->handler->step(connection->handler)) {}
+    while (connection->running && connection->handler->step(connection->handler)) {}
     return 0;
 }
 
@@ -210,14 +231,10 @@ static void maybe_start_handler(struct bmu_node_connection *connection)
     if (connection->seek_pos >= 0) {
         connection->handler->seek(connection->handler, connection->seek_pos);
     }
+    connection->running = 1;
     aki_thread_create(&connection->thread, handler_thread, connection);
-    connection->running = true;
     struct bmu_node *node = connection->node;
     if (connection->paused == BIMU_NOT_PAUSED) {
-        if (node->start == -1) {
-            node->paused = false;
-            node->start = aki_get_timestamp() - START_DELAY;
-        }
         send_resume_packet(connection, connection->seek_pos, node->start);
         connection->paused = BIMU_PLAYING;
     } else if (connection->paused == BIMU_PAUSED) {
@@ -254,9 +271,10 @@ static void identify_packet_callback(void *userdata, struct aki_packet_stream *s
         connection->node = node;
         connection->control = NULL;
         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 = BIMU_PLAYING;
-        connection->running = false;
+        connection->paused = BIMU_NOT_PAUSED;
+        connection->running = 0;
     }
 
     switch (type) {
@@ -274,6 +292,7 @@ static void identify_packet_callback(void *userdata, struct aki_packet_stream *s
         // 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?
+        al_assert(!connection->data);
         connection->data = stream;
         stream->packet_callback = data_packet_callback;
         stream->packet_sent_callback = packet_sent_callback;
@@ -323,7 +342,6 @@ u16 bmu_server_create_node(struct bmu_server *server, struct cch_entry *entry)
     node->paused = false;
     node->start = -1;
     node->seek_pos = 0;
-    aki_packet_pool_init(&node->pool, 16, server->loop, packet_pool_callback, node);
     node->connection_id = 1;
     al_array_init(node->connections);
     node->server = server;
diff --git a/src/bimu/server.h b/src/bimu/server.h
index 754ab7f..d954fa9 100644
--- a/src/bimu/server.h
+++ b/src/bimu/server.h
@@ -22,7 +22,8 @@ struct bmu_node_connection {
     struct bmu_node *node;
     struct aki_packet_stream *control;
     struct aki_packet_stream *data;
-    bool running;
+    struct aki_packet_pool pool;
+    s32 running;
     u8 paused;
     s64 seek_pos;
     struct cch_handle handle;
@@ -36,7 +37,6 @@ struct bmu_node {
     bool paused;
     s64 start;
     s64 seek_pos;
-    struct aki_packet_pool pool;
     u16 connection_id;
     array(struct bmu_node_connection *) connections;
     struct bmu_server *server;