10 files changed, 603 insertions, 82 deletions

diff --git a/src/bimu/client.c b/src/bimu/client.c
index f679c0d..d35e3a1 100644
--- a/src/bimu/client.c
+++ b/src/bimu/client.c
@@ -1 +1,188 @@
+#include "../codec/libav/packet_ext.h"
+
 #include "client.h"
+
+static void data_connection_callback(void *userdata, struct aki_packet_stream *stream)
+{
+    struct bmu_client *client = (struct bmu_client *)userdata;
+    (void)stream;
+    struct aki_packet *packet = aki_packet_create();
+    aki_packet_write_u8(packet, BIMU_CONNECTION_DATA);
+    aki_packet_write_u16(packet, client->connection_id);
+    aki_packet_write_u16(packet, client->node_id);
+    aki_packet_stream_send_packet(&client->data, packet);
+}
+
+static void data_connection_closed_callback(void *userdata, struct aki_packet_stream *stream)
+{
+    (void)userdata;
+    (void)stream;
+}
+
+static struct bmu_client_stream *stream_from_index(struct bmu_client *client, s32 index)
+{
+    struct bmu_client_stream *stream;
+    al_array_foreach(client->streams, i, stream) {
+        if (stream->index == index) return stream;
+    }
+    return NULL;
+}
+
+static void handle_data_packet_internal(struct bmu_client *client, struct aki_packet *packet)
+{
+    struct bmu_client_stream *stream;
+    switch (aki_packet_read_u8(packet)) {
+    case 0:
+        stream = stream_from_index(client, aki_packet_read_s32(packet));
+        stream->client->handle_data_packet(stream->client, packet);
+        break;
+    case 1:
+        al_array_foreach(client->streams, i, stream) {
+            stream->client->handle_eof(stream->client);
+        }
+        break;
+    case 2:
+        break;
+    }
+}
+
+static void data_packet_callback(void *userdata, struct aki_packet_stream *stream,
+    struct aki_packet *packet)
+{
+    struct bmu_client *client = (struct bmu_client *)userdata;
+    (void)stream;
+    handle_data_packet_internal(client, packet);
+    aki_packet_free(packet);
+}
+
+static void packet_sent_callback(void *userdata, struct aki_packet *packet)
+{
+    (void)userdata;
+    aki_packet_free(packet);
+}
+
+static 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);
+    aki_packet_stream_connect(&client->data, client->loop, &client->addr, client->port);
+}
+
+static void control_connection_callback(void *userdata, struct aki_packet_stream *stream)
+{
+    struct bmu_client *client = (struct bmu_client *)userdata;
+    (void)stream;
+    struct aki_packet *packet = aki_packet_create();
+    aki_packet_write_u8(packet, BIMU_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)
+{
+    (void)userdata;
+    (void)stream;
+}
+
+static void parse_info_packet(struct bmu_client *client, struct aki_packet *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);
+        stream->client = bmu_codec_client_create();
+        stream->client->callback = client->callback;
+        stream->client->userdata = client->userdata;
+        stream->stream.type = aki_packet_read_u8(packet);
+        switch (stream->stream.type) {
+        case CAMU_NORMAL:
+            stream->type = BIMU_STREAM_VIDEO;
+            stream->index = 0;
+            stream->stream.video.width = aki_packet_read_s32(packet);
+            stream->stream.video.height = aki_packet_read_s32(packet);
+            break;
+#ifdef HAVE_FFMPEG
+        case CAMU_FFMPEG_COMPAT: {
+            stream->stream.type = CAMU_FFMPEG_COMPAT;
+            stream->stream.av.format_context = avformat_alloc_context();
+            const AVCodec *codec = avcodec_find_decoder(aki_packet_read_av_codec_id(packet));
+            stream->stream.av.stream = aki_packet_read_av_stream(stream->stream.av.format_context, codec, packet);
+            stream->index = stream->stream.av.stream->index;
+            switch (stream->stream.av.stream->codecpar->codec_type) {
+            case AVMEDIA_TYPE_AUDIO:
+                stream->type = BIMU_STREAM_AUDIO;
+                al_array_push(client->subscribed, stream->index);
+                break;
+            case AVMEDIA_TYPE_VIDEO:
+                stream->type = BIMU_STREAM_VIDEO;
+                al_array_push(client->subscribed, stream->index);
+                break;
+            case AVMEDIA_TYPE_SUBTITLE:
+            default:
+                stream->type = BIMU_STREAM_UNKNOWN;
+                break;
+            }
+            break;
+        }
+#endif
+        }
+        if (!stream->client->init(stream->client, NULL, stream)) {
+        }
+        al_array_push(client->streams, stream);
+    }
+}
+
+static void send_subscribe_packet(struct bmu_client *client)
+{
+    struct aki_packet *packet = aki_packet_create();
+    aki_packet_write_u8(packet, BIMU_CONTROL_SUBSCRIBE);
+    aki_packet_write_u32(packet, client->subscribed.size);
+    s32 index;
+    al_array_foreach(client->subscribed, i, index) {
+        aki_packet_write_s32(packet, index);
+    }
+    aki_packet_stream_send_packet(&client->control, packet);
+}
+
+static void control_packet_callback(void *userdata, struct aki_packet_stream *stream,
+    struct aki_packet *packet)
+{
+    struct bmu_client *client = (struct bmu_client *)userdata;
+    (void)stream;
+
+    switch (aki_packet_read_u8(packet)) {
+    case BIMU_CONTROL_INFO:
+        client->connection_id = aki_packet_read_u16(packet);
+        parse_info_packet(client, packet);
+        send_subscribe_packet(client);
+        break;
+    case BIMU_CONTROL_START:
+        connect_data(client);
+        break;
+    default:
+        break;
+    }
+
+    aki_packet_free(packet);
+}
+
+void bmu_client_connect(struct bmu_client *client, struct aki_event_loop *loop, str *addr, s32 port, u16 node_id)
+{
+    client->loop = loop;
+    client->node_id = node_id;
+    al_str_clone(&client->addr, addr);
+    client->port = port;
+    client->corked = false;
+    al_array_init(client->subscribed);
+    al_array_init(client->streams);
+    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_connect(&client->control, client->loop, &client->addr, client->port);
+}
+
+void bmu_client_cork(struct bmu_client *client, bool cork)
+{
+    if (cork == client->corked) return;
+    aki_packet_stream_cork(&client->data, cork);
+    client->corked = cork;
+}
diff --git a/src/bimu/client.h b/src/bimu/client.h
index ec48275..54c903a 100644
--- a/src/bimu/client.h
+++ b/src/bimu/client.h
@@ -1,11 +1,33 @@
 #pragma once
 
 #include <al/types.h>
+#include <aki/packet_stream.h>
 
 #include "handlers/codec.h"
 
+#include "common.h"
+
 struct bmu_client_stream {
     u8 type;
     s32 index;
     struct camu_stream stream;
+    struct bmu_client_handler *client;
+};
+
+struct bmu_client {
+    struct aki_event_loop *loop;
+    u16 node_id;
+    u16 connection_id;
+    str addr;
+    s32 port;
+    struct aki_packet_stream control;
+    struct aki_packet_stream data;
+    bool corked;
+    array(s32) subscribed;
+    array(struct bmu_client_stream *) streams;
+    void (*callback)(void *, u8, struct bmu_client_stream *stream, void *);
+    void *userdata;
 };
+
+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);
diff --git a/src/bimu/common.h b/src/bimu/common.h
index 1c4897a..fe8e0af 100644
--- a/src/bimu/common.h
+++ b/src/bimu/common.h
@@ -5,3 +5,18 @@ enum {
     BIMU_STREAM_VIDEO,
     BIMU_STREAM_UNKNOWN,
 };
+
+enum {
+    BIMU_CONNECTION_CONTROL = 0,
+    BIMU_CONNECTION_DATA
+};
+
+enum {
+    BIMU_CONTROL_INFO = 0,
+    BIMU_CONTROL_SUBSCRIBE,
+    BIMU_CONTROL_START,
+    BIMU_CONTROL_TOGGLE_PAUSE,
+    BIMU_CONTROL_PAUSE,
+    BIMU_CONTROL_RESUME,
+    BIMU_CONTROL_SEEK
+};
diff --git a/src/bimu/handler.h b/src/bimu/handler.h
index e51bc2d..3881288 100644
--- a/src/bimu/handler.h
+++ b/src/bimu/handler.h
@@ -6,12 +6,14 @@
 #include "../cache/handle.h"
 
 struct bmu_server_handler {
-    bool (*init)(struct bmu_server_handler *, struct cch_handle *, struct aki_packet_pool *);
+    bool (*init)(struct bmu_server_handler *, struct cch_handle *, struct aki_packet_pool *, void *);
     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 *);
     bool (*seek)(struct bmu_server_handler *, s64);
     bool (*step)(struct bmu_server_handler *);
     void (*free)(struct bmu_server_handler **);
+    void *userdata;
 };
 
 enum {
diff --git a/src/bimu/handlers/codec_client.c b/src/bimu/handlers/codec_client.c
index ffe8c16..eda453f 100644
--- a/src/bimu/handlers/codec_client.c
+++ b/src/bimu/handlers/codec_client.c
@@ -19,9 +19,6 @@ static bool codec_client_init(struct bmu_client_handler *handler, struct camu_re
 {
     struct bmu_codec_client *codec = (struct bmu_codec_client *)handler;
     codec->dec = camu_lav_decoder_create();
-    //codec->dec = camu_spng_decoder_create();
-    //codec->dec = camu_stbi_decoder_create();
-    //codec->dec = camu_wuffs_decoder_create();
     codec->handler.stream = stream;
     if (!codec->dec->init(codec->dec, renderer, &stream->stream)) {
         return false;
diff --git a/src/bimu/handlers/codec_server.c b/src/bimu/handlers/codec_server.c
index b555370..3c27179 100644
--- a/src/bimu/handlers/codec_server.c
+++ b/src/bimu/handlers/codec_server.c
@@ -9,13 +9,11 @@
 
 #include "codec.h"
 
-static bool codec_server_init(struct bmu_server_handler *handler, struct cch_handle *handle, struct aki_packet_pool *pool)
+static bool codec_server_init(struct bmu_server_handler *handler, struct cch_handle *handle,
+    struct aki_packet_pool *pool, void *userdata)
 {
     struct bmu_codec_server *codec = (struct bmu_codec_server *)handler;
     codec->demux = camu_lav_demuxer_create();
-    //codec->demux = camu_spng_demuxer_create();
-    //codec->demux = camu_stbi_demuxer_create();
-    //codec->demux = camu_wuffs_demuxer_create();
     if (!codec->demux->init(codec->demux, handle)) {
         return false;
     }
@@ -29,6 +27,7 @@ static bool codec_server_init(struct bmu_server_handler *handler, struct cch_han
 #endif
     }
     codec->pool = pool;
+    codec->handler.userdata = userdata;
     return true;
 }
 
@@ -54,6 +53,14 @@ static void codec_server_write_info(struct bmu_server_handler *handler, struct a
     }
 }
 
+static void codec_server_subscribe(struct bmu_server_handler *handler, s32 index, bool subscribe)
+{
+    struct bmu_codec_server *codec = (struct bmu_codec_server *)handler;
+    if (subscribe) {
+        al_array_push(codec->demux->subscribed, index);
+    }
+}
+
 static s64 codec_server_get_duration(struct bmu_server_handler *handler)
 {
     struct bmu_codec_server *codec = (struct bmu_codec_server *)handler;
@@ -71,61 +78,38 @@ 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;
-    aki_packet_write_u8(packet, 0);
+    packet->userdata = codec->handler.userdata;
     s32 ret = codec->demux->get_packet(codec->demux, &codec->packet);
-    do {
-        if (ret == CAMU_OK) {
-            switch (codec->packet.type) {
-            case CAMU_NORMAL: {
-                aki_packet_write_s32(packet, 0);
-                aki_packet_write_u8(packet, 0);
-                aki_packet_write_u8(packet, codec->packet.type);
-                aki_packet_write_buffer(packet, codec->packet.buffer);
-                break;
-            }
-#ifdef HAVE_FFMPEG
-            case CAMU_FFMPEG_COMPAT: {
-                AVPacket *pkt = codec->packet.av.pkt;
-                aki_packet_write_s32(packet, pkt->stream_index);
-                aki_packet_write_u8(packet, 0);
-                aki_packet_write_u8(packet, codec->packet.type);
-                aki_packet_write_av_packet(packet, pkt);
-                av_packet_unref(pkt);
-                break;
-            }
-#endif
-            }
-            aki_packet_pool_submit(codec->pool, packet);
-        } else if (ret == CAMU_ERR_EOF) {
-            struct camu_stream *stream;
-            al_array_foreach_ptr(codec->demux->streams, i, stream) {
-                if (i != 0) { // HACK
-                    packet = aki_packet_pool_get(codec->pool);
-                    if (!packet) return false;
-                    aki_packet_write_u8(packet, 0);
-                }
-                switch (stream->type) {
-                case CAMU_NORMAL:
-                    aki_packet_write_s32(packet, 0);
-                    break;
+    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_u8(packet, codec->packet.type);
+            aki_packet_write_buffer(packet, codec->packet.buffer);
+            break;
+        }
 #ifdef HAVE_FFMPEG
-                case CAMU_FFMPEG_COMPAT:
-                    aki_packet_write_s32(packet, stream->av.stream->index);
-                    break;
+        case CAMU_FFMPEG_COMPAT: {
+            AVPacket *pkt = codec->packet.av.pkt;
+            aki_packet_write_s32(packet, pkt->stream_index);
+            aki_packet_write_u8(packet, codec->packet.type);
+            aki_packet_write_av_packet(packet, pkt);
+            av_packet_unref(pkt);
+            break;
+        }
 #endif
-                }
-                aki_packet_write_u8(packet, 1);
-                aki_packet_pool_submit(codec->pool, packet);
-            }
-            return false;
-        } else {
-            aki_packet_write_s32(packet, -2);
-            aki_packet_write_u8(packet, 2);
-            aki_packet_pool_submit(codec->pool, packet);
-            return false;
         }
-        break;
-    } while (1);
+        aki_packet_pool_submit(codec->pool, packet);
+    } else if (ret == CAMU_ERR_EOF) {
+        aki_packet_write_u8(packet, 1);
+        aki_packet_pool_submit(codec->pool, packet);
+        return false;
+    } else {
+        aki_packet_write_u8(packet, 2);
+        aki_packet_pool_submit(codec->pool, packet);
+        return false;
+    }
     return true;
 }
 
@@ -151,6 +135,7 @@ struct bmu_server_handler *bmu_codec_server_create(void)
     struct bmu_codec_server *codec = al_alloc_object(struct bmu_codec_server);
     codec->handler.init = codec_server_init;
     codec->handler.write_info = codec_server_write_info;
+    codec->handler.subscribe = codec_server_subscribe;
     codec->handler.get_duration = codec_server_get_duration;
     codec->handler.seek = codec_server_seek;
     codec->handler.step = codec_server_step;
diff --git a/src/bimu/local.c b/src/bimu/local.c
index 3f62f75..76824ed 100644
--- a/src/bimu/local.c
+++ b/src/bimu/local.c
@@ -34,10 +34,10 @@ static aki_thread_result AKI_THREADCALL stream_thread(void *userdata)
         }
         switch (aki_packet_read_u8(packet)) {
         case 0:
-            stream->client->handle_data_packet(stream->client, packet);
+            stream->s.client->handle_data_packet(stream->s.client, packet);
             break;
         case 1:
-            stream->client->handle_eof(stream->client);
+            stream->s.client->handle_eof(stream->s.client);
             break;
         }
         aki_packet_pool_return(&stream->runner->pool, packet);
@@ -76,7 +76,7 @@ bool bmu_local_init(struct bmu_local *runner, struct cch_entry *entry)
     cch_entry_get_handle(runner->entry, &runner->handle);
     aki_event_loop_init(&runner->loop);
     aki_packet_pool_init(&runner->pool, PACKETS, &runner->loop, packet_pool_callback, runner);
-    if (!runner->server->init(runner->server, &runner->handle, &runner->pool)) {
+    if (!runner->server->init(runner->server, &runner->handle, &runner->pool, NULL)) {
         return false;
     }
     al_array_init(runner->streams);
@@ -94,9 +94,9 @@ bool bmu_local_prepare_clients(struct bmu_local *runner, struct camu_renderer *r
         struct bmu_local_stream *local = al_alloc_object(struct bmu_local_stream);
         struct bmu_client_stream *stream = &local->s;
         local->runner = runner;
-        local->client = bmu_codec_client_create();
-        local->client->callback = runner->callback;
-        local->client->userdata = runner->userdata;
+        local->s.client = bmu_codec_client_create();
+        local->s.client->callback = runner->callback;
+        local->s.client->userdata = runner->userdata;
         aki_cond_init(&local->cond);
         aki_mutex_init(&local->mutex);
         aki_packet_cache_init(&local->cache, PACKETS);
@@ -131,7 +131,7 @@ bool bmu_local_prepare_clients(struct bmu_local *runner, struct camu_renderer *r
         }
 #endif
         }
-        if (!local->client->init(local->client, renderer, stream)) {
+        if (!local->s.client->init(local->s.client, renderer, stream)) {
             aki_packet_free(packet);
             return false;
         }
@@ -224,7 +224,7 @@ void bmu_local_seek(struct bmu_local *runner, f64 percent)
     al_atomic_bool_store(&runner->closed, false, AL_ATOMIC_RELAXED);
     aki_packet_pool_enable(&runner->pool);
     al_array_foreach(runner->streams, i, stream) {
-        stream->client->flush(stream->client);
+        stream->s.client->flush(stream->s.client);
         al_atomic_s32_store(&stream->state, BIMU_LOCAL_RUNNING, AL_ATOMIC_RELAXED);
         aki_packet_cache_enable(&stream->cache);
         aki_thread_create(&stream->thread, stream_thread, stream);
@@ -254,7 +254,7 @@ void bmu_local_close(struct bmu_local *runner)
 {
     struct bmu_local_stream *stream = NULL;
     al_array_foreach(runner->streams, i, stream) {
-        stream->client->free(&stream->client);
+        stream->s.client->free(&stream->s.client);
         switch (stream->s.stream.type) {
         case CAMU_NORMAL:
             break;
diff --git a/src/bimu/local.h b/src/bimu/local.h
index 1928249..10c0cc8 100644
--- a/src/bimu/local.h
+++ b/src/bimu/local.h
@@ -18,7 +18,6 @@ enum {
 
 struct bmu_local_stream {
     struct bmu_client_stream s;
-    struct bmu_client_handler *client;
     atomic_s32 state;
     struct aki_cond cond;
     struct aki_mutex mutex;
diff --git a/src/bimu/server.c b/src/bimu/server.c
index c1b0c38..f572290 100644
--- a/src/bimu/server.c
+++ b/src/bimu/server.c
@@ -1,46 +1,327 @@
+#include <al/random.h>
+#include <aki/common.h>
+
+#include "handlers/codec.h"
+
 #include "server.h"
 
-static void packet_callback(void *userdata, struct aki_packet_stream *stream, struct aki_packet *packet)
+#define START_DELAY 700000 // 700ms
+#define DELAY_GIVE 200000 // 200ms
+
+static void control_connection_closed_callback(void *userdata, struct aki_packet_stream *stream)
 {
     (void)userdata;
     (void)stream;
-    aki_packet_free(packet);
 }
 
-static void connection_closed_callback(void *userdata, struct aki_packet_stream *stream)
+static void data_connection_closed_callback(void *userdata, struct aki_packet_stream *stream)
 {
     (void)userdata;
     (void)stream;
 }
 
+static void default_packet_sent_callback(void *userdata, struct aki_packet *packet)
+{
+    (void)userdata;
+    aki_packet_free(packet);
+}
+
 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);
+}
+
+static void data_packet_callback(void *userdata, struct aki_packet_stream *stream, struct aki_packet *packet)
+{
+    (void)userdata;
+    (void)stream;
+    aki_packet_free(packet);
+}
+
+static u8 packet_pool_callback(void *userdata, struct aki_packet *packet)
+{
     (void)userdata;
-    (void)packet;
+    struct bmu_node_connection *connection = (struct bmu_node_connection *)packet->userdata;
+    aki_packet_stream_send_packet(connection->data, packet);
+    return AKI_PACKET_POOL_KEEP;
+}
+
+static void send_resume_packet(struct bmu_node_connection *connection, u64 seek_pos, u64 ts)
+{
+    struct aki_packet *packet = aki_packet_create();
+    aki_packet_write_u8(packet, BIMU_CONTROL_RESUME);
+    aki_packet_write_u64(packet, seek_pos);
+    aki_packet_write_u64(packet, ts);
+    aki_packet_stream_send_packet(connection->control, packet);
+}
+
+static void send_pause_packet(struct bmu_node_connection *connection, u64 seek_pos)
+{
+    struct aki_packet *packet = aki_packet_create();
+    aki_packet_write_u8(packet, BIMU_CONTROL_PAUSE);
+    aki_packet_write_u64(packet, seek_pos);
+    aki_packet_stream_send_packet(connection->control, packet);
+}
+
+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;
+    s64 seek_pos = connection->node->seek_pos;
+    if (!connection->node->paused) {
+        if (connection->node->start >= 0) {
+            if ((s64)ts < (connection->node->start - DELAY_GIVE)) {
+                ats = connection->node->start;
+            } else {
+                seek_pos += ats - connection->node->start;
+            }
+        } else {
+            connection->node->start = ats;
+        }
+    }
+    s64 duration = connection->handler->get_duration(connection->handler);
+    if (seek_pos > duration) seek_pos = duration;
+    // -1 signals to not call seek. We have to distinguish between a seek to 0
+    // 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_u8(packet, connection->node->paused);
+    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;
+    (void)stream;
+
+    switch (aki_packet_read_u8(packet)) {
+    case BIMU_CONTROL_SUBSCRIBE: {
+        u32 count = aki_packet_read_u32(packet);
+        for (u32 i = 0; i < count; i++) {
+            connection->handler->subscribe(connection->handler, aki_packet_read_s32(packet), true);
+        }
+        send_start_packet(connection);
+        break;
+    }
+    case BIMU_CONTROL_TOGGLE_PAUSE: {
+        struct bmu_node *node = connection->node;
+        u64 pos = aki_packet_read_u64(packet);
+        u8 paused = connection->paused;
+        if (paused == BIMU_PLAYING) {
+            node->seek_pos = pos;
+            node->paused = true;
+            // If node is paused, start will not be used for new connections.
+            // So, set it to -1 and update it on the first resume from any connection.
+            node->start = -1;
+        } else if (paused == BIMU_PAUSED_AND_RUNNING) {
+            node->paused = false;
+            node->start = aki_get_timestamp() + START_DELAY;
+        }
+        struct bmu_node_connection *rconnection;
+        al_array_foreach(node->connections, i, rconnection) {
+            switch (rconnection->paused) {
+            case BIMU_PLAYING:
+                al_assert(paused == BIMU_PLAYING || paused == BIMU_NOT_PAUSED);
+                rconnection->paused = BIMU_PAUSED;
+                send_pause_packet(rconnection, node->seek_pos);
+                break;
+            case BIMU_NOT_PAUSED:
+                al_assert(paused != BIMU_PAUSED);
+                rconnection->paused = BIMU_PAUSED;
+                break;
+            case BIMU_PAUSED:
+                al_assert(paused != BIMU_NOT_PAUSED);
+                rconnection->paused = BIMU_NOT_PAUSED;
+                break;
+            case BIMU_PAUSED_AND_RUNNING:
+                al_assert(paused != BIMU_NOT_PAUSED);
+                send_resume_packet(rconnection, node->seek_pos, node->start);
+                rconnection->paused = BIMU_PLAYING;
+                break;
+            }
+        }
+        break;
+    }
+    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);
+        break;
+    }
+    }
+    aki_packet_free(packet);
 }
 
-static void flushed_callback(void *userdata)
+static struct bmu_node *get_node_by_id(struct bmu_server *server, u16 id)
+{
+    struct bmu_node *node;
+    al_array_foreach(server->nodes, i, node) {
+        if (node->id == id) return node;
+    }
+    return NULL;
+}
+
+static struct bmu_node_connection *get_connection_by_id(struct bmu_node *node, u16 id)
+{
+    struct bmu_node_connection *connection;
+    al_array_foreach(node->connections, i, connection) {
+        if (connection->id == id) return connection;
+    }
+    return NULL;
+}
+
+static void send_control_info_packet(struct bmu_node_connection *connection)
+{
+    struct aki_packet *packet = aki_packet_create();
+    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->write_info(connection->handler, packet);
+    aki_packet_stream_send_packet(connection->control, packet);
+}
+
+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)) {}
+    return 0;
+}
+
+static void maybe_start_handler(struct bmu_node_connection *connection)
+{
+    al_assert(!connection->running);
+    if (connection->seek_pos >= 0) {
+        connection->handler->seek(connection->handler, connection->seek_pos);
+    }
+    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) {
+        connection->paused = BIMU_PAUSED_AND_RUNNING;
+    }
+}
+
+static void identify_packet_callback(void *userdata, struct aki_packet_stream *stream, struct aki_packet *packet)
+{
+    struct bmu_server *server = (struct bmu_server *)userdata;
+
+    u8 type = aki_packet_read_u8(packet);
+
+    u16 connection_id = aki_packet_read_u16(packet);
+    u16 node_id = aki_packet_read_u16(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.
+    //   - Node with id doesn't exist.
+    //   - Control connected more than once and/or after info is sent.
+    //   - Data connected before control.
+    //   - Data connected more than once.
+
+    struct bmu_node *node = get_node_by_id(server, node_id);
+    al_assert(node);
+
+    struct bmu_node_connection *connection = get_connection_by_id(node, connection_id);
+    if (!connection) {
+        al_assert(type == BIMU_CONNECTION_CONTROL);
+        connection = al_alloc_object(struct bmu_node_connection);
+        al_array_push(node->connections, connection);
+        connection->id = node->connection_id++;
+        connection->node = node;
+        connection->control = NULL;
+        connection->data = NULL;
+        cch_entry_get_handle(connection->node->entry, &connection->handle);
+        connection->paused = BIMU_PLAYING;
+        connection->running = false;
+    }
+
+    switch (type) {
+    case BIMU_CONNECTION_CONTROL:
+        aki_packet_stream_set_no_delay(stream, true);
+        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);
+        break;
+    case BIMU_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?
+        connection->data = stream;
+        stream->packet_callback = data_packet_callback;
+        stream->packet_sent_callback = packet_sent_callback;
+        stream->connection_closed_callback = data_connection_closed_callback;
+        stream->userdata = connection;
+        maybe_start_handler(connection);
+        break;
+    }
+
+    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)
 {
-    stream->userdata = userdata;
-    stream->packet_callback = packet_callback;
-    stream->packet_sent_callback = packet_sent_callback;
+    struct bmu_server *server = (struct bmu_server *)userdata;
     stream->connection_closed_callback = connection_closed_callback;
-    stream->flushed_callback = flushed_callback;
+    stream->packet_callback = identify_packet_callback;
+    stream->packet_sent_callback = default_packet_sent_callback;
+    stream->flushed_callback = NULL;
+    stream->userdata = server;
 }
 
 bool bmu_server_init(struct bmu_server *server)
 {
-    return aki_packet_stream_init(&server->server, AKI_SOCKET_TCP, connection_callback,
-        NULL, NULL, NULL, server);
+    al_array_init(server->nodes);
+    return aki_packet_stream_init(&server->server, AKI_SOCKET_TCP, connection_callback, NULL, NULL, NULL, server);
 }
 
 void bmu_server_listen(struct bmu_server *server, struct aki_event_loop *loop, str *addr, s32 port)
 {
     server->loop = loop;
-    aki_packet_stream_listen(&server->server, server->loop, addr, port);
+    aki_packet_stream_listen(&server->server, loop, addr, port);
+}
+
+u16 bmu_server_create_node(struct bmu_server *server, struct cch_entry *entry)
+{
+    struct bmu_node *node = al_alloc_object(struct bmu_node);
+    al_array_push(server->nodes, node);
+    node->id = al_rand_u16();
+    node->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);
+    //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;
+    return node->id;
+}
+
+void bmu_server_close(struct bmu_server *server)
+{
+    (void)server;
 }
diff --git a/src/bimu/server.h b/src/bimu/server.h
index ba3093e..754ab7f 100644
--- a/src/bimu/server.h
+++ b/src/bimu/server.h
@@ -1,21 +1,54 @@
 #pragma once
 
+#include <al/array.h>
 #include <aki/packet_stream.h>
+#include <aki/packet_pool.h>
 
 #include "../cache/entry.h"
+#include "../cache/handle.h"
+
+#include "common.h"
+#include "handler.h"
+
+enum {
+    BIMU_PLAYING = 0,
+    BIMU_NOT_PAUSED,
+    BIMU_PAUSED,
+    BIMU_PAUSED_AND_RUNNING
+};
 
 struct bmu_node_connection {
+    u16 id;
+    struct bmu_node *node;
+    struct aki_packet_stream *control;
+    struct aki_packet_stream *data;
+    bool running;
+    u8 paused;
+    s64 seek_pos;
     struct cch_handle handle;
+    struct bmu_server_handler *handler;
+    struct aki_thread thread;
 };
 
 struct bmu_node {
+    u16 id;
     struct cch_entry *entry;
+    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;
 };
 
 struct bmu_server {
-    struct aki_packet_stream server;
     struct aki_event_loop *loop;
+    struct aki_packet_stream server;
+    array(struct bmu_node *) nodes;
 };
 
 bool bmu_server_init(struct bmu_server *server);
 void bmu_server_listen(struct bmu_server *server, struct aki_event_loop *loop, str *addr, s32 port);
+u16 bmu_server_create_node(struct bmu_server *server, struct cch_entry *entry);
+void bmu_server_close(struct bmu_server *server);