Initial synced playback

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

5 files changed, 61 insertions, 18 deletions

diff --git a/src/bimu/client.c b/src/bimu/client.c
index d35e3a1..b05d950 100644
--- a/src/bimu/client.c
+++ b/src/bimu/client.c
@@ -144,6 +144,19 @@ static void send_subscribe_packet(struct bmu_client *client)
     aki_packet_stream_send_packet(&client->control, packet);
 }
 
+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);
+    u8 paused = aki_packet_read_u8(packet);
+    (void)paused;
+    struct bmu_seek_req req = {
+        .base = seek_pos,
+        .start = ats
+    };
+    client->callback(client->userdata, BIMU_CLIENT_SET, NULL, &req);
+}
+
 static void control_packet_callback(void *userdata, struct aki_packet_stream *stream,
     struct aki_packet *packet)
 {
@@ -157,8 +170,13 @@ static void control_packet_callback(void *userdata, struct aki_packet_stream *st
         send_subscribe_packet(client);
         break;
     case BIMU_CONTROL_START:
+        parse_start_packet(client, packet);
         connect_data(client);
         break;
+    case BIMU_CONTROL_PAUSE:
+        break;
+    case BIMU_CONTROL_RESUME:
+        break;
     default:
         break;
     }
@@ -186,3 +204,9 @@ void bmu_client_cork(struct bmu_client *client, bool cork)
     aki_packet_stream_cork(&client->data, cork);
     client->corked = cork;
 }
+
+void bmu_client_free(struct bmu_client *client)
+{
+    aki_packet_stream_free(&client->control);
+    aki_packet_stream_free(&client->data);
+}
diff --git a/src/bimu/client.h b/src/bimu/client.h
index 54c903a..464ee03 100644
--- a/src/bimu/client.h
+++ b/src/bimu/client.h
@@ -31,3 +31,4 @@ 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_free(struct bmu_client *client);
diff --git a/src/bimu/handler.h b/src/bimu/handler.h
index 3881288..f9e8440 100644
--- a/src/bimu/handler.h
+++ b/src/bimu/handler.h
@@ -17,13 +17,18 @@ struct bmu_server_handler {
 };
 
 enum {
-    BIMU_CLIENT_CONFIGURE = 0,
+    BIMU_CLIENT_SET = 0,
+    BIMU_CLIENT_CONFIGURE,
     BIMU_CLIENT_DATA,
     BIMU_CLIENT_SEEK,
     BIMU_CLIENT_EOF,
     BIMU_CLIENT_CLOSED
 };
 
+struct bmu_seek_req {
+    u64 base, start;
+};
+
 struct bmu_client_stream;
 struct bmu_client_handler {
     bool (*init)(struct bmu_client_handler *, struct camu_renderer *, struct bmu_client_stream *);
diff --git a/src/bimu/meson.build b/src/bimu/meson.build
index 086114d..dce7f7b 100644
--- a/src/bimu/meson.build
+++ b/src/bimu/meson.build
@@ -1,8 +1,12 @@
-bimu_src = [
-  'local.c',
+bimu_server_src = [
   'server.c',
+  'handlers/codec_server.c'
+]
+
+bimu_client_src = [
   'client.c',
-  'handlers/codec_server.c',
   'handlers/codec_client.c'
 ]
-bimu = declare_dependency(sources: bimu_src)
+
+bimu_server = declare_dependency(sources: bimu_server_src)
+bimu_client = declare_dependency(sources: bimu_client_src)
diff --git a/src/bimu/server.c b/src/bimu/server.c
index f572290..565f0b1 100644
--- a/src/bimu/server.c
+++ b/src/bimu/server.c
@@ -10,14 +10,16 @@
 
 static void control_connection_closed_callback(void *userdata, struct aki_packet_stream *stream)
 {
-    (void)userdata;
+    struct bmu_node_connection *connection = (struct bmu_node_connection *)userdata;
     (void)stream;
+    aki_packet_stream_free(connection->control);
 }
 
 static void data_connection_closed_callback(void *userdata, struct aki_packet_stream *stream)
 {
-    (void)userdata;
+    struct bmu_node_connection *connection = (struct bmu_node_connection *)userdata;
     (void)stream;
+    aki_packet_stream_free(connection->data);
 }
 
 static void default_packet_sent_callback(void *userdata, struct aki_packet *packet)
@@ -43,8 +45,11 @@ static u8 packet_pool_callback(void *userdata, struct aki_packet *packet)
 {
     (void)userdata;
     struct bmu_node_connection *connection = (struct bmu_node_connection *)packet->userdata;
-    aki_packet_stream_send_packet(connection->data, packet);
-    return AKI_PACKET_POOL_KEEP;
+    if (connection->data->connected) {
+        aki_packet_stream_send_packet(connection->data, packet);
+        return AKI_PACKET_POOL_KEEP;
+    }
+    return AKI_PACKET_POOL_RETURN;
 }
 
 static void send_resume_packet(struct bmu_node_connection *connection, u64 seek_pos, u64 ts)
@@ -69,11 +74,11 @@ 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 ats = ts - START_DELAY;
     s64 seek_pos = connection->node->seek_pos;
     if (!connection->node->paused) {
         if (connection->node->start >= 0) {
-            if ((s64)ts < (connection->node->start - DELAY_GIVE)) {
+            if ((ats - DELAY_GIVE) < (u64)connection->node->start) {
                 ats = connection->node->start;
             } else {
                 seek_pos += ats - connection->node->start;
@@ -93,6 +98,13 @@ static void send_start_packet(struct bmu_node_connection *connection)
     aki_packet_stream_send_packet(connection->control, packet);
 }
 
+static void send_seek_packet(struct bmu_node_connection *connection)
+{
+    struct aki_packet *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)
 {
     struct bmu_node_connection *connection = (struct bmu_node_connection *)userdata;
@@ -119,7 +131,7 @@ static void control_packet_callback(void *userdata, struct aki_packet_stream *st
             node->start = -1;
         } else if (paused == BIMU_PAUSED_AND_RUNNING) {
             node->paused = false;
-            node->start = aki_get_timestamp() + START_DELAY;
+            node->start = aki_get_timestamp() - START_DELAY;
         }
         struct bmu_node_connection *rconnection;
         al_array_foreach(node->connections, i, rconnection) {
@@ -149,9 +161,7 @@ static void control_packet_callback(void *userdata, struct aki_packet_stream *st
     case BIMU_CONTROL_SEEK: {
         u64 pos = aki_packet_read_u64(packet);
         connection->seek_pos = pos;
-        struct aki_packet *npacket = aki_packet_create();
-        aki_packet_write_u8(npacket, BIMU_CONTROL_SEEK);
-        aki_packet_stream_send_packet(connection->control, npacket);
+        send_seek_packet(connection);
         break;
     }
     }
@@ -206,9 +216,9 @@ static void maybe_start_handler(struct bmu_node_connection *connection)
     if (connection->paused == BIMU_NOT_PAUSED) {
         if (node->start == -1) {
             node->paused = false;
-            node->start = aki_get_timestamp() + START_DELAY;
+            node->start = aki_get_timestamp() - START_DELAY;
         }
-        //send_resume_packet(connection, connection->seek_pos, node->start);
+        send_resume_packet(connection, connection->seek_pos, node->start);
         connection->paused = BIMU_PLAYING;
     } else if (connection->paused == BIMU_PAUSED) {
         connection->paused = BIMU_PAUSED_AND_RUNNING;
@@ -314,7 +324,6 @@ u16 bmu_server_create_node(struct bmu_server *server, struct cch_entry *entry)
     node->start = -1;
     node->seek_pos = 0;
     aki_packet_pool_init(&node->pool, 16, server->loop, packet_pool_callback, node);
-    //aki_packet_pool_init(&node->pool, 1400, server->loop, packet_pool_callback, node);
     node->connection_id = 1;
     al_array_init(node->connections);
     node->server = server;