Massive restructure and many changes

- The server-side list concept is still a wip

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

16 files changed, 1485 insertions, 0 deletions

diff --git a/src/shrub/client.c b/src/shrub/client.c
new file mode 100644
index 0000000..015f47f
--- /dev/null
+++ b/src/shrub/client.c
@@ -0,0 +1,309 @@
+#include <al/log.h>
+
+#include "../codec/ffmpeg/packet_ext.h"
+
+#include "handlers/codec.h"
+
+#include "client.h"
+#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();
+    aki_packet_write_u8(packet, SHRUB_CONTROL_RECONNECT);
+    aki_packet_write_u32(packet, client->level);
+    aki_packet_stream_send_packet(&client->control, packet);
+}
+
+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)
+{
+    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);
+    }
+}
+
+static void control_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_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;
+    }
+    aki_packet_stream_free(&client->control);
+    // TODO: Reconnect.
+}
+
+static void parse_info_packet(struct shrb_client *client, struct aki_packet *packet)
+{
+    client->duration = aki_packet_read_u64(packet);
+    u32 count = aki_packet_read_u32(packet);
+    for (u32 i = 0; i < count; i++) {
+        struct shrb_vcr_stream *stream = al_alloc_object(struct shrb_vcr_stream);
+        stream->client = shrb_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 = SHRUB_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 CAMU_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 = SHRUB_STREAM_AUDIO;
+                al_array_push(client->subscribed, stream->index);
+                break;
+            case AVMEDIA_TYPE_VIDEO:
+                stream->type = SHRUB_STREAM_VIDEO;
+                al_array_push(client->subscribed, stream->index);
+                break;
+            case AVMEDIA_TYPE_SUBTITLE:
+            default:
+                stream->type = SHRUB_STREAM_UNKNOWN;
+                break;
+            }
+            break;
+        }
+#endif
+        }
+        if (!stream->client->init(stream->client, NULL, stream)) {
+            // TODO: Handle this.
+        }
+        shrb_vcr_add_stream(&client->vcr, stream);
+    }
+}
+
+static void disconnect_data_internal(struct shrb_client *client)
+{
+    client->corked = false;
+    aki_packet_stream_disconnect(&client->data);
+}
+
+static void send_subscribe_packet(struct shrb_client *client)
+{
+    struct aki_packet *packet = aki_packet_create();
+    aki_packet_write_u8(packet, SHRUB_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 parse_start_packet(struct shrb_client *client, struct aki_packet *packet)
+{
+    u64 ts = aki_packet_read_u64(packet);
+    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 };
+    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)
+{
+    struct shrb_client *client = (struct shrb_client *)userdata;
+    (void)stream;
+
+    switch (aki_packet_read_u8(packet)) {
+    case SHRUB_CONTROL_INFO:
+        client->connection_id = aki_packet_read_u16(packet);
+        parse_info_packet(client, packet);
+        send_subscribe_packet(client);
+        break;
+    case SHRUB_CONTROL_START:
+        parse_start_packet(client, packet);
+        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);
+        break;
+    }
+    case SHRUB_CONTROL_SEEK:
+        client->level = 0;
+        disconnect_data_internal(client);
+        break;
+    default:
+        break;
+    }
+
+    aki_packet_free(packet);
+}
+
+void 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;
+    al_str_clone(&client->addr, addr);
+    client->port = port;
+    client->closed = false;
+    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);
+    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);
+}
+
+void shrb_client_reseek(struct shrb_client *client)
+{
+    client->level++;
+    al_log_debug("bimu", "Attempting re-seek at level %i.", client->level);
+    disconnect_data_internal(client);
+}
+
+void shrb_client_seek(struct shrb_client *client, f64 percent)
+{
+    u64 pos = client->duration * percent;
+    struct aki_packet *packet = aki_packet_create();
+    aki_packet_write_u8(packet, SHRUB_CONTROL_SEEK);
+    aki_packet_write_u64(packet, pos);
+    aki_packet_stream_send_packet(&client->control, packet);
+}
+
+void shrb_client_close(struct shrb_client *client)
+{
+    struct shrb_vcr_stream *stream;
+    al_array_foreach(client->vcr.streams, i, stream) {
+        shrb_vcr_stream_close(stream);
+    }
+    client->closed = true;
+    aki_packet_stream_disconnect(&client->control);
+    aki_packet_stream_disconnect(&client->data);
+}
+
+void shrb_client_free(struct shrb_client *client)
+{
+    aki_packet_stream_free(&client->control);
+    aki_packet_stream_free(&client->data);
+    al_str_free(&client->addr);
+    al_array_free(client->subscribed);
+    shrb_vcr_free(&client->vcr);
+}
diff --git a/src/shrub/client.h b/src/shrub/client.h
new file mode 100644
index 0000000..cb3250d
--- /dev/null
+++ b/src/shrub/client.h
@@ -0,0 +1,34 @@
+#pragma once
+
+#include <al/types.h>
+#include <aki/packet_stream.h>
+
+#include "vcr.h"
+
+struct shrb_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 closed;
+    bool reconnect;
+    u32 level;
+    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);
+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);
+void shrb_client_close(struct shrb_client *client);
+void shrb_client_free(struct shrb_client *client);
diff --git a/src/shrub/common.h b/src/shrub/common.h
new file mode 100644
index 0000000..0be00e4
--- /dev/null
+++ b/src/shrub/common.h
@@ -0,0 +1,27 @@
+#pragma once
+
+enum {
+    SHRUB_STREAM_AUDIO = 0,
+    SHRUB_STREAM_VIDEO,
+    SHRUB_STREAM_UNKNOWN,
+};
+
+enum {
+    SHRUB_CONNECTION_CONTROL = 0,
+    SHRUB_CONNECTION_DATA
+};
+
+enum {
+    SHRUB_CONTROL_INFO = 0,
+    SHRUB_CONTROL_SUBSCRIBE,
+    SHRUB_CONTROL_RECONNECT,
+    SHRUB_CONTROL_START,
+    SHRUB_CONTROL_TOGGLE_PAUSE,
+    SHRUB_CONTROL_PAUSE,
+    SHRUB_CONTROL_RESUME,
+    SHRUB_CONTROL_SEEK
+};
+
+#define SHRUB_BASE_DELAY 500000Lu
+#define SHRUB_PAUSE_DELAY 60000Lu
+#define SHRUB_DELAY_IGNORE 0Lu
diff --git a/src/shrub/handler.h b/src/shrub/handler.h
new file mode 100644
index 0000000..4ff2f2d
--- /dev/null
+++ b/src/shrub/handler.h
@@ -0,0 +1,52 @@
+#pragma once
+
+#include <aki/packet_pool.h>
+
+#include "../codec/codec.h"
+#include "../cache/handle.h"
+
+#include "vcr.h"
+
+struct shrb_server_handler {
+    bool (*init)(struct shrb_server_handler *, struct cch_handle *, struct aki_packet_pool *);
+    void (*write_info)(struct shrb_server_handler *, struct aki_packet *);
+    void (*subscribe)(struct shrb_server_handler *, s32, bool);
+    u64 (*get_duration)(struct shrb_server_handler *);
+    bool (*seek)(struct shrb_server_handler *, u64);
+    bool (*step)(struct shrb_server_handler *);
+    void (*free)(struct shrb_server_handler **);
+};
+
+enum {
+    SHRUB_PACKET_DATA = 0,
+    SHRUB_PACKET_EOF,
+    SHRUB_PACKET_ERROR
+};
+
+enum {
+    SHRUB_CLIENT_CONFIGURE = 0,
+    SHRUB_CLIENT_SET,
+    SHRUB_CLIENT_PAUSE,
+    SHRUB_CLIENT_RESUME,
+    SHRUB_CLIENT_DATA,
+    SHRUB_CLIENT_REMOVE_BUFFERS,
+    SHRUB_CLIENT_EOF,
+    SHRUB_CLIENT_CLOSED
+};
+
+struct shrb_seek_req {
+    u64 base;
+    u64 ts;
+    u64 delay;
+};
+
+struct shrb_client_stream;
+struct shrb_client_handler {
+    bool (*init)(struct shrb_client_handler *, struct camu_renderer *, struct shrb_vcr_stream *);
+    void (*handle_packet)(struct shrb_client_handler *, struct aki_packet *);
+    void (*flush)(struct shrb_client_handler *);
+    void (*free)(struct shrb_client_handler **);
+    struct shrb_vcr_stream *stream;
+    void (*callback)(void *, u8, struct shrb_vcr_stream *stream, void *);
+    void *userdata;
+};
diff --git a/src/shrub/handlers/cdio.h b/src/shrub/handlers/cdio.h
new file mode 100644
index 0000000..0f614af
--- /dev/null
+++ b/src/shrub/handlers/cdio.h
@@ -0,0 +1,17 @@
+#pragma once
+
+#include <cdio/paranoia/cdda.h>
+#include <cdio/cd_types.h>
+
+#include "../handler.h"
+
+struct bmu_cdio_server {
+    struct bmu_server_handler handler;
+};
+
+struct bmu_cdio_client {
+    struct bmu_client_handler handler;
+};
+
+struct bmu_server_handler *bmu_cdio_server_create(void);
+struct bmu_client_handler *bmu_cdio_client_create(void);
diff --git a/src/shrub/handlers/cdio_server.c b/src/shrub/handlers/cdio_server.c
new file mode 100644
index 0000000..151b94f
--- /dev/null
+++ b/src/shrub/handlers/cdio_server.c
@@ -0,0 +1,21 @@
+#include "cdio.h"
+
+static bool cdio_server_init(struct bmu_server_handler *handler, struct cch_handle *handle,
+    struct aki_packet_pool *pool)
+{
+    struct bmu_cdio_server *cdio = (struct bmu_cdio_server *)handler;
+    return true;
+}
+
+struct bmu_server_handler *bmu_cdio_server_create(void)
+{
+    struct bmu_cdio_server *cdio = al_alloc_object(struct bmu_cdio_server);
+    cdio->handler.init = cdio_server_init;
+//    cdio->handler.write_info = cdio_server_write_info;
+//    cdio->handler.subscribe = cdio_server_subscribe;
+//    cdio->handler.get_duration = cdio_server_get_duration;
+//    cdio->handler.seek = cdio_server_seek;
+//    cdio->handler.step = cdio_server_step;
+//    cdio->handler.free = cdio_server_free;
+    return (struct bmu_server_handler *)cdio;
+}
diff --git a/src/shrub/handlers/codec.h b/src/shrub/handlers/codec.h
new file mode 100644
index 0000000..f17b17d
--- /dev/null
+++ b/src/shrub/handlers/codec.h
@@ -0,0 +1,20 @@
+#pragma once
+
+#include "../handler.h"
+
+#include "../../codec/codec.h"
+
+struct shrb_codec_server {
+    struct shrb_server_handler handler;
+    struct camu_demuxer *demux;
+    struct camu_packet packet;
+    struct aki_packet_pool *pool;
+};
+
+struct shrb_codec_client {
+    struct shrb_client_handler handler;
+    struct camu_decoder *dec;
+};
+
+struct shrb_server_handler *shrb_codec_server_create(void);
+struct shrb_client_handler *shrb_codec_client_create(void);
diff --git a/src/shrub/handlers/codec_client.c b/src/shrub/handlers/codec_client.c
new file mode 100644
index 0000000..f55a740
--- /dev/null
+++ b/src/shrub/handlers/codec_client.c
@@ -0,0 +1,102 @@
+#include "../../codec/codec.h"
+#include "../../codec/ffmpeg/decoder.h"
+#include "../../codec/ffmpeg/packet_ext.h"
+
+#include "../vcr.h"
+
+#include "codec.h"
+
+static void data_callback(void *userdata, struct camu_frame *frame)
+{
+    struct shrb_codec_client *codec = (struct shrb_codec_client *)userdata;
+    codec->handler.callback(codec->handler.userdata, SHRUB_CLIENT_DATA, codec->handler.stream, frame);
+}
+
+static bool codec_client_init(struct shrb_client_handler *handler, struct camu_renderer *renderer,
+    struct shrb_vcr_stream *stream)
+{
+    struct shrb_codec_client *codec = (struct shrb_codec_client *)handler;
+    codec->dec = camu_ff_decoder_create();
+    codec->handler.stream = stream;
+    if (!codec->dec->init(codec->dec, renderer, &stream->stream)) {
+        return false;
+    }
+    codec->dec->stream = &stream->stream;
+    codec->handler.callback(codec->handler.userdata, SHRUB_CLIENT_CONFIGURE, codec->handler.stream, NULL);
+    codec->dec->set_callback(codec->dec, data_callback, codec);
+    return true;
+}
+
+#ifdef CAMU_HAVE_FFMPEG
+static void push_av_packet(struct shrb_codec_client *codec, AVPacket *pkt)
+{
+    struct camu_packet packet;
+    packet.av.pkt = pkt;
+    s32 ret = codec->dec->push(codec->dec, &packet);
+    av_packet_unref(pkt);
+    av_packet_free(&pkt);
+    al_assert(ret == CAMU_OK);
+}
+#endif
+
+static void push_packet(struct shrb_codec_client *codec, struct aki_buffer *buffer)
+{
+    struct camu_packet packet;
+    packet.buffer = buffer;
+    s32 ret = codec->dec->push(codec->dec, &packet);
+    al_assert(ret == CAMU_OK);
+}
+
+static void codec_client_handle_packet(struct shrb_client_handler *handler, struct aki_packet *packet)
+{
+    struct shrb_codec_client *codec = (struct shrb_codec_client *)handler;
+    if (!packet) {
+        struct shrb_codec_client *codec = (struct shrb_codec_client *)handler;
+        s32 ret = codec->dec->push(codec->dec, NULL);
+        // Flush returns success.
+        ret = codec->dec->process(codec->dec);
+        al_assert(ret == CAMU_ERR_EOF);
+        codec->handler.callback(codec->handler.userdata, SHRUB_CLIENT_EOF, codec->handler.stream, NULL);
+        return;
+    }
+    switch (aki_packet_read_u8(packet)) {
+    case CAMU_NORMAL: {
+        struct aki_buffer buffer;
+        aki_packet_read_buffer(packet, &buffer);
+        push_packet(codec, &buffer);
+        break;
+    }
+#ifdef CAMU_HAVE_FFMPEG
+    case CAMU_FFMPEG_COMPAT: {
+        push_av_packet(codec, aki_packet_read_av_packet(packet));
+        break;
+    }
+#endif
+    }
+    s32 ret = codec->dec->process(codec->dec);
+    al_assert(ret == CAMU_ERR_AGAIN || ret == CAMU_ERR_EOF);
+}
+
+static void codec_client_flush(struct shrb_client_handler *handler)
+{
+    struct shrb_codec_client *codec = (struct shrb_codec_client *)handler;
+    codec->dec->flush(codec->dec);
+}
+
+static void codec_client_free(struct shrb_client_handler **handler)
+{
+    struct shrb_codec_client *codec = (struct shrb_codec_client *)*handler;
+    codec->dec->free(&codec->dec);
+    al_free(codec);
+    *handler = NULL;
+}
+
+struct shrb_client_handler *shrb_codec_client_create(void)
+{
+    struct shrb_codec_client *codec = al_alloc_object(struct shrb_codec_client);
+    codec->handler.init = codec_client_init;
+    codec->handler.handle_packet = codec_client_handle_packet;
+    codec->handler.flush = codec_client_flush;
+    codec->handler.free = codec_client_free;
+    return (struct shrb_client_handler *)codec;
+}
diff --git a/src/shrub/handlers/codec_server.c b/src/shrub/handlers/codec_server.c
new file mode 100644
index 0000000..cab615a
--- /dev/null
+++ b/src/shrub/handlers/codec_server.c
@@ -0,0 +1,140 @@
+#include "../../codec/codec.h"
+#include "../../codec/ffmpeg/demuxer.h"
+#include "../../codec/ffmpeg/packet_ext.h"
+
+#include "../handler.h"
+
+#include "codec.h"
+
+static bool codec_server_init(struct shrb_server_handler *handler, struct cch_handle *handle,
+    struct aki_packet_pool *pool)
+{
+    struct shrb_codec_server *codec = (struct shrb_codec_server *)handler;
+    codec->demux = camu_ff_demuxer_create();
+    if (!codec->demux->init(codec->demux, handle)) {
+        return false;
+    }
+    switch (codec->demux->type) {
+    case CAMU_NORMAL:
+        break;
+#ifdef CAMU_HAVE_FFMPEG
+    case CAMU_FFMPEG_COMPAT:
+        codec->packet.av.pkt = av_packet_alloc();
+        break;
+#endif
+    }
+    codec->pool = pool;
+    return true;
+}
+
+static void codec_server_write_info(struct shrb_server_handler *handler, struct aki_packet *packet)
+{
+    struct shrb_codec_server *codec = (struct shrb_codec_server *)handler;
+    aki_packet_write_u64(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) {
+        aki_packet_write_u8(packet, stream->type);
+        switch (stream->type) {
+        case CAMU_NORMAL:
+            aki_packet_write_s32(packet, stream->video.width);
+            aki_packet_write_s32(packet, stream->video.height);
+            break;
+#ifdef CAMU_HAVE_FFMPEG
+        case CAMU_FFMPEG_COMPAT:
+            aki_packet_write_av_codec_id(packet, stream->av.stream->codecpar->codec_id);
+            aki_packet_write_av_stream(packet, stream->av.stream);
+            break;
+#endif
+        }
+    }
+}
+
+static void codec_server_subscribe(struct shrb_server_handler *handler, s32 index, bool subscribe)
+{
+    struct shrb_codec_server *codec = (struct shrb_codec_server *)handler;
+    if (subscribe) {
+        al_array_push(codec->demux->subscribed, index);
+    }
+}
+
+static u64 codec_server_get_duration(struct shrb_server_handler *handler)
+{
+    struct shrb_codec_server *codec = (struct shrb_codec_server *)handler;
+    return codec->demux->get_duration(codec->demux);
+}
+
+static bool codec_server_seek(struct shrb_server_handler *handler, u64 pos)
+{
+    struct shrb_codec_server *codec = (struct shrb_codec_server *)handler;
+    return codec->demux->seek(codec->demux, pos);
+}
+
+static bool codec_server_step(struct shrb_server_handler *handler)
+{
+    struct shrb_codec_server *codec = (struct shrb_codec_server *)handler;
+    struct aki_packet *packet = aki_packet_pool_get(codec->pool);
+    if (!packet) return false;
+    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, SHRUB_PACKET_DATA);
+            aki_packet_write_u8(packet, codec->packet.type);
+            aki_packet_write_buffer(packet, codec->packet.buffer);
+            break;
+        }
+#ifdef CAMU_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, 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) {
+        aki_packet_write_u8(packet, SHRUB_PACKET_EOF);
+        aki_packet_pool_submit(codec->pool, packet);
+        return false;
+    } else {
+        aki_packet_write_u8(packet, SHRUB_PACKET_ERROR);
+        aki_packet_pool_submit(codec->pool, packet);
+        return false;
+    }
+    return true;
+}
+
+static void codec_server_free(struct shrb_server_handler **handler)
+{
+    struct shrb_codec_server *codec = (struct shrb_codec_server *)*handler;
+    switch (codec->demux->type) {
+    case CAMU_NORMAL:
+        break;
+#ifdef CAMU_HAVE_FFMPEG
+    case CAMU_FFMPEG_COMPAT:
+        av_packet_free(&codec->packet.av.pkt);
+        break;
+#endif
+    }
+    codec->demux->free(&codec->demux);
+    al_free(codec);
+    *handler = NULL;
+}
+
+struct shrb_server_handler *shrb_codec_server_create(void)
+{
+    struct shrb_codec_server *codec = al_alloc_object(struct shrb_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;
+    codec->handler.free = codec_server_free;
+    return (struct shrb_server_handler *)codec;
+}
diff --git a/src/shrub/handlers/dvd.h b/src/shrub/handlers/dvd.h
new file mode 100644
index 0000000..db9bfe9
--- /dev/null
+++ b/src/shrub/handlers/dvd.h
@@ -0,0 +1,14 @@
+#pragma once
+
+#include "../handler.h"
+
+struct bmu_dvd_server {
+    struct bmu_server_handler handler;
+};
+
+struct bmu_dvd_client {
+    struct bmu_client_handler handler;
+};
+
+struct bmu_server_handler *bmu_dvd_server_create(void);
+struct bmu_client_handler *bmu_dvd_client_create(void);
diff --git a/src/shrub/handlers/dvd_server.c b/src/shrub/handlers/dvd_server.c
new file mode 100644
index 0000000..ea3a9d9
--- /dev/null
+++ b/src/shrub/handlers/dvd_server.c
@@ -0,0 +1,21 @@
+#include "dvd.h"
+
+static bool dvd_server_init(struct bmu_server_handler *handler, struct cch_handle *handle,
+    struct aki_packet_pool *pool)
+{
+    struct bmu_dvd_server *dvd = (struct bmu_dvd_server *)handler;
+    return true;
+}
+
+struct bmu_server_handler *bmu_dvd_server_create(void)
+{
+    struct bmu_dvd_server *dvd = al_alloc_object(struct bmu_dvd_server);
+    dvd->handler.init = dvd_server_init;
+//    dvd->handler.write_info = dvd_server_write_info;
+//    dvd->handler.subscribe = dvd_server_subscribe;
+//    dvd->handler.get_duration = dvd_server_get_duration;
+//    dvd->handler.seek = dvd_server_seek;
+//    dvd->handler.step = dvd_server_step;
+//    dvd->handler.free = dvd_server_free;
+    return (struct bmu_server_handler *)dvd;
+}
diff --git a/src/shrub/meson.build b/src/shrub/meson.build
new file mode 100644
index 0000000..a3fc77b
--- /dev/null
+++ b/src/shrub/meson.build
@@ -0,0 +1,29 @@
+shrub_server_src = [
+  'server.c',
+  'handlers/codec_server.c'
+]
+shrub_client_src = [
+  'client.c',
+  'vcr.c',
+  'handlers/codec_client.c'
+]
+shrub_server_deps = [ffmpeg_server]
+shrub_client_deps = [ffmpeg_client]
+
+#libcdio_paranoia = dependency('libcdio_paranoia', required: false, allow_fallback: true)
+#libcdio_cdda = dependency('libcdio_cdda', required: false, allow_fallback: true)
+#if libcdio_paranoia.found() and libcdio_cdda.found()
+#  shrub_server_src += ['handlers/cdio_server.c']
+#  shrub_server_deps += [libcdio_paranoia, libcdio_cdda]
+#endif
+#
+#libdvdcss = dependency('libdvdcss', required: false, allow_fallback: true)
+#libdvdread = dependency('libdvdread', required: false, allow_fallback: true)
+#libdvdnav = dependency('libdvdnav', required: false, allow_fallback: true)
+#if libdvdcss.found() and libdvdread.found() and libdvdnav.found()
+#  shrub_server_src += ['handlers/dvd_server.c']
+#  shrub_server_deps += [libdvdcss]
+#endif
+
+shrub_server = declare_dependency(sources: shrub_server_src, dependencies: shrub_server_deps)
+shrub_client = declare_dependency(sources: shrub_client_src, dependencies: shrub_client_deps)
diff --git a/src/shrub/server.c b/src/shrub/server.c
new file mode 100644
index 0000000..ba7259c
--- /dev/null
+++ b/src/shrub/server.c
@@ -0,0 +1,419 @@
+#include <al/random.h>
+#include <aki/common.h>
+
+#include "handlers/codec.h"
+
+#include "server.h"
+#include "common.h"
+
+static void control_connection_closed_callback(void *userdata, struct aki_packet_stream *stream)
+{
+    struct shrb_node_connection *connection = (struct shrb_node_connection *)userdata;
+    (void)stream;
+    struct shrb_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;
+    //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)
+{
+    struct shrb_node_connection *connection = (struct shrb_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)
+{
+    (void)userdata;
+    aki_packet_free(packet);
+}
+
+static void packet_sent_callback(void *userdata, struct aki_packet *packet)
+{
+    struct shrb_node_connection *connection = (struct shrb_node_connection *)userdata;
+    aki_packet_pool_return(&connection->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)
+{
+    struct shrb_node_connection *connection = (struct shrb_node_connection *)userdata;
+    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 shrb_node_connection *connection, u64 ts)
+{
+    struct aki_packet *packet = aki_packet_create();
+    aki_packet_write_u8(packet, SHRUB_CONTROL_RESUME);
+    aki_packet_write_u64(packet, ts);
+    aki_packet_stream_send_packet(connection->control, packet);
+}
+
+static void send_pause_packet(struct shrb_node_connection *connection, u64 seek_pos)
+{
+    struct aki_packet *packet = aki_packet_create();
+    aki_packet_write_u8(packet, SHRUB_CONTROL_PAUSE);
+    aki_packet_write_u64(packet, seek_pos);
+    aki_packet_stream_send_packet(connection->control, packet);
+}
+
+static struct shrb_node *get_node_by_id(struct shrb_server *server, u16 id)
+{
+    struct shrb_node *node;
+    al_array_foreach(server->nodes, i, node) {
+        if (node->id == id) return node;
+    }
+    return NULL;
+}
+
+// Possible connection senarios:
+//   - initial connection and start
+//   - connected while playing
+//   - connected while paused
+//   - connected while queued
+//   - connected while queued after a paused stream
+// Questions:
+//   - can the server estimate the position of the stream? I think yes
+
+static void send_start_packet(struct shrb_node_connection *connection)
+{
+    struct shrb_node *node = connection->node;
+    struct aki_packet *packet = aki_packet_create();
+    aki_packet_write_u8(packet, SHRUB_CONTROL_START);
+    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;
+    }
+    seek_pos += delay - SHRUB_BASE_DELAY;
+    aki_packet_write_u64(packet, start);
+    aki_packet_write_u64(packet, delay);
+    aki_packet_write_u64(packet, seek_pos);
+    aki_packet_write_u8(packet, node->paused ? 1 : 0);
+    aki_packet_stream_send_packet(connection->control, packet);
+}
+
+static void send_seek_packet(struct shrb_node_connection *connection, u64 pos)
+{
+    connection->seek_pos = pos;
+    struct aki_packet *packet = aki_packet_create();
+    aki_packet_write_u8(packet, SHRUB_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 shrb_node_connection *connection = (struct shrb_node_connection *)userdata;
+    (void)stream;
+
+    switch (aki_packet_read_u8(packet)) {
+    case SHRUB_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 SHRUB_CONTROL_RECONNECT: {
+        connection->level = aki_packet_read_u32(packet);
+        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;
+        connection->node->seeked = true;
+        //send_seek_packet(connection->node);
+        break;
+    }
+    }
+    aki_packet_free(packet);
+}
+
+static struct shrb_node_connection *get_connection_by_id(struct shrb_node *node, u16 id)
+{
+    struct shrb_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 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);
+    }
+    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 shrb_node_connection *connection = (struct shrb_node_connection *)userdata;
+    while (connection->running && connection->handler->step(connection->handler)) {}
+    return 0;
+}
+
+static void maybe_start_handler(struct shrb_node_connection *connection)
+{
+    al_assert(!connection->running);
+    // 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)
+{
+    struct shrb_server *server = (struct shrb_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 shrb_node *node = get_node_by_id(server, node_id);
+    al_assert(node);
+
+    struct shrb_node_connection *connection = get_connection_by_id(node, connection_id);
+    if (!connection) {
+        al_assert(type == SHRUB_CONNECTION_CONTROL);
+        connection = al_alloc_object(struct shrb_node_connection);
+        al_array_push(node->connections, connection);
+        connection->id = node->connection_id++;
+        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 = SHRUB_PLAYING;
+        connection->level = 0;
+        connection->running = 0;
+    }
+
+    switch (type) {
+    case SHRUB_CONNECTION_CONTROL:
+        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;
+        stream->connection_closed_callback = control_connection_closed_callback;
+        stream->userdata = connection;
+        send_control_info_packet(connection);
+        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?
+        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;
+        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)
+{
+    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)
+{
+    al_array_init(server->nodes);
+    return aki_packet_stream_init(&server->server, AKI_SOCKET_TCP, connection_callback, NULL, NULL, NULL, server);
+}
+
+void shrb_server_listen(struct shrb_server *server, struct aki_event_loop *loop, str *addr, s32 port)
+{
+    server->loop = loop;
+    aki_packet_stream_listen(&server->server, loop, addr, port);
+}
+
+struct shrb_node *shrb_server_create_node(struct shrb_server *server, u16 prev, struct cch_entry *entry)
+{
+    struct shrb_node *node = al_alloc_object(struct shrb_node);
+    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);
+        return NULL;
+    }
+    struct shrb_server_handler *handler = shrb_codec_server_create();
+    if (!handler->init(handler, &handle, NULL)) {
+        al_free(node);
+        return NULL;
+    }
+    node->duration = handler->get_duration(handler);
+    handler->free(&handler);
+    cch_entry_return_handle(node->entry, &handle);
+    al_array_push(server->nodes, node);
+    return node;
+}
+
+void shrb_node_set_start(struct shrb_node *node, u64 start)
+{
+    node->start = start;
+}
+
+void shrb_node_set_offset(struct shrb_node *node, u64 offset)
+{
+    node->offset = offset;
+}
+
+void shrb_node_toggle_pause(struct shrb_node *node, u64 pos)
+{
+    if (!node->paused) {
+        node->seek_pos = pos;
+        node->seek_pos += SHRUB_PAUSE_DELAY;
+    } else {
+        node->start = aki_get_timestamp() + 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);
+        }
+    }
+    if (node->paused) {
+        node->start -= SHRUB_BASE_DELAY;
+        node->offset = 0;
+    }
+    node->paused = !node->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) {
+        send_seek_packet(connection, node->seek_pos);
+    }
+}
+
+void shrb_server_close(struct shrb_server *server)
+{
+    (void)server;
+}
diff --git a/src/shrub/server.h b/src/shrub/server.h
new file mode 100644
index 0000000..2bce369
--- /dev/null
+++ b/src/shrub/server.h
@@ -0,0 +1,62 @@
+#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 "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;
+    struct aki_packet_stream *control;
+    struct aki_packet_stream *data;
+    struct aki_packet_pool pool;
+    s32 running;
+    //u8 paused;
+    u32 level;
+    u64 seek_pos;
+    struct cch_handle handle;
+    struct shrb_server_handler *handler;
+    struct aki_thread thread;
+};
+
+struct shrb_node {
+    u16 id;
+    u16 prev;
+    struct cch_entry *entry;
+    bool paused;
+    bool seeked;
+    u64 start;
+    u64 seek_pos;
+    u64 offset;
+    u64 duration;
+    u16 connection_id;
+    array(struct shrb_node_connection *) connections;
+    struct shrb_server *server;
+};
+
+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);
+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_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
new file mode 100644
index 0000000..1816142
--- /dev/null
+++ b/src/shrub/vcr.c
@@ -0,0 +1,173 @@
+#include <al/log.h>
+
+#include "vcr.h"
+#include "handler.h"
+
+#define VCR_BUFFER_HIGH 1024
+#define VCR_BUFFER_BUFFERED (VCR_BUFFER_HIGH - 64)
+#define VCR_BUFFER_LOW (VCR_BUFFER_HIGH - 256)
+
+static void signal_callback(void *userdata)
+{
+    struct shrb_vcr *vcr = (struct shrb_vcr *)userdata;
+    aki_packet_stream_cork(vcr->data, false);
+}
+
+void shrb_vcr_init(struct shrb_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 shrb_vcr_stream *stream = (struct shrb_vcr_stream *)userdata;
+    struct shrb_vcr *vcr = stream->vcr;
+    while (aki_packet_cache_wait(&stream->cache)) {
+        struct aki_packet *packet = aki_packet_cache_pop(&stream->cache);
+        if (!packet) {
+            stream->client->handle_packet(stream->client, NULL);
+            break;
+        }
+        if (al_atomic_u64_sub(&vcr->count, 1, AL_ATOMIC_RELAXED) < VCR_BUFFER_LOW && stream->buffered) {
+            aki_signal_send(&vcr->signal);
+        }
+        if (al_atomic_s32_load(&stream->state, AL_ATOMIC_RELAXED) == SHRUB_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) == SHRUB_STREAM_CLOSED) {
+            aki_packet_free(packet);
+            break;
+        }
+        stream->client->handle_packet(stream->client, packet);
+        aki_packet_free(packet);
+    }
+    return 0;
+}
+
+void shrb_vcr_add_stream(struct shrb_vcr *vcr, struct shrb_vcr_stream *stream)
+{
+    stream->vcr = vcr;
+    aki_cond_init(&stream->cond);
+    aki_mutex_init(&stream->mutex);
+    aki_packet_cache_init(&stream->cache, VCR_BUFFER_HIGH);
+    stream->buffered = 0;
+    al_array_push(vcr->streams, stream);
+    al_atomic_s32_store(&stream->state, SHRUB_STREAM_RUNNING, AL_ATOMIC_RELAXED);
+}
+
+static struct shrb_vcr_stream *get_stream_from_index(struct shrb_vcr *vcr, s32 index)
+{
+    struct shrb_vcr_stream *stream;
+    al_array_foreach(vcr->streams, i, stream) {
+        if (stream->index == index) return stream;
+    }
+    return NULL;
+}
+
+bool shrb_vcr_push_packet(struct shrb_vcr *vcr, struct aki_packet *packet)
+{
+    struct shrb_vcr_stream *stream;
+    switch (aki_packet_read_u8(packet)) {
+    case SHRUB_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) >= VCR_BUFFER_BUFFERED) {
+            aki_packet_stream_cork(vcr->data, true);
+            al_array_foreach(vcr->streams, i, stream) {
+                stream->buffered = 1;
+            }
+        }
+        break;
+    case SHRUB_PACKET_EOF:
+        al_array_foreach(vcr->streams, i, stream) {
+            aki_packet_cache_send_packet(&stream->cache, NULL);
+        }
+        aki_packet_free(packet);
+        break;
+    case SHRUB_PACKET_ERROR:
+        al_log_warn("bimu", "Unhandled error packet.");
+        aki_packet_free(packet);
+        break;
+    default:
+        al_assert(false);
+    }
+    return true;
+}
+
+static void vcr_stream_close_internal(struct shrb_vcr_stream *stream)
+{
+    al_atomic_s32_store(&stream->state, SHRUB_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);
+    if (stream->running) {
+        aki_thread_join(&stream->thread);
+    }
+    struct aki_packet *packet;
+    while ((packet = aki_packet_cache_pop(&stream->cache))) {
+        aki_packet_free(packet);
+    }
+}
+
+void shrb_vcr_stream_close(struct shrb_vcr_stream *stream)
+{
+    vcr_stream_close_internal(stream);
+}
+
+void shrb_vcr_flush(struct shrb_vcr *vcr)
+{
+    struct shrb_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, SHRUB_STREAM_RUNNING, AL_ATOMIC_RELAXED);
+    }
+    al_atomic_u64_store(&vcr->count, 0, AL_ATOMIC_RELAXED);
+}
+
+void shrb_vcr_stream_cork(struct shrb_vcr_stream *stream)
+{
+    al_atomic_s32_store(&stream->state, SHRUB_STREAM_STOPPED, AL_ATOMIC_RELAXED);
+}
+
+void shrb_vcr_stream_uncork(struct shrb_vcr_stream *stream)
+{
+    al_atomic_s32_store(&stream->state, SHRUB_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 shrb_vcr_free(struct shrb_vcr *vcr)
+{
+    aki_signal_stop(&vcr->signal);
+    struct shrb_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/shrub/vcr.h b/src/shrub/vcr.h
new file mode 100644
index 0000000..4fae2c0
--- /dev/null
+++ b/src/shrub/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 {
+    SHRUB_STREAM_RUNNING = 0,
+    SHRUB_STREAM_STOPPED,
+    SHRUB_STREAM_CLOSED
+};
+
+struct shrb_vcr_stream {
+    u8 type;
+    s32 index;
+    struct camu_stream stream;
+    struct shrb_client_handler *client;
+    atomic_s32 state;
+    struct aki_packet_cache cache;
+    s32 buffered;
+    struct aki_cond cond;
+    struct aki_mutex mutex;
+    bool running;
+    struct aki_thread thread;
+    struct shrb_vcr *vcr;
+};
+
+struct shrb_vcr {
+    array(struct shrb_vcr_stream *) streams;
+    atomic_u64 count;
+    struct aki_packet_stream *data;
+    struct aki_signal signal;
+};
+
+void shrb_vcr_init(struct shrb_vcr *vcr, struct aki_event_loop *loop, struct aki_packet_stream *data);
+void shrb_vcr_add_stream(struct shrb_vcr *vcr, struct shrb_vcr_stream *stream);
+bool shrb_vcr_push_packet(struct shrb_vcr *vcr, struct aki_packet *packet);
+void shrb_vcr_stream_close(struct shrb_vcr_stream *stream);
+void shrb_vcr_flush(struct shrb_vcr *vcr);
+void shrb_vcr_stream_cork(struct shrb_vcr_stream *stream);
+void shrb_vcr_stream_uncork(struct shrb_vcr_stream *stream);
+void shrb_vcr_free(struct shrb_vcr *vcr);