diff options
Diffstat (limited to 'src')
103 files changed, 3372 insertions, 2662 deletions
diff --git a/src/bimu/client.c b/src/bimu/client.c index f679c0d..d35e3a1 100644 --- a/src/bimu/client.c +++ b/src/bimu/client.c @@ -1 +1,188 @@ +#include "../codec/libav/packet_ext.h" + #include "client.h" + +static void data_connection_callback(void *userdata, struct aki_packet_stream *stream) +{ + struct bmu_client *client = (struct bmu_client *)userdata; + (void)stream; + struct aki_packet *packet = aki_packet_create(); + aki_packet_write_u8(packet, BIMU_CONNECTION_DATA); + aki_packet_write_u16(packet, client->connection_id); + aki_packet_write_u16(packet, client->node_id); + aki_packet_stream_send_packet(&client->data, packet); +} + +static void data_connection_closed_callback(void *userdata, struct aki_packet_stream *stream) +{ + (void)userdata; + (void)stream; +} + +static struct bmu_client_stream *stream_from_index(struct bmu_client *client, s32 index) +{ + struct bmu_client_stream *stream; + al_array_foreach(client->streams, i, stream) { + if (stream->index == index) return stream; + } + return NULL; +} + +static void handle_data_packet_internal(struct bmu_client *client, struct aki_packet *packet) +{ + struct bmu_client_stream *stream; + switch (aki_packet_read_u8(packet)) { + case 0: + stream = stream_from_index(client, aki_packet_read_s32(packet)); + stream->client->handle_data_packet(stream->client, packet); + break; + case 1: + al_array_foreach(client->streams, i, stream) { + stream->client->handle_eof(stream->client); + } + break; + case 2: + break; + } +} + +static void data_packet_callback(void *userdata, struct aki_packet_stream *stream, + struct aki_packet *packet) +{ + struct bmu_client *client = (struct bmu_client *)userdata; + (void)stream; + handle_data_packet_internal(client, packet); + aki_packet_free(packet); +} + +static void packet_sent_callback(void *userdata, struct aki_packet *packet) +{ + (void)userdata; + aki_packet_free(packet); +} + +static void connect_data(struct bmu_client *client) +{ + aki_packet_stream_init(&client->data, AKI_SOCKET_TCP, data_connection_callback, + data_connection_closed_callback, data_packet_callback, packet_sent_callback, client); + aki_packet_stream_connect(&client->data, client->loop, &client->addr, client->port); +} + +static void control_connection_callback(void *userdata, struct aki_packet_stream *stream) +{ + struct bmu_client *client = (struct bmu_client *)userdata; + (void)stream; + struct aki_packet *packet = aki_packet_create(); + aki_packet_write_u8(packet, BIMU_CONNECTION_CONTROL); + aki_packet_write_u16(packet, 0); + aki_packet_write_u16(packet, client->node_id); + aki_packet_stream_send_packet(&client->control, packet); +} + +static void control_connection_closed_callback(void *userdata, struct aki_packet_stream *stream) +{ + (void)userdata; + (void)stream; +} + +static void parse_info_packet(struct bmu_client *client, struct aki_packet *packet) +{ + u32 count = aki_packet_read_u32(packet); + for (u32 i = 0; i < count; i++) { + struct bmu_client_stream *stream = al_alloc_object(struct bmu_client_stream); + stream->client = bmu_codec_client_create(); + stream->client->callback = client->callback; + stream->client->userdata = client->userdata; + stream->stream.type = aki_packet_read_u8(packet); + switch (stream->stream.type) { + case CAMU_NORMAL: + stream->type = BIMU_STREAM_VIDEO; + stream->index = 0; + stream->stream.video.width = aki_packet_read_s32(packet); + stream->stream.video.height = aki_packet_read_s32(packet); + break; +#ifdef HAVE_FFMPEG + case CAMU_FFMPEG_COMPAT: { + stream->stream.type = CAMU_FFMPEG_COMPAT; + stream->stream.av.format_context = avformat_alloc_context(); + const AVCodec *codec = avcodec_find_decoder(aki_packet_read_av_codec_id(packet)); + stream->stream.av.stream = aki_packet_read_av_stream(stream->stream.av.format_context, codec, packet); + stream->index = stream->stream.av.stream->index; + switch (stream->stream.av.stream->codecpar->codec_type) { + case AVMEDIA_TYPE_AUDIO: + stream->type = BIMU_STREAM_AUDIO; + al_array_push(client->subscribed, stream->index); + break; + case AVMEDIA_TYPE_VIDEO: + stream->type = BIMU_STREAM_VIDEO; + al_array_push(client->subscribed, stream->index); + break; + case AVMEDIA_TYPE_SUBTITLE: + default: + stream->type = BIMU_STREAM_UNKNOWN; + break; + } + break; + } +#endif + } + if (!stream->client->init(stream->client, NULL, stream)) { + } + al_array_push(client->streams, stream); + } +} + +static void send_subscribe_packet(struct bmu_client *client) +{ + struct aki_packet *packet = aki_packet_create(); + aki_packet_write_u8(packet, BIMU_CONTROL_SUBSCRIBE); + aki_packet_write_u32(packet, client->subscribed.size); + s32 index; + al_array_foreach(client->subscribed, i, index) { + aki_packet_write_s32(packet, index); + } + aki_packet_stream_send_packet(&client->control, packet); +} + +static void control_packet_callback(void *userdata, struct aki_packet_stream *stream, + struct aki_packet *packet) +{ + struct bmu_client *client = (struct bmu_client *)userdata; + (void)stream; + + switch (aki_packet_read_u8(packet)) { + case BIMU_CONTROL_INFO: + client->connection_id = aki_packet_read_u16(packet); + parse_info_packet(client, packet); + send_subscribe_packet(client); + break; + case BIMU_CONTROL_START: + connect_data(client); + break; + default: + break; + } + + aki_packet_free(packet); +} + +void bmu_client_connect(struct bmu_client *client, struct aki_event_loop *loop, str *addr, s32 port, u16 node_id) +{ + client->loop = loop; + client->node_id = node_id; + al_str_clone(&client->addr, addr); + client->port = port; + client->corked = false; + al_array_init(client->subscribed); + al_array_init(client->streams); + aki_packet_stream_init(&client->control, AKI_SOCKET_TCP, control_connection_callback, + control_connection_closed_callback, control_packet_callback, packet_sent_callback, client); + aki_packet_stream_connect(&client->control, client->loop, &client->addr, client->port); +} + +void bmu_client_cork(struct bmu_client *client, bool cork) +{ + if (cork == client->corked) return; + aki_packet_stream_cork(&client->data, cork); + client->corked = cork; +} diff --git a/src/bimu/client.h b/src/bimu/client.h index ec48275..54c903a 100644 --- a/src/bimu/client.h +++ b/src/bimu/client.h @@ -1,11 +1,33 @@ #pragma once #include <al/types.h> +#include <aki/packet_stream.h> #include "handlers/codec.h" +#include "common.h" + struct bmu_client_stream { u8 type; s32 index; struct camu_stream stream; + struct bmu_client_handler *client; +}; + +struct bmu_client { + struct aki_event_loop *loop; + u16 node_id; + u16 connection_id; + str addr; + s32 port; + struct aki_packet_stream control; + struct aki_packet_stream data; + bool corked; + array(s32) subscribed; + array(struct bmu_client_stream *) streams; + void (*callback)(void *, u8, struct bmu_client_stream *stream, void *); + void *userdata; }; + +void bmu_client_connect(struct bmu_client *client, struct aki_event_loop *loop, str *addr, s32 port, u16 node_id); +void bmu_client_cork(struct bmu_client *client, bool cork); diff --git a/src/bimu/common.h b/src/bimu/common.h index 1c4897a..fe8e0af 100644 --- a/src/bimu/common.h +++ b/src/bimu/common.h @@ -5,3 +5,18 @@ enum { BIMU_STREAM_VIDEO, BIMU_STREAM_UNKNOWN, }; + +enum { + BIMU_CONNECTION_CONTROL = 0, + BIMU_CONNECTION_DATA +}; + +enum { + BIMU_CONTROL_INFO = 0, + BIMU_CONTROL_SUBSCRIBE, + BIMU_CONTROL_START, + BIMU_CONTROL_TOGGLE_PAUSE, + BIMU_CONTROL_PAUSE, + BIMU_CONTROL_RESUME, + BIMU_CONTROL_SEEK +}; diff --git a/src/bimu/handler.h b/src/bimu/handler.h index e51bc2d..3881288 100644 --- a/src/bimu/handler.h +++ b/src/bimu/handler.h @@ -6,12 +6,14 @@ #include "../cache/handle.h" struct bmu_server_handler { - bool (*init)(struct bmu_server_handler *, struct cch_handle *, struct aki_packet_pool *); + bool (*init)(struct bmu_server_handler *, struct cch_handle *, struct aki_packet_pool *, void *); void (*write_info)(struct bmu_server_handler *, struct aki_packet *); + void (*subscribe)(struct bmu_server_handler *, s32, bool); s64 (*get_duration)(struct bmu_server_handler *); bool (*seek)(struct bmu_server_handler *, s64); bool (*step)(struct bmu_server_handler *); void (*free)(struct bmu_server_handler **); + void *userdata; }; enum { diff --git a/src/bimu/handlers/codec_client.c b/src/bimu/handlers/codec_client.c index ffe8c16..eda453f 100644 --- a/src/bimu/handlers/codec_client.c +++ b/src/bimu/handlers/codec_client.c @@ -19,9 +19,6 @@ static bool codec_client_init(struct bmu_client_handler *handler, struct camu_re { struct bmu_codec_client *codec = (struct bmu_codec_client *)handler; codec->dec = camu_lav_decoder_create(); - //codec->dec = camu_spng_decoder_create(); - //codec->dec = camu_stbi_decoder_create(); - //codec->dec = camu_wuffs_decoder_create(); codec->handler.stream = stream; if (!codec->dec->init(codec->dec, renderer, &stream->stream)) { return false; diff --git a/src/bimu/handlers/codec_server.c b/src/bimu/handlers/codec_server.c index b555370..3c27179 100644 --- a/src/bimu/handlers/codec_server.c +++ b/src/bimu/handlers/codec_server.c @@ -9,13 +9,11 @@ #include "codec.h" -static bool codec_server_init(struct bmu_server_handler *handler, struct cch_handle *handle, struct aki_packet_pool *pool) +static bool codec_server_init(struct bmu_server_handler *handler, struct cch_handle *handle, + struct aki_packet_pool *pool, void *userdata) { struct bmu_codec_server *codec = (struct bmu_codec_server *)handler; codec->demux = camu_lav_demuxer_create(); - //codec->demux = camu_spng_demuxer_create(); - //codec->demux = camu_stbi_demuxer_create(); - //codec->demux = camu_wuffs_demuxer_create(); if (!codec->demux->init(codec->demux, handle)) { return false; } @@ -29,6 +27,7 @@ static bool codec_server_init(struct bmu_server_handler *handler, struct cch_han #endif } codec->pool = pool; + codec->handler.userdata = userdata; return true; } @@ -54,6 +53,14 @@ static void codec_server_write_info(struct bmu_server_handler *handler, struct a } } +static void codec_server_subscribe(struct bmu_server_handler *handler, s32 index, bool subscribe) +{ + struct bmu_codec_server *codec = (struct bmu_codec_server *)handler; + if (subscribe) { + al_array_push(codec->demux->subscribed, index); + } +} + static s64 codec_server_get_duration(struct bmu_server_handler *handler) { struct bmu_codec_server *codec = (struct bmu_codec_server *)handler; @@ -71,61 +78,38 @@ static bool codec_server_step(struct bmu_server_handler *handler) struct bmu_codec_server *codec = (struct bmu_codec_server *)handler; struct aki_packet *packet = aki_packet_pool_get(codec->pool); if (!packet) return false; - aki_packet_write_u8(packet, 0); + packet->userdata = codec->handler.userdata; s32 ret = codec->demux->get_packet(codec->demux, &codec->packet); - do { - if (ret == CAMU_OK) { - switch (codec->packet.type) { - case CAMU_NORMAL: { - aki_packet_write_s32(packet, 0); - aki_packet_write_u8(packet, 0); - aki_packet_write_u8(packet, codec->packet.type); - aki_packet_write_buffer(packet, codec->packet.buffer); - break; - } -#ifdef HAVE_FFMPEG - case CAMU_FFMPEG_COMPAT: { - AVPacket *pkt = codec->packet.av.pkt; - aki_packet_write_s32(packet, pkt->stream_index); - aki_packet_write_u8(packet, 0); - aki_packet_write_u8(packet, codec->packet.type); - aki_packet_write_av_packet(packet, pkt); - av_packet_unref(pkt); - break; - } -#endif - } - aki_packet_pool_submit(codec->pool, packet); - } else if (ret == CAMU_ERR_EOF) { - struct camu_stream *stream; - al_array_foreach_ptr(codec->demux->streams, i, stream) { - if (i != 0) { // HACK - packet = aki_packet_pool_get(codec->pool); - if (!packet) return false; - aki_packet_write_u8(packet, 0); - } - switch (stream->type) { - case CAMU_NORMAL: - aki_packet_write_s32(packet, 0); - break; + if (ret == CAMU_OK) { + aki_packet_write_u8(packet, 0); + switch (codec->packet.type) { + case CAMU_NORMAL: { + aki_packet_write_s32(packet, 0); + aki_packet_write_u8(packet, codec->packet.type); + aki_packet_write_buffer(packet, codec->packet.buffer); + break; + } #ifdef HAVE_FFMPEG - case CAMU_FFMPEG_COMPAT: - aki_packet_write_s32(packet, stream->av.stream->index); - break; + case CAMU_FFMPEG_COMPAT: { + AVPacket *pkt = codec->packet.av.pkt; + aki_packet_write_s32(packet, pkt->stream_index); + aki_packet_write_u8(packet, codec->packet.type); + aki_packet_write_av_packet(packet, pkt); + av_packet_unref(pkt); + break; + } #endif - } - aki_packet_write_u8(packet, 1); - aki_packet_pool_submit(codec->pool, packet); - } - return false; - } else { - aki_packet_write_s32(packet, -2); - aki_packet_write_u8(packet, 2); - aki_packet_pool_submit(codec->pool, packet); - return false; } - break; - } while (1); + aki_packet_pool_submit(codec->pool, packet); + } else if (ret == CAMU_ERR_EOF) { + aki_packet_write_u8(packet, 1); + aki_packet_pool_submit(codec->pool, packet); + return false; + } else { + aki_packet_write_u8(packet, 2); + aki_packet_pool_submit(codec->pool, packet); + return false; + } return true; } @@ -151,6 +135,7 @@ struct bmu_server_handler *bmu_codec_server_create(void) struct bmu_codec_server *codec = al_alloc_object(struct bmu_codec_server); codec->handler.init = codec_server_init; codec->handler.write_info = codec_server_write_info; + codec->handler.subscribe = codec_server_subscribe; codec->handler.get_duration = codec_server_get_duration; codec->handler.seek = codec_server_seek; codec->handler.step = codec_server_step; diff --git a/src/bimu/local.c b/src/bimu/local.c index 3f62f75..76824ed 100644 --- a/src/bimu/local.c +++ b/src/bimu/local.c @@ -34,10 +34,10 @@ static aki_thread_result AKI_THREADCALL stream_thread(void *userdata) } switch (aki_packet_read_u8(packet)) { case 0: - stream->client->handle_data_packet(stream->client, packet); + stream->s.client->handle_data_packet(stream->s.client, packet); break; case 1: - stream->client->handle_eof(stream->client); + stream->s.client->handle_eof(stream->s.client); break; } aki_packet_pool_return(&stream->runner->pool, packet); @@ -76,7 +76,7 @@ bool bmu_local_init(struct bmu_local *runner, struct cch_entry *entry) cch_entry_get_handle(runner->entry, &runner->handle); aki_event_loop_init(&runner->loop); aki_packet_pool_init(&runner->pool, PACKETS, &runner->loop, packet_pool_callback, runner); - if (!runner->server->init(runner->server, &runner->handle, &runner->pool)) { + if (!runner->server->init(runner->server, &runner->handle, &runner->pool, NULL)) { return false; } al_array_init(runner->streams); @@ -94,9 +94,9 @@ bool bmu_local_prepare_clients(struct bmu_local *runner, struct camu_renderer *r struct bmu_local_stream *local = al_alloc_object(struct bmu_local_stream); struct bmu_client_stream *stream = &local->s; local->runner = runner; - local->client = bmu_codec_client_create(); - local->client->callback = runner->callback; - local->client->userdata = runner->userdata; + local->s.client = bmu_codec_client_create(); + local->s.client->callback = runner->callback; + local->s.client->userdata = runner->userdata; aki_cond_init(&local->cond); aki_mutex_init(&local->mutex); aki_packet_cache_init(&local->cache, PACKETS); @@ -131,7 +131,7 @@ bool bmu_local_prepare_clients(struct bmu_local *runner, struct camu_renderer *r } #endif } - if (!local->client->init(local->client, renderer, stream)) { + if (!local->s.client->init(local->s.client, renderer, stream)) { aki_packet_free(packet); return false; } @@ -224,7 +224,7 @@ void bmu_local_seek(struct bmu_local *runner, f64 percent) al_atomic_bool_store(&runner->closed, false, AL_ATOMIC_RELAXED); aki_packet_pool_enable(&runner->pool); al_array_foreach(runner->streams, i, stream) { - stream->client->flush(stream->client); + stream->s.client->flush(stream->s.client); al_atomic_s32_store(&stream->state, BIMU_LOCAL_RUNNING, AL_ATOMIC_RELAXED); aki_packet_cache_enable(&stream->cache); aki_thread_create(&stream->thread, stream_thread, stream); @@ -254,7 +254,7 @@ void bmu_local_close(struct bmu_local *runner) { struct bmu_local_stream *stream = NULL; al_array_foreach(runner->streams, i, stream) { - stream->client->free(&stream->client); + stream->s.client->free(&stream->s.client); switch (stream->s.stream.type) { case CAMU_NORMAL: break; diff --git a/src/bimu/local.h b/src/bimu/local.h index 1928249..10c0cc8 100644 --- a/src/bimu/local.h +++ b/src/bimu/local.h @@ -18,7 +18,6 @@ enum { struct bmu_local_stream { struct bmu_client_stream s; - struct bmu_client_handler *client; atomic_s32 state; struct aki_cond cond; struct aki_mutex mutex; diff --git a/src/bimu/server.c b/src/bimu/server.c index c1b0c38..f572290 100644 --- a/src/bimu/server.c +++ b/src/bimu/server.c @@ -1,46 +1,327 @@ +#include <al/random.h> +#include <aki/common.h> + +#include "handlers/codec.h" + #include "server.h" -static void packet_callback(void *userdata, struct aki_packet_stream *stream, struct aki_packet *packet) +#define START_DELAY 700000 // 700ms +#define DELAY_GIVE 200000 // 200ms + +static void control_connection_closed_callback(void *userdata, struct aki_packet_stream *stream) { (void)userdata; (void)stream; - aki_packet_free(packet); } -static void connection_closed_callback(void *userdata, struct aki_packet_stream *stream) +static void data_connection_closed_callback(void *userdata, struct aki_packet_stream *stream) { (void)userdata; (void)stream; } +static void default_packet_sent_callback(void *userdata, struct aki_packet *packet) +{ + (void)userdata; + aki_packet_free(packet); +} + static void packet_sent_callback(void *userdata, struct aki_packet *packet) { + struct bmu_node_connection *connection = (struct bmu_node_connection *)userdata; + aki_packet_pool_return(&connection->node->pool, packet); +} + +static void data_packet_callback(void *userdata, struct aki_packet_stream *stream, struct aki_packet *packet) +{ + (void)userdata; + (void)stream; + aki_packet_free(packet); +} + +static u8 packet_pool_callback(void *userdata, struct aki_packet *packet) +{ (void)userdata; - (void)packet; + struct bmu_node_connection *connection = (struct bmu_node_connection *)packet->userdata; + aki_packet_stream_send_packet(connection->data, packet); + return AKI_PACKET_POOL_KEEP; +} + +static void send_resume_packet(struct bmu_node_connection *connection, u64 seek_pos, u64 ts) +{ + struct aki_packet *packet = aki_packet_create(); + aki_packet_write_u8(packet, BIMU_CONTROL_RESUME); + aki_packet_write_u64(packet, seek_pos); + aki_packet_write_u64(packet, ts); + aki_packet_stream_send_packet(connection->control, packet); +} + +static void send_pause_packet(struct bmu_node_connection *connection, u64 seek_pos) +{ + struct aki_packet *packet = aki_packet_create(); + aki_packet_write_u8(packet, BIMU_CONTROL_PAUSE); + aki_packet_write_u64(packet, seek_pos); + aki_packet_stream_send_packet(connection->control, packet); +} + +static void send_start_packet(struct bmu_node_connection *connection) +{ + struct aki_packet *packet = aki_packet_create(); + aki_packet_write_u8(packet, BIMU_CONTROL_START); + u64 ts = aki_get_timestamp(); + u64 ats = ts + START_DELAY; + s64 seek_pos = connection->node->seek_pos; + if (!connection->node->paused) { + if (connection->node->start >= 0) { + if ((s64)ts < (connection->node->start - DELAY_GIVE)) { + ats = connection->node->start; + } else { + seek_pos += ats - connection->node->start; + } + } else { + connection->node->start = ats; + } + } + s64 duration = connection->handler->get_duration(connection->handler); + if (seek_pos > duration) seek_pos = duration; + // -1 signals to not call seek. We have to distinguish between a seek to 0 + // and the start of a stream. + connection->seek_pos = seek_pos > 0 ? seek_pos : -1; + aki_packet_write_u64(packet, seek_pos); + aki_packet_write_u64(packet, ats); + aki_packet_write_u8(packet, connection->node->paused); + aki_packet_stream_send_packet(connection->control, packet); +} + +static void control_packet_callback(void *userdata, struct aki_packet_stream *stream, struct aki_packet *packet) +{ + struct bmu_node_connection *connection = (struct bmu_node_connection *)userdata; + (void)stream; + + switch (aki_packet_read_u8(packet)) { + case BIMU_CONTROL_SUBSCRIBE: { + u32 count = aki_packet_read_u32(packet); + for (u32 i = 0; i < count; i++) { + connection->handler->subscribe(connection->handler, aki_packet_read_s32(packet), true); + } + send_start_packet(connection); + break; + } + case BIMU_CONTROL_TOGGLE_PAUSE: { + struct bmu_node *node = connection->node; + u64 pos = aki_packet_read_u64(packet); + u8 paused = connection->paused; + if (paused == BIMU_PLAYING) { + node->seek_pos = pos; + node->paused = true; + // If node is paused, start will not be used for new connections. + // So, set it to -1 and update it on the first resume from any connection. + node->start = -1; + } else if (paused == BIMU_PAUSED_AND_RUNNING) { + node->paused = false; + node->start = aki_get_timestamp() + START_DELAY; + } + struct bmu_node_connection *rconnection; + al_array_foreach(node->connections, i, rconnection) { + switch (rconnection->paused) { + case BIMU_PLAYING: + al_assert(paused == BIMU_PLAYING || paused == BIMU_NOT_PAUSED); + rconnection->paused = BIMU_PAUSED; + send_pause_packet(rconnection, node->seek_pos); + break; + case BIMU_NOT_PAUSED: + al_assert(paused != BIMU_PAUSED); + rconnection->paused = BIMU_PAUSED; + break; + case BIMU_PAUSED: + al_assert(paused != BIMU_NOT_PAUSED); + rconnection->paused = BIMU_NOT_PAUSED; + break; + case BIMU_PAUSED_AND_RUNNING: + al_assert(paused != BIMU_NOT_PAUSED); + send_resume_packet(rconnection, node->seek_pos, node->start); + rconnection->paused = BIMU_PLAYING; + break; + } + } + break; + } + case BIMU_CONTROL_SEEK: { + u64 pos = aki_packet_read_u64(packet); + connection->seek_pos = pos; + struct aki_packet *npacket = aki_packet_create(); + aki_packet_write_u8(npacket, BIMU_CONTROL_SEEK); + aki_packet_stream_send_packet(connection->control, npacket); + break; + } + } + aki_packet_free(packet); } -static void flushed_callback(void *userdata) +static struct bmu_node *get_node_by_id(struct bmu_server *server, u16 id) +{ + struct bmu_node *node; + al_array_foreach(server->nodes, i, node) { + if (node->id == id) return node; + } + return NULL; +} + +static struct bmu_node_connection *get_connection_by_id(struct bmu_node *node, u16 id) +{ + struct bmu_node_connection *connection; + al_array_foreach(node->connections, i, connection) { + if (connection->id == id) return connection; + } + return NULL; +} + +static void send_control_info_packet(struct bmu_node_connection *connection) +{ + struct aki_packet *packet = aki_packet_create(); + aki_packet_write_u8(packet, BIMU_CONTROL_INFO); + aki_packet_write_u16(packet, connection->id); + connection->handler = bmu_codec_server_create(); + connection->handler->init(connection->handler, &connection->handle, &connection->node->pool, connection); + connection->handler->write_info(connection->handler, packet); + aki_packet_stream_send_packet(connection->control, packet); +} + +static aki_thread_result AKI_THREADCALL handler_thread(void *userdata) +{ + struct bmu_node_connection *connection = (struct bmu_node_connection *)userdata; + while (connection->handler->step(connection->handler)) {} + return 0; +} + +static void maybe_start_handler(struct bmu_node_connection *connection) +{ + al_assert(!connection->running); + if (connection->seek_pos >= 0) { + connection->handler->seek(connection->handler, connection->seek_pos); + } + aki_thread_create(&connection->thread, handler_thread, connection); + connection->running = true; + struct bmu_node *node = connection->node; + if (connection->paused == BIMU_NOT_PAUSED) { + if (node->start == -1) { + node->paused = false; + node->start = aki_get_timestamp() + START_DELAY; + } + //send_resume_packet(connection, connection->seek_pos, node->start); + connection->paused = BIMU_PLAYING; + } else if (connection->paused == BIMU_PAUSED) { + connection->paused = BIMU_PAUSED_AND_RUNNING; + } +} + +static void identify_packet_callback(void *userdata, struct aki_packet_stream *stream, struct aki_packet *packet) +{ + struct bmu_server *server = (struct bmu_server *)userdata; + + u8 type = aki_packet_read_u8(packet); + + u16 connection_id = aki_packet_read_u16(packet); + u16 node_id = aki_packet_read_u16(packet); + + // Disconnecting a stream at any point before setting the new callbacks should + // be valid and handled properly by the client. + // Generalize erronious connection checking/handling. + // - Node with id doesn't exist. + // - Control connected more than once and/or after info is sent. + // - Data connected before control. + // - Data connected more than once. + + struct bmu_node *node = get_node_by_id(server, node_id); + al_assert(node); + + struct bmu_node_connection *connection = get_connection_by_id(node, connection_id); + if (!connection) { + al_assert(type == BIMU_CONNECTION_CONTROL); + connection = al_alloc_object(struct bmu_node_connection); + al_array_push(node->connections, connection); + connection->id = node->connection_id++; + connection->node = node; + connection->control = NULL; + connection->data = NULL; + cch_entry_get_handle(connection->node->entry, &connection->handle); + connection->paused = BIMU_PLAYING; + connection->running = false; + } + + switch (type) { + case BIMU_CONNECTION_CONTROL: + aki_packet_stream_set_no_delay(stream, true); + connection->control = stream; + stream->packet_callback = control_packet_callback; + stream->packet_sent_callback = default_packet_sent_callback; + stream->connection_closed_callback = control_connection_closed_callback; + stream->userdata = connection; + send_control_info_packet(connection); + break; + case BIMU_CONNECTION_DATA: + // TODO: Account for control connection being in a reconnect state. + // TODO: is it possible for data->data to be non-NULL during normal operation + // (disconnect and reconnect very fast?). + // - Call maybe_cleanup_data in this function? + connection->data = stream; + stream->packet_callback = data_packet_callback; + stream->packet_sent_callback = packet_sent_callback; + stream->connection_closed_callback = data_connection_closed_callback; + stream->userdata = connection; + maybe_start_handler(connection); + break; + } + + aki_packet_free(packet); +} + +static void connection_closed_callback(void *userdata, struct aki_packet_stream *stream) { (void)userdata; + (void)stream; } static void connection_callback(void *userdata, struct aki_packet_stream *stream) { - stream->userdata = userdata; - stream->packet_callback = packet_callback; - stream->packet_sent_callback = packet_sent_callback; + struct bmu_server *server = (struct bmu_server *)userdata; stream->connection_closed_callback = connection_closed_callback; - stream->flushed_callback = flushed_callback; + stream->packet_callback = identify_packet_callback; + stream->packet_sent_callback = default_packet_sent_callback; + stream->flushed_callback = NULL; + stream->userdata = server; } bool bmu_server_init(struct bmu_server *server) { - return aki_packet_stream_init(&server->server, AKI_SOCKET_TCP, connection_callback, - NULL, NULL, NULL, server); + al_array_init(server->nodes); + return aki_packet_stream_init(&server->server, AKI_SOCKET_TCP, connection_callback, NULL, NULL, NULL, server); } void bmu_server_listen(struct bmu_server *server, struct aki_event_loop *loop, str *addr, s32 port) { server->loop = loop; - aki_packet_stream_listen(&server->server, server->loop, addr, port); + aki_packet_stream_listen(&server->server, loop, addr, port); +} + +u16 bmu_server_create_node(struct bmu_server *server, struct cch_entry *entry) +{ + struct bmu_node *node = al_alloc_object(struct bmu_node); + al_array_push(server->nodes, node); + node->id = al_rand_u16(); + node->entry = entry; + node->paused = false; + node->start = -1; + node->seek_pos = 0; + aki_packet_pool_init(&node->pool, 16, server->loop, packet_pool_callback, node); + //aki_packet_pool_init(&node->pool, 1400, server->loop, packet_pool_callback, node); + node->connection_id = 1; + al_array_init(node->connections); + node->server = server; + return node->id; +} + +void bmu_server_close(struct bmu_server *server) +{ + (void)server; } diff --git a/src/bimu/server.h b/src/bimu/server.h index ba3093e..754ab7f 100644 --- a/src/bimu/server.h +++ b/src/bimu/server.h @@ -1,21 +1,54 @@ #pragma once +#include <al/array.h> #include <aki/packet_stream.h> +#include <aki/packet_pool.h> #include "../cache/entry.h" +#include "../cache/handle.h" + +#include "common.h" +#include "handler.h" + +enum { + BIMU_PLAYING = 0, + BIMU_NOT_PAUSED, + BIMU_PAUSED, + BIMU_PAUSED_AND_RUNNING +}; struct bmu_node_connection { + u16 id; + struct bmu_node *node; + struct aki_packet_stream *control; + struct aki_packet_stream *data; + bool running; + u8 paused; + s64 seek_pos; struct cch_handle handle; + struct bmu_server_handler *handler; + struct aki_thread thread; }; struct bmu_node { + u16 id; struct cch_entry *entry; + bool paused; + s64 start; + s64 seek_pos; + struct aki_packet_pool pool; + u16 connection_id; + array(struct bmu_node_connection *) connections; + struct bmu_server *server; }; struct bmu_server { - struct aki_packet_stream server; struct aki_event_loop *loop; + struct aki_packet_stream server; + array(struct bmu_node *) nodes; }; bool bmu_server_init(struct bmu_server *server); void bmu_server_listen(struct bmu_server *server, struct aki_event_loop *loop, str *addr, s32 port); +u16 bmu_server_create_node(struct bmu_server *server, struct cch_entry *entry); +void bmu_server_close(struct bmu_server *server); diff --git a/src/buffer/audio.c b/src/buffer/audio.c index 939b98a..9799a95 100644 --- a/src/buffer/audio.c +++ b/src/buffer/audio.c @@ -2,16 +2,16 @@ #include "audio.h" -#define BUFFER_USEC (10 * 1000000L) +#define BUFFER_USEC (12 * 1000000L) #define BUFFER_WATERMARK_LOW (3 * 1000000L) // Must be a most half of the buffer size. -#define BUFFER_WATERMARK_HIGH (4 * 1000000L) +#define BUFFER_WATERMARK_HIGH (0.75 * 1000000L) #define DESYNC_PTS 0.022 #ifdef CAMU_AUDIO_BUFFER_FADE -#define FADE 0.9999 +#define FADE 1.0 #define FADE_LENGTH 6 -#define FADE_STEP(rate) ((FADE / (rate)) / (FADE_LENGTH - 1)) +#define FADE_STEP(rate) ((FADE / ((f64)rate)) / (FADE_LENGTH - 1.0)) #endif enum { @@ -23,7 +23,6 @@ enum { enum { PAUSE_PRE = 0, PAUSE_UNPAUSED, - PAUSE_IGNORE_DESYNC, #ifdef CAMU_AUDIO_BUFFER_FADE PAUSE_FADING, #endif @@ -73,8 +72,8 @@ bool camu_audio_buffer_init(struct camu_audio_buffer *buf, struct camu_clock *cl buf->clock = clock; buf->pts = camu_mixer_get_latency(buf->mixer); buf->buffered = false; - //buf->pause = PAUSE_PRE; - buf->pause = PAUSE_IGNORE_DESYNC; + buf->pause = PAUSE_PRE; + buf->ignore_desync = true; #ifdef CAMU_AUDIO_BUFFER_FADE buf->fade_period = 0; buf->fade_offset = 0; @@ -93,23 +92,29 @@ bool camu_audio_buffer_configure(struct camu_audio_buffer *buf, struct camu_stre av_channel_layout_copy(&buf->fmt.in_channel_layout, &stream->av.stream->codecpar->ch_layout); buf->fmt.in_channel_count = buf->fmt.in_channel_layout.nb_channels; buf->fmt.in_sample_rate = stream->av.stream->codecpar->sample_rate; + camu_mixer_pick_format(buf->mixer, &buf->fmt); av_channel_layout_default(&buf->fmt.req_channel_layout, buf->fmt.req_channel_count); if (!setup_optimal_resampler(buf, &buf->resamp, &buf->fmt)) { return false; } + const char *in_format_name = av_get_sample_fmt_name(buf->fmt.in_format); const char *req_format_name = av_get_sample_fmt_name(buf->fmt.req_format); al_log_info("audio_buffer", "Stream: %s (%dch) %dHz -> %s (%dch) %dHz.", in_format_name, buf->fmt.in_channel_layout.nb_channels, buf->fmt.in_sample_rate, req_format_name, buf->fmt.req_channel_layout.nb_channels, buf->fmt.req_sample_rate); + buf->size = camu_lav_resample_fmt_usec_to_bytes(&buf->fmt, BUFFER_USEC); buf->data = (u8 *)al_malloc(buf->size); al_ring_buffer_init(&buf->rb, buf->data, buf->size); + buf->watermark.low = camu_lav_resample_fmt_usec_to_bytes(&buf->fmt, BUFFER_WATERMARK_LOW); buf->watermark.high = camu_lav_resample_fmt_usec_to_bytes(&buf->fmt, BUFFER_WATERMARK_HIGH); camu_peak_buffer_init(&buf->peak); + buf->stream = stream; + return true; } @@ -144,9 +149,9 @@ static bool push_internal(struct camu_audio_buffer *buf, u8 **data, s32 sample_c } // Continuing based on an outdated continue_mark value is safe as long as the // peak buffer is smaller than the low watermark and the low watermark is - // less than or equal to half the buffer. + // less than or equal to half the buffer size. al_atomic_size_t_store(&buf->continue_mark, buf->watermark.low + peak, AL_ATOMIC_RELAXED); - buf->callback(buf->userdata, CAMU_BUFFER_STOP); + buf->callback(buf->userdata, CAMU_BUFFER_CORK); return false; } @@ -195,6 +200,7 @@ void camu_audio_buffer_reset(struct camu_audio_buffer *buf) al_atomic_u8_store(&buf->flow, FLOWING, AL_ATOMIC_RELAXED); buf->pts = camu_clock_get_base_pts(buf->clock); buf->buffered = false; + buf->pause = PAUSE_PRE; al_ring_buffer_reset(&buf->rb); } @@ -212,10 +218,11 @@ void camu_audio_buffer_flush(struct camu_audio_buffer *buf) #define TYPED_CAST(f, type) (f64)(*((type *)f)) #define TYPED_CLAMP(f, type, min, max, ret) *((type *)ret) = (type)AL_CLAMP(f, min, max) +// This is obviously bad for optimization, should just make a function for each type. static void handle_fade(struct camu_audio_buffer *buf, u8 *data, size_t size, bool out) { - u8 bits[MAX_BYTES]; - u8 *ptr = bits + (MAX_BYTES - buf->bytes_per_sample); + u8 bytes[MAX_BYTES]; + u8 *ptr = bytes + (MAX_BYTES - buf->bytes_per_sample); f64 value; u32 sample_count = size / buf->bytes_per_sample; u32 channel_count = 2; @@ -248,14 +255,14 @@ static void handle_fade(struct camu_audio_buffer *buf, u8 *data, size_t size, bo } #endif -#define NOT_UNPAUSED(pause) (pause != PAUSE_UNPAUSED && pause != PAUSE_PRE && pause != PAUSE_IGNORE_DESYNC) +#define NOT_UNPAUSED(pause) (pause != PAUSE_UNPAUSED && pause != PAUSE_PRE) size_t camu_audio_buffer_read(struct camu_audio_buffer *buf, u8 *data, size_t req) { - f64 pts = camu_clock_get_pts(buf->clock); - if (pts == -1.0) { + if (camu_clock_is_paused(buf->clock)) { #ifdef CAMU_AUDIO_BUFFER_FADE if (buf->pause == PAUSE_PLAYING) { + // Fade out. Signified by >0 fade_period and PAUSE_FADING. buf->volume = FADE; buf->fade_period = FADE_LENGTH; buf->fade_offset = 0; @@ -269,10 +276,13 @@ size_t camu_audio_buffer_read(struct camu_audio_buffer *buf, u8 *data, size_t re return req; } } else if (buf->pause == PAUSE_FADING || buf->pause == PAUSE_UNPAUSED) { + // Fade in. Signified by >0 fade_period and PAUSE_PLAYING. buf->volume = 1.0 - FADE; buf->fade_period = FADE_LENGTH; buf->fade_offset = 0; - buf->pause = PAUSE_PLAYING; + // We don't want to consider sync if reversing an active + // fade but we do if PAUSE_UNPAUSED. + if (buf->pause == PAUSE_FADING) buf->pause = PAUSE_PLAYING; #else al_memset(data, 0, req); if (NOT_UNPAUSED(buf->pause)) { @@ -289,23 +299,30 @@ size_t camu_audio_buffer_read(struct camu_audio_buffer *buf, u8 *data, size_t re #ifdef CAMU_AUDIO_BUFFER_FADE if (buf->pause != PAUSE_FADING) { #endif - pts -= buf->pts; - if (UNLIKELY(buf->pause == PAUSE_PRE || buf->pause == PAUSE_UNPAUSED)) { + f64 pts = camu_clock_get_pts(buf->clock) - buf->pts; + // We can assume a call to audio_buffer_read will happen before + // the user NEEDS the data, so, we can't accurately assess the sync here. + if (UNLIKELY(!buf->ignore_desync && (buf->pause == PAUSE_PRE || buf->pause == PAUSE_UNPAUSED))) { + if (fabs(pts) >= 0.322) { + al_log_warn("audio_buffer", "Abnormally large audio desync of %.5fs", pts); + } if (pts > 0.0) { ret = camu_lav_resample_fmt_sec_to_bytes(&buf->fmt, pts); ret = AL_MIN(ret, size); - al_log_debug("audio_buffer", "Skipping %.5fs of audio.", pts); + al_log_debug("audio_buffer", "Skipping %.5fs of audio (%zu bytes).", pts, ret); ret = al_ring_buffer_discard(&buf->rb, ret); size -= ret; buf->pts += camu_lav_resample_fmt_bytes_to_sec(&buf->fmt, ret); + // Could go on to underrun. } else if (pts < 0.0) { pts = -pts; ret = camu_lav_resample_fmt_sec_to_bytes(&buf->fmt, pts); ret = AL_MIN(ret, req); - al_log_debug("audio_buffer", "Delaying audio by %.5fs.", pts); + al_log_debug("audio_buffer", "Delaying audio by %.5fs (%zu bytes).", pts, ret); al_memset(data, 0, ret); data += ret; - req -= ret; + // We can continue to delay. + if ((req -= ret) == 0) return signal; } } if (buf->pause != PAUSE_PLAYING) buf->pause = PAUSE_PLAYING; @@ -339,7 +356,7 @@ size_t camu_audio_buffer_read(struct camu_audio_buffer *buf, u8 *data, size_t re if (flow == FLOWING) { ret = al_atomic_size_t_load(&buf->continue_mark, AL_ATOMIC_RELAXED); if (ret && ((buf->size - size) - req) >= ret) { - buf->callback(buf->userdata, CAMU_BUFFER_CONTINUE); + buf->callback(buf->userdata, CAMU_BUFFER_UNCORK); } } #ifdef CAMU_AUDIO_BUFFER_FADE diff --git a/src/buffer/audio.h b/src/buffer/audio.h index 1a51807..7bd30a2 100644 --- a/src/buffer/audio.h +++ b/src/buffer/audio.h @@ -20,6 +20,7 @@ struct camu_audio_buffer { f64 pts; u8 pause; + bool ignore_desync; struct camu_clock *clock; f64 volume; diff --git a/src/buffer/clock.c b/src/buffer/clock.c index 5f4133e..c9b18ce 100644 --- a/src/buffer/clock.c +++ b/src/buffer/clock.c @@ -4,24 +4,26 @@ void camu_clock_init(struct camu_clock *clock) { f64 tick = aki_get_tick(); al_atomic_f64_store(&clock->pause, tick, AL_ATOMIC_RELAXED); + al_atomic_f64_store(&clock->start, tick, AL_ATOMIC_RELAXED); al_atomic_f64_store(&clock->base, 0.0, AL_ATOMIC_RELAXED); - al_atomic_f64_store(&clock->start, 0.0, AL_ATOMIC_RELAXED); - al_atomic_f64_store(&clock->offset, 0.0, AL_ATOMIC_RELAXED); } void camu_clock_pause(struct camu_clock *clock) { + f64 pause = al_atomic_f64_load(&clock->pause, AL_ATOMIC_RELAXED); + if (pause != -1.0) return; al_atomic_f64_store(&clock->pause, aki_get_tick(), AL_ATOMIC_RELAXED); } void camu_clock_resume(struct camu_clock *clock) { f64 pause = al_atomic_f64_load(&clock->pause, AL_ATOMIC_RELAXED); + if (pause == -1.0) return; f64 start = al_atomic_f64_load(&clock->start, AL_ATOMIC_RELAXED); f64 base = al_atomic_f64_load(&clock->base, AL_ATOMIC_RELAXED); if (start == 0.0) start = (pause - base); - al_atomic_f64_store(&clock->start, aki_get_tick() - (pause - start), AL_ATOMIC_RELAXED); al_atomic_f64_store(&clock->pause, -1.0, AL_ATOMIC_RELAXED); + al_atomic_f64_store(&clock->start, aki_get_tick() - (pause - start), AL_ATOMIC_RELAXED); } void camu_clock_seek(struct camu_clock *clock, f64 pos) @@ -47,7 +49,7 @@ f64 camu_clock_get_base_pts(struct camu_clock *clock) f64 camu_clock_get_pts(struct camu_clock *clock) { f64 pause = al_atomic_f64_load(&clock->pause, AL_ATOMIC_RELAXED); - if (pause != -1.0) return -1.0; f64 start = al_atomic_f64_load(&clock->start, AL_ATOMIC_RELAXED); + if (pause != -1.f) return pause - start; return aki_get_tick() - start; } diff --git a/src/buffer/clock.h b/src/buffer/clock.h index 072f7d3..9f34807 100644 --- a/src/buffer/clock.h +++ b/src/buffer/clock.h @@ -4,7 +4,7 @@ #include <al/atomic.h> struct camu_clock { - atomic_f64 base, start, pause, offset; + atomic_f64 base, start, pause; }; void camu_clock_init(struct camu_clock *clock); diff --git a/src/buffer/common.h b/src/buffer/common.h index 8a99c96..64168ff 100644 --- a/src/buffer/common.h +++ b/src/buffer/common.h @@ -2,8 +2,8 @@ enum { CAMU_BUFFER_BUFFERED = 0, - CAMU_BUFFER_STOP, - CAMU_BUFFER_CONTINUE, + CAMU_BUFFER_CORK, + CAMU_BUFFER_UNCORK, CAMU_BUFFER_PAUSED, CAMU_BUFFER_EOF }; diff --git a/src/buffer/video.c b/src/buffer/video.c index d5d2c12..6c8d6e3 100644 --- a/src/buffer/video.c +++ b/src/buffer/video.c @@ -4,13 +4,17 @@ //#define CAMU_VIDEO_BUFFER_FORCE_SCALER +#define BUFFER_WATERMARK_LOW 16 // frames. +#define BUFFER_WATERMARK_BUFFERED 24 +#define BUFFER_WATERMARK_HIGH 48 +#define BUFFER_WATERMARK_RESET BUFFER_WATERMARK_HIGH + 16. + bool camu_video_buffer_init(struct camu_video_buffer *buf, struct camu_clock *clock, struct camu_renderer *renderer) { buf->queue = renderer->create_queue(renderer); buf->buffered = false; buf->clock = clock; - buf->last_pts = 0.0; #ifdef CAMU_SCREEN_THREADED al_atomic_bool_store(&buf->ref, false, AL_ATOMIC_RELAXED); #endif @@ -32,20 +36,24 @@ bool camu_video_buffer_configure(struct camu_video_buffer *buf, struct camu_stre s32 height = stream->av.stream->codecpar->height; buf->fmt.in_width = buf->stream->video.width = width; buf->fmt.in_height = buf->stream->video.height = height; + buf->fmt.in_format = stream->av.stream->codecpar->format; buf->single_frame = stream->av.stream->duration == 0 || stream->av.stream->avg_frame_rate.den == 0; - buf->avg_frame_duration = buf->single_frame ? 0.0 : - av_q2d(av_inv_q(stream->av.stream->avg_frame_rate)); - buf->fmt.in_format = stream->av.stream->codecpar->format; + AVRational frame_rate = buf->single_frame ? (AVRational){ 0, 1 } : + stream->av.stream->avg_frame_rate; + buf->avg_frame_duration = av_q2d(av_inv_q(frame_rate)); const char *format_name = av_get_pix_fmt_name(buf->fmt.in_format); - al_log_info("video_buffer", "Stream: %s (%dx%d) %s %.2ffps.", - format_name, width, height, buf->single_frame ? "IMAGE" : "VIDEO", - buf->single_frame ? 0.0 : 1.0 / buf->avg_frame_duration); + al_log_info("video_buffer", "Stream: %s (%dx%d) %s %.2ffps.", format_name, + width, height, buf->single_frame ? "IMAGE" : "VIDEO", av_q2d(frame_rate)); #ifdef CAMU_VIDEO_BUFFER_FORCE_SCALER buf->fmt.req_width = width; buf->fmt.req_height = height; buf->fmt.req_format = AV_PIX_FMT_RGBA; - if (!camu_lav_scaler_init(&buf->scale, &buf->fmt)) { + if (camu_lav_scaler_init(&buf->scale, &buf->fmt) && buf->fmt.scaler_needed) { + format_name = av_get_pix_fmt_name(buf->fmt.req_format); + al_log_info("video_buffer", "Scaling stream to: %s (%dx%d)", format_name, + buf->fmt.req_width, buf->fmt.req_height); + } else { return false; } #endif @@ -59,40 +67,38 @@ bool camu_video_buffer_configure(struct camu_video_buffer *buf, struct camu_stre static void after_push_internal(struct camu_video_buffer *buf) { s32 count = buf->queue->count(buf->queue); - if (!buf->buffered && (buf->single_frame || count >= 15)) { + if (!buf->buffered && (buf->single_frame || count >= BUFFER_WATERMARK_BUFFERED)) { buf->callback(buf->userdata, CAMU_BUFFER_BUFFERED); buf->buffered = true; } - if (count >= 40) buf->queue->reset(buf->queue); - else if (count >= 30) buf->callback(buf->userdata, CAMU_BUFFER_STOP); + if (count >= BUFFER_WATERMARK_RESET) buf->queue->reset(buf->queue); + else if (count >= BUFFER_WATERMARK_HIGH) buf->callback(buf->userdata, CAMU_BUFFER_CORK); if (buf->single_frame) camu_video_buffer_flush(buf); } #ifdef HAVE_FFMPEG static void push_av_frame_internal(struct camu_video_buffer *buf, AVFrame *frame) { - AVStream *stream = buf->stream->av.stream; - f64 pts = buf->single_frame ? 0.0 : - (frame->best_effort_timestamp - stream->start_time) * av_q2d(stream->time_base); - if (pts + buf->avg_frame_duration < camu_clock_get_base_pts(buf->clock)) { - av_frame_free(&frame); - return; + f64 pts = 0.0; + if (!buf->single_frame) { + AVStream *stream = buf->stream->av.stream; + pts = (frame->best_effort_timestamp - stream->start_time) * av_q2d(stream->time_base); + if (pts + buf->avg_frame_duration < camu_clock_get_base_pts(buf->clock)) { + av_frame_free(&frame); + return; + } } - AVFrame *scaled_frame; #ifdef CAMU_VIDEO_BUFFER_FORCE_SCALER if (buf->fmt.scaler_needed) { if (!camu_lav_scaler_scale(&buf->scale, (const u8 **)frame->data, frame->linesize)) { return; } - scaled_frame = buf->scale.frame; - } else { - scaled_frame = frame; + av_frame_free(&frame); + frame = av_frame_clone(buf->scale.frame); } -#else - scaled_frame = frame; #endif - if (scaled_frame) scaled_frame->opaque = buf; - buf->queue->push_av_frame(buf->queue, scaled_frame, pts); + if (frame) frame->opaque = buf; + buf->queue->push_av_frame(buf->queue, frame, pts); } #endif @@ -130,19 +136,18 @@ void camu_video_buffer_flush(struct camu_video_buffer *buf) bool camu_video_buffer_read(struct camu_video_buffer *buf, void *out) { - f64 pts = (buf->single_frame) ? 0.0 : camu_clock_get_pts(buf->clock); - if (pts == -1.0) pts = buf->last_pts; - buf->last_pts = pts; + f64 pts = buf->single_frame ? 0.0 : camu_clock_get_pts(buf->clock); u8 ret = buf->queue->read(buf->queue, pts, out); if (ret == CAMU_QUEUE_EOF || (ret == CAMU_QUEUE_OK && buf->single_frame)) { buf->callback(buf->userdata, CAMU_BUFFER_EOF); - } else if (buf->queue->count(buf->queue) <= 20) { - buf->callback(buf->userdata, CAMU_BUFFER_CONTINUE); + } else if (buf->queue->count(buf->queue) <= BUFFER_WATERMARK_LOW) { + buf->callback(buf->userdata, CAMU_BUFFER_UNCORK); } return ret == CAMU_QUEUE_OK || ret == CAMU_QUEUE_MORE; } void camu_video_buffer_free(struct camu_video_buffer *buf) { + if (buf->fmt.scaler_needed) camu_lav_scaler_close(&buf->scale); buf->queue->free(&buf->queue); } diff --git a/src/buffer/video.h b/src/buffer/video.h index 534f797..a6349f8 100644 --- a/src/buffer/video.h +++ b/src/buffer/video.h @@ -14,9 +14,10 @@ struct camu_video_buffer { struct camu_stream *stream; struct camu_clock *clock; + f32 start_time; + f32 avg_frame_duration; + bool single_frame; - f64 last_pts; - f64 avg_frame_duration; struct camu_lav_scaler scale; struct camu_lav_scale_fmt fmt; diff --git a/src/cache/backings/file_common.c b/src/cache/backings/file_common.c index c0b54c0..232580a 100644 --- a/src/cache/backings/file_common.c +++ b/src/cache/backings/file_common.c @@ -9,7 +9,8 @@ static off_t file_backing_get_size_estimate(struct cch_backing *backing) static bool file_open_internal(struct cch_backing_file *file, str *path, size_t size) { - if (!aki_file_open(&file->file, path, size != 0)) return false; + s32 flags = size != 0 ? AKI_FILE_CREATE : 0; + if (!aki_file_open(&file->file, path, flags)) return false; file->filesize = aki_file_get_filesize(&file->file); if (!size) { file->size = file->filesize; diff --git a/src/cache/backings/file_mapped.c b/src/cache/backings/file_mapped.c index 1b06ad2..4277c21 100644 --- a/src/cache/backings/file_mapped.c +++ b/src/cache/backings/file_mapped.c @@ -43,6 +43,7 @@ static void file_backing_resize(struct cch_backing *backing, size_t size) aki_mutex_lock(&file->mutex); file->size = size; if ((off_t)size > file->filesize) { + // TODO: Unmap, truncate, map? file->u.map = aki_file_mremap(&file->file, size, file->u.map); aki_file_truncate(&file->file, size); file->filesize = size; diff --git a/src/cache/handlers/http.c b/src/cache/handlers/http.c index 1daf231..dd3c8ac 100644 --- a/src/cache/handlers/http.c +++ b/src/cache/handlers/http.c @@ -80,12 +80,11 @@ static size_t http_callback(void *userdata, u8 op, u8 *buf, s64 int0) static void signal_callback(void *userdata) { struct cch_handler_http *http = (struct cch_handler_http *)userdata; + u32 size; + s32 index; do { - u32 size; - camu_queue_size(http->queue, size); + camu_queue_try_pop(http->queue, size, index); if (size == 0) break; - s32 index; - camu_queue_pop(http->queue, index); if (index == -1) { aki_event_loop_break(&http->loop); break; diff --git a/src/codec/codec.h b/src/codec/codec.h index d30a648..1309210 100644 --- a/src/codec/codec.h +++ b/src/codec/codec.h @@ -61,8 +61,7 @@ struct camu_stream { s32 height; struct camu_icc_profile icc; } video; - //struct { - //} audio; + //struct { } audio; #ifdef HAVE_FFMPEG struct { AVStream *stream; @@ -94,13 +93,14 @@ struct camu_frame { }; struct camu_demuxer { + u8 type; bool (*init)(struct camu_demuxer *, struct cch_handle *); s32 (*get_packet)(struct camu_demuxer *, struct camu_packet *); s64 (*get_duration)(struct camu_demuxer *); bool (*seek)(struct camu_demuxer *, s64); void (*free)(struct camu_demuxer **); - u8 type; array(struct camu_stream) streams; + array(s32) subscribed; }; struct camu_renderer; diff --git a/src/codec/libav/demuxer.c b/src/codec/libav/demuxer.c index f7ff35c..d823d23 100644 --- a/src/codec/libav/demuxer.c +++ b/src/codec/libav/demuxer.c @@ -65,47 +65,30 @@ static bool lav_demuxer_init(struct camu_demuxer *demux, struct cch_handle *hand av->duration = 0; - bool audio_selected = false; - bool video_selected = false; +#define GUESS_STREAM_IS_IMAGE(stream) \ + (stream->duration >= 0 && stream->nb_frames <= 1 && (stream->avg_frame_rate.den == 0 && stream->r_frame_rate.den > 0)) for (u32 i = 0; i < av->format_context->nb_streams; i++) { AVStream *stream = av->format_context->streams[i]; enum AVMediaType type = stream->codecpar->codec_type; - if (!(type == AVMEDIA_TYPE_AUDIO || type == AVMEDIA_TYPE_VIDEO)) { - continue; - } - if (type == AVMEDIA_TYPE_AUDIO) { - if (audio_selected) continue; - else audio_selected = true; - } - if (type == AVMEDIA_TYPE_VIDEO) { - if (video_selected) continue; - else video_selected = true; - } // Try to detect attached images. - if (type == AVMEDIA_TYPE_VIDEO && stream->duration >= 0 && stream->nb_frames <= 1 && - (stream->avg_frame_rate.den == 0 && stream->r_frame_rate.den > 0)) { - stream->duration = 0; + if (type == AVMEDIA_TYPE_VIDEO && GUESS_STREAM_IS_IMAGE(stream)) { al_log_info("lav_demux", "Assuming stream #%i is an image.", i); + stream->duration = 0; } else if (stream->duration < 0) { + al_log_warn("lav_demux", "Stream #%i has invalid duration (%ld).", av->format_context->duration); if (av->format_context->duration < 0) { stream->duration = 0; } else { - // This could be completely wrong. In that case, it's still probably - // better than 0 (assuming the stream is not an image). - stream->duration = av_rescale_q(av->format_context->duration, - AV_TIME_BASE_Q, stream->time_base); + stream->duration = av_rescale_q(av->format_context->duration, AV_TIME_BASE_Q, stream->time_base); } - al_log_info("lav_demux", "Setting stream #%i to a duration of %.3fs.", - i, stream->duration * av_q2d(stream->time_base)); + al_log_info("lav_demux", "Setting stream #%i to a duration of %.3fs.", i, stream->duration * av_q2d(stream->time_base)); } if (stream->start_time < 0) { - stream->start_time = 0; + al_log_warn("lav_demux", "Stream has a negative start_time (%ld).", stream->start_time); } s64 duration = av_rescale_q(stream->duration, stream->time_base, AV_TIME_BASE_Q); - if (duration > av->duration) { - av->duration = duration; - } + if (duration > av->duration) av->duration = duration; al_array_push(av->demux.streams, ((struct camu_stream){ .type = CAMU_FFMPEG_COMPAT, .av.stream = stream @@ -125,6 +108,15 @@ err: return false; } +static bool subscribed_to_index(struct camu_demuxer *demux, s32 index) +{ + s32 subscribed; + al_array_foreach(demux->subscribed, i, subscribed) { + if (subscribed == index) return true; + } + return false; +} + static s32 lav_demuxer_get_packet(struct camu_demuxer *demux, struct camu_packet *packet) { struct camu_lav_demuxer *av = (struct camu_lav_demuxer *)demux; @@ -132,12 +124,19 @@ static s32 lav_demuxer_get_packet(struct camu_demuxer *demux, struct camu_packet s32 ret = 0; do { ret = av_read_frame(av->format_context, packet->av.pkt); - if (ret < 0 && ret != AVERROR_EOF) { + if (ret == AVERROR_EOF) { + av->eof = true; + break; + } + if (ret < 0) { al_log_error("lav_demuxer", "Failed to read frame (%s).", av_err2str(ret)); } + if (!subscribed_to_index(demux, packet->av.pkt->stream_index)) { + av_packet_unref(packet->av.pkt); + continue; + } break; } while (1); - if (ret == AVERROR_EOF) av->eof = true; return ret; } @@ -175,6 +174,7 @@ struct camu_demuxer *camu_lav_demuxer_create(void) struct camu_lav_demuxer *av = al_alloc_object(struct camu_lav_demuxer); av->demux.type = CAMU_FFMPEG_COMPAT; al_array_init(av->demux.streams); + al_array_init(av->demux.subscribed); av->demux.init = lav_demuxer_init; av->demux.get_packet = lav_demuxer_get_packet; av->demux.get_duration = lav_demuxer_get_duration; diff --git a/src/codec/libav/packet_ext.h b/src/codec/libav/packet_ext.h index e2f7cd5..ff6e337 100644 --- a/src/codec/libav/packet_ext.h +++ b/src/codec/libav/packet_ext.h @@ -1,6 +1,7 @@ #pragma once #include <aki/common.h> +#include <aki/event_loop.h> #include <libavcodec/packet.h> #include <libavcodec/codec_par.h> diff --git a/src/codec/libav/scaler.c b/src/codec/libav/scaler.c index 6146c69..882116f 100644 --- a/src/codec/libav/scaler.c +++ b/src/codec/libav/scaler.c @@ -9,6 +9,12 @@ static AVFrame *alloc_picture(enum AVPixelFormat pix_fmt, s32 width, s32 height) { AVFrame *frame = av_frame_alloc(); if (!frame) return NULL; + s32 ret = av_image_alloc(frame->data, frame->linesize, width, height, pix_fmt, 32); + if (ret < 0) { + al_log_error("lav_scaler", "Failed to allocate picture buffer (%s).", av_err2str(ret)); + return NULL; + } + /* s32 size = av_image_get_buffer_size(pix_fmt, width, height, 1); u8 *buffer = (u8 *)av_malloc(size); if (!buffer) { @@ -21,6 +27,7 @@ static AVFrame *alloc_picture(enum AVPixelFormat pix_fmt, s32 width, s32 height) av_free(frame); return NULL; } + */ frame->width = width; frame->height = height; frame->format = pix_fmt; @@ -31,32 +38,28 @@ bool camu_lav_scaler_init(struct camu_lav_scaler *scale, struct camu_lav_scale_f { al_memset(scale, 0, sizeof(struct camu_lav_scaler)); - /* if (fmt->in_width == fmt->req_width && fmt->in_height == fmt->req_height && fmt->in_format == fmt->req_format) { fmt->scaler_needed = false; return true; } - */ - - fmt->scaler_needed = true; + s32 flags = SWS_BITEXACT; scale->scaler_context = sws_getContext(fmt->in_width, fmt->in_height, fmt->in_format, - fmt->req_width, fmt->req_height, fmt->req_format, - SWS_BITEXACT, NULL, NULL, NULL); - + fmt->req_width, fmt->req_height, fmt->req_format, flags, NULL, NULL, NULL); if (!scale->scaler_context) { al_log_error("lav_scaler", "Failed to create scaler context."); return false; } scale->frame = alloc_picture(fmt->req_format, fmt->req_width, fmt->req_height); - if (!scale->frame) { al_log_error("lav_scaler", "Failed to allocate frame."); return false; } + fmt->scaler_needed = true; + scale->fmt = *fmt; return true; @@ -64,8 +67,8 @@ bool camu_lav_scaler_init(struct camu_lav_scaler *scale, struct camu_lav_scale_f bool camu_lav_scaler_scale(struct camu_lav_scaler *scale, const u8 **in_slice, s32 *in_strides) { - s32 ret = sws_scale(scale->scaler_context, in_slice, in_strides, 0, - scale->fmt.in_height, scale->frame->data, scale->frame->linesize); + s32 ret = sws_scale(scale->scaler_context, in_slice, in_strides, 0, scale->fmt.in_height, + scale->frame->data, scale->frame->linesize); if (ret != scale->fmt.req_height) { al_log_error("lav_scaler", "Failed to scale frame (%s).", av_err2str(ret)); return false; @@ -75,5 +78,7 @@ bool camu_lav_scaler_scale(struct camu_lav_scaler *scale, const u8 **in_slice, s void camu_lav_scaler_close(struct camu_lav_scaler *scale) { + av_freep(&scale->frame->data[0]); + av_frame_free(&scale->frame); sws_freeContext(scale->scaler_context); } diff --git a/src/codec/meson.build b/src/codec/meson.build index e1fd5f7..b4472fb 100644 --- a/src/codec/meson.build +++ b/src/codec/meson.build @@ -54,14 +54,11 @@ if libavutil.found() and libavformat.found() and libavcodec.found() and libswres add_project_arguments('-DHAVE_FFMPEG', language: ['c', 'cpp']) endif -stb_image = subproject('stb_image').get_variable('stb_image') -#stb_image_decoder = declare_dependency(sources: ['stb_image/impl.c'], -# dependencies: stb_image) -# -wuffs = subproject('wuffs').get_variable('wuffs') -wuffs_decoder = declare_dependency(sources: ['wuffs/impl.c'], - dependencies: wuffs) +stb_image = declare_dependency(sources: ['stb_image/impl.c'], + dependencies: subproject('stb').get_variable('stb')) -#spng = dependency('spng') -#spng_decoder = declare_dependency(sources: ['spng/impl.c'], -# dependencies: spng) +#wuffs = declare_dependency(sources: ['wuffs/impl.c'], +# dependencies: subproject('wuffs').get_variable('wuffs')) + +#spng = declare_dependency(sources: ['spng/impl.c'], +# dependencies: dependency('spng')) diff --git a/src/codec/stb_image/impl.c b/src/codec/stb_image/impl.c index 953cca8..b7ed928 100644 --- a/src/codec/stb_image/impl.c +++ b/src/codec/stb_image/impl.c @@ -2,9 +2,9 @@ #include <aki/common.h> #include "../common.h" #define STB_IMAGE_IMPLEMENTATION +#define STB_IMAGE_STATIC #define STBI_NO_STDIO #define STBI_NO_FAILURE_STRINGS -#define STB_IMAGE_STATIC #define STBI_MALLOC camu_page_alloc #define STBI_FREE al_free #define STBI_REALLOC al_realloc diff --git a/src/fruits/cap/cap.c b/src/fruits/cap/cap.c index 317abf4..3f5f5e8 100644 --- a/src/fruits/cap/cap.c +++ b/src/fruits/cap/cap.c @@ -1,20 +1,20 @@ #include <al/random.h> -#include "cap.h" - -#if CAP_USE_PYTHON -#include "../../shoki/src/search.h" -#endif #include "../../cache/handlers/file.h" #ifdef AKIYO_HAS_CURL #include "../../cache/handlers/http.h" #endif +#include "cap.h" + void cap_init(struct cap_runner *cap, bool (*callback)(void *, u8, str *, str *, void *), void *userdata) { cap->callback = callback; cap->userdata = userdata; al_array_init(cap->lists); +#if CAP_USE_PYTHON + sho_client_init(&cap->search, NULL); +#endif } static struct cap_list *list_from_name(struct cap_runner *cap, str *name) @@ -28,7 +28,7 @@ static struct cap_list *list_from_name(struct cap_runner *cap, str *name) return NULL; } -static struct cch_entry *entry_for_unique_id(str *unique_id) +static struct cch_entry *entry_for_unique_id(struct cap_runner *cap, str *unique_id) { struct cch_entry *entry = NULL; #ifdef AKIYO_HAS_CURL @@ -36,16 +36,21 @@ static struct cch_entry *entry_for_unique_id(str *unique_id) if (is_search || al_str_cmp(unique_id, al_str_c("https://"), 0, 8) == 0 || al_str_cmp(unique_id, al_str_c("http://"), 0, 7) == 0) { #if CAP_USE_PYTHON - struct sho_search s; - sho_search_init(&s); + str provider; str query; + al_str_from(&provider, ""); al_str_from(&query, ""); if (al_str_cmp(unique_id, al_str_c("https://twitter.com"), 0, 19) == 0 || al_str_cmp(unique_id, al_str_c("https://x.com"), 0, 13) == 0) { al_str_cat(&query, al_str_c("tweet:")); al_str_cat(&query, unique_id); - if (!sho_search_more_results(&s, al_str_c("twitter"), &query)) { + al_str_cat(&provider, al_str_c("twitter")); + } else if (al_str_cmp(unique_id, al_str_c("https://instagram.com"), 0, 21) == 0) { + s32 slash = al_str_rfind(unique_id, '/'); + if (slash >= 0) { + al_str_cat(&query, al_str_substr(unique_id, slash + 1, unique_id->len)); } + al_str_cat(&provider, al_str_c("instagram")); } else { if (is_search) { al_str_cat(&query, al_str_substr(unique_id, 1, unique_id->len)); @@ -53,26 +58,27 @@ static struct cch_entry *entry_for_unique_id(str *unique_id) al_str_cat(&query, al_str_c("link:")); al_str_cat(&query, unique_id); } - if (!sho_search_more_results(&s, al_str_c("youtube"), &query)) { - } + al_str_cat(&provider, al_str_c("youtube")); } + s32 id = sho_client_create_search(&cap->search, &provider, &query); + al_str_free(&provider); al_str_free(&query); - entry = NULL; - struct sho_result_page *page; - al_array_foreach_ptr(s.result.pages, k, page) { - struct sho_post *post; - al_array_foreach_ptr(page->posts, i, post) { - struct sho_post_media *media; - al_array_foreach_ptr(post->media, j, media) { - if (media->url.len > 0) { - entry = cch_handler_http_create(&media->url); - break; - } + if (id < 0) return NULL; + struct sho_search *search = sho_client_get_search(&cap->search, id); + if (!search || !sho_search_from_page(search, 0)) return NULL; + struct sho_result_page *page = &al_array_at(search->pages, 0); + struct sho_post *post; + al_array_foreach_ptr(page->posts, i, post) { + struct sho_post_media *media; + al_array_foreach_ptr(post->media, j, media) { + if (media->url.len > 0) { + entry = cch_handler_http_create(&media->url); + break; } - if (entry) break; } + if (entry) break; } - sho_search_free(&s); + sho_client_discard_search(&cap->search, id); #else if (!is_search) entry = cch_handler_http_create(unique_id); #endif @@ -88,9 +94,8 @@ static bool buffer_entry(struct cap_runner *cap, struct cap_list *list, struct c { if (entry->errored) return false; if (entry->buffer_requested) return true; - entry->entry = entry_for_unique_id(&entry->unique_id); - if (!entry->entry) return false; - if (!cap->callback(cap->userdata, CAP_BUFFER, &list->name, &entry->unique_id, entry->entry)) { + entry->entry = entry_for_unique_id(cap, &entry->unique_id); + if (!entry->entry || !cap->callback(cap->userdata, CAP_BUFFER, &list->name, &entry->unique_id, entry->entry)) { entry->errored = true; return false; } diff --git a/src/fruits/cap/cap.h b/src/fruits/cap/cap.h index 57f6889..69d4d6d 100644 --- a/src/fruits/cap/cap.h +++ b/src/fruits/cap/cap.h @@ -1,11 +1,15 @@ #pragma once -#define CAP_USE_PYTHON 1 +#define CAP_USE_PYTHON 0 #include <al/array.h> #include <al/str.h> #include <aki/thread.h> +#if CAP_USE_PYTHON +#include "../../shoki/src/search.h" +#endif + struct cap_list_entry { str unique_id; struct cch_entry *entry; @@ -34,6 +38,9 @@ enum { struct cap_runner { array(struct cap_list) lists; +#if CAP_USE_PYTHON + struct sho_client search; +#endif bool (*callback)(void *, u8, str *, str *, void *); void *userdata; }; diff --git a/src/fruits/cap/meson.build b/src/fruits/cap/meson.build index b5930ac..f85ab00 100644 --- a/src/fruits/cap/meson.build +++ b/src/fruits/cap/meson.build @@ -1,4 +1,8 @@ cap_src = ['cap.c'] -cap_deps = [common_deps, shoki] +cap_deps = [common_deps] +use_python = true +if use_python + cap_deps += [shoki] +endif cap = declare_dependency(sources: cap_src, dependencies: cap_deps) diff --git a/src/fruits/cmc/cmc.c b/src/fruits/cmc/cmc.c new file mode 100644 index 0000000..7fd8691 --- /dev/null +++ b/src/fruits/cmc/cmc.c @@ -0,0 +1,781 @@ +#include <al/lib.h> +#include <al/wstr.h> +#include <al/random.h> +#include <aki/common.h> +#include <aki/event_loop.h> +#include <notcurses/notcurses.h> +#include <notcurses/nckeys.h> + +#define STB_IMAGE_IMPLEMENTATION +#include <stb_image.h> + +#include "../../libclient/client.h" +#include "../../libclient/resource_client.h" +#include "../../tree/common.h" + +enum { + CMC_LIST_SEARCHES = 0, + CMC_CREATE_SEARCH, + CMC_OPEN_SEARCH, + CMC_LIST_LISTS, + CMC_CREATE_LIST +}; + +struct cmc_view { + struct sho_post *post; + struct ncplane *container; + bool has_visual_data; +}; + +struct cmc_view_resource { + u16 id; + str unique_id; + u32 index; + struct ncvisual *v; + struct ncvgeom geom; + struct ncvisual_options vopts; +}; + +enum { + CMC_VIEW_NONE = 0, + CMC_VIEW_SINGLE, + CMC_VIEW_PARTITION +}; + +struct cmc_twitter_view { + struct cmc_view v; + struct sho_post *repost; + struct { + u32 width; + u32 height; + u32 y; + struct ncplane *p; + } text; + struct { + struct ncplane *p; + struct cmc_view_resource *resource; + } profile; + struct { + u8 mode; + u32 width; + u32 height; + u32 x; + struct ncplane *p[4]; + struct cmc_view_resource *resources[4]; + } preview; + struct { + struct ncplane *p; + } stats; +}; + +struct cmc { + u8 mode; + struct aki_event_loop loop; + struct camu_client client; + struct camu_resource_client resource_client; + u32 pending_requests; + struct notcurses *nc; + struct aki_poll input; + u32 rows; + u32 cols; + struct { + str provider; + str query; + s32 id; + struct camu_search *search; + } search; + struct { + struct ncplane *p; + array(struct cmc_view *) views; + u32 offset; + u32 roffset; + u32 prev; + u32 page; + u32 selected; + } ui; + array(struct cmc_view_resource *) resource_cache; +}; + +/* +u64 color = 0; +u32 background = 0; +ncchannel_set_rgb8(&background, 255, 255, 255); +ncchannel_set_alpha(&background, 0); +//ncchannels_set_bchannel(&color, background); +ncchannels_set_fchannel(&color, background); +*/ + +//struct nccell cell = NCCELL_INITIALIZER(L'█', ncplane_styles(view->v.container), ncplane_channels(view->v.container)); +//ncplane_vline(view->v.container, &cell, rows - 1); +/* +//ncplane_gradient(view->v.container, 0, 0, rows, width, "", 0, color, color, color, color); +struct nccell ul = NCCELL_TRIVIAL_INITIALIZER, ur = NCCELL_TRIVIAL_INITIALIZER; +struct nccell ll = NCCELL_TRIVIAL_INITIALIZER, lr = NCCELL_TRIVIAL_INITIALIZER; +struct nccell hl = NCCELL_TRIVIAL_INITIALIZER, vl = NCCELL_TRIVIAL_INITIALIZER; +nccells_ascii_box(view->v.container, 0, color, &ul, &ur, &ll, &lr, &hl, &vl); +ncplane_box(view->v.container, &ul, &ur, &ll, &lr, &hl, &vl, rows, width, 0); +nccell_release(view->v.container, &ul); nccell_release(view->v.container, &ur); +nccell_release(view->v.container, &ll); nccell_release(view->v.container, &lr); +nccell_release(view->v.container, &hl); nccell_release(view->v.container, &vl); +//ncplane_double_box(view->v.container, NCSTYLE_NONE, color, rows, width, 0); +*/ + +static u32 count_wrapped_lines(wstr *w, u32 width) +{ + u32 counter = 0, wraps = 1; + for (u32 i = 0; i < w->len; i++) { + wchar_t wc = al_wstr_at(w, i); + s32 wcwidth = al_wchar_width(wc); + bool newline = wc == L'\n'; + if (newline || counter + wcwidth > width) { + wraps++; + counter = 0; + } + if (newline) continue; + counter += wcwidth; + } + return wraps; +} + +static void date_to_relative_date(struct sho_post *post, char *out) +{ + struct sho_post_date *created = NULL; + struct sho_post_date *date; + al_array_foreach_ptr(post->dates, i, date) { + if (date->type == SHOKI_DATE_CREATED) { + created = date; + break; + } + } + if (!created) return; + f64 seconds = (aki_get_timestamp() / 1000000.0 - created->timestamp); + s64 minutes = seconds / 60L; + s64 hours = minutes / 60L; + s64 days = hours / 24L; + if (days > 0) { + al_sprintf(out, "%lud", days); + } else if (hours > 0) { + al_sprintf(out, "%luh", hours); + } else if (minutes > 0) { + al_sprintf(out, "%lum", minutes); + } else { + al_sprintf(out, "%lus", (s64)seconds); + } +} + +static void number_to_shorthand(s64 number, char *out) +{ + if (number >= 1000000) { + al_sprintf(out, "%.1fM", number / 1000000.f); + } else if (number >= 1000) { + al_sprintf(out, "%.1fK", number / 1000.f); + } else { + al_sprintf(out, "%ld", number); + } +} + +#define TEXT_OFFSET 10 + +#define PROFILE_SIZE 5 +#define PREVIEW_OFFSET_WIDTH 26 + +/* +static u32 calc_height_for_display(struct ncvgeom *geom, u32 width) +{ + f32 ratio = geom->pixx / (f32)geom->pixy; + if (ratio >= 1.f) { + return ceil((width / ratio) / 2.f) - 2; + } else { + return ceil((width / ratio) / 2.f) - 2; + } +} +*/ + +static void resource_center_zoom(struct cmc_view_resource *resource, s32 width, s32 height) +{ + struct ncvgeom *geom = &resource->geom; + struct ncvisual_options *vopts = &resource->vopts; + + f32 ratio = geom->pixx / (f32)geom->pixy; + f32 preview_ratio = width / (f32)height; + f32 diff; + + f32 x0 = 0.f; + f32 y0 = 0.f; + f32 x1 = geom->pixx; + f32 y1 = geom->pixy; + + // Not properly tested. + if (preview_ratio > ratio) { + diff = (geom->pixy - (geom->pixy / (preview_ratio * ratio))) / 2.f; + y0 += diff; + y1 -= diff; + } else { + diff = (geom->pixx - (geom->pixx / (preview_ratio / ratio))) / 2.f; + x0 += diff; + x1 -= diff; + } + + vopts->begx = x0; + vopts->begy = y0; + vopts->lenx = x1 - x0; + vopts->leny = y1 - y0; +} + +static void get_post_and_maybe_repost(struct cmc *c, str *unique_id, struct sho_post **post, struct sho_post **repost) +{ + *post = sho_post_cache_get(&c->client.cache, unique_id); + *repost = NULL; + if ((*post)->type == SHOKI_POST_REPOST) { + *repost = *post; + *post = sho_post_cache_get(&c->client.cache, &(*post)->post.unique_id); + } +} + +static u32 layout_tweet(struct cmc *c, struct cmc_twitter_view *view, u32 width) +{ + struct sho_post *post = view->v.post; + struct sho_post *repost = view->repost; + + u32 rows = 1; + + view->text.width = width - TEXT_OFFSET; + view->text.height = 1 + count_wrapped_lines(&post->text, view->text.width); + if (repost) { + view->text.y = 1; + } else { + view->text.y = 0; + } + view->text.height += view->text.y; + rows += view->text.height; + + view->v.has_visual_data = true; + + for (u32 i = 0; i < 4; i++) { + if (view->preview.resources[i] && !view->preview.resources[i]->v) { + view->v.has_visual_data = false; + break; + } + } + + if (view->preview.mode == CMC_VIEW_NONE || !view->v.has_visual_data) return rows; + + view->preview.width = view->text.width; + u32 partition_width = view->preview.width / 2; + view->preview.x = TEXT_OFFSET + 1; + view->preview.height = 0; + + struct ncplane_options opts = { 0 }; + opts.cols = view->preview.width; + opts.rows = ncplane_dim_y(notcurses_stdplane(c->nc)) / 3; + opts.x = 0; + opts.y = 0; + struct ncplane *inter = ncplane_create(notcurses_stdplane(c->nc), &opts); + + u32 left = 0, right = 0; + struct cmc_view_resource *resource; + switch (view->preview.mode) { + case CMC_VIEW_SINGLE: { + resource = view->preview.resources[0]; + resource->vopts.n = inter; + ncvisual_geom(c->nc, resource->v, &resource->vopts, &resource->geom); + left += resource->geom.rcelly + 1; + break; + } + case CMC_VIEW_PARTITION: { + ncplane_resize_simple(inter, opts.rows, partition_width); + + resource = view->preview.resources[0]; + resource->vopts.n = inter; + ncvisual_geom(c->nc, resource->v, &resource->vopts, &resource->geom); + left += resource->geom.rcelly + 1; + + resource = view->preview.resources[1]; + resource->vopts.n = inter; + ncvisual_geom(c->nc, resource->v, &resource->vopts, &resource->geom); + right += resource->geom.rcelly + 1; + + if (view->preview.resources[2]) { + resource = view->preview.resources[2]; + resource->vopts.n = inter; + ncvisual_geom(c->nc, resource->v, &resource->vopts, &resource->geom); + left += resource->geom.rcelly + 1; + } + + if (view->preview.resources[3]) { + resource = view->preview.resources[3]; + resource->vopts.n = inter; + ncvisual_geom(c->nc, resource->v, &resource->vopts, &resource->geom); + right += resource->geom.rcelly + 1; + } + + break; + } + } + view->preview.height = AL_MAX(right, left); + + ncplane_destroy(inter); + + return rows + view->preview.height; +} + +static char format_buf[32]; + +static void draw_tweet(struct cmc *c, struct cmc_twitter_view *view, struct ncplane *container, u32 rows, u32 width) +{ + struct sho_post *post = view->v.post; + struct sho_post *repost = view->repost; + + ncplane_resize_simple(container, rows, width); + + if (view->v.has_visual_data) { + u32 partition_width = view->preview.width / 2; + struct cmc_view_resource *resource; + switch (view->preview.mode) { + case CMC_VIEW_NONE: + break; + case CMC_VIEW_SINGLE: { + resource = view->preview.resources[0]; + ncplane_resize_simple(view->preview.p[0], resource->geom.rcelly + 1, view->preview.width); + ncplane_move_yx(view->preview.p[0], view->text.height, view->preview.x); + resource->vopts.n = view->preview.p[0]; + ncvisual_blit(c->nc, resource->v, &resource->vopts); + break; + } + case CMC_VIEW_PARTITION: { + resource = view->preview.resources[0]; + resource->vopts.n = view->preview.p[0]; + ncplane_resize_simple(view->preview.p[0], resource->geom.rcelly + 1, partition_width); + ncplane_move_yx(view->preview.p[0], view->text.height, view->preview.x); + ncvisual_blit(c->nc, resource->v, &resource->vopts); + + resource = view->preview.resources[1]; + ncplane_resize_simple(view->preview.p[1], resource->geom.rcelly + 1, partition_width); + ncplane_move_yx(view->preview.p[1], view->text.height, (view->preview.x + view->preview.resources[0]->geom.rcellx) + 2); + resource->vopts.n = view->preview.p[1]; + ncvisual_blit(c->nc, resource->v, &resource->vopts); + + if (view->preview.resources[2]) { + resource = view->preview.resources[2]; + ncplane_resize_simple(view->preview.p[2], resource->geom.rcelly + 1, partition_width); + ncplane_move_yx(view->preview.p[2], view->preview.resources[0]->geom.rcelly + view->text.height + 2, view->preview.x); + resource->vopts.n = view->preview.p[2]; + ncvisual_blit(c->nc, resource->v, &resource->vopts); + } + + if (view->preview.resources[3]) { + resource = view->preview.resources[3]; + ncplane_resize_simple(view->preview.p[3], resource->geom.rcelly + 1, partition_width); + ncplane_move_yx(view->preview.p[3], view->preview.resources[2]->geom.rcelly + view->text.height + 2, (view->preview.x + view->preview.resources[2]->geom.rcellx) + 2); + resource->vopts.n = view->preview.p[3]; + ncvisual_blit(c->nc, resource->v, &resource->vopts); + } + + break; + } + } + } + + ncplane_resize_simple(view->text.p, view->text.height, width); + ncplane_move_yx(view->text.p, 0, 0); + + s32 offset = TEXT_OFFSET; + + if (repost) { + const wchar_t *retweet = L"⮤⮧"; + ncplane_set_styles(view->text.p, NCSTYLE_ITALIC); + ncplane_putwstr_yx(view->text.p, 0, offset - 3, retweet); + ncplane_set_styles(view->text.p, NCSTYLE_NONE); + ncplane_putwstr_yx(view->text.p, 0, offset, &al_wstr_at(&repost->author.display_name, 0)); + } + + ncplane_set_styles(view->text.p, NCSTYLE_BOLD); + + ncplane_putwstr_yx(view->text.p, view->text.y, offset, &al_wstr_at(&post->author.display_name, 0)); + offset += al_wstr_width(&post->author.display_name) + 1; + + ncplane_set_styles(view->text.p, NCSTYLE_NONE); + ncplane_putwstr_yx(view->text.p, view->text.y, offset, &al_wstr_at(&post->author.username, 0)); + offset += al_wstr_width(&post->author.username) + 1; + + ncplane_putwstr_yx(view->text.p, view->text.y, offset, L"∙"); + offset += 2; + + date_to_relative_date(post, format_buf); + ncplane_putstr_yx(view->text.p, view->text.y, offset, format_buf); + + u32 text_y = view->text.y + 1, text_x = TEXT_OFFSET; + for (u32 i = 0; i < post->text.len; i++) { + wchar_t wc = al_wstr_at(&post->text, i); + s32 wcwidth = al_wchar_width(wc); + bool newline = wc == L'\n'; + if (newline || text_x + wcwidth > width) { + text_y++; + text_x = TEXT_OFFSET; + } + if (newline) continue; + ncplane_putwc_yx(view->text.p, text_y, text_x, wc); + text_x += wcwidth; + } + + ncplane_resize_simple(view->stats.p, 1, view->text.width); + ncplane_move_yx(view->stats.p, rows - 1, TEXT_OFFSET); + + u32 part = 10; + offset = 1; + + if (post->comments.set) { + const wchar_t *comment = L"🗩"; + //const wchar_t *comment = L"[]"; + //const wchar_t *comment = L"｡○"; + //const wchar_t *comment = L"❑"; + //const wchar_t *comment = L"｡ロ"; + //const wchar_t *comment = L".❍"; + ncplane_set_styles(view->stats.p, NCSTYLE_ITALIC); + ncplane_putwstr_yx(view->stats.p, 0, offset, comment); + ncplane_set_styles(view->stats.p, NCSTYLE_NONE); + number_to_shorthand(post->comments.i, format_buf); + ncplane_putstr_yx(view->stats.p, 0, offset + 3, format_buf); + } + offset += part; + + if (post->reposts.set) { + const wchar_t *retweet = L"⮤⮧"; + ncplane_set_styles(view->stats.p, NCSTYLE_ITALIC); + ncplane_putwstr_yx(view->stats.p, 0, offset, retweet); + ncplane_set_styles(view->stats.p, NCSTYLE_NONE); + number_to_shorthand(post->reposts.i, format_buf); + ncplane_putstr_yx(view->stats.p, 0, offset + 3, format_buf); + } + offset += part; + + if (post->likes.set) { + const wchar_t *like = L"❤"; + ncplane_set_styles(view->stats.p, NCSTYLE_ITALIC); + ncplane_putwstr_yx(view->stats.p, 0, offset, like); + ncplane_set_styles(view->stats.p, NCSTYLE_NONE); + number_to_shorthand(post->likes.i, format_buf); + ncplane_putstr_yx(view->stats.p, 0, offset + 3, format_buf); + } + + if (view->profile.resource && view->profile.resource->v) { + struct cmc_view_resource *resource = view->profile.resource; + ncplane_resize_simple(view->profile.p, (PROFILE_SIZE + 1) / 2, PROFILE_SIZE); + ncplane_move_yx(view->profile.p, view->text.y, PROFILE_SIZE - 2); + resource->vopts.n = view->profile.p; + ncvisual_blit(c->nc, resource->v, &resource->vopts); + } +} + +static void layout_tweets(struct cmc *c) +{ + struct cmc_twitter_view *view; + for (u32 i = 0; i < c->ui.views.size; i++) { + view = (struct cmc_twitter_view *)al_array_at(c->ui.views, i); + ncplane_erase(view->v.container); + ncplane_erase(view->text.p); + ncplane_erase(view->profile.p); + for (u32 i = 0; i < 4; i++) { + ncplane_erase(view->preview.p[i]); + } + ncplane_erase(view->stats.p); + ncplane_reparent_family(view->v.container, view->v.container); + } + + u32 y = 0; + u32 width = ncplane_dim_x(c->ui.p); + u32 rows = 0; + u32 i = c->ui.offset; + c->ui.page = 0; + for (; i < c->ui.views.size; i++) { + view = (struct cmc_twitter_view *)al_array_at(c->ui.views, i); + rows = layout_tweet(c, view, width); + if (y + rows > ncplane_dim_y(c->ui.p)) break; + c->ui.page++; + draw_tweet(c, view, view->v.container, rows, width); + ncplane_reparent_family(view->v.container, c->ui.p); + ncplane_move_yx(view->v.container, y, 0); + if (i - c->ui.offset == c->ui.selected) { + wchar_t box = L'█'; + u32 i = view->repost ? 1 : 0; + for (; i < rows - 1; i++) { + ncplane_cursor_move_yx(view->v.container, i, 0); + ncplane_putwc(view->v.container, box); + } + } + y += rows + 1; + } + + if (!c->ui.page) c->ui.page = 1; + + if (c->ui.roffset != c->ui.offset) { + notcurses_refresh(c->nc, NULL, NULL); + } + notcurses_render(c->nc); + + c->ui.roffset = c->ui.offset; + + if (c->pending_requests == 0 && c->ui.offset + c->ui.page >= c->ui.views.size) { + camu_client_more_results(&c->client, c->search.search); + } +} + +static void resource_client_callback(void *userdata, u16 id, struct aki_buffer *buffer) +{ + struct cmc *c = (struct cmc *)userdata; + struct cmc_view_resource *resource; + al_array_foreach(c->resource_cache, i, resource) { + if (resource->id == id) { + s32 w, h, channels; + u8 *pixels = stbi_load_from_memory(aki_buffer_get_ptr(buffer, 0), buffer->size, &w, &h, &channels, 4); + resource->v = ncvisual_from_rgba(pixels, h, w * 4, w); + stbi_image_free(pixels); + ncvisual_geom(c->nc, resource->v, &resource->vopts, &resource->geom); + break; + } + } + if (--c->pending_requests == 0) layout_tweets(c); +} + +static void input_poll_callback(void *userdata, s32 revents) +{ + struct cmc *c = (struct cmc *)userdata; + (void)revents; + struct ncinput input; + u32 ret; + bool layout = false; + do { + ret = notcurses_get_nblock(c->nc, &input); + if (ret == (u32)-1 || ret == 0) break; + if (input.evtype == NCTYPE_PRESS || input.evtype == NCTYPE_UNKNOWN) { + if (input.id == 'j') { + if (c->ui.selected + 1 >= c->ui.page) { + if (c->ui.offset + c->ui.page >= c->ui.views.size) { + return; + } + c->ui.offset += c->ui.page; + c->ui.prev = c->ui.page; + c->ui.page = 0; + c->ui.selected = 0; + } else { + c->ui.selected++; + } + layout = true; + } else if (input.id == 'k') { + if ((s32)c->ui.selected - 1 < 0) { + if (c->ui.offset == 0 || (s32)c->ui.offset - (s32)c->ui.prev < 0) { + return; + } + c->ui.offset -= c->ui.prev; + c->ui.selected = c->ui.prev - 1; + c->ui.prev = 0; + } else { + c->ui.selected--; + } + layout = true; + } else if (input.id == NCKEY_RETURN) { + if (c->ui.views.size > 0) { + u32 selected = c->ui.offset + c->ui.selected; + struct cmc_twitter_view *view = (struct cmc_twitter_view *)al_array_at(c->ui.views, selected); + camu_client_add(&c->client, &view->v.post->unique_id, 0); + } + } else if (input.id == 'q') { + aki_event_loop_break(&c->loop); + break; + } + } + } while (1); + if (layout) layout_tweets(c); +} + +static struct cmc_view_resource *create_resource(struct cmc *c, str *unique_id, u32 index) +{ + struct cmc_view_resource *resource = al_alloc_object(struct cmc_view_resource); + al_array_push(c->resource_cache, resource); + resource->vopts.blitter = NCBLIT_PIXEL; + //resource->vopts.blitter = NCBLIT_2x1; + resource->vopts.scaling = NCSCALE_SCALE_HIRES; + //resource->vopts.scaling = NCSCALE_STRETCH; + resource->id = al_rand_u16(); + al_str_clone(&resource->unique_id, unique_id); + resource->index = index; + c->pending_requests++; + camu_resource_request(&c->resource_client, unique_id, index, resource->id); + return resource; +} + +static struct cmc_view_resource *get_resource_from_cache(struct cmc *c, str *unique_id, u32 index) +{ + struct cmc_view_resource *resource; + al_array_foreach(c->resource_cache, i, resource) { + if (al_str_eq(&resource->unique_id, unique_id) && resource->index == index) { + return resource; + } + } + return create_resource(c, unique_id, index); +} + +static int resize_cb(struct ncplane *p) +{ + struct cmc *c = (struct cmc *)ncplane_userptr(p); + u32 width, height; + notcurses_stddim_yx(c->nc, &height, &width); + struct ncplane_options nopts = { 0 }; + nopts.rows = height - 1; + nopts.cols = width - 4; + nopts.x = 3; + nopts.y = 1; + nopts.flags = 0; + struct ncplane *oldp = c->ui.p; + c->ui.p = ncplane_create(p, &nopts); + if (oldp) { + ncplane_destroy(oldp); + layout_tweets(c); + } + return 0; +} + +static void client_callback(void *userdata, u8 op, void *opaque) +{ + struct cmc *c = (struct cmc *)userdata; + switch (op) { + case CAMU_CLIENT_STATUS_UPDATED: { + switch (c->mode) { + case CMC_LIST_SEARCHES: { + struct camu_search *search; + al_array_foreach(c->client.state.searches, i, search) { + al_printf("%i %.*s\n", search->id, AL_STR_PRINTF(&search->query)); + } + aki_event_loop_break(&c->loop); + break; + } + case CMC_LIST_LISTS: { + struct camu_list *list; + al_array_foreach(c->client.state.lists, i, list) { + al_printf("%.*s\n", AL_STR_PRINTF(&list->name)); + } + aki_event_loop_break(&c->loop); + break; + } + case CMC_CREATE_SEARCH: { + camu_client_create_search(&c->client, &c->search.provider, &c->search.query); + break; + } + case CMC_OPEN_SEARCH: { + if (!(c->nc = notcurses_init(NULL, stdin))) return; + if (c->client.state.searches.size == 0) return; + c->search.search = al_array_at(c->client.state.searches, 0); + struct ncplane *stdplane = notcurses_stdplane(c->nc); + ncplane_set_resizecb(stdplane, resize_cb); + ncplane_set_userptr(stdplane, c); + resize_cb(stdplane); + camu_client_resume_search(&c->client, c->search.search); + aki_poll_init(&c->input, input_poll_callback, c); + aki_poll_set(&c->input, notcurses_inputready_fd(c->nc), AKI_POLL_READ); + aki_poll_start(&c->input, &c->loop); + break; + } + } + break; + } + case CAMU_CLIENT_SEARCH_CREATED: { + aki_event_loop_break(&c->loop); + break; + } + case CAMU_CLIENT_RESULTS: { + struct camu_search_results *results = (struct camu_search_results *)opaque; + str *unique_id; + al_array_foreach_ptr(results->unique_ids, i, unique_id) { + struct cmc_twitter_view *view = al_alloc_object(struct cmc_twitter_view); + al_array_push(c->ui.views, (struct cmc_view *)view); + get_post_and_maybe_repost(c, unique_id, &view->v.post, &view->repost); + + struct ncplane_options default_options = { 0 }; + default_options.cols = 1; + default_options.rows = 1; + default_options.x = 0; + default_options.y = 0; + + view->v.container = ncpile_create(c->nc, &default_options); + view->text.p = ncplane_create(view->v.container, &default_options); + view->profile.p = ncplane_create(view->v.container, &default_options); + for (u32 i = 0; i < 4; i++) { + view->preview.p[i] = ncplane_create(view->v.container, &default_options); + } + view->stats.p = ncplane_create(view->v.container, &default_options); + + u32 prev_pending = c->pending_requests; + + struct sho_post *post = view->v.post; + view->profile.resource = get_resource_from_cache(c, &post->author.unique_id, 0); + struct sho_post_media *media; + al_array_foreach_ptr(post->media, i, media) { + (void)media; + view->preview.resources[i] = get_resource_from_cache(c, &post->unique_id, i); + } + + view->v.has_visual_data = prev_pending == c->pending_requests; + + if (post->media.size == 0) { + view->preview.mode = CMC_VIEW_NONE; + } else if (post->media.size == 1) { + view->preview.mode = CMC_VIEW_SINGLE; + } else { + view->preview.mode = CMC_VIEW_PARTITION; + } + } + if (c->pending_requests == 0) layout_tweets(c); + break; + } + } +} + +s32 main(s32 argc, char *argv[]) +{ + if (argc < 2 || !aki_common_init()) return EXIT_FAILURE; + + struct cmc c = { 0 }; + + al_array_init(c.ui.views); + al_array_init(c.resource_cache); + c.pending_requests = 0; + c.ui.offset = 0; + + str *cmd = al_str_c(argv[1]); + if (al_str_eq(cmd, al_str_c("search"))) { + if (argc == 2) { + c.mode = CMC_LIST_SEARCHES; + } else { + c.mode = CMC_CREATE_SEARCH; + al_str_from(&c.search.provider, "twitter"); + al_str_from(&c.search.query, argv[2]); + } + } else if (al_str_eq(cmd, al_str_c("open"))) { + c.mode = CMC_OPEN_SEARCH; + } else if (al_str_eq(cmd, al_str_c("list"))) { + if (argc == 2) { + c.mode = CMC_LIST_LISTS; + } else { + } + } else { + return EXIT_FAILURE; + } + + aki_event_loop_init(&c.loop); + + camu_client_init(&c.client, &c.loop, client_callback, &c); + camu_client_login(&c.client, al_str_c("andrew"), al_str_c("127.0.0.1"), TREE_PORT); + + camu_resource_client_init(&c.resource_client, &c.loop, resource_client_callback, &c); + camu_resource_client_connect(&c.resource_client, al_str_c("127.0.0.1"), TREE_RESOURCE_PORT); + + aki_event_loop_run(&c.loop); + + if (c.mode == CMC_OPEN_SEARCH) notcurses_stop(c.nc); + + camu_client_close(&c.client); + + aki_common_close(); + + return EXIT_SUCCESS; +} diff --git a/src/fruits/cmc/meson.build b/src/fruits/cmc/meson.build new file mode 100644 index 0000000..45ac9c2 --- /dev/null +++ b/src/fruits/cmc/meson.build @@ -0,0 +1,7 @@ +cmc_src = ['cmc.c'] +cmc_deps = [common_deps, libclient, cache, stb_image] + +notcurses = dependency('notcurses') +cmc_deps += [notcurses] + +executable('cmc', cmc_src, dependencies: cmc_deps) diff --git a/src/fruits/cmv/cmv2.c b/src/fruits/cmv/cmv2.c index 74a4113..89e3427 100644 --- a/src/fruits/cmv/cmv2.c +++ b/src/fruits/cmv/cmv2.c @@ -1,7 +1,8 @@ -#define CMV_USE_TUI 1 -#define CMV_USE_SOCKET 1 +#define CMV_USE_TUI 0 +#define CMV_USE_SOCKET 0 #include <al/log.h> +#include <aki/event_loop.h> #if CMV_USE_SOCKET #include <aki/line_processor.h> #endif @@ -14,7 +15,8 @@ #include "../../render/renderer_libplacebo.h" //#include "../../render/renderer_tiger.h" -#include "../../libsink/sink.h" +#include "../../libsink/sink2.h" +#include "../../tree/common.h" #include "../../shoki/src/search.h" @@ -162,14 +164,7 @@ static void screen_callback(void *userdata, u8 op, f64 float0) } } -static aki_thread_result AKI_THREADCALL event_loop_thread(void *userdata) -{ - struct cmv *c = (struct cmv *)userdata; - aki_event_loop_run(&c->loop); - aki_event_loop_destroy(&c->loop); - return 0; -} - +/* static bool cap_callback(void *userdata, u8 op, str *name, str *unique_id, void *opaque) { struct cmv *c = (struct cmv *)userdata; @@ -190,6 +185,7 @@ static bool cap_callback(void *userdata, u8 op, str *name, str *unique_id, void } return true; } +*/ #if CMV_USE_SOCKET static u8 line_callback(void *userdata, str *line) @@ -214,7 +210,7 @@ static void timer_callback(void *userdata, struct aki_timer *timer) struct cmv *c = (struct cmv *)userdata; struct camu_sink_entry *entry = camu_sink_get_current(&c->sink); if (entry) { - tui_draw(&c->tui, &entry->runner, &entry->clock); + tui_draw(&c->tui, &((struct camu_sink_local *)entry)->runner, &entry->clock); } else { tui_draw(&c->tui, NULL, NULL); } @@ -233,22 +229,39 @@ static s32 log_callback(void *userdata, char *s) static void lav_log_callback(void *userdata, s32 level, const char *fmt, va_list args) { (void)userdata; - if (level < AV_LOG_VERBOSE) al_logv(warn, "lav_internal", (char *)fmt, args); + if (level < AV_LOG_VERBOSE) al_logv("warn", "lav_internal", (char *)fmt, args); } #endif #endif +static aki_thread_result AKI_THREADCALL event_loop_thread(void *userdata) +{ + struct cmv *c = (struct cmv *)userdata; + aki_event_loop_run(&c->loop); + aki_event_loop_destroy(&c->loop); + return 0; +} + +static struct cmv c = { 0 }; + +static void sigint_handler(s32 signum) +{ + (void)signum; + c.quit = 1; +} + s32 main(s32 argc, char *argv[]) { - aki_common_init(); + if (!aki_common_init()) return EXIT_FAILURE; - struct cmv c = { }; + signal(SIGINT, sigint_handler); + + aki_event_loop_init(&c.loop); #if CMV_USE_TUI - if (!tui_init(&c.tui)) { - aki_common_close(); - return EXIT_FAILURE; - } + if (!tui_init(&c.tui, &c.loop)) { + goto err; + }; al_set_print(log_callback, &c); #ifdef HAVE_FFMPEG camu_lav_set_log_callback(lav_log_callback); @@ -257,37 +270,35 @@ s32 main(s32 argc, char *argv[]) c.scr.callback = screen_callback; c.scr.userdata = &c; - if (!camu_screen_init(&c.scr)) { - aki_common_close(); - return EXIT_FAILURE; - } - - if (!camu_screen_create_window(&c.scr, "cmv")) { - aki_common_close(); - return EXIT_FAILURE; + if (!camu_screen_init(&c.scr) || !camu_screen_create_window(&c.scr, "cmv")) { + goto err; } c.renderer = camu_renderer_lp_create(); //c.renderer = camu_renderer_tiger_create(); - camu_screen_create_renderer(&c.scr, c.renderer); + if (!camu_screen_create_renderer(&c.scr, c.renderer)) { + goto err; + } + c.renderer->render(c.renderer, &c.scr); #if CAP_USE_PYTHON - bool py_init = sho_python_init(); + bool py_init = argc == 1 ? sho_python_init() : false; #endif camu_mixer_init(&c.mixer, (struct camu_audio *)&audio_plugin_miniaudio); c.mixer.audio->configure_stream(c.mixer.audio, NULL); - aki_event_loop_init(&c.loop); - + camu_sink_init(&c.sink, &c.loop, &c.mixer, c.renderer); c.sink.callback = sink_callback; c.sink.userdata = &c; - camu_sink_init(&c.sink, &c.loop, &c.mixer, c.renderer); + camu_sink_connect(&c.sink, al_str_c("127.0.0.1"), TREE_PORT); + /* cap_init(&c.cap, cap_callback, &c); cap_make_list(&c.cap, default_list); for (s32 i = 1; i < argc; i++) { cap_list_add(&c.cap, default_list, al_str_c(argv[i])); } + */ #if CMV_USE_SOCKET c.socket.type = AKI_SOCKET_UNIX; @@ -304,7 +315,7 @@ s32 main(s32 argc, char *argv[]) #if CMV_USE_TUI aki_timer_init(&c.timer, &c.loop, timer_callback, &c); - aki_timer_set_repeat(&c.timer, 0.10); + aki_timer_set_repeat(&c.timer, 0.05); aki_timer_again(&c.timer); #endif @@ -326,10 +337,10 @@ s32 main(s32 argc, char *argv[]) aki_thread_join(&thread0); camu_mixer_close(&c.mixer); - camu_screen_close(&c.scr); c.renderer->free(&c.renderer); + camu_screen_close(&c.scr); - cap_close(&c.cap); + //cap_close(&c.cap); #if CAP_USE_PYTHON if (py_init) sho_python_close(); @@ -342,4 +353,11 @@ s32 main(s32 argc, char *argv[]) aki_common_close(); return EXIT_SUCCESS; +err: +#if CMV_USE_TUI + tui_close(&c.tui); +#endif + aki_event_loop_destroy(&c.loop); + aki_common_close(); + return EXIT_FAILURE; } diff --git a/src/fruits/cmv/meson.build b/src/fruits/cmv/meson.build index b604104..1c9ed2d 100644 --- a/src/fruits/cmv/meson.build +++ b/src/fruits/cmv/meson.build @@ -1,26 +1,10 @@ -cmv2_src = ['cmv2.c', 'tui.c'] -#cmv2_src = ['cmv2.c'] +cmv_src = ['cmv2.c'] cmv_deps = [common_deps, buffer, cache, av, screen, render, mixer, bimu, libsink, cap] -miniaudio_args = [ - '-DMA_NO_JACK', - '-DMA_NO_ALSA', - '-DMA_NO_DECODING', - '-DMA_NO_ENCODING', - '-DMA_NO_GENERATION', - '-DMA_NO_RESOURCE_MANAGER', - '-DMA_NO_NODE_GRAPH', - '-DMA_NO_ENGINE', - '-DMA_NO_RUNTIME_LINKING' -] +use_tui = true +if use_tui + cmv_src += ['tui.c'] + cmv_deps += [dependency('notcurses')] +endif -sekihi_args = [ - '-DSEKIHI_WINDOW_WAYLAND', -# '-DSEKIHI_WINDOW_GLFW', - '-DSEKIHI_API_OPENGL', -# '-DSEKIHI_API_VULKAN', - '-DSEKIHI_POLL_INLINE', - '-DSEKIHI_PAUSE' -] - -executable('cmv2', cmv2_src, dependencies: cmv_deps, c_args: [miniaudio_args, sekihi_args]) +executable('cmv', cmv_src, dependencies: cmv_deps) diff --git a/src/fruits/cmv/tui.c b/src/fruits/cmv/tui.c index 21aeac8..8315a13 100644 --- a/src/fruits/cmv/tui.c +++ b/src/fruits/cmv/tui.c @@ -1,6 +1,7 @@ #include <aki/file.h> #include <al/log.h> #include <sys/ioctl.h> +#include <notcurses/direct.h> #include "tui.h" @@ -23,16 +24,39 @@ static void handle_winch(s32 sig) aki_mutex_unlock(&tui->mutex); } -bool tui_init(struct cmv_tui *tui) +static void input_poll_callback(void *userdata, s32 revents) +{ + struct cmv_tui *tui = (struct cmv_tui *)userdata; + (void)revents; + struct ncinput input; + u32 ret; + while (1) { + ret = ncdirect_get_nblock(tui->dir, &input); + if (ret == (u32)-1 || ret == 0) break; + if (input.evtype == NCTYPE_PRESS || input.evtype == NCTYPE_UNKNOWN) { + } + } +} + +bool tui_init(struct cmv_tui *tui, struct aki_event_loop *loop) { tui->update_size = true; tui->width = 0; tui->height = 0; - //tui->prev_lines = 0; + tui->prev_lines = 0; - tui->fd = fileno(stdin); + tui->fd = fileno(stdout); tui->out = fdopen(tui->fd, "w"); if (!tui->out) return false; + tui->buffer = al_malloc(256); + + //if (!(tui->dir = ncdirect_init(NULL, tui->out, 0))) { + // fclose(tui->out); + // return false; + //} + + //aki_poll_init(&tui->poll, ncdirect_inputready_fd(tui->dir), AKI_POLL_READ, input_poll_callback, tui); + //aki_poll_start(&tui->poll, loop); tui_global = tui; aki_mutex_init(&tui->mutex); @@ -46,81 +70,96 @@ bool tui_init(struct cmv_tui *tui) tui->term.c_lflag &= ~ECHO; tcsetattr(tui->fd, 0, &tui->term); - al_array_init(tui->log_buffer); - tui->last_pts = 0.0; + al_queue_init(&tui->log_buffer, 256); return true; } void tui_push_log_msg(struct cmv_tui *tui, char *msg) { - aki_mutex_lock(&tui->mutex); - al_array_push(tui->log_buffer, msg); - aki_mutex_unlock(&tui->mutex); + al_queue_push(&tui->log_buffer, msg); } -static char buffer[128]; - -static void draw_now_playing(struct cmv_tui *tui, struct camu_clock *clock, struct bmu_local *runner) +static void draw_now_playing(struct cmv_tui *tui, struct bmu_local *runner, struct camu_clock *clock) { + char *buffer = tui->buffer; f64 pts = camu_clock_get_pts(clock); - if (pts == -1) pts = tui->last_pts; - else tui->last_pts = pts; - s32 parts = 0; f64 duration = bmu_local_get_duration(runner); + s32 text = 0; + s64 minute = (s64)(pts / 60L); + s64 hour = minute / 60L; + minute -= hour * 60L; + text += al_sprintf(buffer + text, "["); + if (hour > 0) text += al_sprintf(buffer + text, "%.2ld:", hour); + text += al_sprintf(buffer + text, "%.2ld:%.2ld/", minute, (s64)pts % 60L); + minute = (s64)(duration / 60L); + hour = minute / 60L; + minute -= hour * 60L; + if (hour > 0) text += al_sprintf(buffer + text, "%.2ld:", hour); + text += al_sprintf(buffer + text, "%.2ld:%.2ld", minute, (s64)duration % 60L); + text += al_sprintf(buffer + text, "]"); + s32 parts = 0; if (pts > 0.0 && duration != 0.0) { if (pts > duration) pts = duration; f64 percent = pts / duration; - parts = tui->width * percent; + parts = ((tui->width - text) * percent); } else if (duration == 0.0) { - parts = tui->width; + parts = (tui->width - text); } - al_memset(buffer, '-', parts); - buffer[parts] = '\0'; + al_memset(buffer + text, '-', parts); + buffer[text + parts] = '\0'; fprintf(tui->out, "%s", buffer); } void tui_draw(struct cmv_tui *tui, struct bmu_local *runner, struct camu_clock *clock) { aki_mutex_lock(&tui->mutex); - if (tui->update_size) { update_term_size(tui); tui->update_size = false; } + aki_mutex_unlock(&tui->mutex); + // Erase line. fprintf(tui->out, "\r\033[K"); - //for (u32 i = 1; i < tui->prev_lines; i++) { - // fprintf(tui->out, "\033[A\r\033[K"); - //} + for (u32 i = 0; i < tui->prev_lines; i++) { + // Up one, erase line. + fprintf(tui->out, "\033[A\r\033[K"); + } char *msg; - for (u32 i = 0; i < tui->log_buffer.size; i++) { - al_array_pop_at(tui->log_buffer, 0, msg); + while (al_queue_pop(&tui->log_buffer, (void **)&msg)) { fprintf(tui->out, "%s\n", msg); al_free(msg); } - //tui->prev_lines = 0; + // Simply printing each log message with a newline will leave + // a one character high space in the bottom of the terminal window. + // We use that to display a very basic status line (like mpv). + // Trying to use more than that one line space without a more complete + // terminal interface gets scuffed very fast. + + tui->prev_lines = 0; if (runner) { - draw_now_playing(tui, clock, runner); + draw_now_playing(tui, runner, clock); } - - aki_mutex_unlock(&tui->mutex); } void tui_close(struct cmv_tui *tui) { + if (!tui->out) return; + al_free(tui->buffer); char *msg; - al_array_foreach(tui->log_buffer, i, msg) { + while (al_queue_pop(&tui->log_buffer, (void **)&msg)) { al_free(msg); } - al_array_free(tui->log_buffer); + al_queue_free(&tui->log_buffer); aki_mutex_destroy(&tui->mutex); fprintf(tui->out, "\n"); fflush(tui->out); tui->term.c_lflag |= ECHO; tcsetattr(tui->fd, 0, &tui->term); + //ncdirect_stop(tui->dir); fclose(tui->out); } diff --git a/src/fruits/cmv/tui.h b/src/fruits/cmv/tui.h index fb4215e..6898539 100644 --- a/src/fruits/cmv/tui.h +++ b/src/fruits/cmv/tui.h @@ -1,24 +1,28 @@ #pragma once +#include <al/queue.h> +#include <aki/event_loop.h> #include <termios.h> #include "../../bimu/local.h" -#include "../../buffer/clock.h" +#include "../../buffer/video.h" struct cmv_tui { s32 fd; FILE *out; + char *buffer; + struct ncdirect *dir; + struct aki_poll poll; struct aki_mutex mutex; struct termios term; bool update_size; u32 width; u32 height; - //u32 prev_lines; - array(char *) log_buffer; - f64 last_pts; + u32 prev_lines; + queue log_buffer; }; -bool tui_init(struct cmv_tui *tui); +bool tui_init(struct cmv_tui *tui, struct aki_event_loop *loop); void tui_push_log_msg(struct cmv_tui *tui, char *msg); void tui_draw(struct cmv_tui *tui, struct bmu_local *runner, struct camu_clock *clock); void tui_close(struct cmv_tui *tui); diff --git a/src/libclient/client.c b/src/libclient/client.c index 60998c6..e42f662 100644 --- a/src/libclient/client.c +++ b/src/libclient/client.c @@ -1,6 +1,3 @@ -#include "../tree/commands.h" -#include "../shoki/src/packet_ext.h" - #include "client.h" static void identifed_callback(void *userdata, struct aki_packet *packet) @@ -16,7 +13,7 @@ static void connection_callback(void *userdata, struct aki_rpc_connection *conn) client->conn = conn; struct aki_packet *packet = aki_rpc_get_packet(&client->client, TREE_CMD_IDENTIFY); aki_packet_write_u8(packet, TREE_CLIENT); - aki_packet_write_string(packet, &client->username); + aki_packet_write_str(packet, &client->username); aki_rpc_connection_command(client->conn, packet, identifed_callback, client); } @@ -33,66 +30,161 @@ static u8 packet_pool_callback(void *userdata, struct aki_packet *packet) return AKI_PACKET_POOL_NOP; } -bool camu_client_init(struct camu_client *client) -{ - aki_event_loop_init(&client->loop); - aki_packet_pool_init_ex(&client->pool, 0, &client->loop, - AKI_PACKET_POOL_MODE_PASSTHROUGH, packet_pool_callback, client); - return aki_rpc_init(&client->client, AKI_SOCKET_TCP, connection_callback, - connection_closed_callback, client); -} - -static aki_thread_result AKI_THREADCALL event_loop_thread(void *userdata) +static bool status_command_callback(void *userdata, struct aki_rpc_connection *conn, + struct aki_packet *packet, struct aki_packet *rpacket) { struct camu_client *client = (struct camu_client *)userdata; - aki_event_loop_run(&client->loop); - return 0; + (void)conn; + (void)rpacket; + + str s; + u32 size = aki_packet_read_u32(packet); + for (u32 i = 0; i < size; i++) { + struct camu_list *list = al_alloc_object(struct camu_list); + aki_packet_read_str(packet, &s); + al_str_clone(&list->name, &s); + al_array_push(client->state.lists, list); + } + + size = aki_packet_read_u32(packet); + for (u32 i = 0; i < size; i++) { + struct camu_search *search = al_alloc_object(struct camu_search); + search->id = aki_packet_read_s32(packet); + aki_packet_read_str(packet, &s); + al_str_clone(&search->provider, &s); + aki_packet_read_str(packet, &s); + al_str_clone(&search->query, &s); + search->page = aki_packet_read_s32(packet); + al_array_push(client->state.searches, search); + } + + client->callback(client->userdata, CAMU_CLIENT_STATUS_UPDATED, NULL); + + return false; } -bool camu_client_login(struct camu_client *client, str *username, str *addr, s32 port, +static struct aki_rpc_command commands[] = { + { .op = TREE_CMD_STATUS, .callback = status_command_callback, .userdata = NULL } +}; + +bool camu_client_init(struct camu_client *client, struct aki_event_loop *loop, void (*callback)(void *, u8, void *), void *userdata) { - al_str_clone(&client->username, username); + client->loop = loop; client->callback = callback; client->userdata = userdata; - if (!aki_rpc_connect(&client->client, &client->loop, addr, port)) { + client->username = AL_STR_EMPTY; + al_array_init(client->state.lists); + aki_packet_pool_init_ex(&client->pool, 0, client->loop, + AKI_PACKET_POOL_MODE_PASSTHROUGH, packet_pool_callback, client); + if (!aki_rpc_init(&client->client, AKI_SOCKET_TCP, connection_callback, + connection_closed_callback, client)) { + return false; + } + for (u32 i = 0; i < AL_ARRAY_SIZE(commands); i++) { + commands[i].userdata = client; + aki_rpc_add_command(&client->client, &commands[i]); + } + sho_post_cache_init(&client->cache); + return true; +} + +bool camu_client_login(struct camu_client *client, str *username, str *addr, s32 port) +{ + al_str_clone(&client->username, username); + if (!aki_rpc_connect(&client->client, client->loop, addr, port)) { return false; } - aki_thread_create(&client->thread, event_loop_thread, client); return true; } -static void search_callback(void *userdata, struct aki_packet *packet) +static void create_search_callback(void *userdata, struct aki_packet *packet) { struct camu_client *client = (struct camu_client *)userdata; - struct camu_search_results *results = al_alloc_object(struct camu_search_results); - al_array_init(results->results); - s32 search_id = aki_packet_read_s32(packet); - if (search_id == -1) return; - results->page = aki_packet_read_s32(packet); + s32 id = aki_packet_read_s32(packet); + client->callback(client->userdata, CAMU_CLIENT_SEARCH_CREATED, &id); +} + +void camu_client_create_search(struct camu_client *client, str *provider, str *query) +{ + struct aki_packet *packet = aki_rpc_get_packet(&client->client, TREE_CMD_SEARCH); + aki_packet_write_str(packet, provider); + aki_packet_write_str(packet, query); + packet->userdata = create_search_callback; + aki_packet_pool_submit(&client->pool, packet); +} + +static struct camu_search *search_from_id(struct camu_client *client, s32 id) +{ + struct camu_search *search; + al_array_foreach(client->state.searches, i, search) { + if (search->id == id) return search; + } + return NULL; +} + +static void results_callback(void *userdata, struct aki_packet *packet) +{ + struct camu_client *client = (struct camu_client *)userdata; + s32 ok = aki_packet_read_s32(packet); + if (ok != 0) return; + s32 id = aki_packet_read_s32(packet); + struct camu_search *search = search_from_id(client, id); + if (!search) return; + s32 page = aki_packet_read_s32(packet); + while (search->pages.size <= (u32)page) { + struct camu_search_results result; + result.page = page; + al_array_init(result.unique_ids); + al_array_push(search->pages, result); + } u32 size = aki_packet_read_u32(packet); for (u32 i = 0; i < size; i++) { struct sho_post post; aki_packet_read_sho_post(packet, &post); - sho_post_cache_push(&client->search.cache, &post); + sho_post_cache_push(&client->cache, &post); } + struct camu_search_results *result = &al_array_at(search->pages, page); + str s, unique_id; size = aki_packet_read_u32(packet); for (u32 i = 0; i < size; i++) { - str s, unique_id; - aki_packet_read_string(packet, &s); + aki_packet_read_str(packet, &s); al_str_clone(&unique_id, &s); - al_array_push(results->results, unique_id); + al_array_push(result->unique_ids, unique_id); } - client->callback(client->userdata, CAMU_CLIENT_RESULTS, results); - aki_packet_free(packet); + client->callback(client->userdata, CAMU_CLIENT_RESULTS, result); +} + +void camu_client_resume_search(struct camu_client *client, struct camu_search *search) +{ + struct aki_packet *packet = aki_rpc_get_packet(&client->client, TREE_CMD_RESUME_SEARCH); + aki_packet_write_s32(packet, search->id); + aki_packet_write_s32(packet, search->page); + packet->userdata = results_callback; + aki_packet_pool_submit(&client->pool, packet); } void camu_client_more_results(struct camu_client *client, struct camu_search *search) { - struct aki_packet *packet = aki_rpc_get_packet(&client->client, TREE_CMD_SEARCH); + struct aki_packet *packet = aki_rpc_get_packet(&client->client, TREE_CMD_RESUME_SEARCH); aki_packet_write_s32(packet, search->id); - aki_packet_write_string(packet, &search->provider); - aki_packet_write_string(packet, &search->query); - packet->userdata = search_callback; + aki_packet_write_s32(packet, -1); + packet->userdata = results_callback; aki_packet_pool_submit(&client->pool, packet); } + +void camu_client_add(struct camu_client *client, str *unique_id, u32 index) +{ + struct aki_packet *packet = aki_rpc_get_packet(&client->client, TREE_CMD_ADD); + aki_packet_write_str(packet, unique_id); + aki_packet_write_u32(packet, index); + packet->userdata = NULL; + aki_packet_pool_submit(&client->pool, packet); +} + +void camu_client_close(struct camu_client *client) +{ + aki_packet_pool_free(&client->pool); + al_str_free(&client->username); + al_array_free(client->state.lists); +} diff --git a/src/libclient/client.h b/src/libclient/client.h index 989c839..cdcaffe 100644 --- a/src/libclient/client.h +++ b/src/libclient/client.h @@ -3,29 +3,56 @@ #include <aki/rpc2.h> #include <aki/thread.h> #include <aki/packet_pool.h> +#include <sho/post.h> +#include <sho/post_cache.h> -#include "search.h" +#include "../tree/common.h" enum { CAMU_CLIENT_CONNECTED = 0, - // Search client. + CAMU_CLIENT_STATUS_UPDATED, + CAMU_CLIENT_SEARCH_CREATED, CAMU_CLIENT_RESULTS }; +struct camu_list { + str name; + array(str) entries; +}; + +struct camu_search_results { + s32 page; + array(str) unique_ids; +}; + +struct camu_search { + s32 id; + str provider; + str query; + s32 page; + array(struct camu_search_results) pages; +}; + struct camu_client { - struct aki_event_loop loop; + struct aki_event_loop *loop; struct aki_rpc client; str username; + struct { + array(struct camu_list *) lists; + array(struct camu_search *) searches; + } state; + struct sho_post_cache cache; struct aki_rpc_connection *conn; - struct camu_search_client search; struct aki_packet_pool pool; - struct aki_thread thread; void (*callback)(void *, u8, void *); void *userdata; }; -bool camu_client_init(struct camu_client *client); -bool camu_client_login(struct camu_client *client, str *username, str *addr, s32 port, +bool camu_client_init(struct camu_client *client, struct aki_event_loop *loop, void (*callback)(void *, u8, void *), void *userdata); +bool camu_client_login(struct camu_client *client, str *username, str *addr, s32 port); +void camu_client_create_search(struct camu_client *client, str *provider, str *query); +void camu_client_resume_search(struct camu_client *client, struct camu_search *search); void camu_client_more_results(struct camu_client *client, struct camu_search *search); +void camu_client_add(struct camu_client *client, str *unique_id, u32 index); void camu_client_close(struct camu_client *client); diff --git a/src/libclient/meson.build b/src/libclient/meson.build index 99eabb2..a2f23b0 100644 --- a/src/libclient/meson.build +++ b/src/libclient/meson.build @@ -1,3 +1,3 @@ -libclient_src = ['client.c', 'search.c', 'resource_client.c'] +libclient_src = ['client.c', 'resource_client.c'] libclient_deps = [shoki] libclient = declare_dependency(sources: libclient_src, dependencies: libclient_deps) diff --git a/src/libclient/resource_client.c b/src/libclient/resource_client.c index 1069755..77c19c7 100644 --- a/src/libclient/resource_client.c +++ b/src/libclient/resource_client.c @@ -28,6 +28,7 @@ static void packet_callback(void *userdata, struct aki_packet_stream *stream, aki_buffer_init(&buffer); aki_packet_read_buffer(packet, &buffer); client->callback(client->userdata, id, &buffer); + aki_packet_free(packet); } static void packet_sent_callback(void *userdata, struct aki_packet *packet) @@ -42,41 +43,37 @@ static void signal_callback(void *userdata) u32 size; struct camu_resource_request request; do { - camu_queue_size(client->queue, size); - if (!size) break; - camu_queue_pop(client->queue, request); + camu_queue_try_pop(client->queue, size, request); + if (size == 0) break; struct aki_packet *packet = aki_packet_create(); aki_packet_write_u16(packet, request.id); - aki_packet_write_string(packet, &request.unique_id); + aki_packet_write_str(packet, &request.unique_id); aki_packet_write_u32(packet, request.index); aki_packet_stream_send_packet(&client->client, packet); } while (1); } -static aki_thread_result AKI_THREADCALL event_loop_thread(void *userdata) -{ - struct camu_resource_client *client = (struct camu_resource_client *)userdata; - aki_event_loop_run(&client->loop); - return 0; -} - -void camu_resource_client_run(struct camu_resource_client *client, - void (*callback)(void *, u16, struct aki_buffer *), void *userdata) +void camu_resource_client_init(struct camu_resource_client *client, + struct aki_event_loop *loop, void (*callback)(void *, u16, struct aki_buffer *), void *userdata) { + client->loop = loop; client->callback = callback; client->userdata = userdata; camu_queue_init(client->queue); - aki_event_loop_init(&client->loop); aki_signal_init(&client->signal, signal_callback, client); - aki_signal_start(&client->signal, &client->loop); aki_packet_stream_init(&client->client, AKI_SOCKET_TCP, connection_callback, connection_closed_callback, packet_callback, packet_sent_callback, client); - aki_packet_stream_connect(&client->client, &client->loop, al_str_c("127.0.0.1"), TREE_RESOURCE_PORT); +} + + +void camu_resource_client_connect(struct camu_resource_client *client, str *addr, s32 port) +{ + aki_signal_start(&client->signal, client->loop); + aki_packet_stream_connect(&client->client, client->loop, addr, port); client->connected = false; while (!client->connected) { - aki_event_loop_run_once(&client->loop); + aki_event_loop_run_once(client->loop); } - aki_thread_create(&client->thread, event_loop_thread, client); } void camu_resource_request(struct camu_resource_client *client, diff --git a/src/libclient/resource_client.h b/src/libclient/resource_client.h index 4cc1bbf..17ac31e 100644 --- a/src/libclient/resource_client.h +++ b/src/libclient/resource_client.h @@ -13,7 +13,7 @@ struct camu_resource_request { }; struct camu_resource_client { - struct aki_event_loop loop; + struct aki_event_loop *loop; struct aki_packet_stream client; bool connected; struct aki_thread thread; @@ -23,7 +23,8 @@ struct camu_resource_client { void *userdata; }; -void camu_resource_client_run(struct camu_resource_client *client, - void (*callback)(void *, u16, struct aki_buffer *), void *userdata); +void camu_resource_client_init(struct camu_resource_client *client, + struct aki_event_loop *loop, void (*callback)(void *, u16, struct aki_buffer *), void *userdata); +void camu_resource_client_connect(struct camu_resource_client *client, str *addr, s32 port); void camu_resource_request(struct camu_resource_client *client, str *unique_id, u32 index, u16 id); diff --git a/src/libclient/search.c b/src/libclient/search.c deleted file mode 100644 index f74f83d..0000000 --- a/src/libclient/search.c +++ /dev/null @@ -1,31 +0,0 @@ -#include "search.h" - -void camu_search_client_init(struct camu_search_client *client) -{ - client->id = 0; - sho_post_cache_init(&client->cache); -} - -void camu_search_set(struct camu_search_client *client, struct camu_search *search, - str *provider, str *query) -{ - search->id = client->id++; - al_str_clone(&search->provider, provider); - al_str_clone(&search->query, query); -} - -void camu_search_free(struct camu_search *search) -{ - al_str_free(&search->provider); - al_str_free(&search->query); -} - -void camu_search_results_free(struct camu_search_results *results) -{ - str *unique_id; - al_array_foreach_ptr(results->results, i, unique_id) { - al_str_free(unique_id); - } - al_array_free(results->results); - al_free(results); -} diff --git a/src/libclient/search.h b/src/libclient/search.h deleted file mode 100644 index b0d8554..0000000 --- a/src/libclient/search.h +++ /dev/null @@ -1,28 +0,0 @@ -#pragma once - -#include <al/array.h> -#include <al/str.h> - -#include "../shoki/src/post_cache.h" - -struct camu_search { - s32 id; - str provider; - str query; -}; - -struct camu_search_results { - u32 page; - array(str) results; -}; - -struct camu_search_client { - s32 id; - struct sho_post_cache cache; -}; - -void camu_search_client_init(struct camu_search_client *client); -void camu_search_set(struct camu_search_client *client, struct camu_search *search, - str *provider, str *query); -void camu_search_free(struct camu_search *search); -void camu_search_results_free(struct camu_search_results *results); diff --git a/src/libsink/common.h b/src/libsink/common.h new file mode 100644 index 0000000..1c701c8 --- /dev/null +++ b/src/libsink/common.h @@ -0,0 +1,6 @@ +#pragma once + +enum { + CAMU_SINK_CMD_BUFFER = 0, + CAMU_SINK_CMD_SET +}; diff --git a/src/libsink/meson.build b/src/libsink/meson.build index 5cbe36d..f212b0c 100644 --- a/src/libsink/meson.build +++ b/src/libsink/meson.build @@ -1,2 +1,2 @@ -libsink_src = ['sink.c'] +libsink_src = ['sink2.c'] libsink = declare_dependency(sources: libsink_src) diff --git a/src/libsink/sink.c b/src/libsink/sink.c index bae6c77..3692066 100644 --- a/src/libsink/sink.c +++ b/src/libsink/sink.c @@ -32,12 +32,12 @@ enum { }; #define ENTRY_AUDIO_BUFFER_HELD(entry) \ - (al_atomic_bool_load(&entry->audio.buf.ref, AL_ATOMIC_RELAXED)) + (al_atomic_bool_load(&(entry)->audio.buf.ref, AL_ATOMIC_RELAXED)) #ifdef CAMU_SINK_NO_VIDEO #define ENTRY_BUFFERS_HELD(entry) ENTRY_AUDIO_BUFFER_HELD(entry) #else #define ENTRY_VIDEO_BUFFER_HELD(entry) \ - (al_atomic_bool_load(&entry->video.buf.ref, AL_ATOMIC_RELAXED)) + (al_atomic_bool_load(&(entry)->video.buf.ref, AL_ATOMIC_RELAXED)) #define ENTRY_BUFFERS_HELD(entry) (ENTRY_AUDIO_BUFFER_HELD(entry) || ENTRY_VIDEO_BUFFER_HELD(entry)) #endif @@ -160,7 +160,13 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) } case SEEK: { struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque; - bmu_local_seek(&entry->runner, cmd->value.f); + switch (entry->type) { + case CAMU_SINK_LOCAL: + bmu_local_seek(&((struct camu_sink_local *)entry)->runner, cmd->value.f); + break; + default: + break; + } break; } case CLOSE: { @@ -176,9 +182,8 @@ static void queue_signal_callback(void *userdata) u32 size; struct camu_sink_cmd cmd; do { - camu_queue_size(sink->queue, size); - if (!size) break; - camu_queue_pop(sink->queue, cmd); + camu_queue_try_pop(sink->queue, size, cmd); + if (size == 0) break; handle_sink_cmd(sink, &cmd); } while (1); } @@ -227,10 +232,22 @@ static void audio_buffer_callback(void *userdata, u8 op) aki_mutex_unlock(&entry->sink->mutex); break; case CAMU_BUFFER_STOP: - bmu_local_stream_stop((struct bmu_local_stream *)entry->audio.stream); + switch (entry->type) { + case CAMU_SINK_LOCAL: + bmu_local_stream_stop((struct bmu_local_stream *)entry->audio.stream); + break; + default: + break; + } break; case CAMU_BUFFER_CONTINUE: - bmu_local_stream_continue((struct bmu_local_stream *)entry->audio.stream); + switch (entry->type) { + case CAMU_SINK_LOCAL: + bmu_local_stream_continue((struct bmu_local_stream *)entry->audio.stream); + break; + default: + break; + } break; case CAMU_BUFFER_PAUSED: queue_cmd(entry->sink, (struct camu_sink_cmd){ @@ -245,15 +262,20 @@ static void audio_buffer_callback(void *userdata, u8 op) entry->audio.state = BUFFER_SET_OR_BUFFERED; aki_mutex_unlock(&entry->sink->mutex); if (ret != CAMU_SINK_BUFFERS_SWAPPED) { + // TODO: Can cause popping. Should run a timer + // and if any action would queue_cmd(AUDIO_START) before the + // the timer, don't queue the stop. queue_cmd(entry->sink, (struct camu_sink_cmd){ .op = STOP, .value.i = CAMU_SINK_AUDIO, .opaque = entry->sink }); +#ifndef CAMU_SINK_NO_VIDEO queue_cmd(entry->sink, (struct camu_sink_cmd){ .op = SET_BUFFERED, .value.i = CAMU_SINK_VIDEO }); +#endif } break; } @@ -295,10 +317,22 @@ static void video_buffer_callback(void *userdata, u8 op) aki_mutex_unlock(&entry->sink->mutex); break; case CAMU_BUFFER_STOP: - bmu_local_stream_stop((struct bmu_local_stream *)entry->video.stream); + switch (entry->type) { + case CAMU_SINK_LOCAL: + bmu_local_stream_stop((struct bmu_local_stream *)entry->video.stream); + break; + default: + break; + } break; case CAMU_BUFFER_CONTINUE: - bmu_local_stream_continue((struct bmu_local_stream *)entry->video.stream); + switch (entry->type) { + case CAMU_SINK_LOCAL: + bmu_local_stream_continue((struct bmu_local_stream *)entry->video.stream); + break; + default: + break; + } break; case CAMU_BUFFER_EOF: queue_cmd(entry->sink, (struct camu_sink_cmd){ @@ -409,14 +443,32 @@ static void client_callback(void *userdata, u8 op, struct bmu_client_stream *str break; } case BIMU_CLIENT_CLOSED: { - bmu_local_close(&entry->runner); + switch (entry->type) { + case CAMU_SINK_LOCAL: + bmu_local_close(&((struct camu_sink_local *)entry)->runner); + break; + default: + break; + } camu_audio_buffer_free(&entry->audio.buf); #ifndef CAMU_SINK_NO_VIDEO camu_video_buffer_free(&entry->video.buf); #endif - cch_entry_free(&entry->entry); + switch (entry->type) { + case CAMU_SINK_LOCAL: + cch_entry_free(&((struct camu_sink_local *)entry)->entry); + break; + default: + break; + } al_str_free(&entry->unique_id); - al_free(entry); + switch (entry->type) { + case CAMU_SINK_LOCAL: + al_free((struct camu_sink_local *)entry); + break; + default: + break; + } al_log_debug("sink", "Entry closed."); break; } @@ -533,92 +585,109 @@ void camu_sink_close(struct camu_sink *sink) aki_mutex_lock(&sink->mutex); struct camu_sink_entry *entry; al_array_foreach(sink->entries, i, entry) { - bmu_local_stop(&entry->runner); + switch (entry->type) { + case CAMU_SINK_LOCAL: + bmu_local_stop(&((struct camu_sink_local *)entry)->runner); + break; + default: + break; + } } - aki_mutex_unlock(&sink->mutex); al_array_free(sink->entries); + aki_mutex_unlock(&sink->mutex); aki_signal_stop(&sink->signal); camu_queue_free(sink->queue); aki_mutex_destroy(&sink->mutex); } -static struct camu_sink_entry *entry_from_unique_id(struct camu_sink *sink, str *unique_id) +// Local bimu band-aid compat. + +static struct camu_sink_local *entry_from_unique_id(struct camu_sink *sink, str *unique_id) { struct camu_sink_entry *entry; al_array_foreach(sink->entries, i, entry) { if (al_str_eq(&entry->unique_id, unique_id)) { - aki_mutex_unlock(&sink->mutex); - return entry; + return (struct camu_sink_local *)entry; } } return NULL; } +static void camu_sink_buffer_internal(struct camu_sink *sink, struct camu_sink_entry *entry, str *unique_id) +{ + entry->sink = sink; + al_str_clone(&entry->unique_id, unique_id); + + camu_clock_init(&entry->clock); + + entry->audio.state = BUFFER_INIT; + camu_audio_buffer_init(&entry->audio.buf, &entry->clock, sink->audio.mixer); + entry->audio.buf.callback = audio_buffer_callback; + entry->audio.buf.userdata = entry; + + struct camu_renderer *renderer = NULL; +#ifndef CAMU_SINK_NO_VIDEO + renderer = sink->video.renderer; + entry->video.state = BUFFER_INIT; + camu_video_buffer_init(&entry->video.buf, &entry->clock, renderer); + entry->video.buf.callback = video_buffer_callback; + entry->video.buf.userdata = entry; +#endif + + al_array_push(sink->entries, entry); +} + bool camu_sink_local_buffer(struct camu_sink *sink, str *unique_id, struct cch_entry *centry) { aki_mutex_lock(&sink->mutex); - bool add_entry = false; - struct camu_sink_entry *entry = entry_from_unique_id(sink, unique_id); - if (!entry) { - entry = al_alloc_object(struct camu_sink_entry); - add_entry = true; + bool repeat = true; + struct camu_sink_local *local = entry_from_unique_id(sink, unique_id); + if (!local) { + local = al_alloc_object(struct camu_sink_local); + local->e.type = CAMU_SINK_LOCAL; + repeat = false; } - entry->state = ENTRY_LOADED; - if (!add_entry) { + local->e.state = ENTRY_LOADED; + if (repeat) { queue_cmd(sink, (struct camu_sink_cmd){ .op = SEEK, .value.f = 0, - .opaque = entry + .opaque = &local->e }); aki_mutex_unlock(&sink->mutex); return true; } - entry->entry = centry; - if (!bmu_local_init(&entry->runner, centry)) { - al_free(entry); + local->entry = centry; + if (!bmu_local_init(&local->runner, centry)) { + al_free(local); aki_mutex_unlock(&sink->mutex); return false; } - entry->sink = sink; - al_str_clone(&entry->unique_id, unique_id); - - camu_clock_init(&entry->clock); - - entry->audio.state = BUFFER_INIT; - camu_audio_buffer_init(&entry->audio.buf, &entry->clock, sink->audio.mixer); - entry->audio.buf.callback = audio_buffer_callback; - entry->audio.buf.userdata = entry; + camu_sink_buffer_internal(sink, &local->e, unique_id); struct camu_renderer *renderer = NULL; #ifndef CAMU_SINK_NO_VIDEO renderer = sink->video.renderer; - entry->video.state = BUFFER_INIT; - camu_video_buffer_init(&entry->video.buf, &entry->clock, renderer); - entry->video.buf.callback = video_buffer_callback; - entry->video.buf.userdata = entry; #endif - - entry->runner.callback = client_callback; - entry->runner.userdata = entry; - bmu_local_prepare_clients(&entry->runner, renderer); - bmu_local_run(&entry->runner); - - if (add_entry) al_array_push(sink->entries, entry); + local->runner.callback = client_callback; + local->runner.userdata = &local->e; + bmu_local_prepare_clients(&local->runner, renderer); + bmu_local_run(&local->runner); aki_mutex_unlock(&sink->mutex); return true; } -static void maybe_cleanup_entries(struct camu_sink *sink) +static void maybe_cleanup_local_entries(struct camu_sink *sink) { struct camu_sink_entry *entry; al_array_foreach(sink->entries, i, entry) { if (entry->state == ENTRY_DISREGUARDED && !ENTRY_BUFFERS_HELD(entry)) { - bmu_local_stop(&entry->runner); + bmu_local_stop(&((struct camu_sink_local *)entry)->runner); al_array_remove_at_iter(sink->entries, i); } } @@ -627,12 +696,12 @@ static void maybe_cleanup_entries(struct camu_sink *sink) void camu_sink_local_set(struct camu_sink *sink, str *unique_id) { aki_mutex_lock(&sink->mutex); - struct camu_sink_entry *entry = entry_from_unique_id(sink, unique_id); + struct camu_sink_local *local = entry_from_unique_id(sink, unique_id); struct camu_sink_entry *previous = sink->current; - sink->current = entry; - add_audio_if_set_and_buffered(entry); + sink->current = &local->e; + add_audio_if_set_and_buffered(&local->e); #ifndef CAMU_SINK_NO_VIDEO - add_video_if_set_and_buffered(entry); + add_video_if_set_and_buffered(&local->e); #endif if (previous) { if (!camu_clock_is_paused(&previous->clock)) { @@ -640,24 +709,24 @@ void camu_sink_local_set(struct camu_sink *sink, str *unique_id) } remove_entry_buffers(sink, previous); } - maybe_cleanup_entries(sink); + maybe_cleanup_local_entries(sink); aki_mutex_unlock(&sink->mutex); } void camu_sink_local_swap(struct camu_sink *sink, str *unique_id) { - struct camu_sink_entry *entry = entry_from_unique_id(sink, unique_id); - sink->current = entry; - add_audio_if_set_and_buffered(entry); + struct camu_sink_local *local = entry_from_unique_id(sink, unique_id); + sink->current = &local->e; + add_audio_if_set_and_buffered(&local->e); #ifndef CAMU_SINK_NO_VIDEO - add_video_if_set_and_buffered(entry); + add_video_if_set_and_buffered(&local->e); #endif } void camu_sink_local_unload(struct camu_sink *sink, str *unique_id) { aki_mutex_lock(&sink->mutex); - struct camu_sink_entry *entry = entry_from_unique_id(sink, unique_id); - if (entry && entry->state == ENTRY_LOADED) entry->state = ENTRY_DISREGUARDED; + struct camu_sink_local *local = entry_from_unique_id(sink, unique_id); + if (local && local->e.state == ENTRY_LOADED) local->e.state = ENTRY_DISREGUARDED; aki_mutex_unlock(&sink->mutex); } diff --git a/src/libsink/sink.h b/src/libsink/sink.h index 92c18d9..efef97b 100644 --- a/src/libsink/sink.h +++ b/src/libsink/sink.h @@ -27,20 +27,24 @@ enum { }; enum { + CAMU_SINK_REMOTE = 0, + CAMU_SINK_LOCAL +}; + +enum { CAMU_SINK_ADD_BUFFER = 0, CAMU_SINK_REMOVE_BUFFER, CAMU_SINK_SWAP_BUFFER, - CAMU_SINK_SET_BUFFERED, CAMU_SINK_START, CAMU_SINK_STOP, + CAMU_SINK_ENTRY_BUFFERED, CAMU_SINK_EXIT }; struct camu_sink_entry { + u8 type; u8 state; str unique_id; - struct cch_entry *entry; - struct bmu_local runner; struct camu_clock clock; struct { u8 state; @@ -57,6 +61,12 @@ struct camu_sink_entry { struct camu_sink *sink; }; +struct camu_sink_local { + struct camu_sink_entry e; + struct cch_entry *entry; + struct bmu_local runner; +}; + struct camu_sink_cmd { u8 op; union { s64 i; f64 f; } value; diff --git a/src/libsink/sink2.c b/src/libsink/sink2.c new file mode 100644 index 0000000..e9b51f2 --- /dev/null +++ b/src/libsink/sink2.c @@ -0,0 +1,688 @@ +#include <al/log.h> + +#include "../tree/common.h" + +#include "sink2.h" + +enum { + SINK_EMPTY = 0, + SINK_PAUSED, + SINK_PLAYING +}; + +enum { + ENTRY_LOADED = 0, + ENTRY_DISREGUARDED +}; + +enum { + BUFFER_INIT = 0, + BUFFER_QUEUED, + BUFFER_CONFIGURED, + BUFFER_SET_OR_BUFFERED, + BUFFER_ADDED, +}; + +enum { + // TODO: Add and remove buffer can be done from + // any thread. Does this always need to be the case? + // Is it beneficial for this to be the case? + ADD_BUFFER = 0, + REMOVE_BUFFER, + START, + STOP, + TOGGLE_PAUSE, + SEEK, + ENTRY_BUFFERED, // Currently set entry is buffered. + CLOSE, + // Internal. + CORK, + UNCORK +}; + +#define ENTRY_AUDIO_BUFFER_HELD(entry) \ + (al_atomic_bool_load(&(entry)->audio.buf.ref, AL_ATOMIC_RELAXED)) +#ifdef CAMU_SINK_NO_VIDEO +#define ENTRY_BUFFERS_HELD(entry) ENTRY_AUDIO_BUFFER_HELD(entry) +#else +#define ENTRY_VIDEO_BUFFER_HELD(entry) \ + (al_atomic_bool_load(&(entry)->video.buf.ref, AL_ATOMIC_RELAXED)) +#define ENTRY_BUFFERS_HELD(entry) (ENTRY_AUDIO_BUFFER_HELD(entry) || ENTRY_VIDEO_BUFFER_HELD(entry)) +#endif + +#define ENTRY_VIDEO_EMPTY(entry) \ + (entry->video.state == BUFFER_INIT || entry->video.state == BUFFER_QUEUED) +#define ENTRY_AUDIO_EMPTY(entry) \ + (entry->audio.state == BUFFER_INIT || entry->audio.state == BUFFER_QUEUED) + +static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) +{ + switch (cmd->op) { + case ADD_BUFFER: { + struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque; + switch (cmd->value.i) { + case CAMU_SINK_AUDIO: + sink->callback(sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf); + break; +#ifndef CAMU_SINK_NO_VIDEO + case CAMU_SINK_VIDEO: + sink->callback(sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_VIDEO, &entry->video.buf); + break; +#endif + } + break; + } + case REMOVE_BUFFER: { + struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque; + switch (cmd->value.i) { + case CAMU_SINK_AUDIO: + sink->callback(sink->userdata, CAMU_SINK_REMOVE_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf); + break; +#ifndef CAMU_SINK_NO_VIDEO + case CAMU_SINK_VIDEO: + sink->callback(sink->userdata, CAMU_SINK_REMOVE_BUFFER, CAMU_SINK_VIDEO, &entry->video.buf); + break; +#endif + } + break; + } + case START: { + switch (cmd->value.i) { + case CAMU_SINK_AUDIO: + if (sink->audio.state == SINK_PAUSED) { + sink->callback(sink->userdata, CAMU_SINK_START, CAMU_SINK_AUDIO, NULL); + sink->audio.state = SINK_PLAYING; + } + break; +#ifndef CAMU_SINK_NO_VIDEO + case CAMU_SINK_VIDEO: + if (sink->video.state == SINK_PAUSED) { + sink->callback(sink->userdata, CAMU_SINK_START, CAMU_SINK_VIDEO, NULL); + sink->video.state = SINK_PLAYING; + } + break; +#endif + } + break; + } + case STOP: { + switch (cmd->value.i) { + case CAMU_SINK_AUDIO: + if (sink->audio.state == SINK_PLAYING) { + sink->callback(sink->userdata, CAMU_SINK_STOP, CAMU_SINK_AUDIO, NULL); + sink->audio.state = SINK_PAUSED; + } + break; +#ifndef CAMU_SINK_NO_VIDEO + case CAMU_SINK_VIDEO: + if (sink->video.state == SINK_PLAYING) { + sink->callback(sink->userdata, CAMU_SINK_STOP, CAMU_SINK_VIDEO, NULL); + sink->video.state = SINK_PAUSED; + } + break; +#endif + } + break; + } + case TOGGLE_PAUSE: { + struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque; + if (camu_clock_is_paused(&entry->clock)) { + camu_clock_resume(&entry->clock); + if (sink->audio.state == SINK_PAUSED) { + sink->callback(sink->userdata, CAMU_SINK_START, CAMU_SINK_AUDIO, NULL); + sink->audio.state = SINK_PLAYING; + } +#ifndef CAMU_SINK_NO_VIDEO + if (sink->video.state == SINK_PAUSED) { + sink->callback(sink->userdata, CAMU_SINK_START, CAMU_SINK_VIDEO, NULL); + sink->video.state = SINK_PLAYING; + } +#endif + } else { + camu_clock_pause(&entry->clock); +#ifndef CAMU_SINK_NO_VIDEO + if (sink->video.state == SINK_PLAYING) { + sink->callback(sink->userdata, CAMU_SINK_STOP, CAMU_SINK_VIDEO, NULL); + sink->video.state = SINK_PAUSED; + } +#endif + } + break; + } + case SEEK: { + struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque; + (void)entry; + break; + } + case ENTRY_BUFFERED: { + switch (cmd->value.i) { + case CAMU_SINK_AUDIO: + sink->callback(sink->userdata, CAMU_SINK_SET_BUFFERED, CAMU_SINK_AUDIO, NULL); + break; +#ifndef CAMU_SINK_NO_VIDEO + case CAMU_SINK_VIDEO: + sink->callback(sink->userdata, CAMU_SINK_SET_BUFFERED, CAMU_SINK_VIDEO, NULL); + break; +#endif + } + break; + } + case CLOSE: { + sink->callback(sink->userdata, CAMU_SINK_EXIT, 0, NULL); + return; + } + case CORK: { + struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque; + bmu_client_cork(&entry->client, true); + break; + } + case UNCORK: { + struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque; + bmu_client_cork(&entry->client, false); + break; + } + } +} + +static void queue_signal_callback(void *userdata) +{ + struct camu_sink *sink = (struct camu_sink *)userdata; + u32 size; + struct camu_sink_cmd cmd; + do { + camu_queue_try_pop(sink->queue, size, cmd); + if (size == 0) break; + handle_sink_cmd(sink, &cmd); + } while (1); +} + +static void queue_cmd(struct camu_sink *sink, struct camu_sink_cmd cmd) +{ + camu_queue_push(sink->queue, cmd); + aki_signal_send(&sink->signal); +} + +static void add_audio_if_set_and_buffered(struct camu_sink_entry *entry) +{ + u8 state = entry->audio.state; + if (state == BUFFER_SET_OR_BUFFERED) { + entry->sink->callback(entry->sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf); + queue_cmd(entry->sink, (struct camu_sink_cmd){ + .op = ENTRY_BUFFERED, + .value.i = CAMU_SINK_AUDIO + }); + queue_cmd(entry->sink, (struct camu_sink_cmd){ + .op = START, + .value.i = CAMU_SINK_AUDIO + }); + if (ENTRY_VIDEO_EMPTY(entry) || entry->video.state == BUFFER_ADDED) { + camu_clock_resume(&entry->clock); + } + state = BUFFER_ADDED; + } else if (state == BUFFER_CONFIGURED) { + state = BUFFER_SET_OR_BUFFERED; + } + entry->audio.state = state; +} + +static void audio_buffer_callback(void *userdata, u8 op) +{ + struct camu_sink_entry *entry = (struct camu_sink_entry *)userdata; + switch (op) { + case CAMU_BUFFER_BUFFERED: + aki_mutex_lock(&entry->sink->mutex); + add_audio_if_set_and_buffered(entry); + aki_mutex_unlock(&entry->sink->mutex); + break; + case CAMU_BUFFER_CORK: + queue_cmd(entry->sink, (struct camu_sink_cmd){ + .op = CORK, + .opaque = entry + }); + break; + case CAMU_BUFFER_UNCORK: + queue_cmd(entry->sink, (struct camu_sink_cmd){ + .op = UNCORK, + .opaque = entry + }); + break; + case CAMU_BUFFER_PAUSED: + queue_cmd(entry->sink, (struct camu_sink_cmd){ + .op = STOP, + .value.i = CAMU_SINK_AUDIO + }); + break; + case CAMU_BUFFER_EOF: { + aki_mutex_lock(&entry->sink->mutex); + u8 ret = entry->sink->callback(entry->sink->userdata, CAMU_SINK_SWAP_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf); + entry->audio.state = BUFFER_SET_OR_BUFFERED; + aki_mutex_unlock(&entry->sink->mutex); + if (ret != CAMU_SINK_BUFFERS_SWAPPED) { + // TODO: Can cause popping. Should run a timer + // and if any action would queue_cmd(AUDIO_START) before the + // the timer, don't queue the stop. + queue_cmd(entry->sink, (struct camu_sink_cmd){ + .op = STOP, + .value.i = CAMU_SINK_AUDIO + }); +#ifndef CAMU_SINK_NO_VIDEO + queue_cmd(entry->sink, (struct camu_sink_cmd){ + .op = ENTRY_BUFFERED, + .value.i = CAMU_SINK_VIDEO + }); +#endif + } + break; + } + } +} + +#ifndef CAMU_SINK_NO_VIDEO +static void add_video_if_set_and_buffered(struct camu_sink_entry *entry) +{ + u8 state = entry->video.state; + if (state == BUFFER_SET_OR_BUFFERED) { + entry->sink->callback(entry->sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_VIDEO, &entry->video.buf); + queue_cmd(entry->sink, (struct camu_sink_cmd){ + .op = ENTRY_BUFFERED, + .value.i = CAMU_SINK_VIDEO + }); + queue_cmd(entry->sink, (struct camu_sink_cmd){ + .op = START, + .value.i = CAMU_SINK_VIDEO + }); + if (ENTRY_AUDIO_EMPTY(entry) || entry->audio.state == BUFFER_ADDED) { + camu_clock_resume(&entry->clock); + } + state = BUFFER_ADDED; + } else if (state == BUFFER_CONFIGURED) { + state = BUFFER_SET_OR_BUFFERED; + } + entry->video.state = state; +} + +static void video_buffer_callback(void *userdata, u8 op) +{ + struct camu_sink_entry *entry = (struct camu_sink_entry *)userdata; + switch (op) { + case CAMU_BUFFER_BUFFERED: + aki_mutex_lock(&entry->sink->mutex); + add_video_if_set_and_buffered(entry); + aki_mutex_unlock(&entry->sink->mutex); + break; + case CAMU_BUFFER_CORK: + queue_cmd(entry->sink, (struct camu_sink_cmd){ + .op = CORK, + .opaque = entry + }); + break; + case CAMU_BUFFER_UNCORK: + queue_cmd(entry->sink, (struct camu_sink_cmd){ + .op = UNCORK, + .opaque = entry + }); + break; + case CAMU_BUFFER_EOF: + queue_cmd(entry->sink, (struct camu_sink_cmd){ + .op = STOP, + .value.i = CAMU_SINK_VIDEO + }); + break; + } +} +#endif + +static void client_callback(void *userdata, u8 op, struct bmu_client_stream *stream, void *opaque) +{ + struct camu_sink_entry *entry = (struct camu_sink_entry *)userdata; + switch (op) { + // A lot of logic here assumes that no data will be sent + // until all active streams are configured. This is extremely important, + // if it does not hold true many confusing errors will arise. + case BIMU_CLIENT_CONFIGURE: { + switch (stream->type) { + case BIMU_STREAM_AUDIO: + entry->audio.stream = stream; + camu_audio_buffer_configure(&entry->audio.buf, &stream->stream); + aki_mutex_lock(&entry->sink->mutex); + if (entry->audio.state == BUFFER_QUEUED) { + entry->audio.state = BUFFER_SET_OR_BUFFERED; + } else { + entry->audio.state = BUFFER_CONFIGURED; + } + aki_mutex_unlock(&entry->sink->mutex); + break; +#ifndef CAMU_SINK_NO_VIDEO + case BIMU_STREAM_VIDEO: + entry->video.stream = stream; + camu_video_buffer_configure(&entry->video.buf, &stream->stream); + aki_mutex_lock(&entry->sink->mutex); + if (entry->video.state == BUFFER_QUEUED) { + entry->video.state = BUFFER_SET_OR_BUFFERED; + } else { + entry->video.state = BUFFER_CONFIGURED; + } + aki_mutex_unlock(&entry->sink->mutex); + break; +#endif + } + break; + } + case BIMU_CLIENT_DATA: { + struct camu_frame *frame = (struct camu_frame *)opaque; + switch (stream->type) { + case BIMU_STREAM_AUDIO: + camu_audio_buffer_push(&entry->audio.buf, frame); + break; +#ifndef CAMU_SINK_NO_VIDEO + case BIMU_STREAM_VIDEO: + camu_video_buffer_push(&entry->video.buf, frame); + break; +#endif + default: +#ifdef HAVE_FFMPEG + if (frame->type == CAMU_FFMPEG_COMPAT) { + av_frame_free(&frame->av.frame); + } +#endif + al_free(frame); + break; + } + break; + } + case BIMU_CLIENT_SEEK: { + // Must ensure no more data from the previous + // stream position is sent during or after this call to seek. + f64 seek_pos = *(f64 *)opaque; + aki_mutex_lock(&entry->sink->mutex); + if (entry->audio.state == BUFFER_ADDED) { + entry->sink->callback(entry->sink->userdata, CAMU_SINK_REMOVE_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf); + entry->audio.state = BUFFER_SET_OR_BUFFERED; + } +#ifndef CAMU_SINK_NO_VIDEO + bool single_frame = camu_video_buffer_is_single_frame(&entry->video.buf); + if (entry->video.state == BUFFER_ADDED && !single_frame) { + entry->sink->callback(entry->sink->userdata, CAMU_SINK_REMOVE_BUFFER, CAMU_SINK_VIDEO, &entry->video.buf); + entry->video.state = BUFFER_SET_OR_BUFFERED; + } +#endif + aki_mutex_unlock(&entry->sink->mutex); +#ifndef CAMU_SINK_NO_VIDEO + if (single_frame) { + while (ENTRY_AUDIO_BUFFER_HELD(entry)) { + aki_thread_sleep(AKI_TS_FROM_USEC(200)); + } + } else { +#endif + while (ENTRY_BUFFERS_HELD(entry)) { + aki_thread_sleep(AKI_TS_FROM_USEC(200)); + } +#ifndef CAMU_SINK_NO_VIDEO + } +#endif + camu_clock_seek(&entry->clock, seek_pos); + camu_audio_buffer_reset(&entry->audio.buf); +#ifndef CAMU_SINK_NO_VIDEO + if (!single_frame) { + camu_video_buffer_reset(&entry->video.buf); + } +#endif + break; + } + case BIMU_CLIENT_EOF: { + switch (stream->type) { + case BIMU_STREAM_AUDIO: { + camu_audio_buffer_flush(&entry->audio.buf); + break; + } +#ifndef CAMU_SINK_NO_VIDEO + case BIMU_STREAM_VIDEO: { + bool single_frame = camu_video_buffer_is_single_frame(&entry->video.buf); + if (!single_frame) { + camu_video_buffer_flush(&entry->video.buf); + } + break; + } +#endif + } + break; + } + case BIMU_CLIENT_CLOSED: { + // close/free client. + camu_audio_buffer_free(&entry->audio.buf); +#ifndef CAMU_SINK_NO_VIDEO + camu_video_buffer_free(&entry->video.buf); +#endif + al_free(entry); + al_log_debug("sink", "Entry closed."); + break; + } + } +} + +bool camu_sink_init(struct camu_sink *sink, struct aki_event_loop *loop, + struct camu_mixer *mixer +#ifndef CAMU_SINK_NO_VIDEO + , struct camu_renderer *renderer +#endif + ) +{ + sink->loop = loop; + aki_signal_init(&sink->signal, queue_signal_callback, sink); + aki_signal_start(&sink->signal, sink->loop); + camu_queue_init(sink->queue); + aki_mutex_init(&sink->mutex); + sink->current = NULL; + al_array_init(sink->entries); + sink->audio.mixer = mixer; + sink->audio.state = SINK_PAUSED; +#ifndef CAMU_SINK_NO_VIDEO + sink->video.renderer = renderer; + sink->video.state = SINK_PAUSED; +#endif + return true; +} + +static struct camu_sink_entry *entry_from_node_id(struct camu_sink *sink, u16 node_id) +{ + struct camu_sink_entry *entry; + al_array_foreach(sink->entries, i, entry) { + if (entry->client.node_id == node_id) return entry; + } + return NULL; +} + +static bool buffer_command_callback(void *userdata, struct aki_rpc_connection *conn, + struct aki_packet *packet, struct aki_packet *rpacket) +{ + struct camu_sink *sink = (struct camu_sink *)userdata; + (void)conn; + (void)rpacket; + + str addr; + aki_packet_read_str(packet, &addr); + s32 port = aki_packet_read_s32(packet); + u16 node_id = aki_packet_read_u16(packet); + + struct camu_sink_entry *entry = al_alloc_object(struct camu_sink_entry); + entry->sink = sink; + al_array_push(sink->entries, entry); + + entry->state = ENTRY_LOADED; + camu_clock_init(&entry->clock); + + entry->audio.state = BUFFER_INIT; + camu_audio_buffer_init(&entry->audio.buf, &entry->clock, sink->audio.mixer); + entry->audio.buf.callback = audio_buffer_callback; + entry->audio.buf.userdata = entry; + + struct camu_renderer *renderer = NULL; +#ifndef CAMU_SINK_NO_VIDEO + renderer = sink->video.renderer; + entry->video.state = BUFFER_INIT; + camu_video_buffer_init(&entry->video.buf, &entry->clock, renderer); + entry->video.buf.callback = video_buffer_callback; + entry->video.buf.userdata = entry; +#endif + + entry->client.callback = client_callback; + entry->client.userdata = entry; + bmu_client_connect(&entry->client, sink->loop, &addr, port, node_id); + + aki_packet_free(packet); + return false; +} + +static void remove_entry_buffers(struct camu_sink *sink, struct camu_sink_entry *entry) +{ +#ifndef CAMU_SINK_NO_VIDEO + if (entry->video.state == BUFFER_ADDED) { + sink->callback(sink->userdata, CAMU_SINK_REMOVE_BUFFER, CAMU_SINK_VIDEO, &entry->video.buf); + entry->video.state = BUFFER_SET_OR_BUFFERED; + } +#endif + if (entry->audio.state == BUFFER_ADDED) { + sink->callback(sink->userdata, CAMU_SINK_REMOVE_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf); + entry->audio.state = BUFFER_SET_OR_BUFFERED; + } +} + +static bool set_command_callback(void *userdata, struct aki_rpc_connection *conn, + struct aki_packet *packet, struct aki_packet *rpacket) +{ + struct camu_sink *sink = (struct camu_sink *)userdata; + (void)conn; + (void)rpacket; + + u16 node_id = aki_packet_read_u16(packet); + aki_mutex_lock(&sink->mutex); + struct camu_sink_entry *entry = entry_from_node_id(sink, node_id); + struct camu_sink_entry *previous = sink->current; + sink->current = entry; + if (entry->audio.state == BUFFER_INIT) { + entry->audio.state = BUFFER_QUEUED; + } else { + add_audio_if_set_and_buffered(entry); + } +#ifndef CAMU_SINK_NO_VIDEO + if (entry->video.state == BUFFER_INIT) { + entry->video.state = BUFFER_QUEUED; + } else { + add_video_if_set_and_buffered(entry); + } +#endif + if (previous) { + camu_clock_pause(&previous->clock); + remove_entry_buffers(sink, previous); + } + aki_mutex_unlock(&sink->mutex); + + aki_packet_free(packet); + return false; +} + +static struct aki_rpc_command commands[] = { + { .op = CAMU_SINK_CMD_BUFFER, .callback = buffer_command_callback, .userdata = NULL }, + { .op = CAMU_SINK_CMD_SET, .callback = set_command_callback, .userdata = NULL } +}; + +static void connection_callback(void *userdata, struct aki_rpc_connection *conn) +{ + struct camu_sink *sink = (struct camu_sink *)userdata; + sink->conn = conn; + struct aki_packet *packet = aki_rpc_get_packet(&sink->client, TREE_CMD_IDENTIFY); + aki_packet_write_u8(packet, TREE_SINK); + aki_rpc_connection_command(sink->conn, packet, NULL, NULL); +} + +static void connection_closed_callback(void *userdata, struct aki_rpc_connection *conn) +{ + (void)userdata; + (void)conn; +} + +bool camu_sink_connect(struct camu_sink *sink, str *addr, s32 port) +{ + aki_rpc_init(&sink->client, AKI_SOCKET_TCP, connection_callback, + connection_closed_callback, sink); + for (u32 i = 0; i < AL_ARRAY_SIZE(commands); i++) { + commands[i].userdata = sink; + al_assert(sink->callback); + aki_rpc_add_command(&sink->client, &commands[i]); + } + return aki_rpc_connect(&sink->client, sink->loop, addr, port); +} + +void camu_sink_toggle_pause(struct camu_sink *sink) +{ + aki_mutex_lock(&sink->mutex); + if (sink->current) { + queue_cmd(sink, (struct camu_sink_cmd){ + .op = TOGGLE_PAUSE, + .opaque = sink->current + }); + } + aki_mutex_unlock(&sink->mutex); +} + +void camu_sink_seek(struct camu_sink *sink, f64 pos) +{ + aki_mutex_lock(&sink->mutex); + if (sink->current) { + queue_cmd(sink, (struct camu_sink_cmd){ + .op = SEEK, + .value.f = pos, + .opaque = sink->current + }); + } + aki_mutex_unlock(&sink->mutex); +} + +void camu_sink_skip(struct camu_sink *sink, s32 n) +{ + (void)sink; + (void)n; +} + +struct camu_sink_entry *camu_sink_get_current(struct camu_sink *sink) +{ + aki_mutex_lock(&sink->mutex); + return sink->current; +} + +void camu_sink_return_current(struct camu_sink *sink) +{ + aki_mutex_unlock(&sink->mutex); +} + +void camu_sink_stop(struct camu_sink *sink) +{ + aki_mutex_lock(&sink->mutex); + if (sink->current) { + remove_entry_buffers(sink, sink->current); + sink->current = NULL; + } + aki_mutex_unlock(&sink->mutex); + queue_cmd(sink, (struct camu_sink_cmd){ + .op = STOP, + .value.i = CAMU_SINK_AUDIO + }); + queue_cmd(sink, (struct camu_sink_cmd){ + .op = CLOSE + }); +} + +void camu_sink_close(struct camu_sink *sink) +{ + aki_mutex_lock(&sink->mutex); + /* + struct camu_sink_entry *entry; + al_array_foreach(sink->entries, i, entry) { + } + */ + al_array_free(sink->entries); + aki_mutex_unlock(&sink->mutex); + aki_signal_stop(&sink->signal); + camu_queue_free(sink->queue); + aki_mutex_destroy(&sink->mutex); +} diff --git a/src/libsink/sink2.h b/src/libsink/sink2.h new file mode 100644 index 0000000..95c280f --- /dev/null +++ b/src/libsink/sink2.h @@ -0,0 +1,106 @@ +#pragma once + +//#define CAMU_SINK_NO_VIDEO + +#include <al/str.h> +#include <al/array.h> +#include <aki/rpc2.h> +#include <aki/signal.h> +#include <aki/timer.h> + +#include "../util/queue.h" + +#include "../buffer/clock.h" +#include "../buffer/audio.h" +#ifndef CAMU_SINK_NO_VIDEO +#include "../buffer/video.h" +#endif + +#include "../bimu/client.h" + +#include "common.h" + +enum { + CAMU_SINK_AUDIO = 0, +#ifndef CAMU_SINK_NO_VIDEO + CAMU_SINK_VIDEO +#endif +}; + +enum { + CAMU_SINK_ADD_BUFFER = 0, + CAMU_SINK_REMOVE_BUFFER, + CAMU_SINK_SWAP_BUFFER, + CAMU_SINK_SET_BUFFERED, + CAMU_SINK_START, + CAMU_SINK_STOP, + CAMU_SINK_EXIT +}; + +struct camu_sink_entry { + u8 state; + struct bmu_client client; + struct camu_clock clock; + struct { + u8 state; + struct camu_audio_buffer buf; + struct bmu_client_stream *stream; + } audio; +#ifndef CAMU_SINK_NO_VIDEO + struct { + u8 state; + struct camu_video_buffer buf; + struct bmu_client_stream *stream; + } video; +#endif + struct camu_sink *sink; +}; + +struct camu_sink_cmd { + u8 op; + union { s64 i; f64 f; } value; + void *opaque; +}; + +enum { + CAMU_SINK_OK = 0, + CAMU_SINK_BUFFERS_SWAPPED = 1 +}; + +struct camu_sink { + struct aki_event_loop *loop; + struct aki_rpc client; + struct aki_rpc_connection *conn; + struct aki_signal signal; + queue(struct camu_sink_cmd) queue; + struct aki_mutex mutex; + struct camu_sink_entry *current; + array(struct camu_sink_entry *) entries; + struct { + u8 state; + struct camu_mixer *mixer; + } audio; +#ifndef CAMU_SINK_NO_VIDEO + struct { + u8 state; + struct camu_renderer *renderer; + } video; +#endif + u8 (*callback)(void *, u8, u8, void *); + void *userdata; +}; + +bool camu_sink_init(struct camu_sink *sink, struct aki_event_loop *loop, + struct camu_mixer *mixer +#ifndef CAMU_SINK_NO_VIDEO + , struct camu_renderer *renderer +#endif + ); +bool camu_sink_connect(struct camu_sink *sink, str *addr, s32 port); +void camu_sink_toggle_pause(struct camu_sink *sink); +void camu_sink_seek(struct camu_sink *sink, f64 pos); +void camu_sink_skip(struct camu_sink *sink, s32 n); +struct camu_sink_entry *camu_sink_get_current(struct camu_sink *sink); +void camu_sink_return_current(struct camu_sink *sink); +void camu_sink_stop(struct camu_sink *sink); +void camu_sink_close(struct camu_sink *sink); diff --git a/src/mixer/audio_miniaudio.c b/src/mixer/audio_miniaudio.c index 1388fc4..38cb91d 100644 --- a/src/mixer/audio_miniaudio.c +++ b/src/mixer/audio_miniaudio.c @@ -179,12 +179,12 @@ struct camu_audio_miniaudio audio_plugin_miniaudio = { .stop = audio_miniaudio_stop, .free = audio_miniaudio_free }, - .log = { }, - .context_config = { }, - .context = { }, + .log = { 0 }, + .context_config = { 0 }, + .context = { 0 }, .infos = NULL, .info_count = 0, .selected_device = NULL, - .config = { }, - .device = { } + .config = { 0 }, + .device = { 0 } }; diff --git a/src/mixer/meson.build b/src/mixer/meson.build index 4053210..53389eb 100644 --- a/src/mixer/meson.build +++ b/src/mixer/meson.build @@ -9,4 +9,17 @@ if host_machine.system() != 'windows' mixer_deps += [libpulse] endif -mixer = declare_dependency(sources: mixer_src, dependencies: mixer_deps) +miniaudio_args = [ + '-DMA_NO_JACK', + '-DMA_NO_ALSA', + '-DMA_NO_DECODING', + '-DMA_NO_ENCODING', + '-DMA_NO_GENERATION', + '-DMA_NO_RESOURCE_MANAGER', + '-DMA_NO_NODE_GRAPH', + '-DMA_NO_ENGINE', + '-DMA_NO_RUNTIME_LINKING' +] + +mixer = declare_dependency(sources: mixer_src, dependencies: mixer_deps, + compile_args: miniaudio_args) diff --git a/src/mixer/mixer.c b/src/mixer/mixer.c index 8cfc2cc..646f056 100644 --- a/src/mixer/mixer.c +++ b/src/mixer/mixer.c @@ -60,6 +60,7 @@ void camu_mixer_pick_format(struct camu_mixer *mixer, struct camu_lav_resample_f f64 camu_mixer_get_latency(struct camu_mixer *mixer) { + if (!mixer->audio) return 0; return mixer->audio->get_latency(mixer->audio) / 1000000.0; } diff --git a/src/render/meson.build b/src/render/meson.build index cc11eea..e9c093a 100644 --- a/src/render/meson.build +++ b/src/render/meson.build @@ -23,7 +23,7 @@ libplacebo = dependency('libplacebo', 'd3d11=disabled', 'gl-proc-addr=disabled', 'vk-proc-addr=disabled', - 'lcms=disabled', + 'lcms=enabled', 'dovi=disabled', 'libdovi=disabled' ], required: false, allow_fallback: true) diff --git a/src/render/queue_libplacebo.c b/src/render/queue_libplacebo.c index ee64920..327397f 100644 --- a/src/render/queue_libplacebo.c +++ b/src/render/queue_libplacebo.c @@ -1,5 +1,9 @@ #include <al/lib.h> #include <al/log.h> +#ifdef HAVE_FFMPEG +#define PL_LIBAV_IMPLEMENTATION 1 +#include <libplacebo/utils/libav.h> +#endif #include "../buffer/video.h" diff --git a/src/render/queue_libplacebo.h b/src/render/queue_libplacebo.h index 7da99ca..829ef87 100644 --- a/src/render/queue_libplacebo.h +++ b/src/render/queue_libplacebo.h @@ -4,9 +4,6 @@ #include <libplacebo/utils/frame_queue.h> #include <libplacebo/utils/upload.h> -#ifdef HAVE_FFMPEG -#include <libplacebo/utils/libav.h> -#endif struct camu_frame_queue_lp { struct camu_frame_queue q; diff --git a/src/render/renderer.h b/src/render/renderer.h index 61965ed..64dc691 100644 --- a/src/render/renderer.h +++ b/src/render/renderer.h @@ -18,7 +18,6 @@ struct camu_screen; struct camu_renderer { bool (*create_renderer)( struct camu_renderer *, s32 *, s32 *, - bool (*poll)(struct skh_window *, bool), #if defined CAMU_RENDERER_VULKAN VKAPI_ATTR PFN_vkVoidFunction VKAPI_CALL (*get_vk_proc_address)(VkInstance, const char *), VkResult (*vk_create_surface)(void *, VkInstance, VkSurfaceKHR *), diff --git a/src/render/renderer_libplacebo.c b/src/render/renderer_libplacebo.c index 019a5ce..148574b 100644 --- a/src/render/renderer_libplacebo.c +++ b/src/render/renderer_libplacebo.c @@ -1,6 +1,9 @@ #include <al/lib.h> #include <al/log.h> #include <aki/file.h> +#ifdef HAVE_FFMPEG +#include <libplacebo/utils/libav.h> +#endif #include "../screen/screen.h" #include "../util/color_palette.h" @@ -16,8 +19,14 @@ static void renderer_lp_resize(struct camu_renderer *renderer, s32 *width, s32 * if (lr->swapchain) pl_swapchain_resize(lr->swapchain, width, height); } +static void log_callback(void *userdata, enum pl_log_level level, const char *msg) +{ + (void)userdata; + (void)level; + al_log_debug("render_libplacebo", msg); +} + static bool renderer_lp_create_renderer(struct camu_renderer *renderer, s32 *width, s32 *height, - bool (*poll)(struct skh_window *, bool), #if defined CAMU_RENDERER_VULKAN VKAPI_ATTR PFN_vkVoidFunction VKAPI_CALL (*get_vk_proc_address)(VkInstance, const char *), VkResult (*vk_create_surface)(void *, VkInstance, VkSurfaceKHR *), @@ -38,29 +47,32 @@ static bool renderer_lp_create_renderer(struct camu_renderer *renderer, s32 *wid clear_color[2] = ((global_color_palette.colors[0]) & 0xff) / 255.f; lr->logger = pl_log_create(PL_API_VER, pl_log_params( - .log_cb = pl_log_color, - .log_level = PL_LOG_NONE, -// .log_level = PL_LOG_INFO, + .log_cb = log_callback, + .log_priv = NULL, + .log_level = PL_LOG_INFO )); #if defined CAMU_RENDERER_VULKAN VkResult err; u32 num; + const char *const *extensions = vk_get_extensions(&num); lr->vk_inst = pl_vk_inst_create(lr->logger, pl_vk_inst_params( .get_proc_addr = get_vk_proc_address, .debug = false, - .extensions = vk_get_extensions(&num), + .extensions = extensions, .num_extensions = num, )); if (!lr->vk_inst) { al_log_error("render_libplacebo", "Failed to create vulkan instance."); + goto err; } err = vk_create_surface(priv, lr->vk_inst->instance, &lr->surface); if (err != VK_SUCCESS) { al_log_error("render_libplacebo", "Failed to create vulkan surface."); + goto err; } lr->vk = pl_vulkan_create(lr->logger, pl_vulkan_params( @@ -72,6 +84,7 @@ static bool renderer_lp_create_renderer(struct camu_renderer *renderer, s32 *wid if (!lr->vk) { al_log_error("render_libplacebo", "Failed to create vulkan device."); + goto err; } lr->swapchain = pl_vulkan_create_swapchain(lr->vk, pl_vulkan_swapchain_params( @@ -81,6 +94,7 @@ static bool renderer_lp_create_renderer(struct camu_renderer *renderer, s32 *wid if (!lr->swapchain) { al_log_error("render_libplacebo", "Failed to create vulkan swapchain."); + goto err; } lr->gpu = lr->vk->gpu; @@ -98,6 +112,7 @@ static bool renderer_lp_create_renderer(struct camu_renderer *renderer, s32 *wid if (!lr->gl) { al_log_error("render_libplacebo", "Failed to create GL device."); + goto err; } lr->swapchain = pl_opengl_create_swapchain(lr->gl, pl_opengl_swapchain_params( @@ -107,6 +122,7 @@ static bool renderer_lp_create_renderer(struct camu_renderer *renderer, s32 *wid if (!lr->swapchain) { al_log_error("render_libplacebo", "Failed to create GL swapchain."); + goto err; } lr->gpu = lr->gl->gpu; @@ -114,32 +130,24 @@ static bool renderer_lp_create_renderer(struct camu_renderer *renderer, s32 *wid lr->renderer = pl_renderer_create(lr->logger, lr->gpu); - renderer_lp_resize(renderer, width, height); - -#define PRE_SWAPS 2 - // See renderer_tiger.c:122. - for (u32 i = 0; i < PRE_SWAPS; i++) { - poll(priv, true); - struct pl_swapchain_frame frame; - pl_swapchain_start_frame(lr->swapchain, &frame); - struct pl_frame target; - pl_frame_from_swapchain(&target, &frame); - pl_frame_clear_rgba(lr->gpu, &target, clear_color); - pl_swapchain_submit_frame(lr->swapchain); - pl_swapchain_swap_buffers(lr->swapchain); - } + lr->r.resize(&lr->r, width, height); al_memset(&lr->params, 0, sizeof(struct pl_render_params)); lr->params = pl_render_default_params; //lr->params = pl_render_high_quality_params; - lr->params.upscaler = &pl_filter_bicubic; + //lr->params.upscaler = &pl_filter_bicubic; + //lr->params.upscaler = &pl_filter_nearest; + // Clear manually so we can draw multiple images per frame. lr->params.skip_target_clearing = true; + // Breaks zoom/pan. lr->params.skip_caching_single_frame = true; + // Prioritize a more consistant image (should be configurable). + lr->params.correct_subpixel_offsets = true; #if 0 struct aki_file file; - if (aki_file_open(&file, al_str_c(""))) { + if (aki_file_open(&file, al_str_c(""), 0)) { char *str; size_t len = aki_file_read_as_c_str(&file, &str); const struct pl_hook *hook = pl_mpv_user_shader_parse(lr->gpu, (const char *)str, len); @@ -157,6 +165,9 @@ static bool renderer_lp_create_renderer(struct camu_renderer *renderer, s32 *wid #endif return true; +err: + lr->r.free((struct camu_renderer **)&lr); + return false; } static struct camu_frame_queue *renderer_lp_create_queue(struct camu_renderer *renderer) @@ -166,9 +177,9 @@ static struct camu_frame_queue *renderer_lp_create_queue(struct camu_renderer *r struct camu_frame_queue_lp *lq = (struct camu_frame_queue_lp *)queue; lq->queue = pl_queue_create(lr->gpu); al_memset(&lq->params, 0, sizeof(struct pl_queue_params)); - lq->params.interpolation_threshold = 0.01; + lq->params.interpolation_threshold = 0.01f; lq->params.radius = pl_frame_mix_radius(&lr->params); - lq->params.timeout = 0; // vsync + lq->params.timeout = 0uL; // vsync return queue; } @@ -211,7 +222,7 @@ static void renderer_lp_render(struct camu_renderer *renderer, struct camu_scree pl_swapchain_swap_buffers(lr->swapchain); } -static void renderer_lp_free(struct camu_renderer **renderer) +void renderer_lp_free(struct camu_renderer **renderer) { struct camu_renderer_lp *lr = (struct camu_renderer_lp *)*renderer; if (lr->renderer) pl_renderer_destroy(&lr->renderer); diff --git a/src/render/renderer_libplacebo.h b/src/render/renderer_libplacebo.h index 661faff..80f4688 100644 --- a/src/render/renderer_libplacebo.h +++ b/src/render/renderer_libplacebo.h @@ -1,7 +1,6 @@ #pragma once #include <libplacebo/common.h> -#include <libplacebo/opengl.h> #include <libplacebo/renderer.h> #include <libplacebo/shaders/lut.h> #include <libplacebo/colorspace.h> @@ -10,13 +9,10 @@ #if defined CAMU_RENDERER_VULKAN #include <libplacebo/vulkan.h> +#elif defined CAMU_RENDERER_OPENGL +#include <libplacebo/opengl.h> #endif -struct camu_renderer_lp_resource { - pl_tex tex; - struct pl_plane plane; -}; - struct camu_renderer_lp { struct camu_renderer r; #if defined CAMU_RENDERER_VULKAN @@ -31,8 +27,6 @@ struct camu_renderer_lp { pl_swapchain swapchain; pl_renderer renderer; struct pl_render_params params; - struct pl_overlay overlays[10]; - s32 overlay_count; }; struct camu_renderer *camu_renderer_lp_create(void); diff --git a/src/render/renderer_tiger.c b/src/render/renderer_tiger.c index 7cbbbb9..e4ae239 100644 --- a/src/render/renderer_tiger.c +++ b/src/render/renderer_tiger.c @@ -96,7 +96,6 @@ static void renderer_tiger_resize(struct camu_renderer *renderer, s32 *width, s3 } static bool renderer_tiger_create_renderer(struct camu_renderer *renderer, s32 *width, s32 *height, - bool (*poll)(struct skh_window *, bool), #if defined CAMU_RENDERER_VULKAN VKAPI_ATTR PFN_vkVoidFunction VKAPI_CALL (*get_vk_proc_address)(VkInstance, const char *), void (*vk_create_surface)(void *, VkInstance, VkSurfaceKHR *), @@ -132,14 +131,6 @@ static bool renderer_tiger_create_renderer(struct camu_renderer *renderer, s32 * glClearColor(clear_color[0], clear_color[1], clear_color[2], 1.f); -#define PRE_SWAPS 2 - // Hack? to make the window appear quicker. Only tested on linux/wayland. - for (u32 i = 0; i < PRE_SWAPS; i++) { - poll(priv, true); - glClear(GL_COLOR_BUFFER_BIT); - tr->swap_buffers(tr->priv); - } - glEnable(GL_BLEND); glBlendFunc(GL_SRC_ALPHA, GL_ONE_MINUS_SRC_ALPHA); diff --git a/src/screen/meson.build b/src/screen/meson.build index cd78988..d0bbd50 100644 --- a/src/screen/meson.build +++ b/src/screen/meson.build @@ -1,4 +1,10 @@ screen_src = ['screen.c', 'view.c'] -sekihi = subproject('sekihi').get_variable('sekihi') -screen = declare_dependency(sources: screen_src, - dependencies: [sekihi]) +sekihi = subproject('sekihi', + default_options: [ + 'api=opengl', + 'window=wayland', + 'poll=inline', + 'event-buffer=false', + 'pause=true' + ]).get_variable('sekihi') +screen = declare_dependency(sources: screen_src, dependencies: [sekihi]) diff --git a/src/screen/screen.c b/src/screen/screen.c index 82c7b12..dd4bdc6 100644 --- a/src/screen/screen.c +++ b/src/screen/screen.c @@ -3,18 +3,6 @@ #include "screen.h" -#define ZOOM_MAX 100.0 -#define ZOOM_MIN 0.02 - -#define PAN_BOUNDS 35.0 - -#define X_IN_BOUNDS(scr, view, x) \ - ((x + PAN_BOUNDS <= scr->width) && (PAN_BOUNDS - x <= view->width * view->zoom)) -#define Y_IN_BOUNDS(scr, view, y) \ - ((y + PAN_BOUNDS <= scr->height) && (PAN_BOUNDS - y <= view->height * view->zoom)) - -#define CAMU_DEFAULT_VIEW CAMU_VIEW_AUTOFIT - static struct camu_view *get_view_from_mouse_pos(struct camu_screen *scr) { struct camu_view *view = NULL; @@ -51,31 +39,30 @@ static void resize_callback(void *userdata, s32 width, s32 height) } } -static void pointer_pos_callback(void *userdata, f64 x, f64 y) +static bool pointer_pos_callback(void *userdata, f64 x, f64 y) { struct camu_screen *scr = (struct camu_screen *)userdata; struct camu_view *view = get_view_from_mouse_pos(scr); + bool queue_refresh = false; if (view) { if (scr->flags & CAMU_SCREEN_DRAGGING) { if (scr->flags & CAMU_SCREEN_ZOOM_PAN_SIMPLE) { - f64 nx = view->x_offset + x - scr->last_mouse_x; - if (X_IN_BOUNDS(scr, view, nx)) { - view->x_offset = nx; + f64 dx = x - scr->last_mouse_x; + f64 dy = y - scr->last_mouse_y; + if (camu_view_pan_simple(view, scr->width, scr->height, dx, dy)) { + view->mode = CAMU_VIEW_DETACHED; + queue_refresh = true; } - f64 ny = view->y_offset + y - scr->last_mouse_y; - if (Y_IN_BOUNDS(scr, view, ny)) { - view->y_offset = ny; - } - view->mode = CAMU_VIEW_DETACHED; } scr->last_click_ts = 0; } } scr->last_mouse_x = x; scr->last_mouse_y = y; + return queue_refresh; } -static void mouse_button_callback(void *userdata, u8 state, u8 button) +static bool mouse_button_callback(void *userdata, u8 state, u8 button) { struct camu_screen *scr = (struct camu_screen *)userdata; switch (button) { @@ -106,55 +93,30 @@ static void mouse_button_callback(void *userdata, u8 state, u8 button) default: break; } + return false; } -static void zoom_on_point(struct camu_screen *scr, struct camu_view *view, f64 x, f64 y, f64 v) -{ - x -= view->x_offset; - y -= view->y_offset; - x /= view->zoom; - y /= view->zoom; - - f64 prev_width = scr->width * view->zoom; - f64 prev_height = scr->height * view->zoom; - - f64 x_ratio = x / (f64)scr->width; - f64 y_ratio = y / (f64)scr->height; - - f64 zoom = view->zoom + (v * view->zoom); - if (zoom > ZOOM_MAX || zoom < ZOOM_MIN) return; - - f64 nx = view->x_offset - x_ratio * ((scr->width * zoom) - prev_width); - f64 ny = view->y_offset - y_ratio * ((scr->height * zoom) - prev_height); - if (X_IN_BOUNDS(scr, view, nx) && Y_IN_BOUNDS(scr, view, ny)) { - view->x_offset = nx; - view->y_offset = ny; - view->zoom = zoom; - } -} +#if defined SEKIHI_WINDOW_WAYLAND +#define SCROLL_MULTIPLIER 87.5 +#else +#define SCROLL_MULTIPLIER 5.0 +#endif -static void scroll_callback(void *userdata, f64 y) +static bool scroll_callback(void *userdata, f64 y) { struct camu_screen *scr = (struct camu_screen *)userdata; - struct camu_view *view = get_view_from_mouse_pos(scr); - if (!view) return; - - if (scr->flags & CAMU_SCREEN_ZOOM_PAN_SIMPLE) { - y = -y; -#if defined SEKIHI_WINDOW_GLFW - y /= 4.5; -#elif defined SEKIHI_WINDOW_WAYLAND - y /= 100.0; -#else - y /= 14.0; -#endif - zoom_on_point(scr, view, scr->last_mouse_x, scr->last_mouse_y, y); - view->mode = CAMU_VIEW_DETACHED; + if (view && scr->flags & CAMU_SCREEN_ZOOM_PAN_SIMPLE) { + y = -y / SCROLL_MULTIPLIER; + if (camu_view_zoom_simple(view, scr->width, scr->height, scr->last_mouse_x, scr->last_mouse_y, y)) { + view->mode = CAMU_VIEW_DETACHED; + return true; + } } + return false; } -static void key_callback(void *userdata, u8 state, u8 button) +static bool key_callback(void *userdata, u8 state, u8 button) { struct camu_screen *scr = (struct camu_screen *)userdata; switch (state) { @@ -183,9 +145,11 @@ static void key_callback(void *userdata, u8 state, u8 button) case 0x1b: case 'r': { struct camu_view *view = get_view_from_mouse_pos(scr); - if (!view) return; - view->mode = CAMU_DEFAULT_VIEW; - camu_view_calculate(view, scr->width, scr->height); + if (view) { + view->mode = CAMU_DEFAULT_VIEW; + camu_view_calculate(view, scr->width, scr->height); + return true; + } break; } case 0x1d: @@ -204,6 +168,7 @@ static void key_callback(void *userdata, u8 state, u8 button) default: break; } + return false; } static void key_immediate_callback(void *userdata, u8 state, u8 button) @@ -251,7 +216,8 @@ bool camu_screen_init(struct camu_screen *scr) bool camu_screen_create_window(struct camu_screen *scr, const char *name) { - if (!scr->window->create_window(scr->window, CAMU_SCREEN_WIDTH, CAMU_SCREEN_HEIGHT, name)) { + s32 flags = SEKIHI_WINDOW_VSYNC; + if (!scr->window->create_window(scr->window, CAMU_SCREEN_WIDTH, CAMU_SCREEN_HEIGHT, NULL, name, flags)) { return false; } scr->width = scr->window->width; @@ -263,14 +229,13 @@ bool camu_screen_create_renderer(struct camu_screen *scr, struct camu_renderer * { scr->renderer = renderer; return renderer->create_renderer(renderer, &scr->width, &scr->height, - scr->window->poll, #if defined CAMU_RENDERER_VULKAN scr->window->get_vk_proc_address, scr->window->vk_create_surface, scr->window->vk_get_extensions, #elif defined CAMU_RENDERER_OPENGL scr->window->get_gl_proc_address, - scr->window->gl_load_loader, + scr->window->gl_loader_load, scr->window->gl_make_current, scr->window->gl_release_current, scr->window->gl_swap_buffers, @@ -367,28 +332,30 @@ void camu_screen_run_queue(struct camu_screen *scr) } #endif -void camu_screen_set_state(struct camu_screen *scr, u8 state) +void camu_screen_set_state(struct camu_screen *scr, s32 state) { al_atomic_s32_store(&scr->state, state, AL_ATOMIC_RELAXED); } -bool camu_screen_tick(struct camu_screen *scr) +bool camu_screen_poll(struct camu_screen *scr, bool block) { -#if defined SEKIHI_POLL_CONSERVATIVE - scr->window->poll(scr->window, - al_atomic_s32_load(&scr->state, AL_ATOMIC_RELAXED) == CAMU_SCREEN_PAUSED); - return skh_window_read_events(scr->window); -#elif defined SEKIHI_POLL_INLINE - return scr->window->poll(scr->window, - al_atomic_s32_load(&scr->state, AL_ATOMIC_RELAXED) == CAMU_SCREEN_PAUSED); + scr->window->poll(scr->window, block); + scr->window->process_events(scr->window); +#if defined SEKIHI_EVENT_BUFFER + skh_window_read_events(scr->window); #endif + return scr->window->should_refresh(scr->window); +} + +bool camu_screen_tick(struct camu_screen *scr) +{ + bool paused = al_atomic_s32_load(&scr->state, AL_ATOMIC_RELAXED) == CAMU_SCREEN_PAUSED; + return camu_screen_poll(scr, paused) || !paused; } void camu_screen_wake(struct camu_screen *scr) { -#ifdef SEKIHI_PAUSE scr->window->wake(scr->window); -#endif } void camu_screen_close(struct camu_screen *scr) @@ -399,4 +366,5 @@ void camu_screen_close(struct camu_screen *scr) aki_mutex_destroy(&scr->mutex); #endif al_array_free(scr->videos); + scr->window->free(&scr->window); } diff --git a/src/screen/screen.h b/src/screen/screen.h index 798fc8e..0ca5779 100644 --- a/src/screen/screen.h +++ b/src/screen/screen.h @@ -19,9 +19,9 @@ #define CAMU_SCREEN_HEIGHT 700 enum { - CAMU_SCREEN_ZOOM_PAN_SIMPLE = 1, + CAMU_SCREEN_MODIFIER = 1, CAMU_SCREEN_DRAGGING = 1 << 1, - CAMU_SCREEN_MODIFIER = 1 << 2, + CAMU_SCREEN_ZOOM_PAN_SIMPLE = 1 << 2, }; enum { @@ -72,7 +72,8 @@ void camu_screen_remove_buffer(struct camu_screen *scr, struct camu_video_buffer #ifdef CAMU_SCREEN_THREADED void camu_screen_run_queue(struct camu_screen *scr); #endif -void camu_screen_set_state(struct camu_screen *scr, u8 state); +void camu_screen_set_state(struct camu_screen *scr, s32 state); +bool camu_screen_poll(struct camu_screen *scr, bool block); bool camu_screen_tick(struct camu_screen *scr); void camu_screen_wake(struct camu_screen *scr); void camu_screen_close(struct camu_screen *scr); diff --git a/src/screen/view.c b/src/screen/view.c index cf23ed4..4cf6007 100644 --- a/src/screen/view.c +++ b/src/screen/view.c @@ -1,5 +1,15 @@ #include "view.h" +#define ZOOM_MAX 100.0 +#define ZOOM_MIN 0.02 + +#define PAN_BOUNDS 35.0 + +#define X_IN_BOUNDS(view, w, x) \ + ((x + PAN_BOUNDS <= w) && (PAN_BOUNDS - x <= view->width * view->zoom)) +#define Y_IN_BOUNDS(view, h, y) \ + ((y + PAN_BOUNDS <= h) && (PAN_BOUNDS - y <= view->height * view->zoom)) + void camu_view_calculate(struct camu_view *view, s32 window_width, s32 window_height) { f64 ratio = window_width / (f64)window_height; @@ -26,3 +36,48 @@ void camu_view_calculate(struct camu_view *view, s32 window_width, s32 window_he view->x_offset = ((window_width - (view->width * view->zoom)) / 2.0); view->y_offset = ((window_height - (view->height * view->zoom)) / 2.0); } + +bool camu_view_pan_simple(struct camu_view *view, s32 window_width, s32 window_height, f64 dx, f64 dy) +{ + f64 nx = view->x_offset + dx; + f64 ny = view->y_offset + dy; + bool view_changed = false; + if (X_IN_BOUNDS(view, window_width, nx)) { + view->x_offset = nx; + view_changed = true; + } + if (Y_IN_BOUNDS(view, window_height, ny)) { + view->y_offset = ny; + view_changed = true; + } + return view_changed; +} + +bool camu_view_zoom_simple(struct camu_view *view, s32 window_width, s32 window_height, f64 x, f64 y, f64 v) +{ + x -= view->x_offset; + y -= view->y_offset; + x /= view->zoom; + y /= view->zoom; + + f64 prev_width = window_width * view->zoom; + f64 prev_height = window_height * view->zoom; + + f64 x_ratio = x / (f64)window_width; + f64 y_ratio = y / (f64)window_height; + + f64 zoom = view->zoom + (v * view->zoom); + if (zoom > ZOOM_MAX || zoom < ZOOM_MIN) return false; + + x = view->x_offset - x_ratio * ((window_width * zoom) - prev_width); + y = view->y_offset - y_ratio * ((window_height * zoom) - prev_height); + if (!(X_IN_BOUNDS(view, window_width, x) && Y_IN_BOUNDS(view, window_height, y))) { + return false; + } + + view->x_offset = x; + view->y_offset = y; + view->zoom = zoom; + + return true; +} diff --git a/src/screen/view.h b/src/screen/view.h index 0546062..e9bec81 100644 --- a/src/screen/view.h +++ b/src/screen/view.h @@ -9,6 +9,8 @@ enum { CAMU_VIEW_DETACHED }; +#define CAMU_DEFAULT_VIEW CAMU_VIEW_AUTOFIT + struct camu_view { u8 mode; s32 width; @@ -21,3 +23,5 @@ struct camu_view { }; void camu_view_calculate(struct camu_view *view, s32 window_width, s32 window_height); +bool camu_view_pan_simple(struct camu_view *view, s32 window_width, s32 window_height, f64 dx, f64 dy); +bool camu_view_zoom_simple(struct camu_view *view, s32 window_width, s32 window_height, f64 x, f64 y, f64 v); diff --git a/src/shoki/cpy/lib.pxd b/src/shoki/cpy/lib.pxd deleted file mode 100644 index 5749141..0000000 --- a/src/shoki/cpy/lib.pxd +++ /dev/null @@ -1,13 +0,0 @@ -include "types.pxd" - -cdef extern from "<al/lib.h>": - void al_free(void *ptr) - void al_malloc(size_t size) - void al_strlen(char *s) - void al_bzero(void *s, size_t n) - -cdef extern from "<al/log.h>": - s32 al_log_info(const char *ns, const char *, ...) - s32 al_log_warn(const char *ns, const char *, ...) - s32 al_log_error(const char *ns, const char *, ...) - s32 al_log_debug(const char *ns, const char *, ...) diff --git a/src/shoki/cpy/post.pxd b/src/shoki/cpy/post.pxd deleted file mode 100644 index 0b0a97e..0000000 --- a/src/shoki/cpy/post.pxd +++ /dev/null @@ -1,58 +0,0 @@ -include "str.pxd" - -cdef extern from "../../shoki/src/post.h": - ctypedef struct sho_optional_int: - s64 i - bool set - - cdef struct sho_post_date: - u8 type - s64 timestamp - - cdef struct sho_post_media: - u8 type - str url - str thumbnail_url - - ctypedef struct sho_dates_array: - u32 size - u32 alloc - sho_post_date *data - - ctypedef struct sho_media_array: - u32 size - u32 alloc - sho_post_media *data - - cdef struct sho_post_user: - str unique_id - str username - str display_name - str profile_image_url - - cdef struct sho_post_ref: - str unique_id - - cdef struct sho_post: - u16 version - u8 type - str unique_id - str raw_responses - str url - sho_dates_array dates - sho_post_user author - str title - str text - sho_optional_int likes - sho_optional_int reposts - sho_optional_int quotes - sho_optional_int comments - sho_optional_int views - sho_media_array media - sho_post_ref post - sho_post_ref quoted - sho_post_ref in_reply_to - - void sho_post_reset(sho_post *post) - void sho_post_add_media(sho_post *post, u8 type, str *url, str *thumbnail_url) - void sho_post_add_date(sho_post *post, u8 type, s64 timestamp) diff --git a/src/shoki/cpy/result.pxd b/src/shoki/cpy/result.pxd deleted file mode 100644 index 1bebb55..0000000 --- a/src/shoki/cpy/result.pxd +++ /dev/null @@ -1,8 +0,0 @@ -include "post.pxd" - -cdef extern from "../../shoki/src/search.h": - cdef struct sho_result: - pass - - void sho_result_add_post(sho_result *res, s32 page, sho_post *post) - void sho_result_add_to_list(sho_result *res, s32 page, str *unique_id) diff --git a/src/shoki/cpy/setup.py b/src/shoki/cpy/setup.py deleted file mode 100644 index 1b491f1..0000000 --- a/src/shoki/cpy/setup.py +++ /dev/null @@ -1,6 +0,0 @@ -from Cython.Build import cythonize -from Cython.Compiler import Options -from setuptools import Extension -Options.embed = True -alabaster_inc = "../../../subprojects/libalabaster/include" -cythonize([Extension("shoki", ["shoki.pyx"], include_dirs=[alabaster_inc], extra_link_args=[])], language_level=3) diff --git a/src/shoki/cpy/shoki.pyx b/src/shoki/cpy/shoki.pyx deleted file mode 100644 index 6d1dd56..0000000 --- a/src/shoki/cpy/shoki.pyx +++ /dev/null @@ -1,132 +0,0 @@ -cimport lib -cimport str -cimport post -cimport result - -import sys -sys.path.append('./py') - -cdef public int log_info(char *msg) except -1: - return lib.al_log_info("shoki", "%s", msg) - -cdef public int log_warn(char *msg) except -1: - return lib.al_log_warn("shoki", "%s", msg) - -cdef public int log_error(char *msg) except -1: - return lib.al_log_error("shoki", "%s", msg) - -cdef public int log_debug(char *msg) except -1: - return lib.al_log_debug("shoki", "%s", msg) - -def encode_str(str): - return str.encode('UTF-8') - -def decode_str(str): - return str.decode('UTF-8') - -class OutputHook(object): - def __init__(self): - self.fd = sys.__stdout__ - - def __getattr__(self, attr): - return getattr(self.fd, attr) - - def write(self, msg): - if len(msg) > 0 and msg != '\n': - log_debug(encode_str(msg)) - -hook = OutputHook() -sys.stdout = hook -sys.stderr = hook - -class ShokiLogger(): - def debug(self, msg): - log_debug(encode_str(msg)) - - def info(self, msg): - log_info(encode_str(msg)) - - def warning(self, msg): - log_warn(encode_str(msg)) - - def error(self, msg): - log_error(encode_str(msg)) - -import log -log.set_logger(ShokiLogger()) - -def exception_handler(exception_type, exception, traceback): - log.error(repr(exception)) -sys.excepthook = exception_handler - -import client -from post import PostType, DateType - -shoki = client.ShokiClient() -for name, p in client.ALL_PROVIDERS.items(): - if p[0].init(): - shoki.register_provider(name, p) - -cdef public int sho_client_search(str.str *provider, str.str *query) except -1: - cdef char *c_str0 = str.al_str_to_c_str(provider) - cdef char *c_str1 = str.al_str_to_c_str(query) - ret = shoki.search(decode_str(c_str0), decode_str(c_str1)) - lib.al_free(c_str0) - lib.al_free(c_str1) - return ret - -cdef public int py_str_to_str(str.str *s, char *py_c_str) except -1: - str.al_str_from(s, py_c_str) - return 0 - -cdef public int sho_client_get_page(result.sho_result *res, int search_id, int num) except -1: - cdef post.sho_post cpost - cdef str.str url - cdef str.str thumbnail_url - cdef str.str unique_id - page = shoki.get_page(search_id, num) - if not page: return -1 - for py_post in page['posts']: - post.sho_post_reset(&cpost) - cpost.version = py_post.version - cpost.type = py_post.type.value - py_str_to_str(&cpost.unique_id, encode_str(py_post.unique_id)) - py_str_to_str(&cpost.url, encode_str(py_post.url)) - for date in py_post.dates: - post.sho_post_add_date(&cpost, date.value, py_post.dates[date]) - py_str_to_str(&cpost.author.unique_id, encode_str(py_post.author.unique_id)) - py_str_to_str(&cpost.author.username, encode_str(py_post.author.username)) - py_str_to_str(&cpost.author.display_name, encode_str(py_post.author.display_name)) - py_str_to_str(&cpost.author.profile_image_url, encode_str(py_post.author.profile_image_url)) - if cpost.type == PostType.REPOST.value: - py_str_to_str(&cpost.post.unique_id, encode_str(py_post.post.unique_id)) - result.sho_result_add_post(res, num, &cpost) - continue - py_str_to_str(&cpost.title, encode_str(py_post.title)) - py_str_to_str(&cpost.text, encode_str(py_post.text)) - cpost.likes.set = py_post.likes != None - if cpost.likes.set: - cpost.likes.i = py_post.likes - cpost.reposts.set = py_post.reposts != None - if cpost.reposts.set: - cpost.reposts.i = py_post.reposts - cpost.quotes.set = py_post.quotes != None - if cpost.quotes.set: - cpost.quotes.i = py_post.quotes - cpost.comments.set = py_post.comments != None - if cpost.comments.set: - cpost.comments.i = py_post.comments - cpost.views.set = py_post.views != None - if cpost.views.set: - cpost.views.i = py_post.views - for media in py_post.media: - py_str_to_str(&url, encode_str(media.url)) - py_str_to_str(&thumbnail_url, encode_str(media.thumbnail_url)) - post.sho_post_add_media(&cpost, 0, &url, &thumbnail_url) - py_str_to_str(&cpost.quoted.unique_id, encode_str(py_post.quoted.unique_id)) - py_str_to_str(&cpost.in_reply_to.unique_id, encode_str(py_post.in_reply_to.unique_id)) - result.sho_result_add_post(res, num, &cpost) - for py_unique_id in page['list']: - py_str_to_str(&unique_id, encode_str(py_unique_id)) - result.sho_result_add_to_list(res, num, &unique_id) - return 0 diff --git a/src/shoki/cpy/str.pxd b/src/shoki/cpy/str.pxd deleted file mode 100644 index a12ced4..0000000 --- a/src/shoki/cpy/str.pxd +++ /dev/null @@ -1,11 +0,0 @@ -include "types.pxd" - -cdef extern from "<al/str.h>": - ctypedef struct str: - u32 len - s32 alloc - char *data - - void al_str_from(str *s, const char *str) - char *al_str_to_c_str(str *s) - diff --git a/src/shoki/cpy/types.pxd b/src/shoki/cpy/types.pxd deleted file mode 100644 index fec35a5..0000000 --- a/src/shoki/cpy/types.pxd +++ /dev/null @@ -1,11 +0,0 @@ -ctypedef unsigned char u8; -ctypedef char s8; -ctypedef unsigned short u16; -ctypedef short s16; -ctypedef unsigned int u32; -ctypedef int s32; -ctypedef unsigned long u64; -ctypedef long s64; -ctypedef float f32; -ctypedef double f64; -ctypedef bint bool diff --git a/src/shoki/meson.build b/src/shoki/meson.build deleted file mode 100644 index 557d497..0000000 --- a/src/shoki/meson.build +++ /dev/null @@ -1,14 +0,0 @@ -shoki_src = [ - 'src/search.c', - 'src/post.c', - 'src/packet_ext.c', - 'src/post_cache.c' -] -shoki_inc = [] -shoki_deps = [common_deps] - -python = dependency('python-3.11-embed', required: true) -shoki_deps += [python] - -shoki = declare_dependency(sources: shoki_src, - include_directories: shoki_inc, dependencies: shoki_deps) diff --git a/src/shoki/py/__init__.py b/src/shoki/py/__init__.py deleted file mode 100644 index e69de29..0000000 --- a/src/shoki/py/__init__.py +++ /dev/null diff --git a/src/shoki/py/client.py b/src/shoki/py/client.py deleted file mode 100644 index 3986060..0000000 --- a/src/shoki/py/client.py +++ /dev/null @@ -1,51 +0,0 @@ -import log -import config -from providers.twitter import TwitterProvider -from providers.youtube import YoutubeProvider - -ALL_PROVIDERS = { - 'twitter': (TwitterProvider( - config.TWITTER_ACCESS_TOKEN, config.TWITTER_ACCESS_TOKEN_SECRET, - config.TWITTER_CONSUMER_TOKEN, config.TWITTER_CONSUMER_TOKEN_SECRET, - config.TWITTER_COOKIES_PATH), []), - 'youtube': (YoutubeProvider(), []) -} - -class ShokiClient(): - def __init__(self): - self.providers = {} - self.searches = {} - self.inc = 0 - - def register_provider(self, name, p): - log.info('Registered provider \'{}\'.'.format(name)) - self.providers[name] = p - - def get_providers(self): - return self.providers.keys() - - def search(self, provider, query): - if provider not in self.providers: - return -1 - search_id = self.inc - p = self.providers[provider] - try: - search = p[0].search(query, *p[1]) - self.searches[search_id] = search - self.inc += 1 - except Exception as e: - log.error(repr(e)) - return -1 - return search_id - - def get_page(self, search_id, num): - if search_id not in self.searches: - return None - try: - return self.searches[search_id].get_page(num) - except Exception as e: - log.error(repr(e)) - return None - - def get_playback_data(self, unique_id): - return '' diff --git a/src/shoki/py/config.def.py b/src/shoki/py/config.def.py deleted file mode 100644 index 237d658..0000000 --- a/src/shoki/py/config.def.py +++ /dev/null @@ -1,19 +0,0 @@ -from pathlib import Path - -BASE_DIR = '' - -TWITTER_ACCESS_TOKEN = '' -TWITTER_ACCESS_TOKEN_SECRET = '' -TWITTER_CONSUMER_TOKEN = '' -TWITTER_CONSUMER_TOKEN_SECRET = '' -TWITTER_COOKIES_PATH = '{}/cookies_twitter.txt'.format(BASE_DIR) -TWITTER_TIMEOUT = 5 # seconds - -FANBOX_SESSID = '' - -PIXIV_REFRESH_TOKEN = '' - -INSTAGRAM_USERNAME = '' -INSTAGRAM_PASSWORD = '' -INSTAGRAM_SETTINGS_PATH = Path('{}/ig_settings.json'.format(BASE_DIR)) -INSTAGRAM_USER_AGENT = 'Instagram 146.0.0.27.125 (iPhone12,1; iOS 13_3; en_US; en-US; scale=2.00; 1656x3584; 190542906)' diff --git a/src/shoki/py/log.py b/src/shoki/py/log.py deleted file mode 100644 index c36c54e..0000000 --- a/src/shoki/py/log.py +++ /dev/null @@ -1,30 +0,0 @@ -class DefaultLogger(): - def debug(self, msg): - print(msg) - - def info(self, msg): - print(msg) - - def warning(self, msg): - print(msg) - - def error(self, msg): - print(msg) - -LOGGER = DefaultLogger() - -def set_logger(logger): - global LOGGER - LOGGER = logger - -def debug(msg): - LOGGER.debug(msg) - -def info(msg): - LOGGER.info(msg) - -def warning(msg): - LOGGER.warning(msg) - -def error(msg): - LOGGER.error(msg) diff --git a/src/shoki/py/post.py b/src/shoki/py/post.py deleted file mode 100644 index d489298..0000000 --- a/src/shoki/py/post.py +++ /dev/null @@ -1,70 +0,0 @@ -import dataclasses -from dataclasses import field -from typing import Optional -from json import JSONEncoder -from enum import Enum - -class PostType(Enum): - UNKNOWN = -1 - POST = 0 - REPOST = 1 - -class DateType(Enum): - CREATED = 0 - EDITED = 1 - ARCHIVED = 2 - -@dataclasses.dataclass -class Media(): - url: str = '' - base64_data: str = '' - -@dataclasses.dataclass -class Image(Media): - type: str = 'image' - thumbnail_url: str = '' - -@dataclasses.dataclass -class Video(Media): - type: str = 'video' - thumbnail_url: str = '' - -@dataclasses.dataclass -class Audio(Media): - type: str = 'audio' - -@dataclasses.dataclass -class User(): - unique_id: str = '' - username: str = '' - display_name: str = '' - profile_image_url: str = '' - -@dataclasses.dataclass -class PostRef(): - unique_id: str = '' - -@dataclasses.dataclass -class Post(): - version: int = 2 - type: 'PostType' = PostType.UNKNOWN - unique_id: str = '' - raw_responses: dict[str, str] = field(default_factory=lambda: {}) - url: str = '' - dates: dict['DateType', int] = field(default_factory=lambda: {}) - author: 'User' = field(default_factory=lambda: User()) - title: str = '' - text: str = '' - likes: Optional[int] = None - reposts: Optional[int] = None - quotes: Optional[int] = None - comments: Optional[int] = None - views: Optional[int] = None - media: list['Media'] = field(default_factory=lambda: []) - post: 'PostRef' = field(default_factory=lambda: PostRef()) - quoted: 'PostRef' = field(default_factory=lambda: PostRef()) - in_reply_to: 'PostRef' = field(default_factory=lambda: PostRef()) - -class PostEncoder(JSONEncoder): - def default(self, obj): - return obj.__dict__ diff --git a/src/shoki/py/provider.py b/src/shoki/py/provider.py deleted file mode 100644 index a719d01..0000000 --- a/src/shoki/py/provider.py +++ /dev/null @@ -1,39 +0,0 @@ -import httpx -import json -import log - -class Provider(): - def __init__(self): - self.cookies = {} - self.headers = {} - self.unique_id_map = {} - - def init(self): - return True - - # TODO: retries. - def get(self, url, params = None): - log.debug('HTTP request {} {}.'.format(url, params)) - r = httpx.get(url, params=params, cookies=self.cookies, headers=self.headers) - # TODO: Non-200 = err, support redirects (if not already done in httpx)? - if r.status_code == 403 or r.status_code == 404: - log.error('HTTP {}.'.format(r.status_code)) - return None - log.debug('HTTP {}.'.format(r.status_code)) - return r - - def get_json(self, url, params = None): - try: - r = self.get(url, params) - if not r: - return None - return json.loads(r.content) - except json.JSONDecodeError as e: - log.error(repr(e)) - return None - - def get_obj_from_unique_id(self, unique_id): - return None - - def search(self, query, *extra_args): - return None diff --git a/src/shoki/py/providers/__init__.py b/src/shoki/py/providers/__init__.py deleted file mode 100644 index e69de29..0000000 --- a/src/shoki/py/providers/__init__.py +++ /dev/null diff --git a/src/shoki/py/providers/fanbox.py b/src/shoki/py/providers/fanbox.py deleted file mode 100644 index 7739be3..0000000 --- a/src/shoki/py/providers/fanbox.py +++ /dev/null @@ -1,130 +0,0 @@ -import urllib.parse -import log -from plugin import Plugin -from search import Search -from post import Post, Image - -BASE_URL = "https://api.fanbox.cc/" - -PAGES_PER_QUERY = 2 - -class FanboxSearch(Search): - def __init__(self, plugin, creator_id): - super().__init__() - self.p = plugin - self.creator_id = creator_id - self.pagination_pages = [] - self.pagination_complete = False - - def paginate(self): - res = self.p.get_json(BASE_URL + 'post.paginateCreator', { 'creatorId': self.creator_id }) - if not res or 'body' not in res or len(res['body']) == 0: - return False - for p in res['body']: - params = urllib.parse.parse_qs(urllib.parse.urlparse(p).query) - page = { - 'url': p, - 'params': params, - 'data': None, - 'limit': -1 if 'limit' not in params else int(params['limit'][0]) - } - self.pagination_pages.append(page) - return True - - def load_pagination_page(self, p): - if not p['data']: - res = self.p.get_json(BASE_URL + 'post.listCreator', p['params']) - if not res: - return False - p['data'] = res['body']['items'] - # The get_post_at_index logic assumes limit will always be exact. - p['limit'] = len(p['data']) - return True - - def get_post_at_index(self, index): - if not self.pagination_complete: - if not self.paginate(): - return None - self.pagination_complete = True - page = None - count = 0 - for p in self.pagination_pages: - if p['limit'] == -1: - if not self.load_pagination_page(p): - return None - ncount = count + p['limit'] - if ncount > index: - page = p - break - count = ncount - if not page or not self.load_pagination_page(page) or count + page['limit'] <= index: - return None - return page['data'][index - count] - - def add_images_to_map(self, body): - if 'imageMap' not in body: - return - for k in body['imageMap'].keys(): - self.p.image_map[k] = body['imageMap'][k]['originalUrl'] - - def load_page(self, index): - self.pages[index] = [] - s = index * PAGES_PER_QUERY - for i in range(s, s + PAGES_PER_QUERY): - p = self.get_post_at_index(i) - # Break to return what we have, which could be less than expected. - if not p: - self.completed = True - break - unique_id = 'fanbox:{}'.format(p['id']) - if unique_id not in self.p.unique_id_map: - res = self.p.get_json( - BASE_URL + 'post.info', { 'postId': p['id'] } - ) - if not res or 'body' not in res: - # TODO: Log this. - continue - if res['body']['type'] == 'article': - self.add_images_to_map(res['body']['body']) - self.p.unique_id_map[unique_id] = res['body'] - # TODO: Should store Post object here instead. - body = self.p.unique_id_map[unique_id] - text = '' - media = [] - # If body is None, that post is restricted (not subbed or from a higher tier than your sub) - if body['body']: - if body['type'] == 'image': - text = body['body']['text'] - for i in body['body']['images']: - media.append(Image(url=i['originalUrl'], thumbnail_url='')) - elif body['type'] == 'article': - for b in body['body']['blocks']: - if b['type'] == 'p': - text += b['text'] + '\n' - elif b['type'] == 'image': - media.append(Image(url=self.p.image_map[b['imageId']], thumbnail_url='')) - else: - log.warn('Unhandled block type.') - else: - log.warn('Unhandled post type.') - # TODO: Raw response - self.pages[index].append(Post( - unique_id=unique_id, url='', title=body['title'], text=text, media=media)) - return True - -class FanboxPlugin(Plugin): - def __init__(self, sessid): - super().__init__() - self.image_map = {} - self.headers.update({ - 'accept': 'application/json, text/plain, */*', - 'accept-encoding': 'gzip, deflate, br', - 'accept-language': 'ja,en-US;q=0.9,en;q=0.8', - 'origin': 'https://www.fanbox.cc', - 'referer': 'https://www.fanbox.cc/', - 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/98.0.4758.102 Safari/537.36' - }) - self.cookies['FANBOXSESSID'] = sessid - - def search(self, query, *extra_args): - return FanboxSearch(self, query) diff --git a/src/shoki/py/providers/instagram.py b/src/shoki/py/providers/instagram.py deleted file mode 100644 index 0146ba3..0000000 --- a/src/shoki/py/providers/instagram.py +++ /dev/null @@ -1,96 +0,0 @@ -import os -import config -import log -from plugin import Plugin -from search import Search -from instagrapi import Client -from instagrapi.exceptions import UserNotFound, LoginRequired, ChallengeRequired -from post import User, Post, Image, Video - -REACH_LIMIT = 3 - -class InstagramSearch(Search): - def __init__(self, cl, pk): - super().__init__() - self.cl = cl - self.pk = pk - self.pagination_page = 0 - self.pagination_cursor = None - - def load_page(self, index): - if index >= self.pagination_page + REACH_LIMIT: - return False - request_satisfied = False - for _ in range(self.pagination_page, index + 1): - if self.pagination_page > 0 and not self.pagination_cursor: - self.completed = True - return False - try: - media, self.pagination_cursor = self.cl.user_medias_paginated_v1( - self.pk, 0, end_cursor=self.pagination_cursor) - except LoginRequired: - return False - except ChallengeRequired: - return False - self.pages[self.pagination_page] = [] - for m in media: - unique_id = 'instagram:p:{}'.format(m.id) - media = [] - if m.media_type == 8: # Album - for r in m.resources: - if r.media_type == 1: # Photo - media.append(Image(url=r.thumbnail_url)) - elif r.media_type == 2: # Video - media.append(Video(url=r.video_url, thumbnail_url=r.thumbnail_url)) - elif m.media_type == 2: # Video - #print(m.product_type) # feed, igtv, or clips - media.append(Video(url=m.video_url, thumbnail_url=m.thumbnail_url)) - elif m.media_type == 1: # Photo - media.append(Image(url=m.thumbnail_url)) - kwargs = {} - kwargs['unique_id'] = unique_id - kwargs['raw_responses'] = { 'tile': m.json() } - kwargs['url'] = 'https://www.instagram.com/p/{}'.format(m.code) - user = User(unique_id='instagram:u:{}'.format(m.user.pk)) - if m.user.username: - user.username = m.user.username - if m.user.full_name: - user.display_name = m.user.full_name - kwargs['author'] = user - kwargs['title'] = m.title - kwargs['text'] = m.caption_text - kwargs['media'] = media - self.pages[self.pagination_page].append(Post(**kwargs)) - if self.pagination_page == index: - request_satisfied = True - self.pagination_page += 1 - return request_satisfied - -class InstagramPlugin(Plugin): - def __init__(self): - super().__init__() - self.cl = Client() - self.cl.set_user_agent(config.INSTAGRAM_USER_AGENT) - - def init(self): - if os.path.exists(config.INSTAGRAM_SETTINGS_PATH): - self.cl.load_settings(config.INSTAGRAM_SETTINGS_PATH) - if self.cl.login(config.INSTAGRAM_USERNAME, config.INSTAGRAM_PASSWORD): - return True - if not self.cl.login(config.INSTAGRAM_USERNAME, config.INSTAGRAM_PASSWORD): - return False - self.cl.dump_settings(config.INSTAGRAM_SETTINGS_PATH) - return True - - def search(self, query, *extra_args): - try: - user = self.cl.user_info_by_username_v1(query) - except UserNotFound: - return None - except LoginRequired: - log.error('Login required.') - return None - except ChallengeRequired: - log.error('Challenge required.') - return None - return InstagramSearch(self.cl, user.pk) diff --git a/src/shoki/py/providers/pixiv.py b/src/shoki/py/providers/pixiv.py deleted file mode 100644 index 9276d61..0000000 --- a/src/shoki/py/providers/pixiv.py +++ /dev/null @@ -1,113 +0,0 @@ -import json -import requests -from post import User, Image, Post -from search import Search -from plugin import Plugin -from query_parser import QueryParser -from pixivpy3 import * - -class PixivBase(Search): - def __init__(self, userdata, arg): - super().__init__() - self.plugin = userdata - self.next_page = 0 - - def api_load_page(self, index): - return False, None - - def load_page(self, index): - request_satisfied = False - for i in range(self.next_page, index + 1): - if not self.next_qs: - break - self.pages[i] = [] - res, self.next_qs = self.api_load_page(i) - if not res: - break - if i == index: - request_satisfied = True - self.next_page += 1 - return request_satisfied - - def make_post(self, post): - if len(post['meta_pages']) == 0: - post['meta_pages'].append({ - 'image_urls': { - "medium": post['image_urls']['medium'], - "original": post['meta_single_page']['original_image_url'] - } - }) - kwargs = {} - kwargs['unique_id'] = 'pixiv:i:{}'.format(post['id']) - kwargs['raw_responses'] = { 'api': json.dumps(post) } - kwargs['url'] = 'https://www.pixiv.net/en/artworks/{}'.format(post['id']) - kwargs['author'] = User( - unique_id='pixiv:u:{}'.format(post['user']['id']), - username=post['user']['account'], display_name=post['user']['name']) - kwargs['title'] = post['title'] - kwargs['text'] = post['caption'] - kwargs['likes'] = post['total_bookmarks'] - if 'total_comments' in post: - kwargs['comments'] = post['total_comments'] - media = [] - for img in post['meta_pages']: - media.append(Image( - url=img['image_urls']['original'], thumbnail_url=img['image_urls']['medium'])) - kwargs['media'] = media - return Post(**kwargs) - -class PixivFollowing(PixivBase): - def __init__(self, userdata, arg): - super().__init__(userdata, arg) - self.next_qs = { "user_id": arg, "restrict": "public" } - - def api_load_page(self, index): - obj = self.plugin.api.user_following(**self.next_qs) - if 'user_previews' not in obj: - return False, None - for u in obj['user_previews']: - user = u['user'] - self.pages[index].append(User( - unique_id='pixiv:u:{}'.format(user['id']), username=user['account'], - display_name=user['name'])) - return True, self.plugin.api.parse_qs(obj['next_url']) - -class PixivUser(PixivBase): - def __init__(self, userdata, arg): - super().__init__(userdata, arg) - self.next_qs = { 'user_id': arg, 'type': 'illust' } - - def api_load_page(self, index): - obj = self.plugin.api.user_illusts(**self.next_qs) - if 'illusts' not in obj: - return False, None - for post in obj['illusts']: - self.pages[index].append(self.make_post(post)) - return True, self.plugin.api.parse_qs(obj['next_url']) - -class PixivBookmarks(PixivBase): - def __init__(self, userdata, arg): - super().__init__(userdata, arg) - self.next_qs = { 'user_id': arg, 'restrict': 'public' } - - def api_load_page(self, index): - obj = self.plugin.api.user_bookmarks_illust(**self.next_qs) - if 'illusts' not in obj: - return False, None - for post in obj['illusts']: - self.pages[index].append(self.make_post(post)) - return True, self.plugin.api.parse_qs(obj['next_url']) - -class PixivPlugin(Plugin): - def __init__(self, refresh_token): - super().__init__() - self.parser = QueryParser(self, 'following') - self.parser.add_command('following', PixivFollowing) - self.parser.add_command('user', PixivUser) - self.parser.add_command('bookmarks', PixivBookmarks) - self.api = AppPixivAPI() - self.api.requests = requests.Session() - self.api.auth(refresh_token=refresh_token) - - def search(self, query, *extra_args): - return self.parser.parse_query(query) diff --git a/src/shoki/py/providers/searx.py b/src/shoki/py/providers/searx.py deleted file mode 100644 index b536f28..0000000 --- a/src/shoki/py/providers/searx.py +++ /dev/null @@ -1,93 +0,0 @@ -import sys -sys.path.append('../../../subprojects/searxng') -import json -from searx import settings -from searx.utils import gen_useragent -from searx.engines import load_engine, register_engine -from plugin import Plugin -from search import Search -from post import Post, Image - -# TODO: Put these on the Plugin object. -ENGINES = {} - -ENGINE_INITIAL_INDEX = { - 'google images': 0, - 'google': 1, - 'bing images': 0 -} - -class SearxSearch(Search): - def __init__(self, plugin, query, engine): - super().__init__() - global ENGINES - self.query = query - self.plugin = plugin - self.key = engine.replace(' ', '_') - self.engine = ENGINES[engine] - self.index = ENGINE_INITIAL_INDEX[engine] - - def load_page(self, index): - self.pages[index] = [] - res = self.plugin.send_http_request(self.engine.request(self.query, { - 'language': 'en-US', - 'safesearch': 0, - 'time_range': None, - 'pageno': self.index + index, - 'cookies': self.plugin.cookies, - 'headers': self.plugin.headers, - 'data': None - })) - if not res: - return False - try: - ret = self.engine.response(res) - except: - return False - if len(ret) == 0: - # TODO: Test. - self.completed = True - return False - for i, p in enumerate(ret): - if 'url' not in p: - continue - # Don't add this to the global map because it's not actually unique. - unique_id = '{}:{}_{}:{}'.format(self.key, self.query, i, index) - media = [] - if 'img_src' in p: - media.append(Image(url=p['img_src'], thumbnail_url='')) - elif 'image_url' in p: - media.append(Image(url=p['image_url'], thumbnail_url='')) - kwargs = {} - kwargs['unique_id'] = unique_id - kwargs['raw_responses'] = { 'result': json.dumps(p) } - kwargs['url'] = p['url'] - kwargs['title'] = p['title'] - kwargs['text'] = p['content'] - kwargs['media'] = media - self.pages[index].append(Post(**kwargs)) - return True - -class SearxPlugin(Plugin): - def __init__(self): - super().__init__() - global ENGINES - self.headers = { - 'User-Agent': gen_useragent(), - 'Accept-Language': 'en-US,en;q=0.5', - } - for engine_data in settings['engines']: - engine = load_engine(engine_data) - if engine: - register_engine(engine) - ENGINES[engine.name] = engine - - def send_http_request(self, params): - self.cookies.update(params['cookies']) - self.headers.update(params['headers']) - return self.get(params['url'], params=params['data']) - - def search(self, query, *extra_args): - if not extra_args: - return None - return SearxSearch(self, query, extra_args[0]) diff --git a/src/shoki/py/providers/twitter.py b/src/shoki/py/providers/twitter.py deleted file mode 100644 index 4770dda..0000000 --- a/src/shoki/py/providers/twitter.py +++ /dev/null @@ -1,424 +0,0 @@ -import json -import config -import http.cookiejar -from datetime import timezone - -import log -from provider import Provider -from search import Search -from post import PostType, DateType, PostRef, Post, User, Image, Video -from query_parser import QueryParser - -# Official Twitter API -#from twitter import Twitter2, TwitterError, OAuth - -# GraphQL API -import snscrape.modules.twitter -from snscrape.base import ScraperException -from snscrape.modules.twitter import (TwitterSearchScraper, TwitterProfileScraper, - TwitterUserScraper, TwitterTweetScraper, TweetRef) - -# Don't attempt to load a page 4 or more pages past the current page. -REACH_LIMIT = 4 - -# Quite incomplete API based backend. Should *not* be used for archiving. -# HOLD: I can't test this without paying $100 for the X API WTFFF. -''' -class TwitterBase(Search): - ALL_PARAMS = { - 'tweet.fields': 'attachments,author_id,context_annotations,conversation_id,created_at,entities,geo,id,in_reply_to_user_id,lang,public_metrics,possibly_sensitive,referenced_tweets,reply_settings,source,text,withheld', - 'user.fields': 'created_at,description,entities,id,location,name,pinned_tweet_id,profile_image_url,protected,public_metrics,url,username,verified,withheld', - 'media.fields': 'duration_ms,height,media_key,preview_image_url,type,url,width,public_metrics', - 'place.fields': 'contained_within,country,country_code,full_name,geo,id,name,place_type', - 'poll.fields': 'duration_minutes,end_datetime,id,options,voting_status', - } - ALL_EXPRESSIONS = 'author_id,referenced_tweets.id,referenced_tweets.id.author_id,entities.mentions.username,attachments.poll_ids,attachments.media_keys,in_reply_to_user_id,geo.place_id' - - SOME_PARAMS = { - 'tweet.fields': 'attachments,author_id,text,entities,referenced_tweets', - 'user.fields': 'id,name,profile_image_url,url,username', - 'media.fields': 'duration_ms,height,media_key,preview_image_url,type,url,width,public_metrics', - 'place.fields': '', - 'poll.fields': '', - } - SOME_EXPRESSIONS = 'author_id,referenced_tweets.id,referenced_tweets.id.author_id,entities.mentions.username,attachments.media_keys,in_reply_to_user_id' - - def __init__(self, userdata, arg): - super().__init__() - self.provider = userdata - self.params = self.SOME_PARAMS.copy() - self.media_map = {} - self.user_id = None - self.pagination_page = 0 - self.pagination_token = None - self.arg = arg - - def add_attachemnts(self, l, r, data): - if 'media_keys' not in data: - return - for m in data['media_keys']: - if m in r: - continue - r.append(m) - if m in self.media_map: - l.append(Image(url=self.media_map[m], thumbnail_url='')) - - def add_entity_urls(self, data): - if 'urls' in data: - for u in data['urls']: - if 'media_key' in u and u['media_key'] not in self.media_map: - self.media_map[u['media_key']] = u['expanded_url'] - - def make_post(self, tweet): - print(json.dumps(tweet, indent=4)) - media = [] - repeats = [] - if 'referenced_tweets' in t: - print(len(referenced_tweets)) - #for rt in t['referenced_tweets']: - # if rt['id'] in self.tweet_map: - # rrt = self.tweet_map[rt['id']] - # if 'entities' in rrt: - # self.add_entity_urls(rrt['entities']) - # if 'attachments' in rrt: - # self.add_attachemnts(media, repeats, rrt['attachments']) - if 'attachments' in t: - self.add_attachemnts(media, repeats, t['attachments']) - unique_id = 'twitter:t:{}'.format(t['id']) - url = 'https://twitter.com/{}/status/{}'.format(t['author_id'], t['id']) - kwargs = {} - kwargs['unique_id'] = unique_id - kwargs['raw_responses'] = { 'tweet' : json.dumps(t) } - kwargs['url'] = url - kwargs['author'] = User(unique_id='twitter:u:{}'.format(t['author_id'])) - kwargs['title'] = '' - kwargs['text'] = t['text'] - kwargs['media'] = media - return Post(**kwargs) - - def add_items_from_search(self, data): - if 'includes' in data: - includes = data['includes'] - if 'media' in includes: - for i in includes['media']: - if 'url' in i: - self.media_map[i['media_key']] = i['url'] - #if 'tweets' in includes: - # for t in includes['tweets']: - # self.tweet_map[t['id']] = t - if 'data' not in data or not data['data']: - return False - l = data['data'] if type(data['data']) == list else [data['data']] - for t in l: - if 'entities' in t: - self.add_entity_urls(t['entities']) - post = self.make_post(t) - self.pages[self.pagination_page].append(post) - return True - - def add_user_ids_from_search(self, data): - if not data['data']: - return False - for u in data['data']: - self.pages[self.pagination_page].append(User( - 'twitter:u:{}'.format(u['id']), username=u['username'], display_name=u['name'])) - return True - - def should_process_index(self, index): - if index - self.pagination_page >= REACH_LIMIT: - return False - if self.pagination_page > 0 and not self.pagination_token: - self.completed = True - return False - if self.pagination_token: - self.params['pagination_token'] = self.pagination_token - self.pages[self.pagination_page] = [] - return True - - def handle_data(self, data, user_ids=False): - if 'next_token' not in data['meta'].keys(): - self.pagination_token = None - else: - self.pagination_token = data['meta']['next_token'] - if user_ids: - if not self.add_user_ids_from_search(data): - return False - else: - if not self.add_items_from_search(data): - return False - self.pagination_page += 1 - return True - - def get_user_id(self, username): - if self.user_id: - return True - try: - data = self.provider.t.users.by.username._username( - _username=username, _timeout=config.TWITTER_TIMEOUT) - except TwitterError as e: - log.error(repr(e)) - return False - self.user_id = data['data']['id'] - return True - -class TwitterSearch(TwitterBase): - def load_page(self, index): - for _ in range(self.pagination_page, index + 1): - if not self.should_process_index(index): - return False - try: - data = self.provider.t.tweets.search.recent( - query=self.arg, expansions=self.SOME_EXPRESSIONS, params=self.params, - sort_order='relevancy', max_results=25, _timeout=config.TWITTER_TIMEOUT) - except TwitterError as e: - log.error(repr(e)) - return False - return self.handle_data(data) - -class TwitterUser(TwitterBase): - def __init__(self, userdata, arg): - if arg.startswith('@'): - arg = arg[1:] - super().__init__(userdata, arg) - - def load_page(self, index): - for _ in range(self.pagination_page, index + 1): - if not self.should_process_index(index): - return False - if not self.get_user_id(self.arg): - return False - try: - data = self.provider.t.users._id.tweets( - _id=self.user_id, expansions=self.SOME_EXPRESSIONS, params=self.params, - max_results=25, _timeout=config.TWITTER_TIMEOUT) - except TwitterError as e: - log.error(repr(e)) - return False - return self.handle_data(data) - -class TwitterTimeline(TwitterBase): - def __init__(self, userdata, arg): - if not arg: - arg = 'pizzabelly' - super().__init__(userdata, arg) - self.params['exclude'] = 'replies' - - def load_page(self, index): - for _ in range(self.pagination_page, index + 1): - if not self.should_process_index(index): - return False - if not self.get_user_id(self.arg): - return False - try: - data = self.provider.t.users._id.timelines.reverse_chronological( - _id=self.user_id, expansions=self.SOME_EXPRESSIONS, params=self.params, - max_results=25, _timeout=config.TWITTER_TIMEOUT) - except TwitterError as e: - log.error(repr(e)) - return False - return self.handle_data(data) - -class TwitterLikes(TwitterBase): - def __init__(self, userdata, arg): - if arg.startswith('@'): - arg = arg[1:] - super().__init__(userdata, arg) - - def load_page(self, index): - for _ in range(self.pagination_page, index + 1): - if not self.should_process_index(index): - return False - if not self.get_user_id(self.arg): - return False - try: - data = self.provider.t.users._id.liked_tweets( - _id=self.user_id, expansions=self.SOME_EXPRESSIONS, params=self.params, - max_results=25, _timeout=config.TWITTER_TIMEOUT) - except TwitterError as e: - log.error(repr(e)) - return False - return self.handle_data(data) - -class TwitterTweet(TwitterBase): - def load_page(self, index): - for _ in range(self.pagination_page, index + 1): - if not self.should_process_index(index): - return False - try: - data = self.provider.t.tweets( - ids=self.arg, expansions=self.SOME_EXPRESSIONS, params=self.params, - _timeout=config.TWITTER_TIMEOUT) - except TwitterError as e: - log.error(repr(e)) - return False - return self.handle_data(data) - -class TwitterFollowing(TwitterBase): - def __init__(self, userdata, arg): - if arg.startswith('@'): - arg = arg[1:] - super().__init__(userdata, arg) - del self.params['media.fields'] - del self.params['place.fields'] - del self.params['poll.fields'] - - def load_page(self, index): - for _ in range(self.pagination_page, index + 1): - if not self.should_process_index(index): - return False - if not self.get_user_id(self.arg): - return False - try: - data = self.provider.t.users._id.following( - _id=self.user_id, params=self.params, max_results=25, _timeout=config.TWITTER_TIMEOUT) - except TwitterError as e: - log.error(repr(e)) - return False - return self.handle_data(data, True) -''' - -class TwitterScrapeBase(Search): - # Posts per emulated page because snscrape returns an iterator. - POSTS_PER_PAGE = 8 - - def __init__(self, userdata, arg): - super().__init__() - self.provider = userdata - self.cookies = self.provider.cookies - self.arg = arg - self.iterator = None - self.iter_page = 0 - self.sent_hashes = [] - - def parse_media(self, l): - media = [] - for m in l: - if isinstance(m, snscrape.modules.twitter.Photo): - media.append(Image(url=m.fullUrl, thumbnail_url=m.previewUrl)) - elif isinstance(m, snscrape.modules.twitter.Video): - videos = sorted(m.variants, key=lambda x: x.bitrate if x.bitrate else 0, reverse=True) - media.append(Video(url=videos[0].url, thumbnail_url=m.thumbnailUrl)) - elif isinstance(m, snscrape.modules.twitter.Gif): - gifs = sorted(m.variants, key=lambda x: x.bitrate if x.bitrate else 0, reverse=True) - media.append(Video(url=gifs[0].url, thumbnail_url=m.thumbnailUrl)) - return media - - def make_post(self, tweet): - kwargs = {} - kwargs['unique_id'] = 'twitter:t:{}'.format(tweet.id) - if isinstance(tweet, TweetRef): - return Post(**kwargs) - kwargs['raw_responses'] = tweet.rawResponses - h = tweet.rawResponses['hash'] - if h not in self.sent_hashes: - self.sent_hashes.append(h) - else: - del kwargs['raw_responses']['api'] - kwargs['url'] = tweet.url - kwargs['dates'] = { - DateType.CREATED: tweet.date.replace(tzinfo=timezone.utc).timestamp() - } - author = User(unique_id='twitter:u:{}'.format(tweet.user.id), username='@' + tweet.user.username, - profile_image_url=tweet.user.profileImageUrl) - if tweet.user.displayname: - author.display_name = tweet.user.displayname - kwargs['author'] = author - if tweet.retweetedTweet: - kwargs['type'] = PostType.REPOST - kwargs['post'] = PostRef(unique_id='twitter:t:{}'.format(tweet.retweetedTweet.id)) - return Post(**kwargs) - kwargs['type'] = PostType.POST - kwargs['text'] = tweet.rawContent - kwargs['likes'] = tweet.likeCount - kwargs['reposts'] = tweet.retweetCount - kwargs['quotes'] = tweet.quoteCount - kwargs['comments'] = tweet.replyCount - kwargs['views'] = tweet.viewCount - if tweet.media: - kwargs['media'] = self.parse_media(tweet.media) - if tweet.inReplyToTweetId: - kwargs['in_reply_to'] = PostRef(unique_id='twitter:t:{}'.format(tweet.inReplyToTweetId)) - if tweet.quotedTweet: - kwargs['quoted'] = PostRef(unique_id='twitter:t:{}'.format(tweet.quotedTweet.id)) - return Post(**kwargs) - - def add_post(self, i, post): - self.pages[i]['posts'].append(post) - - def add_post_to_list(self, i, post): - self.add_post(i, post) - self.pages[i]['list'].append(post.unique_id) - - def load_page(self, index): - if not self.iterator or index - self.iter_page >= REACH_LIMIT: - return False - if index in self.pages: - return True - # It's possible to load some "pages", but not enough to satisfy the requested index. - request_satisfied = False - for i in range(self.iter_page, index + 1): - if self.completed: - break - self.pages[i] = { 'posts': [], 'list': [] } - for _ in range(0, self.POSTS_PER_PAGE): - try: - tweet = next(self.iterator) - except StopIteration: - self.completed = True - break - except ScraperException as e: - log.error(repr(e)) - continue - self.add_post_to_list(i, self.make_post(tweet)) - if tweet.retweetedTweet: - self.add_post(i, self.make_post(tweet.retweetedTweet)) - if tweet.quotedTweet: - self.add_post(i, self.make_post(tweet.quotedTweet)) - if i == index: - request_satisfied = True - self.iter_page += 1 - return request_satisfied - -class TwitterScrapeSearch(TwitterScrapeBase): - def __init__(self, userdata, arg): - super().__init__(userdata, arg) - self.iterator = TwitterSearchScraper(self.arg, top=True, cookies=self.cookies).get_items() - -class TwitterScrapeUser(TwitterScrapeBase): - def __init__(self, userdata, arg): - super().__init__(userdata, arg) - if (self.arg.startswith('@')): - self.arg = self.arg[1:] - self.iterator = TwitterProfileScraper(self.arg, cookies=self.cookies).get_items() - #self.iterator = TwitterUserScraper(self.arg, cookies=self.cookies).get_items() - -class TwitterScrapeTweet(TwitterScrapeBase): - def __init__(self, userdata, arg): - super().__init__(userdata, arg) - self.single = True - if (self.arg.startswith('https://')): - self.arg = self.arg[self.arg.rfind('/') + 1:] - question = self.arg.find('?') - if question >= 0: - self.arg = self.arg[:question] - self.iterator = TwitterTweetScraper(self.arg, cookies=self.cookies).get_items() - -class TwitterProvider(Provider): - def __init__(self, access_key, access_secret, consumer_key, consumer_secret, cookies_path): - super().__init__() - self.parser = QueryParser(self, 'search') - self.parser.add_command('search', TwitterScrapeSearch) - self.parser.add_command('user', TwitterScrapeUser) - self.parser.add_command('tweet', TwitterScrapeTweet) - #self.parser.add_command('timeline', TwitterTimeline) - #self.parser.add_command('likes', TwitterLikes) - #self.parser.add_command('following', TwitterFollowing) - #self.t = Twitter2(auth=OAuth(access_key, access_secret, consumer_key, consumer_secret), retry=True) - # Cookies are used by the scrape backend. - cookie_jar = http.cookiejar.MozillaCookieJar() - cookie_jar.load(filename=cookies_path, ignore_expires=True) - for c in cookie_jar: - self.cookies[c.name] = c.value - - def search(self, query, *extra_args): - return self.parser.parse_query(query) diff --git a/src/shoki/py/providers/youtube.py b/src/shoki/py/providers/youtube.py deleted file mode 100644 index 3cbf0c8..0000000 --- a/src/shoki/py/providers/youtube.py +++ /dev/null @@ -1,122 +0,0 @@ -import log -from provider import Provider -from search import Search -from post import Post, PostType, Image, Video -from query_parser import QueryParser -from yt_dlp import YoutubeDL - -class YDLLogger(): - def debug(self, msg): - if msg.startswith('[debug] '): - log.debug(msg) - else: - log.info(msg) - - def info(self, msg): - pass - - def warning(self, msg): - log.warning(msg) - - def error(self, msg): - log.error(msg) - -ydl_opts = { - 'quiet': False, - 'logger': YDLLogger(), - 'cachedir': False, -# 'cookiefile': '', -} - -ydl = YoutubeDL(ydl_opts) - -def get_playback_url(data, video=True): - if 'entries' in data: - if len(data['entries']) == 0: - return None - data = data['entries'][0] - - if 'formats' not in data: - if 'url' in data: - return data['url'] - return None - - # Filter out hls temporarily. - data['formats'] = list(filter(lambda f: not f['protocol'].startswith('m3u8'), data['formats'])) - - if len(data['formats']) == 0: - return None - - url = data['formats'][0]['url'] - - # audio_ext? - has_audio = list(filter(lambda f: 'acodec' not in f or f['acodec'] != 'none', data['formats'])) - - if video: - if len(has_audio) > 0: - data['formats'] = has_audio - try: - data['formats'] = list(filter(lambda f: 'quality' in f, data['formats'])) - url = max(data['formats'], key=lambda f: f['quality'])['url'] - except: - pass - else: - if len(has_audio) == 0: - return None - data['formats'] = has_audio - try: - audio_only = list(filter(lambda f: f['vcodec'] == 'none', data['formats'])) - if len(audio_only) > 0: - data['formats'] = audio_only - else: - data['formats'] = list(filter(lambda f: f['ext'] in ['mp4'], data['formats'])) - except: - pass - try: - data['formats'] = list(filter(lambda f: 'abr' in f, data['formats'])) - url = max(data['formats'], key=lambda f: f['abr'])['url'] - except: - pass - - return url - -class YoutubeBase(Search): - def __init__(self, userdata, arg): - super().__init__() - self.provider = userdata - self.query = arg - - def load_page(self, index): - self.pages[index] = { 'posts': [], 'list': [] } - res = self.get_info() - if res: - url = get_playback_url(res) - if not url: - return False - media = [Video(url=url)] - self.pages[index]['posts'].append(Post(type=PostType.POST, unique_id='', url='', title='', text='', media=media)) - return True - -class YoutubeSearch(YoutubeBase): - def __init__(self, userdata, arg): - super().__init__(userdata, arg) - - def get_info(self): - return ydl.extract_info('ytsearch1:{}'.format(self.query), download=False) - -class YoutubeLink(YoutubeBase): - def __init__(self, userdata, arg): - super().__init__(userdata, arg) - - def get_info(self): - return ydl.extract_info(self.query, download=False) - -class YoutubeProvider(Provider): - def __init__(self): - super().__init__() - self.parser = QueryParser(self, 'search') - self.parser.add_command('search', YoutubeSearch) - self.parser.add_command('link', YoutubeLink) - - def search(self, query, *extra_args): - return self.parser.parse_query(query) diff --git a/src/shoki/py/query_parser.py b/src/shoki/py/query_parser.py deleted file mode 100644 index ee23bef..0000000 --- a/src/shoki/py/query_parser.py +++ /dev/null @@ -1,39 +0,0 @@ -class QueryParser(): - def __init__(self, userdata, default_command): - self.commands = {} - self.escaped_commands = {} - self.userdata = userdata - self.default_command = default_command - - def add_command(self, name, search): - self.commands[name] = search - self.escaped_commands['\\' + name] = name - - def parse_query(self, query): - arg = query - search = self.commands[self.default_command] - - sp = query.split(' ') - - for i, word in enumerate(sp): - col = word.find(':') - if col == -1: - continue - command = word[0:col] - if command in self.escaped_commands: - query = query.replace(command, self.escaped_commands[command], 1) - continue - if command in self.commands: - if col + 1 < len(word): - arg = word[col+1:] - else: - arg = None - search = self.commands[command] - if i == 0: - if len(sp) > 1: - query = query.replace(word + ' ', '', 1) - else: - query = query.replace(word, '', 1) - else: - query = query.replace(' ' + word + ' ', '', 1) - return search(self.userdata, arg) diff --git a/src/shoki/py/search.py b/src/shoki/py/search.py deleted file mode 100644 index 4134851..0000000 --- a/src/shoki/py/search.py +++ /dev/null @@ -1,35 +0,0 @@ -class Search(): - def __init__(self): - self.pages = {} - self.iter_page = 0 - self.iter_index = 0 - self.single = False - self.completed = False - - def __iter__(self): - self.iter_page = 0 - self.iter_index = 0 - return self - - def __next__(self): - if self.iter_index >= len(self.pages[self.iter_page]): - self.iter_page += 1 - self.iter_index = 0 - if self.iter_page not in self.pages: - if not self.load_page(self.iter_page): - raise StopIteration - return self.pages[self.iter_page][self.iter_index] - - def load_page(self, index): - return False - - def get_page(self, index): - if index not in self.pages: - if self.completed: - if self.single: - return self.pages[0] - else: - return None - if not self.load_page(index): - return None - return self.pages[index] diff --git a/src/shoki/src/packet_ext.c b/src/shoki/src/packet_ext.c deleted file mode 100644 index 26ba9ee..0000000 --- a/src/shoki/src/packet_ext.c +++ /dev/null @@ -1,91 +0,0 @@ -#include "search.h" -#include "packet_ext.h" - -static void aki_packet_write_sho_optional_int(struct aki_packet *packet, sho_optional_int *o) -{ - AKI_PACKET_WRITE_TYPE(packet, s64, o->i); - AKI_PACKET_WRITE_TYPE(packet, bool, o->set); -} - -void aki_packet_write_sho_post(struct aki_packet *packet, struct sho_post *post) -{ - AKI_PACKET_WRITE_TYPE(packet, u16, post->version); - AKI_PACKET_WRITE_TYPE(packet, u8, post->type); - aki_packet_write_string(packet, &post->unique_id); - aki_packet_write_string(packet, &post->url); - AKI_PACKET_WRITE_TYPE(packet, u32, post->dates.size); - struct sho_post_date *date; - al_array_foreach_ptr(post->dates, i, date) { - AKI_PACKET_WRITE_TYPE(packet, u8, date->type); - AKI_PACKET_WRITE_TYPE(packet, s64, date->timestamp); - } - aki_packet_write_string(packet, &post->author.unique_id); - aki_packet_write_string(packet, &post->author.username); - aki_packet_write_string(packet, &post->author.display_name); - aki_packet_write_string(packet, &post->author.profile_image_url); - aki_packet_write_string(packet, &post->title); - aki_packet_write_string(packet, &post->text); - aki_packet_write_sho_optional_int(packet, &post->likes); - aki_packet_write_sho_optional_int(packet, &post->reposts); - aki_packet_write_sho_optional_int(packet, &post->quotes); - aki_packet_write_sho_optional_int(packet, &post->comments); - aki_packet_write_sho_optional_int(packet, &post->views); - AKI_PACKET_WRITE_TYPE(packet, u32, post->media.size); - struct sho_post_media *media; - al_array_foreach_ptr(post->media, i, media) { - AKI_PACKET_WRITE_TYPE(packet, u8, media->type); - aki_packet_write_string(packet, &media->url); - aki_packet_write_string(packet, &media->thumbnail_url); - } - aki_packet_write_string(packet, &post->post.unique_id); - aki_packet_write_string(packet, &post->quoted.unique_id); - aki_packet_write_string(packet, &post->in_reply_to.unique_id); -} - -static void aki_packet_read_sho_optional_int(struct aki_packet *packet, sho_optional_int *o) -{ - AKI_PACKET_READ_TYPE(packet, s64, o->i); - AKI_PACKET_READ_TYPE(packet, bool, o->set); -} - -void aki_packet_read_sho_post(struct aki_packet *packet, struct sho_post *post) -{ - sho_post_reset(post); - AKI_PACKET_READ_TYPE(packet, u16, post->version); - AKI_PACKET_READ_TYPE(packet, u8, post->type); - aki_packet_read_string(packet, &post->unique_id); - aki_packet_read_string(packet, &post->url); - u32 dates_size; - AKI_PACKET_READ_TYPE(packet, u32, dates_size); - al_array_reserve(post->dates, dates_size); - for (u32 i = 0; i < dates_size; i++) { - struct sho_post_date date; - AKI_PACKET_READ_TYPE(packet, u8, date.type); - AKI_PACKET_READ_TYPE(packet, s64, date.timestamp); - al_array_push(post->dates, date); - } - aki_packet_read_string(packet, &post->author.unique_id); - aki_packet_read_string(packet, &post->author.username); - aki_packet_read_string(packet, &post->author.display_name); - aki_packet_read_string(packet, &post->author.profile_image_url); - aki_packet_read_string(packet, &post->title); - aki_packet_read_string(packet, &post->text); - aki_packet_read_sho_optional_int(packet, &post->likes); - aki_packet_read_sho_optional_int(packet, &post->reposts); - aki_packet_read_sho_optional_int(packet, &post->quotes); - aki_packet_read_sho_optional_int(packet, &post->comments); - aki_packet_read_sho_optional_int(packet, &post->views); - u32 media_size; - AKI_PACKET_READ_TYPE(packet, u32, media_size); - al_array_reserve(post->media, media_size); - for (u32 i = 0; i < media_size; i++) { - struct sho_post_media media; - AKI_PACKET_READ_TYPE(packet, u8, media.type); - aki_packet_read_string(packet, &media.url); - aki_packet_read_string(packet, &media.thumbnail_url); - al_array_push(post->media, media); - } - aki_packet_read_string(packet, &post->post.unique_id); - aki_packet_read_string(packet, &post->quoted.unique_id); - aki_packet_read_string(packet, &post->in_reply_to.unique_id); -} diff --git a/src/shoki/src/packet_ext.h b/src/shoki/src/packet_ext.h deleted file mode 100644 index ff0fda1..0000000 --- a/src/shoki/src/packet_ext.h +++ /dev/null @@ -1,8 +0,0 @@ -#pragma once - -#include <aki/common.h> - -#include "post.h" - -void aki_packet_write_sho_post(struct aki_packet *packet, struct sho_post *post); -void aki_packet_read_sho_post(struct aki_packet *packet, struct sho_post *post); diff --git a/src/shoki/src/post.c b/src/shoki/src/post.c deleted file mode 100644 index 22fb7ef..0000000 --- a/src/shoki/src/post.c +++ /dev/null @@ -1,55 +0,0 @@ -#include "post.h" - -void sho_post_reset(struct sho_post *post) -{ - al_bzero(post, sizeof(struct sho_post)); - al_array_init(post->dates); - al_array_init(post->media); -} - -void sho_post_add_media(struct sho_post *post, u8 type, str *url, str *thumbnail_url) -{ - al_array_push(post->media, ((struct sho_post_media){ - .type = type, .url = *url, .thumbnail_url = *thumbnail_url - })); -} - -void sho_post_add_date(struct sho_post *post, u8 type, s64 timestamp) -{ - al_array_push(post->dates, ((struct sho_post_date){ - .type = type, .timestamp = timestamp - })); -} - -void sho_post_clone(struct sho_post *dest, struct sho_post *src) -{ - sho_post_reset(dest); - dest->version = src->version; - dest->type = src->type; - al_str_clone(&dest->unique_id, &src->unique_id); - al_str_clone(&dest->url, &src->url); - al_array_clone(dest->dates, src->dates); - al_str_clone(&dest->author.unique_id, &src->author.unique_id); - al_str_clone(&dest->author.username, &src->author.username); - al_str_clone(&dest->author.display_name, &src->author.display_name); - al_str_clone(&dest->author.profile_image_url, &src->author.profile_image_url); - al_str_clone(&dest->title, &src->title); - al_str_clone(&dest->text, &src->text); - dest->likes = src->likes; - dest->reposts = src->reposts; - dest->quotes = src->quotes; - dest->comments = src->comments; - dest->views = src->views; - struct sho_post_media *media; - struct sho_post_media media_copy; - al_array_foreach_ptr(src->media, i, media) { - al_bzero(&media_copy, sizeof(struct sho_post_media)); - media_copy.type = media->type; - al_str_clone(&media_copy.url, &media->url); - al_str_clone(&media_copy.thumbnail_url, &media->thumbnail_url); - al_array_push(dest->media, media_copy); - } - al_str_clone(&dest->post.unique_id, &src->post.unique_id); - al_str_clone(&dest->quoted.unique_id, &src->quoted.unique_id); - al_str_clone(&dest->in_reply_to.unique_id, &src->in_reply_to.unique_id); -} diff --git a/src/shoki/src/post.h b/src/shoki/src/post.h deleted file mode 100644 index 6814b0f..0000000 --- a/src/shoki/src/post.h +++ /dev/null @@ -1,78 +0,0 @@ -#pragma once - -#include <al/types.h> -#include <al/str.h> -#include <al/array.h> - -enum { - SHOKI_MEDIA_IMAGE = 0, - SHOKI_MEDIA_VIDEO, - SHOKI_MEDIA_AUDIO -}; - -enum { - SHOKI_POST_POST = 0, - SHOKI_POST_REPOST -}; - -enum { - SHOKI_DATE_CREATED = 0, - SHOKI_DATE_EDITED, - SHOKI_DATE_ARCHIVED -}; - -typedef struct { - s64 i; - bool set; -} sho_optional_int; - -struct sho_post_date { - u8 type; - s64 timestamp; -}; - -struct sho_post_media { - u8 type; - str url; - str thumbnail_url; -}; - -typedef array(struct sho_post_date) sho_dates_array; -typedef array(struct sho_post_media) sho_media_array; - -struct sho_post_user { - str unique_id; - str username; - str display_name; - str profile_image_url; -}; - -struct sho_post_ref { - str unique_id; -}; - -struct sho_post { - u16 version; - u8 type; - str unique_id; - str raw_responses; - str url; - sho_dates_array dates; - struct sho_post_user author; - str title; - str text; - sho_optional_int likes; - sho_optional_int reposts; - sho_optional_int quotes; - sho_optional_int comments; - sho_optional_int views; - sho_media_array media; - struct sho_post_ref post; // reposted post. - struct sho_post_ref quoted; - struct sho_post_ref in_reply_to; -}; - -void sho_post_reset(struct sho_post *post); -void sho_post_add_media(struct sho_post *post, u8 type, str *url, str *thumbnail_url); -void sho_post_add_date(struct sho_post *post, u8 type, s64 timestamp); -void sho_post_clone(struct sho_post *dest, struct sho_post *src); diff --git a/src/shoki/src/post_cache.c b/src/shoki/src/post_cache.c deleted file mode 100644 index a8f02eb..0000000 --- a/src/shoki/src/post_cache.c +++ /dev/null @@ -1,30 +0,0 @@ -#include "post_cache.h" - -void sho_post_cache_init(struct sho_post_cache *cache) -{ - al_array_init(cache->cache); -} - -void sho_post_cache_push(struct sho_post_cache *cache, struct sho_post *post) -{ - struct sho_post *check = sho_post_cache_get(cache, &post->unique_id); - if (!check) { - struct sho_post *npost = al_alloc_object(struct sho_post); - sho_post_clone(npost, post); - al_array_push(cache->cache, npost); - } -} - -struct sho_post *sho_post_cache_get(struct sho_post_cache *cache, str *unique_id) -{ - struct sho_post *post; - al_array_foreach(cache->cache, i, post) { - if (al_str_eq(&post->unique_id, unique_id)) { - return post; - } - if (al_str_eq(&post->author.unique_id, unique_id)) { - return post; - } - } - return NULL; -} diff --git a/src/shoki/src/post_cache.h b/src/shoki/src/post_cache.h deleted file mode 100644 index 80cd0b7..0000000 --- a/src/shoki/src/post_cache.h +++ /dev/null @@ -1,13 +0,0 @@ -#pragma once - -#include <al/array.h> - -#include "post.h" - -struct sho_post_cache { - array(struct sho_post *) cache; -}; - -void sho_post_cache_init(struct sho_post_cache *cache); -void sho_post_cache_push(struct sho_post_cache *cache, struct sho_post *post); -struct sho_post *sho_post_cache_get(struct sho_post_cache *cache, str *unique_id); diff --git a/src/shoki/src/search.c b/src/shoki/src/search.c deleted file mode 100644 index b4b2157..0000000 --- a/src/shoki/src/search.c +++ /dev/null @@ -1,99 +0,0 @@ -#include <al/log.h> - -#include "../cpy/shoki.c" - -#include "search.h" - -bool sho_python_init(void) -{ - PyPreConfig pre; - PyPreConfig_InitPythonConfig(&pre); - pre.utf8_mode = 1; - Py_PreInitialize(&pre); - if (PyImport_AppendInittab("shoki", PyInit_shoki) == -1) { - al_log_error("shoki", "Could not extend in-built modules table."); - sho_python_close(); - return false; - } - Py_Initialize(); - PyObject *module = PyImport_ImportModule("shoki"); - if (!module) { - PyErr_Print(); - al_log_error("shoki", "Could not import module."); - sho_python_close(); - return false; - } - return true; -} - -void sho_result_init(struct sho_result *res) -{ - al_array_init(res->pages); -} - -static struct sho_result_page *page_at_index(struct sho_result *res, s32 num) -{ - struct sho_result_page *page; - al_array_foreach_ptr(res->pages, i, page) { - if (page->num == num) { - return page; - } - } - return NULL; -} - -void sho_result_add_post(struct sho_result *res, s32 num, struct sho_post *post) -{ - struct sho_result_page *page = page_at_index(res, num); - if (!page) { - al_array_push(res->pages, (struct sho_result_page){0}); - page = &al_array_last(res->pages); - page->num = num; - al_array_init(page->posts); - al_array_init(page->list); - } - al_array_push(page->posts, *post); -} - -void sho_result_add_to_list(struct sho_result *res, s32 num, str *unique_id) -{ - struct sho_result_page *page = page_at_index(res, num); - al_array_push(page->list, *unique_id); -} - -void sho_search_init(struct sho_search *search) -{ - search->id = -1; - search->page = 0; - sho_result_init(&search->result); -} - -s32 sho_search_more_results(struct sho_search *search, str *provider, str *query) -{ - if (search->id < 0) { - search->id = sho_client_search(provider, query); - } else { - search->page++; - } - if (search->id < 0) return -1; - al_log_info("shoki", "Loading page %i (%.*s).", search->page, AL_STR_PRINTF(query)); - if (sho_client_get_page(&search->result, search->id, search->page) < 0) { - return -1; - } - return search->page; -} - -void sho_search_free(struct sho_search *search) -{ - struct sho_result_page *page; - al_array_foreach_ptr(search->result.pages, i, page) { - al_array_free(page->posts); - al_array_free(page->list); - } - al_array_free(search->result.pages); -} - -void sho_python_close(void) -{ - if (Py_IsInitialized()) Py_Finalize(); -} diff --git a/src/shoki/src/search.h b/src/shoki/src/search.h deleted file mode 100644 index 97e40ae..0000000 --- a/src/shoki/src/search.h +++ /dev/null @@ -1,30 +0,0 @@ -#pragma once - -#include "post.h" - -struct sho_result_page { - s32 num; - array(struct sho_post) posts; - array(str) list; -}; - -struct sho_result { - array(struct sho_result_page) pages; -}; - -struct sho_search { - s32 id; - s32 page; - struct sho_result result; -}; - -bool sho_python_init(void); -void sho_python_close(void); - -void sho_result_init(struct sho_result *res); -void sho_result_add_post(struct sho_result *res, s32 num, struct sho_post *post); -void sho_result_add_to_list(struct sho_result *res, s32 num, str *unique_id); - -void sho_search_init(struct sho_search *search); -s32 sho_search_more_results(struct sho_search *search, str *provider, str *query); -void sho_search_free(struct sho_search *s); diff --git a/src/tree/commands.h b/src/tree/common.h index 1b1187e..4e3ec3a 100644 --- a/src/tree/commands.h +++ b/src/tree/common.h @@ -1,5 +1,9 @@ #pragma once +#define TREE_PORT 4356 +#define TREE_RESOURCE_PORT 4357 +#define TREE_STREAM_PORT 4358 + enum { TREE_NODE = 0, TREE_CLIENT, @@ -8,6 +12,8 @@ enum { enum { TREE_CMD_IDENTIFY = 0, + TREE_CMD_STATUS, TREE_CMD_SEARCH, + TREE_CMD_RESUME_SEARCH, TREE_CMD_ADD }; diff --git a/src/tree/list.c b/src/tree/list.c new file mode 100644 index 0000000..016c0de --- /dev/null +++ b/src/tree/list.c @@ -0,0 +1,112 @@ +#ifdef AKIYO_HAS_CURL +#include "../cache/handlers/http.h" +#endif + +#include "../libsink/common.h" + +#include "list.h" +#include "tree.h" + +void tree_list_init(struct tree_list *list, struct tree_server *tree, str *name) +{ + al_str_clone(&list->name, name); + list->set = -1; + list->current = 0; + al_array_init(list->entries); + al_array_init(list->sinks); + list->tree = tree; +} + +void tree_list_add_sink(struct tree_list *list, struct tree_sink *sink) +{ + al_array_push(list->sinks, sink); +} + +void tree_list_remove_sink(struct tree_list *list, struct tree_sink *sink) +{ + struct tree_sink *rsink; + al_array_foreach(list->sinks, i, rsink) { + if (rsink == sink) { + al_array_remove_at_iter(list->sinks, i); + break; + } + } +} + +bool tree_list_add(struct tree_list *list, str *unique_id, u32 index) +{ + struct tree_server *tree = list->tree; + struct sho_post *post = sho_post_cache_get(&tree->resources.cache, unique_id); + if (!post) return false; + + struct tree_list_entry *entry = al_alloc_object(struct tree_list_entry); + + entry->post = post; + entry->index = index; + + entry->entry = cch_handler_http_create(&al_array_at(entry->post->media, entry->index).url); + if (!entry->entry) { + al_free(entry); + return false; + } + entry->entry->handler->maybe_spawn_worker(entry->entry->handler, 0); + + entry->node_id = bmu_server_create_node(&tree->streams.server, entry->entry); + + entry->buffer_requested = false; + + al_array_push(list->entries, entry); + + tree_list_pump(list); + + return true; +} + +void tree_list_skip(struct tree_list *list, s32 n) +{ + s32 size = (s32)list->entries.size; + if (list->current + n < 0 || list->current + n >= size) { + return; + } + list->current += n; + tree_list_pump(list); +} + +static void send_buffer_cmd(struct tree_list *list, struct tree_list_entry *entry) +{ + struct aki_packet *packet; + struct tree_sink *sink; + al_array_foreach(list->sinks, i, sink) { + packet = aki_rpc_get_packet(&list->tree->server, CAMU_SINK_CMD_BUFFER); + aki_packet_write_str(packet, al_str_c("127.0.0.1")); + aki_packet_write_s32(packet, TREE_STREAM_PORT); + aki_packet_write_u16(packet, entry->node_id); + aki_rpc_connection_command(sink->conn, packet, NULL, NULL); + } +} + +static void send_set_cmd(struct tree_list *list, struct tree_list_entry *entry) +{ + struct aki_packet *packet; + struct tree_sink *sink; + al_array_foreach(list->sinks, i, sink) { + packet = aki_rpc_get_packet(&list->tree->server, CAMU_SINK_CMD_SET); + aki_packet_write_u16(packet, entry->node_id); + aki_rpc_connection_command(sink->conn, packet, NULL, NULL); + } +} + +void tree_list_pump(struct tree_list *list) +{ + s32 size = (s32)list->entries.size; + if (size <= list->current) return; + if (list->set != list->current) { + struct tree_list_entry *entry = al_array_at(list->entries, list->current); + if (!entry->buffer_requested) { + send_buffer_cmd(list, entry); + entry->buffer_requested = true; + } + send_set_cmd(list, entry); + list->set = list->current; + } +} diff --git a/src/tree/list.h b/src/tree/list.h new file mode 100644 index 0000000..ac7f8ee --- /dev/null +++ b/src/tree/list.h @@ -0,0 +1,31 @@ +#pragma once + +#include <al/str.h> +#include <al/array.h> + +#include "../cache/entry.h" + +struct tree_list_entry { + struct sho_post *post; + u32 index; + u16 node_id; + bool buffer_requested; + struct cch_entry *entry; +}; + +struct tree_sink; +struct tree_list { + str name; + s32 set; + s32 current; + array(struct tree_list_entry *) entries; + array(struct tree_sink *) sinks; + struct tree_server *tree; +}; + +void tree_list_init(struct tree_list *list, struct tree_server *tree, str *name); +void tree_list_add_sink(struct tree_list *list, struct tree_sink *sink); +void tree_list_remove_sink(struct tree_list *list, struct tree_sink *sink); +bool tree_list_add(struct tree_list *list, str *unique_id, u32 index); +void tree_list_skip(struct tree_list *list, s32 n); +void tree_list_pump(struct tree_list *list); diff --git a/src/tree/meson.build b/src/tree/meson.build index c7301f9..e9b8670 100644 --- a/src/tree/meson.build +++ b/src/tree/meson.build @@ -1,3 +1,7 @@ -tree_src = ['tree.c', 'resource_manager.c'] +tree_src = [ + 'tree.c', + 'resource_manager.c', + 'list.c' +] tree_deps = [common_deps, shoki, cache, cap, bimu, av] executable('tree', sources: tree_src, dependencies: tree_deps) diff --git a/src/tree/resource_manager.c b/src/tree/resource_manager.c index 0e95fbb..dd24310 100644 --- a/src/tree/resource_manager.c +++ b/src/tree/resource_manager.c @@ -18,7 +18,7 @@ static void packet_callback(void *userdata, struct aki_packet_stream *stream, st request->stream = stream; request->id = aki_packet_read_u16(packet); str unique_id; - aki_packet_read_string(packet, &unique_id); + aki_packet_read_str(packet, &unique_id); al_str_clone(&request->unique_id, &unique_id); request->index = aki_packet_read_u32(packet); request->packet = aki_packet_pool_get(&server->pool); @@ -40,8 +40,8 @@ static void packet_sent_callback(void *userdata, struct aki_packet *packet) { struct tree_resource_server *server = (struct tree_resource_server *)userdata; struct tree_resource_request *request = (struct tree_resource_request *)packet->userdata; - aki_packet_pool_return(&server->pool, packet); aki_http_request_close(&request->request); + aki_packet_pool_return(&server->pool, packet); } static void flushed_callback(void *userdata) @@ -70,6 +70,6 @@ bool tree_resource_server_init(struct tree_resource_server *server, void tree_resource_server_listen(struct tree_resource_server *server, struct aki_event_loop *loop, str *addr, s32 port) { - aki_packet_pool_init(&server->pool, 25, loop, packet_pool_callback, server); + aki_packet_pool_init(&server->pool, 50, loop, packet_pool_callback, server); aki_packet_stream_listen(&server->server, loop, addr, port); } diff --git a/src/tree/tree.c b/src/tree/tree.c index 0947be0..014397f 100644 --- a/src/tree/tree.c +++ b/src/tree/tree.c @@ -24,8 +24,12 @@ static void http_callback(void *userdata, struct aki_http_request *response, boo static bool request_resource(void *userdata, struct tree_resource_request *request) { struct tree_server *tree = (struct tree_server *)userdata; - struct sho_post *post = sho_post_cache_get(&tree->cache, &request->unique_id); - if (!post) return false; + struct sho_post *post = sho_post_cache_get(&tree->resources.cache, &request->unique_id); + struct aki_packet *packet = request->packet; + if (!post) { + aki_packet_pool_return(&request->server->pool, packet); + return false; + } str *url = NULL; if (al_str_eq(&post->author.unique_id, &request->unique_id)) { url = &post->author.profile_image_url; @@ -37,7 +41,10 @@ static bool request_resource(void *userdata, struct tree_resource_request *reque aki_http_request_init(&request->request); aki_http_set_url(&request->request.http, url); aki_http_set_user_agent(&request->request.http, USER_AGENT); - aki_http_request(&request->request, AKI_HTTP_GET, &tree->loop, http_callback, request); + if (!aki_http_request(&request->request, AKI_HTTP_GET, &tree->loop, http_callback, request)) { + aki_packet_pool_return(&request->server->pool, packet); + return false; + } return true; } @@ -66,6 +73,25 @@ static struct tree_user *get_user_by_connection(struct tree_server *tree, struct return NULL; } +static void send_status(struct tree_server *server, struct tree_user *user, struct aki_rpc_connection *conn) +{ + struct aki_packet *packet = aki_rpc_get_packet(&server->server, TREE_CMD_STATUS); + aki_packet_write_u32(packet, user->lists.size); + struct tree_list *list; + al_array_foreach(user->lists, i, list) { + aki_packet_write_str(packet, &list->name); + } + aki_packet_write_u32(packet, user->search.searches.size); + struct sho_search *search; + al_array_foreach(user->search.searches, i, search) { + aki_packet_write_s32(packet, search->id); + aki_packet_write_str(packet, &search->module); + aki_packet_write_str(packet, &search->query); + aki_packet_write_s32(packet, search->page); + } + aki_rpc_connection_command(conn, packet, NULL, NULL); +} + static bool identify_command_callback(void *userdata, struct aki_rpc_connection *conn, struct aki_packet *packet, struct aki_packet *rpacket) { @@ -76,18 +102,21 @@ static bool identify_command_callback(void *userdata, struct aki_rpc_connection struct tree_node *node = al_alloc_object(struct tree_node); node->conn = conn; al_array_push(tree->nodes, node); + al_log_info("tree", "New node."); break; } case TREE_CLIENT: { str username; - aki_packet_read_string(packet, &username); + aki_packet_read_str(packet, &username); struct tree_user *user = get_user_by_username(tree, &username); if (user) { - al_log_info("tree", "User \"%.*s\" logged in.", AL_STR_PRINTF(&user->username)); struct tree_client *client = al_alloc_object(struct tree_client); client->conn = conn; + client->user = user; al_array_push(tree->clients, client); al_array_push(user->clients, client); + send_status(tree, user, conn); + al_log_info("tree", "User \"%.*s\" logged in.", AL_STR_PRINTF(&user->username)); } break; } @@ -95,6 +124,10 @@ static bool identify_command_callback(void *userdata, struct aki_rpc_connection struct tree_sink *sink = al_alloc_object(struct tree_sink); sink->conn = conn; al_array_push(tree->sinks, sink); + struct tree_user *user = al_array_at(tree->users, 0); + struct tree_list *list = al_array_at(user->lists, 0); + tree_list_add_sink(list, sink); + al_log_info("tree", "New sink."); break; } } @@ -102,22 +135,26 @@ static bool identify_command_callback(void *userdata, struct aki_rpc_connection return true; } -static struct tree_query *query_from_id(struct tree_user *user, s32 id) +static bool search_command_callback(void *userdata, struct aki_rpc_connection *conn, + struct aki_packet *packet, struct aki_packet *rpacket) { - struct tree_query *query; - al_array_foreach_ptr(user->queries, i, query) { - if (query->search.id == id) { - return query; - } + struct tree_server *tree = (struct tree_server *)userdata; + struct tree_user *user = get_user_by_connection(tree, conn); + if (!user) { + aki_packet_write_s32(rpacket, -1); + goto out; } - al_array_push(user->queries, (struct tree_query){}); - query = &al_array_last(user->queries); - query->id = id; - sho_search_init(&query->search); - return query; + str module, query; + aki_packet_read_str(packet, &module); + aki_packet_read_str(packet, &query); + s32 id = sho_client_create_search(&user->search, &module, &query); + aki_packet_write_s32(rpacket, id); +out: + aki_packet_free(packet); + return true; } -static bool search_command_callback(void *userdata, struct aki_rpc_connection *conn, +static bool resume_search_command_callback(void *userdata, struct aki_rpc_connection *conn, struct aki_packet *packet, struct aki_packet *rpacket) { struct tree_server *tree = (struct tree_server *)userdata; @@ -127,66 +164,134 @@ static bool search_command_callback(void *userdata, struct aki_rpc_connection *c goto out; } s32 id = aki_packet_read_s32(packet); - struct tree_query *query = query_from_id(user, id); - str provider, query_str; - aki_packet_read_string(packet, &provider); - aki_packet_read_string(packet, &query_str); - s32 page_num = sho_search_more_results(&query->search, &provider, &query_str); - if (page_num < 0) { - al_log_error("tree", "Search failed."); + s32 req_page = aki_packet_read_s32(packet); + struct sho_search *search = sho_client_get_search(&user->search, id); + if (!search) { aki_packet_write_s32(rpacket, -1); goto out; } - struct sho_result_page *page = &al_array_at(query->search.result.pages, page_num); - aki_packet_write_s32(rpacket, query->search.id); - aki_packet_write_s32(rpacket, page_num); + if (req_page >= 0) { + if (!sho_search_from_page(search, req_page)) { + aki_packet_write_s32(rpacket, -1); + goto out; + } + } else { + if (!sho_search_more_results(search)) { + aki_packet_write_s32(rpacket, -1); + goto out; + } + } + aki_packet_write_s32(rpacket, 0); + aki_packet_write_s32(rpacket, search->id); + aki_packet_write_s32(rpacket, search->page); + struct sho_result_page *page = &al_array_at(search->pages, search->page); aki_packet_write_u32(rpacket, page->posts.size); struct sho_post *post; al_array_foreach_ptr(page->posts, i, post) { aki_packet_write_sho_post(rpacket, post); - sho_post_cache_push(&tree->cache, post); } aki_packet_write_u32(rpacket, page->list.size); str *unique_id; al_array_foreach_ptr(page->list, i, unique_id) { - aki_packet_write_string(rpacket, unique_id); + aki_packet_write_str(rpacket, unique_id); } out: aki_packet_free(packet); return true; } +static bool add_command_callback(void *userdata, struct aki_rpc_connection *conn, + struct aki_packet *packet, struct aki_packet *rpacket) +{ + struct tree_server *tree = (struct tree_server *)userdata; + (void)rpacket; + struct tree_user *user = get_user_by_connection(tree, conn); + if (!user) goto out; + + str unique_id; + aki_packet_read_str(packet, &unique_id); + u32 index = aki_packet_read_u32(packet); + + struct tree_list *list = al_array_at(user->lists, 0); + tree_list_add(list, &unique_id, index); + tree_list_skip(list, 1); + +out: + aki_packet_free(packet); + return false; +} + static void connection_callback(void *userdata, struct aki_rpc_connection *conn) { (void)userdata; (void)conn; } +static void cleanup_node(struct tree_node *node) +{ + al_free(node); +} + +static void cleanup_client(struct tree_client *client) +{ + al_free(client); +} + +static void cleanup_sink(struct tree_sink *sink) +{ + al_free(sink); +} + static void connection_closed_callback(void *userdata, struct aki_rpc_connection *conn) { - (void)userdata; - (void)conn; + struct tree_server *tree = (struct tree_server *)userdata; + + struct tree_node *node; + al_array_foreach(tree->nodes, i, node) { + if (node->conn == conn) { + al_log_info("tree", "Node removed."); + cleanup_node(node); + al_array_remove_at_iter(tree->nodes, i); + break; + } + } + + struct tree_client *client; + al_array_foreach(tree->clients, i, client) { + if (client->conn == conn) { + al_log_info("tree", "User \"%.*s\" logged out.", AL_STR_PRINTF(&client->user->username)); + cleanup_client(client); + al_array_remove_at_iter(tree->clients, i); + break; + } + } + + struct tree_sink *sink; + al_array_foreach(tree->sinks, i, sink) { + if (sink->conn == conn) { + al_log_info("tree", "Sink removed."); + cleanup_sink(sink); + al_array_remove_at_iter(tree->sinks, i); + struct tree_user *user = al_array_at(tree->users, 0); + struct tree_list *list = al_array_at(user->lists, 0); + tree_list_remove_sink(list, sink); + break; + } + } } static struct aki_rpc_command commands[] = { { .op = TREE_CMD_IDENTIFY, .callback = identify_command_callback, .userdata = NULL }, // Client commands. { .op = TREE_CMD_SEARCH, .callback = search_command_callback, .userdata = NULL }, + { .op = TREE_CMD_RESUME_SEARCH, .callback = resume_search_command_callback, .userdata = NULL }, + { .op = TREE_CMD_ADD, .callback = add_command_callback, .userdata = NULL } }; -static struct tree_server tree; - -void sigint_handler(s32 signum) -{ - (void)signum; - // explode. - exit(EXIT_SUCCESS); -} - static bool open_user(struct tree_server *tree, struct aki_dir_entry *dir) { struct aki_file file; - if (!aki_file_open(&file, &dir->path, false)) { + if (!aki_file_open(&file, &dir->path, 0)) { return false; } str s; @@ -195,10 +300,10 @@ static bool open_user(struct tree_server *tree, struct aki_dir_entry *dir) json_t *json = json_loadb(s.data, s.len, 0, &error); struct tree_user *user = al_alloc_object(struct tree_user); al_str_from(&user->username, json_string_value(json_object_get(json, "username"))); - al_array_init(user->queries); + sho_client_init(&user->search, &tree->resources.cache); al_array_init(user->lists); - struct tree_list default_list; - al_str_from(&default_list.name, "default"); + struct tree_list *default_list = al_alloc_object(struct tree_list); + tree_list_init(default_list, tree, al_str_c("default")); al_array_push(user->lists, default_list); al_array_init(user->clients); al_log_info("tree", "Loaded user \"%.*s\"", AL_STR_PRINTF(&user->username)); @@ -232,25 +337,15 @@ static bool open_db(struct tree_server *tree, str *path) return true; } -static bool cap_callback(void *userdata, u8 op, str *name, str *unique_id, void *opaque) +static void sigint_handler(s32 signum) { - (void)userdata; - (void)name; - (void)unique_id; - (void)opaque; - switch (op) { - case CAP_BUFFER: - return true; - case CAP_SWAP: - break; - case CAP_SET: - break; - case CAP_UNLOAD: - break; - } - return true; + (void)signum; + // explode. + exit(EXIT_SUCCESS); } +static struct tree_server tree = { 0 }; + s32 main(void) { aki_common_init(); @@ -266,7 +361,7 @@ s32 main(void) return EXIT_FAILURE; } - sho_post_cache_init(&tree.cache); + sho_post_cache_init(&tree.resources.cache); bool py_init = sho_python_init(); @@ -280,13 +375,11 @@ s32 main(void) } aki_rpc_listen(&tree.server, &tree.loop, al_str_c("0.0.0.0"), TREE_PORT); - tree_resource_server_init(&tree.resource_server, request_resource, &tree); - tree_resource_server_listen(&tree.resource_server, &tree.loop, al_str_c("0.0.0.0"), TREE_RESOURCE_PORT); - - cap_init(&tree.cap, cap_callback, &tree); + tree_resource_server_init(&tree.resources.server, request_resource, &tree); + tree_resource_server_listen(&tree.resources.server, &tree.loop, al_str_c("0.0.0.0"), TREE_RESOURCE_PORT); - bmu_server_init(&tree.stream_server); - bmu_server_listen(&tree.stream_server, &tree.loop, al_str_c("0.0.0.0"), TREE_STREAM_PORT); + bmu_server_init(&tree.streams.server); + bmu_server_listen(&tree.streams.server, &tree.loop, al_str_c("0.0.0.0"), TREE_STREAM_PORT); aki_event_loop_run(&tree.loop); diff --git a/src/tree/tree.h b/src/tree/tree.h index 06f98b1..478306a 100644 --- a/src/tree/tree.h +++ b/src/tree/tree.h @@ -1,47 +1,35 @@ #pragma once #include <aki/rpc2.h> +#include <sho/post.h> +#include <sho/post_cache.h> +#include <sho/search.h> -#include "../shoki/src/post_cache.h" -#include "../shoki/src/search.h" -#include "../shoki/src/packet_ext.h" #include "../bimu/server.h" -#include "../fruits/cap/cap.h" -#include "commands.h" +#include "common.h" #include "resource_manager.h" +#include "list.h" #define CAMU_DB_PATH "/home/andrew/c/camu/data/camu_db_test" -#define TREE_PORT 4356 -#define TREE_RESOURCE_PORT 4357 -#define TREE_STREAM_PORT 4358 - struct tree_node { struct aki_rpc_connection *conn; }; struct tree_client { struct aki_rpc_connection *conn; + struct tree_user *user; }; struct tree_sink { struct aki_rpc_connection *conn; }; -struct tree_query { - s32 id; - struct sho_search search; -}; - -struct tree_list { - str name; -}; - struct tree_user { str username; - array(struct tree_query) queries; - array(struct tree_list) lists; + struct sho_client search; + array(struct tree_list *) lists; array(struct tree_client *) clients; }; @@ -52,8 +40,11 @@ struct tree_server { array(struct tree_client *) clients; array(struct tree_sink *) sinks; array(struct tree_user *) users; - struct tree_resource_server resource_server; - struct sho_post_cache cache; - struct cap_runner cap; - struct bmu_server stream_server; + struct { + struct sho_post_cache cache; + struct tree_resource_server server; + } resources; + struct { + struct bmu_server server; + } streams; }; diff --git a/src/util/color_palette.c b/src/util/color_palette.c index e55c9e1..53c2c68 100644 --- a/src/util/color_palette.c +++ b/src/util/color_palette.c @@ -32,7 +32,7 @@ static char *normal_colors[16] = { bool camu_color_palette_init(str *path) { struct aki_file file; - if (!aki_file_open(&file, path, false)) { + if (!aki_file_open(&file, path, 0)) { return false; } str s; diff --git a/src/util/queue.h b/src/util/queue.h index 032bb55..4f184ba 100644 --- a/src/util/queue.h +++ b/src/util/queue.h @@ -36,6 +36,16 @@ aki_mutex_unlock(&(q).mutex); \ } while (0) +#define camu_queue_try_pop(q, s, r) \ + do { \ + aki_mutex_lock(&(q).mutex); \ + s = (q).a.size; \ + if (s > 0) { \ + al_array_pop_at((q).a, 0, r); \ + } \ + aki_mutex_unlock(&(q).mutex); \ + } while (0) + #define camu_queue_free(q) \ do { \ aki_mutex_destroy(&(q).mutex); \ |