diff options
| author | 2024-10-21 19:22:50 -0400 | |
|---|---|---|
| committer | 2024-10-21 19:22:50 -0400 | |
| commit | 60b4ebfbf3be78dba9dc7c65ab2bdaa0b218c0c2 (patch) | |
| tree | 08ff2ce7975f523112e7ad2fe4f797b4fc7db5de /src/shrub | |
| parent | 2f9a0945bfeee3296cec3d38d094e4c49f9cb65f (diff) | |
| download | camu-60b4ebfbf3be78dba9dc7c65ab2bdaa0b218c0c2.tar.gz camu-60b4ebfbf3be78dba9dc7c65ab2bdaa0b218c0c2.tar.bz2 camu-60b4ebfbf3be78dba9dc7c65ab2bdaa0b218c0c2.zip | |
Everything before initial synced list
Signed-off-by: Andrew Opalach <andrew@akon.city>
Diffstat (limited to 'src/shrub')
| -rw-r--r-- | src/shrub/client.c | 283 | ||||
| -rw-r--r-- | src/shrub/client.h | 33 | ||||
| -rw-r--r-- | src/shrub/common.h | 26 | ||||
| -rw-r--r-- | src/shrub/handler.h | 59 | ||||
| -rw-r--r-- | src/shrub/handlers/cdio.h | 17 | ||||
| -rw-r--r-- | src/shrub/handlers/cdio_server.c | 21 | ||||
| -rw-r--r-- | src/shrub/handlers/codec.h | 20 | ||||
| -rw-r--r-- | src/shrub/handlers/codec_client.c | 102 | ||||
| -rw-r--r-- | src/shrub/handlers/codec_server.c | 140 | ||||
| -rw-r--r-- | src/shrub/handlers/dvd.h | 14 | ||||
| -rw-r--r-- | src/shrub/handlers/dvd_server.c | 21 | ||||
| -rw-r--r-- | src/shrub/meson.build | 29 | ||||
| -rw-r--r-- | src/shrub/server.c | 401 | ||||
| -rw-r--r-- | src/shrub/server.h | 62 | ||||
| -rw-r--r-- | src/shrub/vcr.c | 177 | ||||
| -rw-r--r-- | src/shrub/vcr.h | 45 |
16 files changed, 0 insertions, 1450 deletions
diff --git a/src/shrub/client.c b/src/shrub/client.c deleted file mode 100644 index abe2d23..0000000 --- a/src/shrub/client.c +++ /dev/null @@ -1,283 +0,0 @@ -#include <al/log.h> - -#include "../codec/ffmpeg/packet_ext.h" -#include "../server/common.h" - -#include "handlers/codec.h" - -#include "client.h" -#include "handler.h" -#include "common.h" - -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 packet_sent_callback(void *userdata, struct aki_packet *packet) -{ - (void)userdata; - aki_packet_free(packet); -} - -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 data_connection_callback(void *userdata, struct aki_packet_stream *stream) -{ - struct shrb_client *client = (struct shrb_client *)userdata; - (void)stream; - client->data.packet_callback = data_packet_callback; - client->data.packet_sent_callback = packet_sent_callback; - 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 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); - send_reconnect_packet(client); -} - -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, client); - aki_packet_stream_set_multiplex(&client->data, CAMU_MULTIPLEX_SHRUB); - aki_packet_stream_connect(&client->data, client->loop, &client->addr, client->port); -} - -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); - u64 paused_at = aki_packet_read_u64(packet); - struct shrb_seek_req req = { - .base = seek_pos, - .ts = ts, - .delay = delay, - .paused = paused, - .paused_at = paused_at - }; - client->callback(client->userdata, SHRUB_CLIENT_SET, 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: { - f64 pts = aki_packet_read_u64(packet) / 1000000.f; - client->callback(client->userdata, SHRUB_CLIENT_PAUSE, NULL, &pts); - break; - } - case SHRUB_CONTROL_RESUME: { - u64 start = aki_packet_read_u64(packet); - u64 offset = aki_packet_read_u64(packet); - struct shrb_resume_req req = { - .start = start, - .offset = offset - }; - client->callback(client->userdata, SHRUB_CLIENT_RESUME, NULL, &req); - break; - } - case SHRUB_CONTROL_SEEK: - client->level = 0; - disconnect_data_internal(client); - break; - default: - break; - } - - aki_packet_free(packet); -} - -static void control_connection_callback(void *userdata, struct aki_packet_stream *stream) -{ - struct shrb_client *client = (struct shrb_client *)userdata; - (void)stream; - client->control.packet_callback = control_packet_callback; - client->control.packet_sent_callback = packet_sent_callback; - 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; - } - // TODO: Reconnect. -} - -bool 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; - al_array_init(client->subscribed); - shrb_vcr_init(&client->vcr, client->loop, &client->data); - if (!aki_packet_stream_init(&client->control, AKI_SOCKET_TCP, control_connection_callback, - control_connection_closed_callback, client)) { - return false; - } - aki_packet_stream_set_nodelay(&client->control, 1); - aki_packet_stream_set_multiplex(&client->control, CAMU_MULTIPLEX_SHRUB); - aki_packet_stream_connect(&client->control, client->loop, &client->addr, client->port); - return true; -} - -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 deleted file mode 100644 index 14f7da2..0000000 --- a/src/shrub/client.h +++ /dev/null @@ -1,33 +0,0 @@ -#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; - array(s32) subscribed; - struct shrb_vcr vcr; - void (*callback)(void *, u8, struct shrb_vcr_stream *stream, void *); - void *userdata; -}; - -bool 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 deleted file mode 100644 index 882eb49..0000000 --- a/src/shrub/common.h +++ /dev/null @@ -1,26 +0,0 @@ -#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_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 deleted file mode 100644 index 7242130..0000000 --- a/src/shrub/handler.h +++ /dev/null @@ -1,59 +0,0 @@ -#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_resume_req { - u64 start; - u64 offset; -}; - -struct shrb_seek_req { - u64 base; - u64 ts; - u64 delay; - bool paused; - u64 paused_at; -}; - -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 deleted file mode 100644 index 0f614af..0000000 --- a/src/shrub/handlers/cdio.h +++ /dev/null @@ -1,17 +0,0 @@ -#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 deleted file mode 100644 index 151b94f..0000000 --- a/src/shrub/handlers/cdio_server.c +++ /dev/null @@ -1,21 +0,0 @@ -#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 deleted file mode 100644 index f17b17d..0000000 --- a/src/shrub/handlers/codec.h +++ /dev/null @@ -1,20 +0,0 @@ -#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 deleted file mode 100644 index f55a740..0000000 --- a/src/shrub/handlers/codec_client.c +++ /dev/null @@ -1,102 +0,0 @@ -#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 deleted file mode 100644 index cab615a..0000000 --- a/src/shrub/handlers/codec_server.c +++ /dev/null @@ -1,140 +0,0 @@ -#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 deleted file mode 100644 index db9bfe9..0000000 --- a/src/shrub/handlers/dvd.h +++ /dev/null @@ -1,14 +0,0 @@ -#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 deleted file mode 100644 index ea3a9d9..0000000 --- a/src/shrub/handlers/dvd_server.c +++ /dev/null @@ -1,21 +0,0 @@ -#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 deleted file mode 100644 index a3fc77b..0000000 --- a/src/shrub/meson.build +++ /dev/null @@ -1,29 +0,0 @@ -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 deleted file mode 100644 index 31d8787..0000000 --- a/src/shrub/server.c +++ /dev/null @@ -1,401 +0,0 @@ -#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 start, u64 offset) -{ - struct aki_packet *packet = aki_packet_create(); - aki_packet_write_u8(packet, SHRUB_CONTROL_RESUME); - aki_packet_write_u64(packet, start); - aki_packet_write_u64(packet, offset); - 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; - bool paused = node->paused != SHRUB_NOT_PAUSED; - if (!paused) { - s64 diff = aki_get_timestamp() - start; - if (diff > (s64)SHRUB_BASE_DELAY) { - start += diff; - seek_pos += diff; - } - } - // Start is not used if the node is paused. - aki_packet_write_u64(packet, start); - seek_pos += delay - SHRUB_BASE_DELAY; - aki_packet_write_u64(packet, delay); - aki_packet_write_u64(packet, seek_pos); - aki_packet_write_u8(packet, paused); - aki_packet_write_u64(packet, node->paused_at); - 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_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 bool send_control_info_packet(struct shrb_node_connection *connection) -{ - connection->handler = shrb_codec_server_create(); - if (!connection->handler->init(connection->handler, &connection->handle, &connection->pool)) { - return false; - } - struct aki_packet *packet = aki_packet_create(); - aki_packet_write_u8(packet, SHRUB_CONTROL_INFO); - aki_packet_write_u16(packet, connection->id); - connection->handler->write_info(connection->handler, packet); - aki_packet_stream_send_packet(connection->control, packet); - return true; -} - -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); -} - -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); - - // Before this point: - // - We get socket from accept - // - Create packet_stream - // - connection_closed_callback could have been called before this point - // - the socket exists only to wait for an identify 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->level = 0; - connection->running = 0; - } - - switch (type) { - case SHRUB_CONNECTION_CONTROL: - aki_packet_stream_set_nodelay(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; - if (!send_control_info_packet(connection)) { - aki_packet_stream_disconnect(stream); - } - 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? - // ^^^^ Literally any connection could call itself data, this needs to be handled better. - 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_callback(void *userdata, struct aki_packet_stream *stream) -{ - struct shrb_server *server = (struct shrb_server *)userdata; - stream->packet_callback = identify_packet_callback; - stream->packet_sent_callback = default_packet_sent_callback; - stream->userdata = server; -} - -bool shrb_server_init(struct shrb_server *server, struct aki_event_loop *loop) -{ - server->loop = loop; - al_array_init(server->nodes); - return true; -} - -static void connection_closed_callback(void *userdata, struct aki_packet_stream *stream) -{ - (void)userdata; - (void)stream; -} - -void shrb_server_add_socket(struct shrb_server *server, struct aki_socket *s) -{ - struct aki_packet_stream *stream = al_alloc_object(struct aki_packet_stream); - stream->connection_callback = connection_callback; - stream->connection_closed_callback = connection_closed_callback; - stream->userdata = server; - aki_packet_stream_from_socket(stream, server->loop, s); -} - -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; - 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); - node->seeked = false; - node->start = 0; - node->paused = SHRUB_NOT_PAUSED; - node->paused_at = 0; - node->seek_pos = 0; - node->connection_id = 1; - al_array_init(node->connections); - node->server = server; - al_array_push(server->nodes, node); - return node; -} - -void shrb_node_set_start(struct shrb_node *node, u64 start) -{ - if (node->paused == SHRUB_PAUSED) { - node->paused = SHRUB_NOT_PAUSED; - } - node->start = start; -} - -void shrb_node_user_pause(struct shrb_node *node, u64 pos) -{ - al_assert(node->paused != SHRUB_USER_PAUSED); - pos += SHRUB_PAUSE_DELAY; - node->seek_pos = pos; - node->paused_at = pos; - struct shrb_node_connection *connection; - al_array_foreach(node->connections, i, connection) { - send_pause_packet(connection, node->seek_pos); - } - node->paused = SHRUB_USER_PAUSED; -} - -void shrb_node_user_resume(struct shrb_node *node, struct shrb_resume_req *req) -{ - al_assert(node->paused == SHRUB_USER_PAUSED); - node->start = req->start + SHRUB_PAUSE_DELAY; - struct shrb_node_connection *connection; - al_array_foreach(node->connections, i, connection) { - send_resume_packet(connection, node->start, req->offset); - } - node->start -= SHRUB_BASE_DELAY; - node->paused = SHRUB_NOT_PAUSED; -} - -void shrb_node_pause(struct shrb_node *node) -{ - return; - if (node->paused != SHRUB_NOT_PAUSED) return; - u64 start = node->start; - s64 pos = node->seek_pos + (aki_get_timestamp() - (start + SHRUB_BASE_DELAY)); - pos += SHRUB_PAUSE_DELAY; - al_assert(pos >= 0); - node->seek_pos = (u64)pos; - node->paused_at = pos; - node->paused = SHRUB_PAUSED; -} - -void shrb_node_seek(struct shrb_node *node, u64 pos) -{ - node->seek_pos = pos; - 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 deleted file mode 100644 index cb901c5..0000000 --- a/src/shrub/server.h +++ /dev/null @@ -1,62 +0,0 @@ -#pragma once - -#include <al/array.h> -#include <aki/multiplex.h> -#include <aki/packet_stream.h> -#include <aki/packet_pool.h> - -#include "../cache/entry.h" -#include "../cache/handle.h" - -#include "handler.h" - -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; - u32 level; - u64 seek_pos; - struct cch_handle handle; - struct shrb_server_handler *handler; - struct aki_thread thread; -}; - -enum { - SHRUB_NOT_PAUSED = 0, - SHRUB_PAUSED, - SHRUB_USER_PAUSED -}; - -struct shrb_node { - u16 id; - u16 prev; - struct cch_entry *entry; - u64 duration; - bool seeked; - u64 start; // Exact start of the stream minus BASE_DELAY. - u8 paused; - u64 paused_at; - u64 seek_pos; - u16 connection_id; - array(struct shrb_node_connection *) connections; - struct shrb_server *server; -}; - -struct shrb_server { - struct aki_event_loop *loop; - array(struct shrb_node *) nodes; -}; - -bool shrb_server_init(struct shrb_server *server, struct aki_event_loop *loop); -void shrb_server_add_socket(struct shrb_server *server, struct aki_socket *s); -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_user_pause(struct shrb_node *node, u64 pos); -void shrb_node_user_resume(struct shrb_node *node, struct shrb_resume_req *req); -void shrb_node_pause(struct shrb_node *node); -void shrb_node_resume(struct shrb_node *node); -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 deleted file mode 100644 index 7f2b058..0000000 --- a/src/shrub/vcr.c +++ /dev/null @@ -1,177 +0,0 @@ -#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) -{ - if (al_atomic_s32_load(&stream->state, AL_ATOMIC_RELAXED) == SHRUB_STREAM_RUNNING) { - // DEBUG: This will be reached during normal operation. - return; - } - 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 deleted file mode 100644 index 4fae2c0..0000000 --- a/src/shrub/vcr.h +++ /dev/null @@ -1,45 +0,0 @@ -#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); |