wip

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

9 files changed, 313 insertions, 110 deletions

diff --git a/src/bimu/client.c b/src/bimu/client.c
index c592906..03910ee 100644
--- a/src/bimu/client.c
+++ b/src/bimu/client.c
@@ -15,6 +15,7 @@ static void data_connection_callback(void *userdata, struct aki_packet_stream *s
     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_write_u64(packet, client->seek_pos);
     aki_packet_stream_send_packet(&client->data, packet);
 }
 
@@ -22,62 +23,35 @@ 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_write_u32(packet, client->level);
     aki_packet_stream_send_packet(&client->control, packet);
 }
 
-static void flush_streams(struct bmu_client *client)
+static void cleanup_connection_internal(struct bmu_client *client)
 {
-    struct bmu_client_stream *stream;
-    al_array_foreach(client->streams, i, stream) {
-        stream->client->flush(stream->client);
-    }
+    client->callback(client->userdata, BIMU_CLIENT_CLOSED, NULL, NULL);
 }
 
 static void data_connection_closed_callback(void *userdata, struct aki_packet_stream *stream)
 {
     struct bmu_client *client = (struct bmu_client *)userdata;
     (void)stream;
+    if (client->closed) {
+        if (!client->control.connected) cleanup_connection_internal(client);
+        return;
+    }
     aki_packet_stream_free(&client->data);
     client->callback(client->userdata, BIMU_CLIENT_REMOVE_BUFFERS, NULL, NULL);
-    flush_streams(client);
+    bmu_vcr_flush(&client->vcr);
     send_reconnect_packet(client);
 }
 
-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 BIMU_PACKET_DATA:
-        stream = stream_from_index(client, aki_packet_read_s32(packet));
-        stream->client->handle_data_packet(stream->client, packet);
-        break;
-    case BIMU_PACKET_EOF:
-        al_array_foreach(client->streams, i, stream) {
-            stream->client->handle_eof(stream->client);
-        }
-        break;
-    case BIMU_PACKET_ERROR:
-        al_log_warn("bimu", "Unhandled error packet.");
-        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);
+    bmu_vcr_push_packet(&client->vcr, packet);
 }
 
 static void packet_sent_callback(void *userdata, struct aki_packet *packet)
@@ -106,8 +80,14 @@ static void control_connection_callback(void *userdata, struct aki_packet_stream
 
 static void control_connection_closed_callback(void *userdata, struct aki_packet_stream *stream)
 {
-    (void)userdata;
+    struct bmu_client *client = (struct bmu_client *)userdata;
     (void)stream;
+    if (client->closed) {
+        if (!client->data.connected) cleanup_connection_internal(client);
+        return;
+    }
+    aki_packet_stream_free(&client->control);
+    // TODO: Reconnect.
 }
 
 static void parse_info_packet(struct bmu_client *client, struct aki_packet *packet)
@@ -115,7 +95,7 @@ static void parse_info_packet(struct bmu_client *client, struct aki_packet *pack
     client->duration = aki_packet_read_u64(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);
+        struct bmu_vcr_stream *stream = al_alloc_object(struct bmu_vcr_stream);
         stream->client = bmu_codec_client_create();
         stream->client->callback = client->callback;
         stream->client->userdata = client->userdata;
@@ -154,7 +134,7 @@ static void parse_info_packet(struct bmu_client *client, struct aki_packet *pack
         }
         if (!stream->client->init(stream->client, NULL, stream)) {
         }
-        al_array_push(client->streams, stream);
+        bmu_vcr_add_stream(&client->vcr, stream);
     }
 }
 
@@ -172,26 +152,12 @@ 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 ts = aki_packet_read_u64(packet);
-    u8 paused = aki_packet_read_u8(packet);
-    (void)paused;
-    struct bmu_seek_req req = {
-        .base = seek_pos,
-        .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 delay = aki_packet_read_u64(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
-    };
+    struct bmu_seek_req req = { .base = seek_pos, .ts = ts, .delay = delay };
     client->callback(client->userdata, BIMU_CLIENT_SET, NULL, &req);
+    client->seek_pos = req.base;
 }
 
 static void control_packet_callback(void *userdata, struct aki_packet_stream *stream,
@@ -213,9 +179,9 @@ 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->level = 0;
         client->corked = false;
         aki_packet_stream_disconnect(&client->data);
         break;
@@ -232,19 +198,22 @@ void bmu_client_connect(struct bmu_client *client, struct aki_event_loop *loop,
     client->node_id = node_id;
     al_str_clone(&client->addr, addr);
     client->port = port;
+    client->level = 0;
     client->corked = false;
     al_array_init(client->subscribed);
-    al_array_init(client->streams);
+    bmu_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);
     aki_packet_stream_connect(&client->control, client->loop, &client->addr, client->port);
 }
 
-void bmu_client_cork(struct bmu_client *client, bool cork)
+void bmu_client_reseek(struct bmu_client *client)
 {
-    if (cork == client->corked) return;
-    aki_packet_stream_cork(&client->data, cork);
-    client->corked = cork;
+    client->level++;
+    al_log_debug("bimu", "Attempting re-seek at level %i.", client->level);
+    client->corked = false;
+    aki_packet_stream_disconnect(&client->data);
 }
 
 void bmu_client_seek(struct bmu_client *client, f64 percent)
@@ -256,8 +225,22 @@ void bmu_client_seek(struct bmu_client *client, f64 percent)
     aki_packet_stream_send_packet(&client->control, packet);
 }
 
+void bmu_client_close(struct bmu_client *client)
+{
+    struct bmu_vcr_stream *stream;
+    al_array_foreach(client->vcr.streams, i, stream) {
+        bmu_vcr_stream_close(stream);
+    }
+    client->closed = true;
+    aki_packet_stream_disconnect(&client->control);
+    aki_packet_stream_disconnect(&client->data);
+}
+
 void bmu_client_free(struct bmu_client *client)
 {
     aki_packet_stream_free(&client->control);
     aki_packet_stream_free(&client->data);
+    al_str_free(&client->addr);
+    al_array_free(client->subscribed);
+    bmu_vcr_free(&client->vcr);
 }
diff --git a/src/bimu/client.h b/src/bimu/client.h
index 8c9803b..db1f9e5 100644
--- a/src/bimu/client.h
+++ b/src/bimu/client.h
@@ -3,14 +3,7 @@
 #include <al/types.h>
 #include <aki/packet_stream.h>
 
-#include "../codec/codec.h"
-
-struct bmu_client_stream {
-    u8 type;
-    s32 index;
-    struct camu_stream stream;
-    struct bmu_client_handler *client;
-};
+#include "vcr.h"
 
 struct bmu_client {
     struct aki_event_loop *loop;
@@ -20,15 +13,19 @@ struct bmu_client {
     s32 port;
     struct aki_packet_stream control;
     struct aki_packet_stream data;
+    bool closed;
+    u32 level;
     bool corked;
     u64 duration;
+    u64 seek_pos;
     array(s32) subscribed;
-    array(struct bmu_client_stream *) streams;
-    void (*callback)(void *, u8, struct bmu_client_stream *stream, void *);
+    struct bmu_vcr vcr;
+    void (*callback)(void *, u8, struct bmu_vcr_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);
+void bmu_client_reseek(struct bmu_client *client);
 void bmu_client_seek(struct bmu_client *client, f64 percent);
+void bmu_client_close(struct bmu_client *client);
 void bmu_client_free(struct bmu_client *client);
diff --git a/src/bimu/handler.h b/src/bimu/handler.h
index 3646dd6..c6b482d 100644
--- a/src/bimu/handler.h
+++ b/src/bimu/handler.h
@@ -5,6 +5,8 @@
 #include "../codec/codec.h"
 #include "../cache/handle.h"
 
+#include "vcr.h"
+
 struct bmu_server_handler {
     bool (*init)(struct bmu_server_handler *, struct cch_handle *, struct aki_packet_pool *);
     void (*write_info)(struct bmu_server_handler *, struct aki_packet *);
@@ -32,17 +34,19 @@ enum {
 };
 
 struct bmu_seek_req {
-    u64 base, start;
+    u64 base;
+    u64 ts;
+    u64 delay;
 };
 
 struct bmu_client_stream;
 struct bmu_client_handler {
-    bool (*init)(struct bmu_client_handler *, struct camu_renderer *, struct bmu_client_stream *);
+    bool (*init)(struct bmu_client_handler *, struct camu_renderer *, struct bmu_vcr_stream *);
     void (*handle_data_packet)(struct bmu_client_handler *, struct aki_packet *);
     void (*handle_eof)(struct bmu_client_handler *);
     void (*flush)(struct bmu_client_handler *);
     void (*free)(struct bmu_client_handler **);
-    struct bmu_client_stream *stream;
-    void (*callback)(void *, u8, struct bmu_client_stream *stream, void *);
+    struct bmu_vcr_stream *stream;
+    void (*callback)(void *, u8, struct bmu_vcr_stream *stream, void *);
     void *userdata;
 };
diff --git a/src/bimu/handlers/codec_client.c b/src/bimu/handlers/codec_client.c
index 570ef97..79ef27d 100644
--- a/src/bimu/handlers/codec_client.c
+++ b/src/bimu/handlers/codec_client.c
@@ -2,7 +2,7 @@
 #include "../../codec/libav/decoder.h"
 #include "../../codec/libav/packet_ext.h"
 
-#include "../../bimu/client.h"
+#include "../../bimu/vcr.h"
 
 #include "codec.h"
 
@@ -13,7 +13,7 @@ static void data_callback(void *userdata, struct camu_frame *frame)
 }
 
 static bool codec_client_init(struct bmu_client_handler *handler, struct camu_renderer *renderer,
-    struct bmu_client_stream *stream)
+    struct bmu_vcr_stream *stream)
 {
     struct bmu_codec_client *codec = (struct bmu_codec_client *)handler;
     codec->dec = camu_lav_decoder_create();
diff --git a/src/bimu/meson.build b/src/bimu/meson.build
index dce7f7b..c7de266 100644
--- a/src/bimu/meson.build
+++ b/src/bimu/meson.build
@@ -5,6 +5,7 @@ bimu_server_src = [
 
 bimu_client_src = [
   'client.c',
+  'vcr.c',
   'handlers/codec_client.c'
 ]
 
diff --git a/src/bimu/server.c b/src/bimu/server.c
index e1aedce..6db0dd2 100644
--- a/src/bimu/server.c
+++ b/src/bimu/server.c
@@ -4,9 +4,7 @@
 #include "handlers/codec.h"
 
 #include "server.h"
-
-#define START_DELAY 3000000 // 3s
-#define DELAY_GIVE 200000 // 200ms
+#include "common.h"
 
 static void control_connection_closed_callback(void *userdata, struct aki_packet_stream *stream)
 {
@@ -21,6 +19,9 @@ static void control_connection_closed_callback(void *userdata, struct aki_packet
     }
     aki_packet_stream_free(connection->control);
     connection->control = NULL;
+    //cch_entry_return_handle(connection->node->entry, &connection->handle);
+    //connection->handler->free(&connection->handler);
+    //aki_packet_pool_free(&connection->pool);
 }
 
 static void data_connection_closed_callback(void *userdata, struct aki_packet_stream *stream)
@@ -81,31 +82,33 @@ static void send_pause_packet(struct bmu_node_connection *connection, u64 seek_p
     aki_packet_stream_send_packet(connection->control, packet);
 }
 
-static void send_start_packet(struct bmu_node_connection *connection)
+#define DELAY 275000Lu // 275ms
+#define DELAY_GIVE 30000Lu // 30ms
+
+static void send_start_packet(struct bmu_node_connection *connection, u32 level)
 {
     struct aki_packet *packet = aki_packet_create();
     aki_packet_write_u8(packet, BIMU_CONTROL_START);
-    u64 ts = aki_get_timestamp() - START_DELAY;
+    u64 ts = aki_get_timestamp();
+    u64 delay = DELAY * powl(3, level);
     u64 seek_pos = connection->node->seek_pos;
-    if (!connection->node->paused) {
-        if (connection->node->start >= 0) {
+    if (!(connection->node->flags & BIMU_NODE_PAUSED)) {
+        if (connection->node->flags & BIMU_NODE_STARTED) {
             if ((ts - DELAY_GIVE) < (u64)connection->node->start) {
+                // Current time is close enough to node start.
                 ts = connection->node->start;
             } else {
-                seek_pos += ts - connection->node->start;
+                seek_pos += (ts - connection->node->start) + (delay - DELAY);
             }
         } else {
+            // Must be set at level 0, this applies for seeks too.
             connection->node->start = ts;
+            connection->node->flags |= BIMU_NODE_STARTED;
         }
     }
-    u64 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, ts);
-    aki_packet_write_u8(packet, connection->node->paused);
+    aki_packet_write_u64(packet, delay);
+    aki_packet_write_u64(packet, seek_pos);
     aki_packet_stream_send_packet(connection->control, packet);
 }
 
@@ -131,11 +134,11 @@ static void control_packet_callback(void *userdata, struct aki_packet_stream *st
         for (u32 i = 0; i < count; i++) {
             connection->handler->subscribe(connection->handler, aki_packet_read_s32(packet), true);
         }
-        send_start_packet(connection);
+        send_start_packet(connection, 0);
         break;
     }
     case BIMU_CONTROL_RECONNECT: {
-        send_start_packet(connection);
+        send_start_packet(connection, aki_packet_read_u32(packet));
         break;
     }
     case BIMU_CONTROL_TOGGLE_PAUSE: {
@@ -144,13 +147,10 @@ static void control_packet_callback(void *userdata, struct aki_packet_stream *st
         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;
+            node->flags |= BIMU_NODE_PAUSED;
+            // Update node start?
         } else if (paused == BIMU_PAUSED_AND_RUNNING) {
-            node->paused = false;
-            node->start = aki_get_timestamp() - START_DELAY;
+            node->flags &= ~BIMU_NODE_PAUSED;
         }
         struct bmu_node_connection *rconnection;
         al_array_foreach(node->connections, i, rconnection) {
@@ -180,8 +180,8 @@ 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;
+        connection->node->flags |= BIMU_NODE_SEEKED;
+        connection->node->flags &= ~BIMU_NODE_STARTED;
         send_seek_packet(connection->node);
         break;
     }
@@ -228,7 +228,8 @@ static aki_thread_result AKI_THREADCALL handler_thread(void *userdata)
 static void maybe_start_handler(struct bmu_node_connection *connection)
 {
     al_assert(!connection->running);
-    if (connection->seek_pos >= 0) {
+    // Don't seek to 0 unless NODE_SEEKED is set.
+    if (connection->seek_pos > 0 || (connection->node->flags & BIMU_NODE_SEEKED)) {
         connection->handler->seek(connection->handler, connection->seek_pos);
     }
     connection->running = 1;
@@ -279,7 +280,7 @@ static void identify_packet_callback(void *userdata, struct aki_packet_stream *s
 
     switch (type) {
     case BIMU_CONNECTION_CONTROL:
-        aki_packet_stream_set_no_delay(stream, true);
+        aki_packet_stream_set_no_delay(stream, 1);
         connection->control = stream;
         stream->packet_callback = control_packet_callback;
         stream->packet_sent_callback = default_packet_sent_callback;
@@ -293,6 +294,7 @@ static void identify_packet_callback(void *userdata, struct aki_packet_stream *s
         // (disconnect and reconnect very fast?).
         //   - Call maybe_cleanup_data in this function?
         al_assert(!connection->data);
+        connection->seek_pos = aki_packet_read_u64(packet);
         connection->data = stream;
         stream->packet_callback = data_packet_callback;
         stream->packet_sent_callback = packet_sent_callback;
@@ -339,8 +341,8 @@ u16 bmu_server_create_node(struct bmu_server *server, struct cch_entry *entry)
     al_array_push(server->nodes, node);
     node->id = al_rand_u16();
     node->entry = entry;
-    node->paused = false;
-    node->start = -1;
+    node->flags = 0;
+    node->start = 0;
     node->seek_pos = 0;
     node->connection_id = 1;
     al_array_init(node->connections);
diff --git a/src/bimu/server.h b/src/bimu/server.h
index 09fed99..20f73c6 100644
--- a/src/bimu/server.h
+++ b/src/bimu/server.h
@@ -7,7 +7,6 @@
 #include "../cache/entry.h"
 #include "../cache/handle.h"
 
-#include "common.h"
 #include "handler.h"
 
 enum {
@@ -31,10 +30,16 @@ struct bmu_node_connection {
     struct aki_thread thread;
 };
 
+enum {
+    BIMU_NODE_STARTED = 1,
+    BIMU_NODE_SEEKED = 1 << 1,
+    BIMU_NODE_PAUSED = 1 << 2
+};
+
 struct bmu_node {
     u16 id;
     struct cch_entry *entry;
-    bool paused;
+    u8 flags;
     u64 start;
     u64 seek_pos;
     u16 connection_id;
diff --git a/src/bimu/vcr.c b/src/bimu/vcr.c
new file mode 100644
index 0000000..6c39d86
--- /dev/null
+++ b/src/bimu/vcr.c
@@ -0,0 +1,166 @@
+#include <al/log.h>
+
+#include "vcr.h"
+#include "handler.h"
+
+static void signal_callback(void *userdata)
+{
+    struct bmu_vcr *vcr = (struct bmu_vcr *)userdata;
+    aki_packet_stream_cork(vcr->data, false);
+}
+
+void bmu_vcr_init(struct bmu_vcr *vcr, struct aki_event_loop *loop, struct aki_packet_stream *data)
+{
+    al_array_init(vcr->streams);
+    al_atomic_u64_store(&vcr->count, 0, AL_ATOMIC_RELAXED);
+    vcr->data = data;
+    aki_signal_init(&vcr->signal, signal_callback, vcr);
+    aki_signal_start(&vcr->signal, loop);
+}
+
+static aki_thread_result AKI_THREADCALL vcr_stream_thread(void *userdata)
+{
+    struct bmu_vcr_stream *stream = (struct bmu_vcr_stream *)userdata;
+    struct bmu_vcr *vcr = stream->vcr;
+    do {
+        if (!aki_packet_cache_wait(&stream->cache)) break;
+        struct aki_packet *packet = aki_packet_cache_pop(&stream->cache);
+        if (!packet) {
+            stream->client->handle_eof(stream->client);
+            break;
+        }
+        if (al_atomic_u64_sub(&vcr->count, 1, AL_ATOMIC_RELAXED) < 128 && stream->buffered) {
+            aki_signal_send(&vcr->signal);
+        }
+        if (al_atomic_s32_load(&stream->state, AL_ATOMIC_RELAXED) == BIMU_STREAM_STOPPED) {
+            aki_mutex_lock(&stream->mutex);
+            aki_cond_wait(&stream->cond, &stream->mutex);
+            aki_mutex_unlock(&stream->mutex);
+        }
+        if (al_atomic_s32_load(&stream->state, AL_ATOMIC_RELAXED) == BIMU_STREAM_CLOSED) {
+            aki_packet_free(packet);
+            break;
+        }
+        stream->client->handle_data_packet(stream->client, packet);
+        aki_packet_free(packet);
+    } while (1);
+    return 0;
+}
+
+void bmu_vcr_add_stream(struct bmu_vcr *vcr, struct bmu_vcr_stream *stream)
+{
+    stream->vcr = vcr;
+    aki_cond_init(&stream->cond);
+    aki_mutex_init(&stream->mutex);
+    aki_packet_cache_init(&stream->cache, 256);
+    stream->buffered = false;
+    al_array_push(vcr->streams, stream);
+    al_atomic_s32_store(&stream->state, BIMU_STREAM_RUNNING, AL_ATOMIC_RELAXED);
+}
+
+static struct bmu_vcr_stream *get_stream_from_index(struct bmu_vcr *vcr, s32 index)
+{
+    struct bmu_vcr_stream *stream;
+    al_array_foreach(vcr->streams, i, stream) {
+        if (stream->index == index) return stream;
+    }
+    return NULL;
+}
+
+bool bmu_vcr_push_packet(struct bmu_vcr *vcr, struct aki_packet *packet)
+{
+    struct bmu_vcr_stream *stream;
+    switch (aki_packet_read_u8(packet)) {
+    case BIMU_PACKET_DATA:
+        stream = get_stream_from_index(vcr, aki_packet_read_s32(packet));
+        al_assert(stream);
+        if (!stream->running) {
+            aki_thread_create(&stream->thread, vcr_stream_thread, stream);
+            stream->running = true;
+        }
+        if (!aki_packet_cache_send_packet(&stream->cache, packet)) {
+            aki_packet_free(packet);
+        } else if (al_atomic_u64_add(&vcr->count, 1, AL_ATOMIC_RELAXED) >= 196) {
+            aki_packet_stream_cork(vcr->data, true);
+            al_array_foreach(vcr->streams, i, stream) {
+                stream->buffered = true;
+            }
+        }
+        break;
+    case BIMU_PACKET_EOF:
+        al_array_foreach(vcr->streams, i, stream) {
+            aki_packet_cache_send_packet(&stream->cache, NULL);
+        }
+        aki_packet_free(packet);
+        break;
+    case BIMU_PACKET_ERROR:
+        aki_packet_free(packet);
+        al_log_warn("bimu", "Unhandled error packet.");
+        break;
+    }
+    return true;
+}
+
+static void vcr_stream_close_internal(struct bmu_vcr_stream *stream)
+{
+    al_atomic_s32_store(&stream->state, BIMU_STREAM_CLOSED, AL_ATOMIC_RELAXED);
+    aki_packet_cache_disable(&stream->cache);
+    aki_mutex_lock(&stream->mutex);
+    if (aki_cond_is_waiting(&stream->cond)) {
+        aki_cond_signal(&stream->cond);
+    }
+    aki_mutex_unlock(&stream->mutex);
+    aki_thread_join(&stream->thread);
+    struct aki_packet *packet;
+    while ((packet = aki_packet_cache_pop(&stream->cache))) {
+        aki_packet_free(packet);
+    }
+}
+
+void bmu_vcr_stream_close(struct bmu_vcr_stream *stream)
+{
+    vcr_stream_close_internal(stream);
+}
+
+void bmu_vcr_flush(struct bmu_vcr *vcr)
+{
+    struct bmu_vcr_stream *stream;
+    al_array_foreach(vcr->streams, i, stream) {
+        vcr_stream_close_internal(stream);
+        stream->client->flush(stream->client);
+        aki_packet_cache_enable(&stream->cache);
+        stream->running = false;
+        al_atomic_s32_store(&stream->state, BIMU_STREAM_RUNNING, AL_ATOMIC_RELAXED);
+    }
+    al_atomic_u64_store(&vcr->count, 0, AL_ATOMIC_RELAXED);
+}
+
+void bmu_vcr_stream_cork(struct bmu_vcr_stream *stream)
+{
+    al_atomic_s32_store(&stream->state, BIMU_STREAM_STOPPED, AL_ATOMIC_RELAXED);
+}
+
+void bmu_vcr_stream_uncork(struct bmu_vcr_stream *stream)
+{
+    al_atomic_s32_store(&stream->state, BIMU_STREAM_RUNNING, AL_ATOMIC_RELAXED);
+    aki_mutex_lock(&stream->mutex);
+    if (aki_cond_is_waiting(&stream->cond)) {
+        aki_cond_signal(&stream->cond);
+    }
+    aki_mutex_unlock(&stream->mutex);
+}
+
+void bmu_vcr_free(struct bmu_vcr *vcr)
+{
+    aki_signal_stop(&vcr->signal);
+    struct bmu_vcr_stream *stream;
+    al_array_foreach(vcr->streams, i, stream) {
+        aki_packet_cache_free(&stream->cache);
+        stream->client->free(&stream->client);
+        if (stream->stream.type == CAMU_FFMPEG_COMPAT) {
+            avformat_free_context(stream->stream.av.format_context);
+        }
+        al_free(stream);
+    }
+    al_array_free(vcr->streams);
+}
diff --git a/src/bimu/vcr.h b/src/bimu/vcr.h
new file mode 100644
index 0000000..5a6d8d5
--- /dev/null
+++ b/src/bimu/vcr.h
@@ -0,0 +1,45 @@
+#pragma once
+
+#include <aki/packet_cache.h>
+#include <al/atomic.h>
+#include <aki/packet_stream.h>
+#include <aki/signal.h>
+
+#include "../codec/codec.h"
+
+enum {
+    BIMU_STREAM_RUNNING = 0,
+    BIMU_STREAM_STOPPED,
+    BIMU_STREAM_CLOSED
+};
+
+struct bmu_vcr_stream {
+    u8 type;
+    s32 index;
+    struct camu_stream stream;
+    struct bmu_client_handler *client;
+    atomic_s32 state;
+    struct aki_packet_cache cache;
+    bool buffered;
+    struct aki_cond cond;
+    struct aki_mutex mutex;
+    bool running;
+    struct aki_thread thread;
+    struct bmu_vcr *vcr;
+};
+
+struct bmu_vcr {
+    array(struct bmu_vcr_stream *) streams;
+    atomic_u64 count;
+    struct aki_packet_stream *data;
+    struct aki_signal signal;
+};
+
+void bmu_vcr_init(struct bmu_vcr *vcr, struct aki_event_loop *loop, struct aki_packet_stream *data);
+void bmu_vcr_add_stream(struct bmu_vcr *vcr, struct bmu_vcr_stream *stream);
+bool bmu_vcr_push_packet(struct bmu_vcr *vcr, struct aki_packet *packet);
+void bmu_vcr_stream_close(struct bmu_vcr_stream *stream);
+void bmu_vcr_flush(struct bmu_vcr *vcr);
+void bmu_vcr_stream_cork(struct bmu_vcr_stream *stream);
+void bmu_vcr_stream_uncork(struct bmu_vcr_stream *stream);
+void bmu_vcr_free(struct bmu_vcr *vcr);