diff options
Diffstat (limited to 'src')
45 files changed, 901 insertions, 510 deletions
diff --git a/src/buffer/clock.c b/src/buffer/clock.c index 032e44a..d26e4a8 100644 --- a/src/buffer/clock.c +++ b/src/buffer/clock.c @@ -2,29 +2,55 @@ #include "clock.h" -void camu_clock_set(struct camu_clock *clock, u64 base, u64 ts, u64 delay) +void camu_clock_set(struct camu_clock *clock, struct shrb_seek_req *req) { - al_atomic_f64_store(&clock->base, base / 1000000.0, AL_ATOMIC_RELAXED); al_atomic_f64_store(&clock->tick, 0.0, AL_ATOMIC_RELAXED); - al_atomic_u64_store(&clock->start, ts + delay, AL_ATOMIC_RELAXED); - clock->delay = delay; - clock->pause_at = -1.0; + if (req->paused) { + clock->pause_at = req->paused_at / 1000000.0; + al_atomic_f64_store(&clock->base, clock->pause_at, AL_ATOMIC_RELAXED); + al_atomic_u64_store(&clock->start, 0Lu, AL_ATOMIC_RELAXED); + } else { + clock->pause_at = -1.0; + al_atomic_f64_store(&clock->base, req->base / 1000000.0, AL_ATOMIC_RELAXED); + al_atomic_u64_store(&clock->start, req->ts + req->delay, AL_ATOMIC_RELAXED); + clock->ts = req->ts; + } + clock->offset = 0L; + clock->delay = req->delay; } f64 calc_tick_internal(struct camu_clock *clock) { - s64 diff = clock->delay == 0 ? 0 : ((al_atomic_u64_load(&clock->start, AL_ATOMIC_ACQUIRE)) - aki_get_timestamp()); + s64 diff = clock->delay == 0Lu ? 0L + : ((al_atomic_u64_load(&clock->start, AL_ATOMIC_ACQUIRE) + clock->offset) + - aki_get_timestamp()); f64 tick = aki_get_tick() - al_atomic_f64_load(&clock->base, AL_ATOMIC_RELAXED); + //al_printf("DIFFERENCE: %f\n", diff / 1000000.0); tick += diff / 1000000.0; al_atomic_f64_store(&clock->tick, tick, AL_ATOMIC_RELAXED); - al_atomic_u64_store(&clock->start, 0, AL_ATOMIC_RELEASE); + al_atomic_u64_store(&clock->start, 0Lu, AL_ATOMIC_RELEASE); return tick; } bool camu_clock_is_late(struct camu_clock *clock) { s64 diff = (s64)(al_atomic_u64_load(&clock->start, AL_ATOMIC_RELAXED)) - aki_get_timestamp(); - return clock->delay > 0 && diff < 0; + return clock->delay > 0L && diff < 0L; +} + +void camu_clock_ready(struct camu_clock *clock) +{ + if (clock->pause_at != -1.0) { + clock->pause_at = -1.0; + return; + } + camu_clock_resume(clock); +} + +void camu_clock_offset(struct camu_clock *clock, s64 offset) +{ + clock->offset += offset; + //al_printf("OFFSET: %f\n", clock->offset / 1000000.0); } void camu_clock_resume(struct camu_clock *clock) @@ -32,27 +58,31 @@ void camu_clock_resume(struct camu_clock *clock) f64 tick = al_atomic_f64_load(&clock->tick, AL_ATOMIC_ACQUIRE); if (tick != 0.0) return; u64 start = al_atomic_u64_load(&clock->start, AL_ATOMIC_RELAXED); - if (start != 0L) { + if (start != 0Lu) { // No longer report paused, calculate tick on first call to get_pts(). al_atomic_f64_store(&clock->tick, -1.0, AL_ATOMIC_RELEASE); } else { tick = al_atomic_f64_load(&clock->base, AL_ATOMIC_RELAXED); al_atomic_f64_store(&clock->tick, aki_get_tick() - tick, AL_ATOMIC_RELEASE); } + clock->pause_at = -1.0; } void camu_clock_arm_resume(struct camu_clock *clock, u64 ts) { al_atomic_u64_store(&clock->start, ts, AL_ATOMIC_RELAXED); al_atomic_f64_store(&clock->tick, -1.0, AL_ATOMIC_RELAXED); + clock->pause_at = -1.0; } -void camu_clock_pause(struct camu_clock *clock) +void camu_clock_pause(struct camu_clock *clock, f64 base) { f64 tick = al_atomic_f64_load(&clock->tick, AL_ATOMIC_ACQUIRE); if (tick == 0.0) return; - al_atomic_f64_store(&clock->base, aki_get_tick() - tick, AL_ATOMIC_RELAXED); + if (base == -1.0) base = aki_get_tick() - tick; al_atomic_f64_store(&clock->tick, 0.0, AL_ATOMIC_RELEASE); + clock->pause_at = base; + al_atomic_f64_store(&clock->base, clock->pause_at, AL_ATOMIC_RELAXED); } void camu_clock_arm_pause(struct camu_clock *clock, f64 pts) @@ -63,13 +93,13 @@ void camu_clock_arm_pause(struct camu_clock *clock, f64 pts) bool camu_clock_is_paused(struct camu_clock *clock) { f64 tick = al_atomic_f64_load(&clock->tick, AL_ATOMIC_RELAXED); - if (clock->pause_at >= 0.0 && (aki_get_tick() - tick) >= clock->pause_at) { + if (tick == 0.0) return true; + if (clock->pause_at >= 0.0 && aki_get_tick() - tick >= clock->pause_at) { al_atomic_f64_store(&clock->base, clock->pause_at, AL_ATOMIC_RELAXED); al_atomic_f64_store(&clock->tick, 0.0, AL_ATOMIC_RELAXED); - clock->pause_at = -1.0; return true; } - return tick == 0.0; + return false; } f64 camu_clock_get_base_pts(struct camu_clock *clock) diff --git a/src/buffer/clock.h b/src/buffer/clock.h index bdd50d8..6f505be 100644 --- a/src/buffer/clock.h +++ b/src/buffer/clock.h @@ -3,18 +3,24 @@ #include <aki/thread.h> #include <al/atomic.h> +#include "../shrub/handler.h" + struct camu_clock { atomic_f64 base, tick; atomic_u64 start; + u64 ts; + s64 offset; u64 delay; f64 pause_at; }; -void camu_clock_set(struct camu_clock *clock, u64 base, u64 ts, u64 delay); +void camu_clock_set(struct camu_clock *clock, struct shrb_seek_req *req); bool camu_clock_is_late(struct camu_clock *clock); +void camu_clock_ready(struct camu_clock *clock); +void camu_clock_offset(struct camu_clock *clock, s64 offset); void camu_clock_resume(struct camu_clock *clock); void camu_clock_arm_resume(struct camu_clock *clock, u64 ts); -void camu_clock_pause(struct camu_clock *clock); +void camu_clock_pause(struct camu_clock *clock, f64 base); void camu_clock_arm_pause(struct camu_clock *clock, f64 pts); bool camu_clock_is_paused(struct camu_clock *clock); f64 camu_clock_get_base_pts(struct camu_clock *clock); diff --git a/src/buffer/meson.build b/src/buffer/meson.build index d037d6b..a2def3e 100644 --- a/src/buffer/meson.build +++ b/src/buffer/meson.build @@ -1,8 +1,11 @@ -no_video = false +no_video = meson.is_subproject() buffer_src = ['audio.c', 'clock.c', 'peak_buffer.c'] +buffer_deps = [common_deps] +buffer_args = [] if not no_video buffer_src += ['video.c'] +else + buffer_args += ['-DCAMU_SINK_NO_VIDEO'] endif -buffer_deps = [common_deps] buffer = declare_dependency(sources: buffer_src, - dependencies: buffer_deps) + dependencies: buffer_deps, compile_args: buffer_args) diff --git a/src/buffer/video.c b/src/buffer/video.c index 390b181..3b9be20 100644 --- a/src/buffer/video.c +++ b/src/buffer/video.c @@ -22,6 +22,7 @@ bool camu_video_buffer_init(struct camu_video_buffer *buf, struct camu_clock *cl #ifdef CAMU_SCREEN_THREADED al_atomic_bool_store(&buf->ref, false, AL_ATOMIC_RELAXED); #endif + buf->view.mode = CAMU_VIEW_NONE; return true; } @@ -163,8 +164,8 @@ bool camu_video_buffer_read(struct camu_video_buffer *buf, void *out) f64 pts = camu_clock_get_pts(buf->clock, buf->offset); if (pts > base) base = pts; } - u8 ret = buf->queue->read(buf->queue, base, out); u8 flow = al_atomic_u8_load(&buf->flow, AL_ATOMIC_RELAXED); + u8 ret = buf->queue->read(buf->queue, base, out); if (flow == FLUSHED && (ret == CAMU_QUEUE_EOF || (buf->single_frame && ret == CAMU_QUEUE_OK))) { buf->callback(buf->userdata, CAMU_BUFFER_EOF); al_atomic_u8_store(&buf->flow, SIGNALED, AL_ATOMIC_RELAXED); diff --git a/src/buffer/video.h b/src/buffer/video.h index 5b92dc6..5e18b33 100644 --- a/src/buffer/video.h +++ b/src/buffer/video.h @@ -6,6 +6,7 @@ #include "../codec/ffmpeg/scaler.h" #include "../render/renderer.h" #include "../screen/screen.h" +#include "../screen/view.h" #include "clock.h" #include "frame_queue.h" @@ -31,6 +32,7 @@ struct camu_video_buffer { #ifdef CAMU_SCREEN_THREADED atomic_bool ref; #endif + struct camu_view view; void (*callback)(void *, u8); void *userdata; diff --git a/src/cache/handlers/http.c b/src/cache/handlers/http.c index 97abf79..789b247 100644 --- a/src/cache/handlers/http.c +++ b/src/cache/handlers/http.c @@ -6,7 +6,7 @@ #include "http.h" -#define USER_AGENT al_str_c("Mozilla/5.0 (X11; Linux x86_64; rv:96.0) Gecko/20100101 Firefox/96.0") +#define USER_AGENT al_str_c("Mozilla/5.0 (Windows NT 10.0; rv:122.0) Gecko/20100101 Firefox/122.0") static bool handler_http_can_seek(struct cch_handler *handler) { @@ -17,6 +17,8 @@ static bool handler_http_can_seek(struct cch_handler *handler) static void disable_waiters_internal(struct cch_handler_http *http) { aki_mutex_lock(&http->mutex); + // Don't allow any further waits. + http->disabled = true; struct cch_handler_wait *wait; al_array_foreach(http->handler.waits, i, wait) { aki_mutex_lock(&wait->mutex); @@ -37,8 +39,8 @@ static size_t http_callback(void *userdata, u8 op, u8 *buf, s64 int0) break; } case AKI_HTTP_ERROR: { - disable_waiters_internal(http); // TODO: Untested. al_log_debug("cache_handler_http", "Error."); + disable_waiters_internal(http); break; } case AKI_HTTP_RESPONSE_CODE: { @@ -123,8 +125,8 @@ static bool handler_http_wait_for_range(struct cch_handler *handler, struct cch_ { struct cch_handler_http *http = (struct cch_handler_http *)handler; aki_mutex_lock(&http->mutex); - bool canceled = false; - if (cch_entry_get_size(http->handler.entry) < 0 || http->pointer < wait->end) { + bool canceled = http->disabled; + if (!canceled && (cch_entry_get_size(http->handler.entry) < 0 || http->pointer < wait->end)) { aki_mutex_lock(&wait->mutex); if (!wait->disabled) { al_array_push(http->handler.waits, wait); @@ -201,6 +203,7 @@ struct cch_entry *cch_handler_http_create(str *url) struct cch_handler_http *http = (struct cch_handler_http *)entry->handler; al_str_clone(&http->url, url); http->pointer = 0; + http->disabled = false; al_array_init(http->requests); camu_queue_init(http->queue); aki_event_loop_init(&http->loop); diff --git a/src/cache/handlers/http.h b/src/cache/handlers/http.h index f79a39a..e8fb2e0 100644 --- a/src/cache/handlers/http.h +++ b/src/cache/handlers/http.h @@ -13,6 +13,7 @@ struct cch_handler_http { struct cch_handler handler; off_t pointer; str url; + bool disabled; array(struct aki_http) requests; queue(s32) queue; struct aki_event_loop loop; diff --git a/src/codec/ffmpeg/avio.c b/src/codec/ffmpeg/avio.c index 12dd468..143bc1f 100644 --- a/src/codec/ffmpeg/avio.c +++ b/src/codec/ffmpeg/avio.c @@ -1,6 +1,6 @@ #include "avio.h" -#include "../cache/handle.h" +#include "../../cache/handle.h" s32 camu_avio_read(void *data, u8 *buf, s32 buf_size) { diff --git a/src/codec/ffmpeg/encoder.c b/src/codec/ffmpeg/encoder.c index 9ceaa99..4617baf 100644 --- a/src/codec/ffmpeg/encoder.c +++ b/src/codec/ffmpeg/encoder.c @@ -15,6 +15,20 @@ #define FRAME_SIZE 20 #define CHANNELS 2 +static void free_frame_data(struct camu_ff_encoder *enc) +{ + if (enc->frame && enc->frame->nb_samples > 0) av_freep(&enc->frame->data[0]); +} + +static void alloc_frame_data(struct camu_ff_encoder *enc, s32 frame_size) +{ + if (enc->frame->nb_samples >= frame_size) return; + free_frame_data(enc); + s32 nb_channels = enc->frame->ch_layout.nb_channels; + av_samples_alloc(enc->frame->data, NULL, nb_channels, frame_size, enc->frame->format, 0); + enc->frame->nb_samples = frame_size; +} + bool camu_ff_encoder_init(struct camu_ff_encoder *enc, char *encoder_name, char *ext) { al_memset(enc, 0, sizeof(struct camu_ff_encoder)); @@ -99,20 +113,6 @@ void camu_ff_encoder_push(struct camu_ff_encoder *enc, u8 **data, s32 sample_cou av_audio_fifo_write(enc->audio_fifo, (void **)data, sample_count); } -static void free_frame_data(struct camu_ff_encoder *enc) -{ - if (enc->frame && enc->frame->nb_samples > 0) av_freep(&enc->frame->data[0]); -} - -static void alloc_frame_data(struct camu_ff_encoder *enc, s32 frame_size) -{ - if (enc->frame->nb_samples >= frame_size) return; - free_frame_data(enc); - s32 nb_channels = enc->frame->ch_layout.nb_channels; - av_samples_alloc(enc->frame->data, NULL, nb_channels, frame_size, enc->frame->format, 0); - enc->frame->nb_samples = frame_size; -} - static bool receive_packets(struct camu_ff_encoder *enc) { s32 ret; @@ -141,10 +141,15 @@ static s32 send_frame(struct camu_ff_encoder *enc, AVFrame *in_frame) return ret; } +static s32 read_frame(struct camu_ff_encoder *enc, s32 frame_size) +{ + alloc_frame_data(enc, frame_size); + return av_audio_fifo_read(enc->audio_fifo, (void **)enc->frame->data, frame_size); +} + bool camu_ff_encoder_process(struct camu_ff_encoder *enc, bool flush) { s32 samples_available = av_audio_fifo_size(enc->audio_fifo); - if (samples_available == 0 && flush) { // Flush encoder. send_frame(enc, NULL); @@ -154,29 +159,34 @@ bool camu_ff_encoder_process(struct camu_ff_encoder *enc, bool flush) } s32 frame_size = enc->codec_context->frame_size; - if (samples_available == 0 || (samples_available < frame_size && !flush)) { return false; } - if (samples_available < frame_size) frame_size = samples_available; - - alloc_frame_data(enc, frame_size); - av_audio_fifo_read(enc->audio_fifo, (void **)enc->frame->data, frame_size); - s32 ret; do { - ret = send_frame(enc, enc->frame); - // TODO: Check ret < 0 ? - if (!receive_packets(enc)) return false; - } while (ret == AVERROR(EAGAIN)); - - enc->frame->pts += frame_size; + if (samples_available < frame_size) frame_size = samples_available; + read_frame(enc, frame_size); + samples_available -= frame_size; + do { + ret = send_frame(enc, enc->frame); + if (ret < 0) break; + if (!receive_packets(enc)) return false; + } while (ret == AVERROR(EAGAIN)); + if (ret < 0) break; + enc->frame->pts += frame_size; + } while (samples_available > 0); - // ret < 0 means encoder erroed while sending frame. + // ret < 0 means encoder errored while sending frame. return ret == 0; } +void camu_ff_encoder_reset(struct camu_ff_encoder *enc) +{ + avcodec_flush_buffers(enc->codec_context); + av_audio_fifo_reset(enc->audio_fifo); +} + void camu_ff_encoder_close(struct camu_ff_encoder *enc) { if (enc->codec_context) avcodec_free_context(&enc->codec_context); diff --git a/src/codec/ffmpeg/encoder.h b/src/codec/ffmpeg/encoder.h index 6cba9ac..9bfe8ea 100644 --- a/src/codec/ffmpeg/encoder.h +++ b/src/codec/ffmpeg/encoder.h @@ -23,4 +23,5 @@ struct camu_ff_encoder { bool camu_ff_encoder_init(struct camu_ff_encoder *enc, char *encoder_name, char *ext); void camu_ff_encoder_push(struct camu_ff_encoder *enc, u8 **data, s32 sample_count); bool camu_ff_encoder_process(struct camu_ff_encoder *enc, bool flush); +void camu_ff_encoder_reset(struct camu_ff_encoder *enc); void camu_ff_encoder_close(struct camu_ff_encoder *enc); diff --git a/src/fruits/sink/sink.c b/src/fruits/sink/sink.c index cbd82bd..2b3d451 100644 --- a/src/fruits/sink/sink.c +++ b/src/fruits/sink/sink.c @@ -129,7 +129,13 @@ 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_init(&c->loop); + camu_sink_init(&c->sink, &c->loop, &c->mixer, c->renderer); + c->sink.callback = sink_callback; + c->sink.userdata = c; + if (camu_sink_connect(&c->sink, CAMU_SERVER_IP, CAMU_PORT)) { + aki_event_loop_run(&c->loop); + } aki_event_loop_destroy(&c->loop); return 0; } @@ -154,8 +160,6 @@ s32 wmain(s32 argc, wchar_t **argv) signal(SIGINT, sigint_handler); - aki_event_loop_init(&c.loop); - c.scr.callback = screen_callback; c.scr.userdata = &c; if (!camu_screen_init(&c.scr) || !camu_screen_create_window(&c.scr, "cmv")) { @@ -172,13 +176,6 @@ s32 wmain(s32 argc, wchar_t **argv) goto err; } - camu_sink_init(&c.sink, &c.loop, &c.mixer, c.renderer); - c.sink.callback = sink_callback; - c.sink.userdata = &c; - if (!camu_sink_connect(&c.sink, CAMU_SERVER_IP, CAMU_PORT)) { - goto err; - } - struct aki_thread thread0; aki_thread_create(&thread0, event_loop_thread, &c); @@ -206,7 +203,6 @@ s32 wmain(s32 argc, wchar_t **argv) return EXIT_SUCCESS; err: - aki_event_loop_destroy(&c.loop); aki_common_close(); return EXIT_FAILURE; } diff --git a/src/libsink/sink.c b/src/libsink/sink.c index 65a082e..c6727ab 100644 --- a/src/libsink/sink.c +++ b/src/libsink/sink.c @@ -129,8 +129,8 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) break; } case SKIP: { - struct aki_packet *packet = aki_rpc_get_packet(&sink->client, CAMU_SRV_LIST_ACTION); - aki_packet_write_u8(packet, CAMU_LIST_SKIP); + struct aki_packet *packet = aki_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION); + aki_packet_write_u8(packet, CAMU_SKIP); s32 sequence = sink->current ? sink->current->sequence : CAMU_SEQUENCE_INVALID; aki_packet_write_s32(packet, sequence); aki_packet_write_s32(packet, cmd->value.i); @@ -153,7 +153,7 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) } #endif } else { - camu_clock_pause(&entry->clock); + camu_clock_pause(&entry->clock, -1.0); #ifndef CAMU_SINK_NO_VIDEO if (sink->video.state == SINK_PLAYING) { sink->callback(sink->userdata, CAMU_SINK_STOP, CAMU_SINK_VIDEO, NULL); @@ -167,8 +167,8 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) f64 pts = camu_clock_get_pts(&entry->clock, 0.0); if (pts > base) base = pts; } - struct aki_packet *packet = aki_rpc_get_packet(&sink->client, CAMU_SRV_LIST_ACTION); - aki_packet_write_u8(packet, CAMU_LIST_TOGGLE_PAUSE); + struct aki_packet *packet = aki_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION); + aki_packet_write_u8(packet, CAMU_TOGGLE_PAUSE); s32 sequence = sink->current ? sink->current->sequence : CAMU_SEQUENCE_INVALID; aki_packet_write_s32(packet, sequence); aki_packet_write_u64(packet, base * 1000000.0); @@ -177,8 +177,8 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) break; } case SEEK: { - struct aki_packet *packet = aki_rpc_get_packet(&sink->client, CAMU_SRV_LIST_ACTION); - aki_packet_write_u8(packet, CAMU_LIST_SEEK); + struct aki_packet *packet = aki_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION); + aki_packet_write_u8(packet, CAMU_SEEK); s32 sequence = sink->current ? sink->current->sequence : CAMU_SEQUENCE_INVALID; aki_packet_write_s32(packet, sequence); aki_packet_write_f64(packet, cmd->value.f); @@ -186,8 +186,8 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) break; } case FINISHED: { - struct aki_packet *packet = aki_rpc_get_packet(&sink->client, CAMU_SRV_LIST_ACTION); - aki_packet_write_u8(packet, CAMU_LIST_FINISHED); + struct aki_packet *packet = aki_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION); + aki_packet_write_u8(packet, CAMU_FINISHED); aki_packet_write_s32(packet, cmd->value.i); aki_rpc_connection_command(sink->conn, packet, NULL, NULL); break; @@ -257,7 +257,7 @@ void add_audio_if_set_and_buffered(struct camu_sink_entry *entry) camu_audio_buffer_unpause(&entry->audio.buf); entry->sink->callback(entry->sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf); if (can_resume) { - camu_clock_resume(&entry->clock); + camu_clock_ready(&entry->clock); maybe_remove_previous(entry->sink); } //queue_cmd(entry->sink, (struct camu_sink_cmd){ @@ -283,7 +283,7 @@ void add_video_if_set_and_buffered(struct camu_sink_entry *entry) }); entry->sink->callback(entry->sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_VIDEO, &entry->video.buf); if (can_resume) { - camu_clock_resume(&entry->clock); + camu_clock_ready(&entry->clock); maybe_remove_previous(entry->sink); } //queue_cmd(entry->sink, (struct camu_sink_cmd){ @@ -301,11 +301,12 @@ void add_video_if_set_and_buffered(struct camu_sink_entry *entry) static void audio_buffer_callback(void *userdata, u8 op) { struct camu_sink_entry *entry = (struct camu_sink_entry *)userdata; + struct camu_sink *sink = entry->sink; switch (op) { case CAMU_BUFFER_BUFFERED: - aki_mutex_lock(&entry->sink->mutex); + aki_mutex_lock(&sink->mutex); add_audio_if_set_and_buffered(entry); - aki_mutex_unlock(&entry->sink->mutex); + aki_mutex_unlock(&sink->mutex); break; case CAMU_BUFFER_CORK: shrb_vcr_stream_cork(entry->audio.stream); @@ -314,25 +315,22 @@ static void audio_buffer_callback(void *userdata, u8 op) shrb_vcr_stream_uncork(entry->audio.stream); break; case CAMU_BUFFER_PAUSED: - aki_mutex_lock(&entry->sink->mutex); + aki_mutex_lock(&sink->mutex); if (!entry->audio.armed) { - queue_cmd(entry->sink, (struct camu_sink_cmd){ + queue_cmd(sink, (struct camu_sink_cmd){ .op = STOP, .value.i = CAMU_SINK_AUDIO }); } - aki_mutex_unlock(&entry->sink->mutex); + aki_mutex_unlock(&sink->mutex); break; case CAMU_BUFFER_EOF: { - // TODO: Video only case not handled (video not image). - struct camu_sink *sink = entry->sink; bool swapped = false; aki_mutex_lock(&sink->mutex); - s32 sequence = sink->current->sequence; if (sink->queued) { + al_log_info("sink", "Swapping audio buffers (gapless)."); swap_buffers_internal(sink); swapped = true; - al_log_info("sink", "Audio buffers swapped (gapless)."); } aki_mutex_unlock(&sink->mutex); if (!swapped) { @@ -347,9 +345,9 @@ static void audio_buffer_callback(void *userdata, u8 op) //}); #endif } - queue_cmd(entry->sink, (struct camu_sink_cmd){ + queue_cmd(sink, (struct camu_sink_cmd){ .op = FINISHED, - .value.i = sequence + .value.i = entry->sequence }); break; } @@ -360,11 +358,12 @@ static void audio_buffer_callback(void *userdata, u8 op) static void video_buffer_callback(void *userdata, u8 op) { struct camu_sink_entry *entry = (struct camu_sink_entry *)userdata; + struct camu_sink *sink = entry->sink; switch (op) { case CAMU_BUFFER_BUFFERED: - aki_mutex_lock(&entry->sink->mutex); + aki_mutex_lock(&sink->mutex); add_video_if_set_and_buffered(entry); - aki_mutex_unlock(&entry->sink->mutex); + aki_mutex_unlock(&sink->mutex); break; case CAMU_BUFFER_CORK: shrb_vcr_stream_cork(entry->video.stream); @@ -372,13 +371,23 @@ static void video_buffer_callback(void *userdata, u8 op) case CAMU_BUFFER_UNCORK: shrb_vcr_stream_uncork(entry->video.stream); break; - case CAMU_BUFFER_EOF: - queue_cmd(entry->sink, (struct camu_sink_cmd){ + case CAMU_BUFFER_EOF: { + queue_cmd(sink, (struct camu_sink_cmd){ .op = STOP, .value.i = CAMU_SINK_VIDEO }); + aki_mutex_lock(&sink->mutex); + bool no_audio = entry->audio.state == BUFFER_INIT; + aki_mutex_unlock(&sink->mutex); + if (no_audio) { + queue_cmd(sink, (struct camu_sink_cmd){ + .op = FINISHED, + .value.i = entry->sequence + }); + } break; } + } } #endif @@ -419,8 +428,9 @@ static void client_callback(void *userdata, u8 op, struct shrb_vcr_stream *strea } case SHRUB_CLIENT_SET: { struct shrb_seek_req *req = (struct shrb_seek_req *)opaque; - //req->delay = SHRUB_DELAY_IGNORE; - camu_clock_set(&entry->clock, req->base, req->ts, req->delay); + req->delay = SHRUB_DELAY_IGNORE; + camu_clock_set(&entry->clock, req); + if (req->paused) req->base = req->paused_at; break; } case SHRUB_CLIENT_PAUSE: { @@ -432,9 +442,12 @@ static void client_callback(void *userdata, u8 op, struct shrb_vcr_stream *strea break; } case SHRUB_CLIENT_RESUME: { - u64 ts = *(u64 *)opaque; aki_mutex_lock(&entry->sink->mutex); - camu_clock_arm_resume(&entry->clock, ts); + struct shrb_resume_req *req = (struct shrb_resume_req *)opaque; + if (sink->queued) { + camu_clock_offset(&sink->queued->clock, req->offset); + } + camu_clock_arm_resume(&entry->clock, req->start); if (sink->audio.state == SINK_PAUSED) { sink->callback(sink->userdata, CAMU_SINK_START, CAMU_SINK_AUDIO, NULL); sink->audio.state = SINK_PLAYING; @@ -569,17 +582,16 @@ static struct camu_sink_entry *ensure_entry_buffered_internal(struct camu_sink * 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, 0.0, renderer); + camu_video_buffer_init(&entry->video.buf, &entry->clock, 0.0, sink->video.renderer); entry->video.buf.callback = video_buffer_callback; entry->video.buf.userdata = entry; #endif entry->client.callback = client_callback; entry->client.userdata = entry; + // Assume we can error out at any point after connect is called (even from within connect() itself). shrb_client_connect(&entry->client, sink->loop, addr, port, node_id); return entry; @@ -598,16 +610,19 @@ static bool set_command_callback(void *userdata, struct aki_rpc_connection *conn u16 node_id = aki_packet_read_u16(packet); s32 sequence = aki_packet_read_s32(packet); u64 start = aki_packet_read_u64(packet); + u8 paused = aki_packet_read_u8(packet); + u64 paused_at = aki_packet_read_u64(packet); aki_mutex_lock(&sink->mutex); struct camu_sink_entry *entry = ensure_entry_buffered_internal(sink, &addr, port, node_id); entry->sequence = sequence; if (entry == sink->current) goto out; - // This will only have an effect if the clock is already set. - camu_clock_arm_resume(&entry->clock, start + SHRUB_BASE_DELAY); + if (!paused) { + // This will only have an effect if the clock is already set. + camu_clock_arm_resume(&entry->clock, start + SHRUB_BASE_DELAY); + } if (sink->current) { - // TODOODO, pause_at sent from list calc difference in clock and set. - camu_clock_pause(&sink->current->clock); + camu_clock_pause(&sink->current->clock, paused_at / 1000000.0); al_array_push(sink->previous, sink->current); } sink->current = entry; @@ -662,7 +677,7 @@ 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, CAMU_SRV_IDENTIFY); + struct aki_packet *packet = aki_rpc_get_packet(&sink->client, CAMU_SERVER_IDENTIFY); aki_packet_write_u8(packet, CAMU_SINK); aki_rpc_connection_command(sink->conn, packet, NULL, NULL); } @@ -676,14 +691,17 @@ static void connection_closed_callback(void *userdata, struct aki_rpc_connection 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); + aki_rpc_init(&sink->client, sink->loop, 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); + if (!aki_rpc_prepare_client(&sink->client, AKI_SOCKET_TCP, CAMU_MULTIPLEX_RPC)) { + return false; + } + aki_rpc_connect(&sink->client, addr, port); + return true; } struct camu_sink_entry *camu_sink_get_current(struct camu_sink *sink) @@ -697,6 +715,14 @@ void camu_sink_return_current(struct camu_sink *sink) aki_mutex_unlock(&sink->mutex); } +void camu_sink_add(struct camu_sink *sink, str *line) +{ + struct aki_packet *packet = aki_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION); + aki_packet_write_u8(packet, CAMU_ADD); + aki_packet_write_str(packet, line); + aki_rpc_connection_command(sink->conn, packet, NULL, NULL); +} + void camu_sink_skip(struct camu_sink *sink, s32 n) { queue_cmd(sink, (struct camu_sink_cmd){ @@ -763,7 +789,7 @@ void camu_sink_stop(struct camu_sink *sink) void camu_sink_close(struct camu_sink *sink) { // TODO: Make sure no commands can come in after this. - aki_rpc_disconnect(&sink->client); + aki_rpc_conn_disconnect(sink->client.conn); aki_mutex_lock(&sink->mutex); struct camu_sink_entry *entry; al_array_foreach(sink->entries, i, entry) { diff --git a/src/libsink/sink.h b/src/libsink/sink.h index 4d7dc06..b789c9a 100644 --- a/src/libsink/sink.h +++ b/src/libsink/sink.h @@ -1,6 +1,5 @@ #pragma once -//#define CAMU_SINK_NO_VIDEO //#define CAMU_SINK_LOCAL_PAUSE #include <al/str.h> @@ -100,6 +99,7 @@ bool camu_sink_init(struct camu_sink *sink, struct aki_event_loop *loop, bool camu_sink_connect(struct camu_sink *sink, str *addr, s32 port); struct camu_sink_entry *camu_sink_get_current(struct camu_sink *sink); void camu_sink_return_current(struct camu_sink *sink); +void camu_sink_add(struct camu_sink *sink, str *line); void camu_sink_skip(struct camu_sink *sink, s32 n); void camu_sink_toggle_pause(struct camu_sink *sink); void camu_sink_seek(struct camu_sink *sink, f64 pos); diff --git a/src/list/list.c b/src/list/list.c index 3af75e4..d5be3dc 100644 --- a/src/list/list.c +++ b/src/list/list.c @@ -2,6 +2,7 @@ #include <aki/thread.h> #include "../libsink/common.h" +#include "../shrub/handler.h" #include "list.h" @@ -9,6 +10,7 @@ void camu_list_init(struct camu_list *list, str *name) { al_str_clone(&list->name, name); list->current = -1; + list->prev = -1; list->idle = true; list->sequence = 0; al_array_init(list->entries); @@ -24,19 +26,26 @@ static void camu_list_pump(struct camu_list *list) if (list->current + 1 < size) { upcoming = al_array_at(list->entries, list->current + 1); } + struct camu_list_entry *prev = NULL; + if (list->prev >= 0) { + prev = al_array_at(list->entries, list->prev); + list->prev = -1; + } struct camu_list_sink *sink; al_array_foreach(list->sinks, i, sink) { if (sink->set != list->current) { - sink->callback(sink->userdata, CAMU_SINK_SET, entry->opaque, entry->sequence); + void *prev_opaque = NULL; + if (prev) prev_opaque = prev->opaque; + sink->callback(sink->userdata, CAMU_SINK_SET, entry->opaque, prev_opaque, entry->sequence); sink->set = list->current; } if (upcoming) { - sink->callback(sink->userdata, CAMU_SINK_QUEUE, upcoming->opaque, upcoming->sequence); + sink->callback(sink->userdata, CAMU_SINK_QUEUE, upcoming->opaque, entry->opaque, upcoming->sequence); } } } -void camu_list_add_sink(struct camu_list *list, void (*callback)(void *, u8, void *, s32), +void camu_list_add_sink(struct camu_list *list, void (*callback)(void *, u8, void *, void *, s32), void *userdata) { struct camu_list_sink *sink = al_alloc_object(struct camu_list_sink); @@ -68,12 +77,12 @@ void camu_list_add(struct camu_list *list, void *opaque, void (*callback)(void * entry->callback = callback; entry->duration = duration; entry->start = aki_get_timestamp(); + entry->paused = false; if (!list->idle) { struct camu_list_entry *prev = al_array_at(list->entries, size - 1); u64 end = prev->start + prev->duration; if (end > entry->start) { - u64 diff = end - entry->start; - entry->start += diff; + entry->start += end - entry->start; } } al_array_push(list->entries, entry); @@ -83,10 +92,7 @@ void camu_list_add(struct camu_list *list, void *opaque, void (*callback)(void * list->idle = false; } camu_list_pump(list); - struct camu_list_timing timing = { - .start = entry->start - }; - entry->callback(entry->opaque, CAMU_LIST_ENTRY_IMPULSE, &timing); + entry->callback(entry->opaque, CAMU_LIST_IMPULSE, &entry->start); } static s32 index_of_entry_from_sequence(struct camu_list *list, s32 sequence) @@ -106,12 +112,13 @@ void camu_list_skip(struct camu_list *list, s32 sequence, s32 n) s32 index = index_of_entry_from_sequence(list, sequence); if (index == CAMU_SEQUENCE_INVALID) index = list->current; if (index + n < 0 || index + n >= size || index + n == list->current) return; + list->prev = list->current; + struct camu_list_entry *prev = al_array_at(list->entries, list->prev); + prev->callback(prev->opaque, CAMU_LIST_PAUSE, NULL); list->current = index + n; struct camu_list_entry *entry = al_array_at(list->entries, list->current); - struct camu_list_timing timing = { - .start = aki_get_timestamp() - }; - entry->callback(entry->opaque, CAMU_LIST_ENTRY_IMPULSE, &timing); + entry->start = aki_get_timestamp(); + entry->callback(entry->opaque, CAMU_LIST_IMPULSE, &entry->start); camu_list_pump(list); } @@ -128,7 +135,20 @@ void camu_list_toggle_pause(struct camu_list *list, s32 sequence, u64 pos) s32 index = index_of_entry_from_sequence(list, sequence); if (index == CAMU_SEQUENCE_INVALID) return; struct camu_list_entry *entry = al_array_at(list->entries, index); - entry->callback(entry->opaque, CAMU_LIST_ENTRY_TOGGLE_PAUSE, &pos); + if (entry->paused) { + u64 ts = aki_get_timestamp(); + struct shrb_resume_req req = { + .start = ts, + .offset = ts - entry->start + }; + entry->start = ts; + entry->callback(entry->opaque, CAMU_LIST_USER_RESUME, &req); + entry->paused = false; + } else { + entry->start = aki_get_timestamp(); + entry->callback(entry->opaque, CAMU_LIST_USER_PAUSE, &pos); + entry->paused = true; + } } void camu_list_seek(struct camu_list *list, s32 sequence, f64 percent) @@ -137,8 +157,10 @@ void camu_list_seek(struct camu_list *list, s32 sequence, f64 percent) if (index == CAMU_SEQUENCE_INVALID) return; list->idle = false; struct camu_list_entry *entry = al_array_at(list->entries, index); + entry->start = aki_get_timestamp(); + entry->callback(entry->opaque, CAMU_LIST_IMPULSE, &entry->start); u64 pos = percent * entry->duration; - entry->callback(entry->opaque, CAMU_LIST_ENTRY_SEEK, &pos); + entry->callback(entry->opaque, CAMU_LIST_SEEK, &pos); } void camu_list_finished(struct camu_list *list, s32 sequence) @@ -170,3 +192,66 @@ void camu_list_shuffle(struct camu_list *list) } camu_list_pump(list); } + +static s64 camu_db_num_from_path(str *path) +{ + s32 slash = al_str_rfind(path, '/'); + if (slash < 0) return -1; + str s = *al_str_substr(path, slash + 1, path->len); + s32 underscore = al_str_find(&s, '_'); + if (underscore < 0) return -1; + s = *al_str_substr(&s, underscore + 1, s.len); + underscore = al_str_find(&s, '_'); + if (underscore < 0) return -1; + s = *al_str_substr(&s, underscore + 1, s.len); + underscore = al_str_find(&s, '_'); + if (underscore < 0) return -1; + s = *al_str_substr(&s, 0, underscore); + s64 num = al_str_to_long(&s, 10); + if (num == INT64_MIN || num == INT64_MAX) { + return -1; + } + return num; +} + +static s32 camu_db_compare(void *a, void *b) +{ + struct camu_list_entry *aa = *((struct camu_list_entry **)a); + struct camu_list_entry *bb = *((struct camu_list_entry **)b); + str *astr = NULL, *bstr = NULL; + aa->callback(aa->opaque, CAMU_LIST_ENTRY_ID, &astr); + bb->callback(bb->opaque, CAMU_LIST_ENTRY_ID, &bstr); + s64 anum = camu_db_num_from_path(astr); + s64 bnum = camu_db_num_from_path(bstr); + if (anum > bnum) { + return -1; + } else if (anum < bnum) { + return 1; + } + return 0; +} + +void camu_list_sort(struct camu_list *list) +{ + al_array_sort(list->entries, struct camu_list_entry *, camu_db_compare); + struct camu_list_sink *sink; + al_array_foreach(list->sinks, i, sink) { + sink->set = -1; + } + camu_list_pump(list); +} + +void camu_list_reverse(struct camu_list *list) +{ + u32 size = list->entries.size; + for (u32 i = 0; i < size; i++) { + u32 tail = size - (i + 1); + if (tail <= i) break; + AL_SWAP(al_array_at(list->entries, i), al_array_at(list->entries, tail), struct camu_list_entry *); + } + struct camu_list_sink *sink; + al_array_foreach(list->sinks, i, sink) { + sink->set = -1; + } + camu_list_pump(list); +} diff --git a/src/list/list.h b/src/list/list.h index edd0218..4873265 100644 --- a/src/list/list.h +++ b/src/list/list.h @@ -5,14 +5,14 @@ #define CAMU_SEQUENCE_INVALID INT32_MAX +// TODO: Improve naming between this and server LIST_ACTION's enum { - CAMU_LIST_ENTRY_IMPULSE = 0, - CAMU_LIST_ENTRY_TOGGLE_PAUSE, - CAMU_LIST_ENTRY_SEEK -}; - -struct camu_list_timing { - u64 start; + CAMU_LIST_ENTRY_ID = 0, + CAMU_LIST_IMPULSE, + CAMU_LIST_USER_PAUSE, + CAMU_LIST_USER_RESUME, + CAMU_LIST_PAUSE, // Pause at current position. + CAMU_LIST_SEEK }; struct camu_list_entry { @@ -21,27 +21,28 @@ struct camu_list_entry { void (*callback)(void *, u8, void *); u64 duration; u64 start; - u64 paused_at; + bool paused; }; struct camu_list_sink { s32 set; bool (*action)(void *, u8, void *); - void (*callback)(void *, u8, void *, s32); + void (*callback)(void *, u8, void *, void *, s32); void *userdata; }; struct camu_list { str name; s32 current; + s32 prev; bool idle; - s32 sequence; // Next entry sequence. + s32 sequence; // Holds the next entry sequence. array(struct camu_list_entry *) entries; array(struct camu_list_sink *) sinks; }; void camu_list_init(struct camu_list *list, str *name); -void camu_list_add_sink(struct camu_list *list, void (*callback)(void *, u8, void *, s32), +void camu_list_add_sink(struct camu_list *list, void (*callback)(void *, u8, void *, void *, s32), void *userdata); void camu_list_remove_sink(struct camu_list *list, void *userdata); void camu_list_add(struct camu_list *list, void *opaque, void (*callback)(void *, u8, void *), @@ -52,3 +53,5 @@ void camu_list_toggle_pause(struct camu_list *list, s32 sequence, u64 pos); void camu_list_seek(struct camu_list *list, s32 sequence, f64 percent); void camu_list_finished(struct camu_list *list, s32 sequence); void camu_list_shuffle(struct camu_list *list); +void camu_list_sort(struct camu_list *list); +void camu_list_reverse(struct camu_list *list); diff --git a/src/mixer/audio.h b/src/mixer/audio.h index 0e8e105..6624896 100644 --- a/src/mixer/audio.h +++ b/src/mixer/audio.h @@ -17,6 +17,6 @@ struct camu_audio { void (*free)(struct camu_audio **); - s32 (*data_callback)(void *, u8 *, s32); + s32 (*data_callback)(void *, u8 *, s32, bool *); void *userdata; }; diff --git a/src/mixer/audio_miniaudio.c b/src/mixer/audio_miniaudio.c index a7a1889..26c01b2 100644 --- a/src/mixer/audio_miniaudio.c +++ b/src/mixer/audio_miniaudio.c @@ -1,5 +1,6 @@ #include <al/log.h> +#define MINIAUDIO_IMPLEMENTATION #include "audio_miniaudio.h" static ma_backend backends[] = { @@ -107,6 +108,7 @@ static void audio_miniaudio_pick_format(struct camu_audio *audio, struct camu_ff struct camu_audio_miniaudio *ma = (struct camu_audio_miniaudio *)audio; fmt->req_format = av_sample_fmt_from_miniaudio(ma->device.playback.internalFormat); fmt->req_channel_count = ma->device.playback.internalChannels; + av_channel_layout_default(&fmt->req_channel_layout, fmt->req_channel_count); fmt->req_sample_rate = ma->device.playback.internalSampleRate; } @@ -114,7 +116,8 @@ static void data_callback(ma_device *device, void *output, const void *input, u3 { struct camu_audio_miniaudio *ma = (struct camu_audio_miniaudio *)device->pUserData; (void)input; - ma->a.data_callback(ma->a.userdata, output, frame_count); + bool silence; + ma->a.data_callback(ma->a.userdata, output, frame_count, &silence); } #define DEFAULT_PERIODS 3 diff --git a/src/mixer/audio_miniaudio.h b/src/mixer/audio_miniaudio.h index 5d5cd03..3d7f010 100644 --- a/src/mixer/audio_miniaudio.h +++ b/src/mixer/audio_miniaudio.h @@ -1,6 +1,5 @@ #pragma once -#define MINIAUDIO_IMPLEMENTATION #include <miniaudio.h> #include "audio.h" diff --git a/src/mixer/mixer.c b/src/mixer/mixer.c index 31a67b0..5caee1b 100644 --- a/src/mixer/mixer.c +++ b/src/mixer/mixer.c @@ -3,12 +3,13 @@ #include "mixer.h" #include "audio.h" -static s32 data_callback(void *userdata, u8 *data, s32 frame_count) +static s32 data_callback(void *userdata, u8 *data, s32 frame_count, bool *silence) { struct camu_mixer *mixer = (struct camu_mixer *)userdata; size_t req = camu_ff_resample_fmt_samples_to_bytes(&mixer->fmt, (size_t)frame_count); if (UNLIKELY(mixer->paused)) { al_memset(data, 0, req); + *silence = true; } else { #ifdef CAMU_MIXER_THREADED if (al_atomic_bool_load(&mixer->queued, AL_ATOMIC_RELAXED)) { @@ -18,6 +19,7 @@ static s32 data_callback(void *userdata, u8 *data, s32 frame_count) do { if (mixer->buffers.size == 0) { al_memset(data, 0, req); + *silence = true; } else { struct camu_audio_buffer *buf = al_array_last(mixer->buffers); size_t signal; @@ -31,6 +33,7 @@ static s32 data_callback(void *userdata, u8 *data, s32 frame_count) req = req - signal; continue; } + *silence = false; } break; } while (1); @@ -62,7 +65,6 @@ bool camu_mixer_init(struct camu_mixer *mixer, struct camu_audio *audio) void camu_mixer_pick_format(struct camu_mixer *mixer, struct camu_ff_resample_fmt *fmt) { mixer->audio->pick_format(mixer->audio, fmt); - av_channel_layout_default(&fmt->req_channel_layout, fmt->req_channel_count); } f64 camu_mixer_get_latency(struct camu_mixer *mixer) diff --git a/src/portal/py/base.py b/src/portal/py/base.py index c0a0b4f..cc883af 100644 --- a/src/portal/py/base.py +++ b/src/portal/py/base.py @@ -7,12 +7,14 @@ from collections import OrderedDict from typing import Optional, Self, Any from post import Post, User +USER_AGENT = 'Mozilla/5.0 (Windows NT 10.0; rv:122.0) Gecko/20100101 Firefox/122.0' + class Method(Enum): HEAD = "HEAD" GET = "GET" POST = "POST" -ParsedJson = Any +type ParsedJson = Any class Search(): def __init__(self): diff --git a/src/portal/py/config.def.py b/src/portal/py/config.def.py index c4e11c6..8c0dab9 100644 --- a/src/portal/py/config.def.py +++ b/src/portal/py/config.def.py @@ -11,6 +11,8 @@ TWITTER_TIMEOUT = 5 # seconds FANBOX_SESSID = '' +PIXIV_SESSID = '' +PIXIV_USERID = '' PIXIV_REFRESH_TOKEN = '' INSTAGRAM_USERNAME = '' @@ -18,3 +20,7 @@ INSTAGRAM_PASSWORD = '' INSTAGRAM_SESSION_ID = '' INSTAGRAM_SETTINGS_PATH = Path('{}/ig_settings.json'.format(BASE_DIR)) INSTAGRAM_USER_AGENT = '' + +PATREON_USER_ID = '' +PATREON_UUID = '' +PATREON_SESSION_ID = '' diff --git a/src/portal/py/modules/__init__.py b/src/portal/py/modules/__init__.py index 480660c..994b25a 100644 --- a/src/portal/py/modules/__init__.py +++ b/src/portal/py/modules/__init__.py @@ -6,17 +6,19 @@ from modules.pixiv_app import PixivAppModule from modules.pixiv_web import PixivWebModule from modules.fanbox import FanboxModule from modules.instagram import InstagramModule +from modules.patreon import PatreonModule ALL_MODULES = { 'youtube': (YoutubeModule(), []), # 'twitter': (TwitterModule( # config.TWITTER_ACCESS_TOKEN, config.TWITTER_ACCESS_TOKEN_SECRET, # config.TWITTER_CONSUMER_TOKEN, config.TWITTER_CONSUMER_TOKEN_SECRET), []), -# 'twitter': (TwitterScrapeModule(config.TWITTER_COOKIES_PATH), []), -# 'pixiv_web': (PixivWebModule(config.PIXIV_SESSID), []), + 'twitter': (TwitterScrapeModule(config.TWITTER_COOKIES_PATH), []), + 'pixiv_web': (PixivWebModule(config.PIXIV_SESSID, config.PIXIV_USERID), []), # 'pixiv_app': (PixivAppModule(config.PIXIV_REFRESH_TOKEN), []), # 'fanbox': (FanboxModule(config.FANBOX_SESSID), []), # 'instagram': (InstagramModule( # config.INSTAGRAM_USER_AGENT, config.INSTAGRAM_SETTINGS_PATH, # config.INSTAGRAM_SESSION_ID), []) + 'patreon': (PatreonModule(config.PATREON_SESSION_ID, config.PATREON_UUID, config.PATREON_USER_ID), []) } diff --git a/src/portal/py/modules/fanbox.py b/src/portal/py/modules/fanbox.py index 01e66c3..4323d6b 100644 --- a/src/portal/py/modules/fanbox.py +++ b/src/portal/py/modules/fanbox.py @@ -2,7 +2,7 @@ import json import httpx from typing import Optional, Any from json.decoder import JSONDecodeError -from base import Search, Module, Method, ParsedJson +from base import USER_AGENT, Search, Module, Method, ParsedJson from post import MediaUrl, PostType, DateType, Post, User, Image, File, Tag from query_parser import QueryParser from modules.common import parse_pixiv_date, get_current_utc_time @@ -173,7 +173,7 @@ class FanboxModule(Module): self.headers['Accept-Language'] = 'en-US,en;q=0.5' self.headers['Origin'] = 'https://www.fanbox.cc' self.headers['Referer'] = 'https://www.fanbox.cc/' - self.headers['User-Agent'] = 'Mozilla/5.0 (Windows NT 10.0; rv:122.0) Gecko/20100101 Firefox/122.0' + self.headers['User-Agent'] = USER_AGENT self.cookies['FANBOXSESSID'] = sessid def search(self, query: str, *extra_args: Any) -> Optional[Search]: diff --git a/src/portal/py/modules/instagram.py b/src/portal/py/modules/instagram.py index 325caf6..d107566 100644 --- a/src/portal/py/modules/instagram.py +++ b/src/portal/py/modules/instagram.py @@ -3,7 +3,7 @@ import log import email.utils from typing import Optional, Any from pathlib import Path -from base import Module, Search +from base import USER_AGENT, Module, Search from post import MediaUrl, PostType, DateType, Post, User, Image, Video from modules.common import get_current_utc_time from instagrapi import Client @@ -149,6 +149,6 @@ class InstagramModule(Module): return { 'urls': [post.media[key].url], 'headers': { - 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; rv:122.0) Gecko/20100101 Firefox/122.0' + 'User-Agent': USER_AGENT } } diff --git a/src/portal/py/modules/patreon.py b/src/portal/py/modules/patreon.py new file mode 100644 index 0000000..40c3255 --- /dev/null +++ b/src/portal/py/modules/patreon.py @@ -0,0 +1,124 @@ +import json +import httpx +from json import JSONDecodeError +from typing import Optional, Any +from base import USER_AGENT, Search, Module, Method, ParsedJson +from post import MediaUrl, PostType, DateType, Post, User, Media, Image, Animation, Tag, TagType +from query_parser import QueryParser + +BASE_URL = 'https://www.patreon.com/api' + +class PatreonBase(Search): + REACH_LIMIT = 3 + + def __init__(self, userdata: Any): + super().__init__() + self.module: PatreonModule = userdata + self.page: int = 0 + self.cursor: Optional[str] = None + + def check_api_response(self, response: Optional[httpx.Response]) -> Optional[ParsedJson]: + if not response: + return None + try: + obj = response.json() + except JSONDecodeError: + return None + return obj + + def search_user(self, query: str) -> Optional[User]: + url = f'{BASE_URL}/search' + params = { + 'q': query, + 'page[number]': '1', + 'json-api-version': '1.0', + 'json-api-use-default-includes': 'false', + 'include': '[]' + } + obj = self.check_api_response(self.module.do_request(Method.GET, url, params=params)) + if not obj: + return None + for result in obj['data']: + prefix = result['id'].find('campaign_') + if prefix >= 0: + campaign = result['id'][prefix + len('campaign_'):] + unique_id = 'patreon:c:' + campaign + user = User(unique_id=unique_id, username=result['attributes']['creator_name'], display_name=result['attributes']['name']) + return user + return None + +class PatreonUser(PatreonBase): + def __init__(self, userdata: Any, arg: str): + super().__init__(userdata) + self.username: str = arg + self.user: Optional[User] = None + self.campaign: str + + def request_page(self, num: int) -> bool: + if num >= self.page + self.REACH_LIMIT: + return False + if not self.user: + user = self.search_user(self.username) + if not user: + self.errored = True + return False + self.campaign = user.unique_id[len('patreon:c:'):] + request_satisfied = False + for i in range(self.page, num + 1): + url = f'{BASE_URL}/posts' + params = { + 'include': 'campaign,access_rules,attachments,audio,audio_preview.null,drop,images,media,native_video_insights,poll.choices,poll.current_user_responses.user,poll.current_user_responses.choice,poll.current_user_responses.poll,user,user_defined_tags,ti_checks,content_unlock_options.product_variant.null', + 'fields[campaign]': 'currency,show_audio_post_download_links,avatar_photo_url,avatar_photo_image_urls,earnings_visibility,is_nsfw,is_monthly,name,url', + 'fields[post]': 'change_visibility_at,comment_count,commenter_count,content,current_user_can_comment,current_user_can_delete,current_user_can_report,current_user_can_view,current_user_comment_disallowed_reason,current_user_has_liked,embed,image,insights_last_updated_at,is_paid,like_count,meta_image_url,min_cents_pledged_to_view,post_file,post_metadata,published_at,patreon_url,post_type,pledge_url,preview_asset_type,thumbnail,share_images,thumbnail_url,teaser_text,title,upgrade_url,url,was_posted_by_campaign_owner,has_ti_violation,moderation_status,post_level_suspension_removal_date,pls_one_liners_by_category,video_preview,view_count', + 'fields[post_tag]': 'tag_type,value', + 'fields[user]': 'image_url,full_name,url', + 'fields[access_rule]': 'access_rule_type,amount_cents', + 'fields[media]': 'id,image_urls,display,download_url,metadata,file_name', + 'fields[native_video_insights]': 'average_view_duration,average_view_pct,has_preview,id,last_updated_at,num_views,preview_views,video_duration', + 'fields[content-unlock-option]': 'content_unlock_type', + 'fields[product-variant]': 'price_cents,currency_code,checkout_url,is_hidden,published_at_datetime,orders_count,access_metadata', + 'filter[campaign_id]': self.campaign, + 'filter[contains_exclusive_posts]': 'true', + 'filter[is_draft]': 'false', + 'filter[accessible_by_user_id]': self.module.user_id, + 'sort': '-published_at', + 'json-api-version': '1.0', + 'json-api-use-default-includes': 'false' + } + if self.cursor: + params['page[cursor]'] = self.cursor + obj = self.check_api_response(self.module.do_request(Method.GET, url, params=params)) + if not obj: + self.errored = True + return False + print(json.dumps(obj, indent=4)) + return True + +class PatreonModule(Module): + def __init__(self, session_id: str, uuid: str, user_id: str): + super().__init__() + self.user_id: str = user_id + self.parser: QueryParser = QueryParser(self, 'user') + self.parser.add_command('user', PatreonUser) + self.headers['Accept-Encoding'] = 'gzip, deflate, br' + self.headers['Accept-Language'] = 'en-US,en;q=0.5' + self.headers['x-patreon-uuid'] = uuid + self.cookies['patreon_locale_code'] = 'en-US' + self.cookies['patreon_location_country_code'] = 'US' + self.cookies['session_id'] = session_id + + def search(self, query: str, *extra_args: Any) -> Optional[Search]: + return self.parser.parse_query(query) + + def get_download(self, unique_id: str, key: str) -> Optional[dict[str, Any]]: + if unique_id not in self.unique_id_map: + return None + post = self.unique_id_map[unique_id] + if key not in post.media: + return None + return { + 'urls': [post.media[key].url], + 'headers': { + 'User-Agent': USER_AGENT + } + } diff --git a/src/portal/py/modules/pixiv_app.py b/src/portal/py/modules/pixiv_app.py index 8be3f9c..c84030d 100644 --- a/src/portal/py/modules/pixiv_app.py +++ b/src/portal/py/modules/pixiv_app.py @@ -1,6 +1,6 @@ import json from typing import Optional, Any -from base import Search, Module, ParsedJson +from base import USER_AGENT, Search, Module, ParsedJson from post import MediaUrl, PostType, DateType, User, Post, Image, Tag, TagType from query_parser import QueryParser from pixivpy3 import * @@ -195,7 +195,7 @@ class PixivAppModule(Module): return { 'urls': [post.media[key].url], 'headers': { - 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; rv:122.0) Gecko/20100101 Firefox/122.0', + 'User-Agent': USER_AGENT, 'Referer': 'https://www.pixiv.net/' } } diff --git a/src/portal/py/modules/pixiv_web.py b/src/portal/py/modules/pixiv_web.py index 7bc043b..be68b7e 100644 --- a/src/portal/py/modules/pixiv_web.py +++ b/src/portal/py/modules/pixiv_web.py @@ -4,7 +4,7 @@ import httpx import urllib.parse from typing import Optional, Any from json.decoder import JSONDecodeError -from base import Search, Module, Method, ParsedJson +from base import USER_AGENT, Search, Module, Method, ParsedJson from post import MediaUrl, PostType, DateType, Post, User, Media, Image, Animation, Tag, TagType from query_parser import QueryParser from modules.common import parse_pixiv_date, get_current_utc_time @@ -199,13 +199,17 @@ class PixivWebUser(PixivWebBase): self.errored = True return False self.pages[num] = [] + keys = [] + if obj['illusts']: + keys.extend([int(x) for x in obj['illusts'].keys()]) if obj['manga']: - for illust in obj['manga'].keys(): - post = self.request_illust(int(illust)) - if not post: - self.errored = True - return False - self.pages[num].append(post.unique_id) + keys.extend([int(x) for x in obj['manga'].keys()]) + for illust in keys: + post = self.request_illust(illust) + if not post: + self.errored = True + return False + self.pages[num].append(post.unique_id) self.completed = True return True @@ -231,18 +235,18 @@ class PixivWebFollowing(PixivWebBase): return True class PixivWebModule(Module): - def __init__(self, sessid: str): + def __init__(self, sessid: str, user_id: str): super().__init__() self.parser: QueryParser = QueryParser(self, 'search') self.parser.add_command('search', PixivWebSearch) self.parser.add_command('illust', PixivWebIllust) self.parser.add_command('user', PixivWebUser) self.parser.add_command('following', PixivWebFollowing) - self.headers['Accept-Encoding'] = 'gzip, deflate, br' + self.headers['Accept-Encoding'] = 'gzip, deflate, br' # add 'zstd' when httpx update self.headers['Accept-Language'] = 'en-US,en;q=0.5' self.headers['Referer'] = 'https://www.pixiv.net/' - self.headers['User-Agent'] = 'Mozilla/5.0 (Windows NT 10.0; rv:122.0) Gecko/20100101 Firefox/122.0' - self.headers['x-user-id'] = '22781328' + self.headers['User-Agent'] = USER_AGENT + self.headers['x-user-id'] = user_id self.cookies['PHPSESSID'] = sessid def search(self, query: str, *extra_args: Any) -> Optional[Search]: @@ -257,7 +261,7 @@ class PixivWebModule(Module): return { 'urls': [post.media[key].url], 'headers': { - 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; rv:122.0) Gecko/20100101 Firefox/122.0', + 'User-Agent': USER_AGENT, 'Referer': 'https://www.pixiv.net/' } } diff --git a/src/portal/py/modules/twitter.py b/src/portal/py/modules/twitter.py index 7e77253..2996d0e 100644 --- a/src/portal/py/modules/twitter.py +++ b/src/portal/py/modules/twitter.py @@ -2,7 +2,7 @@ import log import http.cookiejar from typing import Optional, Any, Iterator from datetime import timezone -from base import Search, Module +from base import USER_AGENT, Search, Module from post import MediaUrl, PostType, DateType, PostRef, Post, User, Media, Image, Video from query_parser import QueryParser from modules.common import get_current_utc_time @@ -185,6 +185,6 @@ class TwitterScrapeModule(Module): return { 'urls': [post.media[key].url], 'headers': { - 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; rv:122.0) Gecko/20100101 Firefox/122.0' + 'User-Agent': USER_AGENT } } diff --git a/src/portal/py/post.py b/src/portal/py/post.py index b064f97..8149e4c 100644 --- a/src/portal/py/post.py +++ b/src/portal/py/post.py @@ -1,6 +1,6 @@ import dataclasses from dataclasses import field -from typing import Optional +from typing import Optional, Any from enum import Enum from json import JSONEncoder @@ -34,6 +34,9 @@ class TagType(int, Enum): META = 5 DEPRECATED = 6 +def df(c: Any) -> Any: + return field(default_factory=lambda: c) + @dataclasses.dataclass class MediaUrl(): url: str = '' @@ -42,7 +45,7 @@ class MediaUrl(): @dataclasses.dataclass class Media(): type: MediaType = MediaType.UNKNOWN - url: MediaUrl = field(default_factory=lambda: MediaUrl()) + url: MediaUrl = df(MediaUrl()) @dataclasses.dataclass class File(Media): @@ -56,38 +59,38 @@ class Audio(Media): @dataclasses.dataclass class Image(Media): type: MediaType = MediaType.IMAGE - thumbnail_url: MediaUrl = field(default_factory=lambda: MediaUrl()) + thumbnail_url: MediaUrl = df(MediaUrl()) @dataclasses.dataclass class Video(Media): type: MediaType = MediaType.VIDEO - thumbnail_url: MediaUrl = field(default_factory=lambda: MediaUrl()) + thumbnail_url: MediaUrl = df(MediaUrl()) @dataclasses.dataclass class VideoSplit(Media): type: MediaType = MediaType.VIDEO_SPLIT - audio_url: MediaUrl = field(default_factory=lambda: MediaUrl()) - subtitle_url: MediaUrl = field(default_factory=lambda: MediaUrl()) - thumbnail_url: MediaUrl = field(default_factory=lambda: MediaUrl()) + audio_url: MediaUrl = df(MediaUrl()) + subtitle_url: MediaUrl = df(MediaUrl()) + thumbnail_url: MediaUrl = df(MediaUrl()) @dataclasses.dataclass class Animation(Media): type: MediaType = MediaType.ANIMATION - thumbnail_url: MediaUrl = field(default_factory=lambda: MediaUrl()) - frames: list[tuple[str, int]] = field(default_factory=lambda: []) + thumbnail_url: MediaUrl = df(MediaUrl()) + frames: list[tuple[str, int]] = df([]) @dataclasses.dataclass class Tag(): type: TagType = TagType.UNKNOWN name: str = '' - alts: dict[str, str] = field(default_factory=lambda: {}) + alts: dict[str, str] = df({}) @dataclasses.dataclass class User(): unique_id: str = '' username: str = '' display_name: str = '' - profile_picture_url: MediaUrl = field(default_factory=lambda: MediaUrl()) + profile_picture_url: MediaUrl = df(MediaUrl()) @dataclasses.dataclass class PostRef(): @@ -104,10 +107,10 @@ class Post(): version: int = 3 type: PostType = PostType.UNKNOWN unique_id: str = '' - raw_responses: dict[str, str] = field(default_factory=lambda: {}) + raw_responses: dict[str, str] = df({}) url: str = '' - dates: dict[DateType, float] = field(default_factory=lambda: {}) - author: User = field(default_factory=lambda: User()) + dates: dict[DateType, float] = df({}) + author: User = df(User()) title: str = '' text: str = '' likes: Optional[int] = None @@ -116,12 +119,12 @@ class Post(): quotes: Optional[int] = None comments: Optional[int] = None views: Optional[int] = None - tags: list[Tag] = field(default_factory=lambda: []) - links: list[str] = field(default_factory=lambda: []) - media: dict[str, 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()) + tags: list[Tag] = df([]) + links: list[str] = df([]) + media: dict[str, Media] = df({}) + post: PostRef = df(PostRef()) + quoted: PostRef = df(PostRef()) + in_reply_to: PostRef = df(PostRef()) class PostEncoder(JSONEncoder): def default(self, o): diff --git a/src/portal/src/search.c b/src/portal/src/search.c index ec40d0e..7334c41 100644 --- a/src/portal/src/search.c +++ b/src/portal/src/search.c @@ -71,7 +71,7 @@ s32 camu_portal_create_search(struct camu_portal *portal, str *module, str *quer search->portal = portal; al_array_push(portal->searches, search); } - al_log_info("portal", "New search %i (%.*s).", id, AL_STR_PRINTF(query)); + al_log_info("portal", "New search %x (%.*s).", id, AL_STR_PRINTF(query)); return id; } diff --git a/src/portal/tests/test.py b/src/portal/tests/test.py index cdf7178..a5139f6 100644 --- a/src/portal/tests/test.py +++ b/src/portal/tests/test.py @@ -3,7 +3,7 @@ import httpx import sys sys.path.append('../py') from base import Method -from post import DateType, PostEncoder +from post import DateType, PostEncoder, Media from modules import ALL_MODULES #ALL_MODULES['instagram'][0].init() @@ -12,6 +12,10 @@ from modules import ALL_MODULES #print(page) #print(json.dumps(page, indent=4, cls=PostEncoder)) +d = Media() +print(d) +sys.exit(1) + #module = ALL_MODULES['fanbox'][0] #module = ALL_MODULES['pixiv_web'][0] module = ALL_MODULES['instagram'][0] diff --git a/src/render/renderer_libplacebo.c b/src/render/renderer_libplacebo.c index 03f30a5..9cc0917 100644 --- a/src/render/renderer_libplacebo.c +++ b/src/render/renderer_libplacebo.c @@ -133,10 +133,9 @@ static bool renderer_lp_create_renderer(struct camu_renderer *renderer, s32 *wid 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_nearest; + //lr->params = pl_render_default_params; + lr->params = pl_render_fast_params; + lr->params.upscaler = &pl_filter_box; // Clear manually so we can draw multiple images per frame. lr->params.skip_target_clearing = true; @@ -187,39 +186,49 @@ static void renderer_lp_render(struct camu_renderer *renderer, struct camu_scree { struct camu_renderer_lp *lr = (struct camu_renderer_lp *)renderer; - struct pl_swapchain_frame frame; - pl_swapchain_start_frame(lr->swapchain, &frame); - - struct pl_frame target; - pl_frame_from_swapchain(&target, &frame); - #ifdef CAMU_SCREEN_THREADED if (al_atomic_bool_load(&scr->queued, AL_ATOMIC_RELAXED)) { camu_screen_run_queue(scr); } #endif + bool have_frame = false; + struct pl_swapchain_frame frame; + struct pl_frame target; + struct pl_frame_mix mix; + +#if 1 + pl_swapchain_start_frame(lr->swapchain, &frame); + pl_frame_from_swapchain(&target, &frame); pl_frame_clear_rgba(lr->gpu, &target, clear_color); + have_frame = true; +#endif - struct pl_frame_mix mix; if (scr->videos.size > 0) { struct camu_screen_video *video = &al_array_last(scr->videos); - //al_array_foreach_ptr(scr->videos, i, video) { if (camu_video_buffer_read(video->buf, &mix) && mix.frames) { + if (!have_frame) { + pl_swapchain_start_frame(lr->swapchain, &frame); + pl_frame_from_swapchain(&target, &frame); + pl_frame_clear_rgba(lr->gpu, &target, clear_color); + have_frame = true; + } target.crop = mix.frames[0]->crop; - target.crop.x1 *= video->view.zoom; - target.crop.y1 *= video->view.zoom; + target.crop.x1 *= video->view.zoom / video->view.stretch; + target.crop.y1 *= video->view.zoom * video->view.stretch; target.crop.x0 += video->view.x_offset; target.crop.y0 += video->view.y_offset; target.crop.x1 += video->view.x_offset; target.crop.y1 += video->view.y_offset; + target.rotation = video->view.rotation; pl_render_image_mix(lr->renderer, &mix, &target, &lr->params); } - //} } - pl_swapchain_submit_frame(lr->swapchain); - pl_swapchain_swap_buffers(lr->swapchain); + if (have_frame) { + pl_swapchain_submit_frame(lr->swapchain); + pl_swapchain_swap_buffers(lr->swapchain); + } } void renderer_lp_free(struct camu_renderer **renderer) diff --git a/src/screen/screen.c b/src/screen/screen.c index b3ce0de..f92cd98 100644 --- a/src/screen/screen.c +++ b/src/screen/screen.c @@ -117,7 +117,7 @@ static bool mouse_button_callback(void *userdata, u8 state, u8 button) } #if defined SEKIHI_WINDOW_WAYLAND -#define SCROLL_MULTIPLIER 87.5 +#define SCROLL_MULTIPLIER 150.0 #else #define SCROLL_MULTIPLIER 25.0 #endif @@ -149,6 +149,9 @@ static bool key_callback(void *userdata, u8 state, u8 button) case SEKIHI_BUTTON_PRESSED: //al_printf("%x\n", button); switch (button) { + case 0x1d: + scr->flags |= CAMU_SCREEN_MODIFIER; + break; case 0x10: case 0x18: case 'q': @@ -170,24 +173,55 @@ static bool key_callback(void *userdata, u8 state, u8 button) case 0x39: scr->callback(scr->userdata, CAMU_SCREEN_TOGGLE_PAUSE, 0.0); break; + case 0x1f: + case 's': + scr->callback(scr->userdata, CAMU_SCREEN_RESEEK, 0.0); + break; case 0x13: case 0x1b: case 'r': { struct camu_view *view = get_view_from_mouse_pos(scr); if (view) { - view->mode = CAMU_DEFAULT_VIEW; + if (!(scr->flags & CAMU_SCREEN_MODIFIER)) { + view->mode = CAMU_DEFAULT_VIEW; + } else { + switch (view->rotation) { + case CAMU_VIEW_ROTATION_0: + case CAMU_VIEW_ROTATION_360: + view->rotation = CAMU_VIEW_ROTATION_270; + break; + case CAMU_VIEW_ROTATION_270: + view->rotation = CAMU_VIEW_ROTATION_180; + break; + case CAMU_VIEW_ROTATION_180: + view->rotation = CAMU_VIEW_ROTATION_90; + break; + case CAMU_VIEW_ROTATION_90: + view->rotation = CAMU_VIEW_ROTATION_0; + break; + } + } camu_view_calculate(view, scr->width, scr->height); return true; } break; } - case 0x1f: - case 's': - scr->callback(scr->userdata, CAMU_SCREEN_RESEEK, 0.0); + case 0xd: { + struct camu_view *view = get_view_from_mouse_pos(scr); + if (view) { + camu_view_zoom_simple(view, scr->width, scr->height, scr->width / 2.0, scr->height / 2.0, 0.05); + return true; + } break; - case 0x1d: - scr->flags |= CAMU_SCREEN_MODIFIER; + } + case 0xc: { + struct camu_view *view = get_view_from_mouse_pos(scr); + if (view) { + camu_view_zoom_simple(view, scr->width, scr->height, scr->width / 2.0, scr->height / 2.0, -0.05); + return true; + } break; + } default: break; } @@ -287,12 +321,15 @@ static void add_buffer_internal(struct camu_screen *scr, struct camu_video_buffe .mode = CAMU_DEFAULT_VIEW, .width = stream->video.width, .height = stream->video.height, + .stretch = 1.0, + .rotation = CAMU_VIEW_ROTATION_0, .zindex = 0 } }; #ifdef CAMU_SCREEN_THREADED al_atomic_bool_store(&buf->ref, true, AL_ATOMIC_RELAXED); #endif + if (buf->view.mode != CAMU_VIEW_NONE) video.view = buf->view; camu_view_calculate(&video.view, scr->width, scr->height); al_array_push(scr->videos, video); } @@ -305,6 +342,7 @@ static void remove_buffer_internal(struct camu_screen *scr, struct camu_video_bu #ifdef CAMU_SCREEN_THREADED al_atomic_bool_store(&buf->ref, false, AL_ATOMIC_RELAXED); #endif + buf->view = video->view; al_array_remove_at_iter(scr->videos, i); break; } diff --git a/src/screen/view.c b/src/screen/view.c index ab84277..0ca5f4c 100644 --- a/src/screen/view.c +++ b/src/screen/view.c @@ -1,3 +1,5 @@ +#include <al/lib.h> + #include "view.h" #define ZOOM_MAX 100.0 @@ -6,36 +8,48 @@ #define PAN_BOUNDS 35.0 #define X_IN_BOUNDS(view, w, x) \ - ((x + PAN_BOUNDS <= w) && (PAN_BOUNDS - x <= view->width * view->zoom)) + ((x + PAN_BOUNDS <= w) && (PAN_BOUNDS - x <= view->width * view->zoom / view->stretch)) #define Y_IN_BOUNDS(view, h, y) \ - ((y + PAN_BOUNDS <= h) && (PAN_BOUNDS - y <= view->height * view->zoom)) + ((y + PAN_BOUNDS <= h) && (PAN_BOUNDS - y <= view->height * view->zoom * view->stretch)) + +#define VIEW_ON_ITS_SIDE(view) \ + (view->rotation % CAMU_VIEW_ROTATION_180 == CAMU_VIEW_ROTATION_90) void camu_view_calculate(struct camu_view *view, s32 window_width, s32 window_height) { - if (view->mode == CAMU_VIEW_DETACHED) return; + f64 width, height; + if (VIEW_ON_ITS_SIDE(view)) { + width = (f64)view->height; + height = (f64)view->width; + view->stretch = height / width; + } else { + width = (f64)view->width; + height = (f64)view->height; + view->stretch = 1.0; + } + f64 frame_ratio = width / height; f64 ratio = window_width / (f64)window_height; - f64 frame_ratio = view->width / (f64)view->height; view->ratio = ratio / frame_ratio; switch (view->mode) { case CAMU_VIEW_AUTOFIT: if (frame_ratio > ratio) { - view->zoom = (window_width / (f64)view->width); + view->zoom = window_width / width; } else { - view->zoom = (window_height / (f64)view->height); + view->zoom = window_height / height; } break; case CAMU_VIEW_ZOOM: if (frame_ratio < ratio) { - view->zoom = (window_width / (f64)view->width); + view->zoom = window_width / width; } else { - view->zoom = (window_height / (f64)view->height); + view->zoom = window_height / height; } break; default: return; } - view->x_offset = (window_width - (view->width * view->zoom)) / 2.0; - view->y_offset = (window_height - (view->height * view->zoom)) / 2.0; + view->x_offset = (window_width - (width * view->zoom)) / 2.0; + view->y_offset = (window_height - (height * view->zoom)) / 2.0; } bool camu_view_pan_simple(struct camu_view *view, s32 window_width, s32 window_height, f64 dx, f64 dy) diff --git a/src/screen/view.h b/src/screen/view.h index 5f33742..8cd3d28 100644 --- a/src/screen/view.h +++ b/src/screen/view.h @@ -3,12 +3,20 @@ #include <al/types.h> enum { - CAMU_VIEW_HIDDEN = 0, + CAMU_VIEW_NONE = 0, CAMU_VIEW_AUTOFIT, CAMU_VIEW_ZOOM, CAMU_VIEW_DETACHED }; +enum { + CAMU_VIEW_ROTATION_0 = 0, + CAMU_VIEW_ROTATION_90, + CAMU_VIEW_ROTATION_180, + CAMU_VIEW_ROTATION_270, + CAMU_VIEW_ROTATION_360 +}; + #define CAMU_DEFAULT_VIEW CAMU_VIEW_AUTOFIT struct camu_view { @@ -19,7 +27,8 @@ struct camu_view { f64 zoom; f64 x_offset; f64 y_offset; - f64 rotation; + f64 stretch; + s32 rotation; s32 zindex; }; diff --git a/src/server/common.h b/src/server/common.h index 4d3a0ae..26d0edf 100644 --- a/src/server/common.h +++ b/src/server/common.h @@ -6,12 +6,12 @@ //#define CAMU_DB_PATH al_str_c("/home/andrew/c/camu/data/camu_db_test") #define CAMU_PORT 14356 -#define CAMU_RESOURCE_PORT 14357 +#define CAMU_MULTIPLEX_RPC 0x53 +#define CAMU_MULTIPLEX_SHRUB 0x85 -#define CAMU_SERVER_IP al_str_c("127.0.0.1") +//#define CAMU_SERVER_IP al_str_c("127.0.0.1") //#define CAMU_SERVER_IP al_str_c("192.168.1.192") -//#define CAMU_SERVER_IP al_str_c("108.52.160.112") - +#define CAMU_SERVER_IP al_str_c("108.52.160.112") enum { CAMU_NODE = 0, @@ -20,23 +20,15 @@ enum { }; enum { - CAMU_SRV_IDENTIFY = 0, - CAMU_SRV_LIST_ACTION, - CAMU_SRV_CREATE_SEARCH, - CAMU_SRV_GET_PAGE, - CAMU_SRV_CREATE_BROWSE, - CAMU_SRV_GET_PATH -}; - -enum { - CAMU_CONN_STATE + CAMU_SERVER_IDENTIFY = 0, + CAMU_SERVER_LIST_ACTION }; enum { - CAMU_LIST_ADD = 0, - CAMU_LIST_SKIP, - CAMU_LIST_SKIPTO, - CAMU_LIST_TOGGLE_PAUSE, - CAMU_LIST_SEEK, - CAMU_LIST_FINISHED + CAMU_ADD = 0, + CAMU_SKIP, + CAMU_SKIPTO, + CAMU_TOGGLE_PAUSE, + CAMU_SEEK, + CAMU_FINISHED }; diff --git a/src/server/server.c b/src/server/server.c index 0fd1d71..4e11d3e 100644 --- a/src/server/server.c +++ b/src/server/server.c @@ -1,7 +1,6 @@ #include <al/log.h> #include "../libsink/common.h" -#include "../shrub/common.h" #include "server.h" #include "common.h" @@ -21,16 +20,23 @@ static struct camu_user *get_user_by_username(struct camu_server *tree, str *use return NULL; } -static void list_callback(void *userdata, u8 op, void *opaque, s32 sequence) +static void list_callback(void *userdata, u8 op, void *opaque, void *prev_opaque, s32 sequence) { struct camu_srv_sink *sink = (struct camu_srv_sink *)userdata; struct camu_srv_resource *resource = (struct camu_srv_resource *)opaque; + struct camu_srv_resource *prev = (struct camu_srv_resource *)prev_opaque; struct aki_packet *packet = aki_rpc_get_packet(sink->conn->rpc, op); aki_packet_write_str(packet, CAMU_SERVER_IP); - aki_packet_write_s32(packet, CAMU_RESOURCE_PORT); + aki_packet_write_s32(packet, CAMU_PORT); aki_packet_write_u16(packet, resource->node->id); aki_packet_write_s32(packet, sequence); aki_packet_write_u64(packet, resource->node->start); + aki_packet_write_u8(packet, resource->node->paused != SHRUB_NOT_PAUSED); + if (prev) { + aki_packet_write_u64(packet, prev->node->paused_at); + } else { + aki_packet_write_u64(packet, 0); + } aki_rpc_connection_command(sink->conn, packet, NULL, NULL); if (op == CAMU_SINK_SET) { al_printf("Now Playing: %.*s\n", AL_STR_PRINTF(&resource->unique_id)); @@ -81,7 +87,52 @@ static bool identify_command_callback(void *userdata, struct aki_rpc_connection return true; } -static bool list_command_callback(void *userdata, struct aki_rpc_connection *conn, +static void list_entry_callback(void *entry, u8 op, void *opaque) +{ + struct camu_srv_resource *resource = (struct camu_srv_resource *)entry; + switch (op) { + case CAMU_LIST_ENTRY_ID: + *(str **)opaque = &resource->unique_id; + break; + case CAMU_LIST_IMPULSE: + shrb_node_set_start(resource->node, *(u64 *)opaque); + break; + case CAMU_LIST_USER_PAUSE: + shrb_node_user_pause(resource->node, *(u64 *)opaque); + break; + case CAMU_LIST_USER_RESUME: { + struct shrb_resume_req *req = (struct shrb_resume_req *)opaque; + shrb_node_user_resume(resource->node, req); + break; + } + case CAMU_LIST_PAUSE: + shrb_node_pause(resource->node); + break; + case CAMU_LIST_SEEK: + shrb_node_seek(resource->node, *(u64 *)opaque); + break; + } +} + +#ifdef CAMU_LOCAL_SOCKET +static void list_add_local(struct camu_server *srv, struct camu_list *list, str *line) +{ + struct cch_entry *entry = entry_for_external_path(&srv->portal, line); + if (entry) { + struct shrb_node *node = shrb_server_create_node(&srv->resource, 0, entry); + if (node) { + struct camu_srv_resource *resource = al_alloc_object(struct camu_srv_resource); + al_array_push(srv->resources, resource); + al_str_clone(&resource->unique_id, line); + resource->entry = entry; + resource->node = node; + camu_list_add(list, resource, list_entry_callback, resource->node->duration, false); + } + } +} +#endif + +static bool list_action_command_callback(void *userdata, struct aki_rpc_connection *conn, struct aki_packet *packet, struct aki_packet *rpacket) { struct camu_server *srv = (struct camu_server *)userdata; @@ -92,33 +143,38 @@ static bool list_command_callback(void *userdata, struct aki_rpc_connection *con struct camu_list *list = al_array_at(user->lists, 0); switch (aki_packet_read_u8(packet)) { - case CAMU_LIST_ADD: { + case CAMU_ADD: { + str line; + aki_packet_read_str(packet, &line); +#ifdef CAMU_LOCAL_SOCKET + list_add_local(srv, list, &line); +#endif break; } - case CAMU_LIST_SKIP: { + case CAMU_SKIP: { s32 sequence = aki_packet_read_s32(packet); s32 n = aki_packet_read_s32(packet); camu_list_skip(list, sequence, n); break; } - case CAMU_LIST_SKIPTO: { + case CAMU_SKIPTO: { s32 i = aki_packet_read_s32(packet); camu_list_skipto(list, i); break; } - case CAMU_LIST_TOGGLE_PAUSE: { + case CAMU_TOGGLE_PAUSE: { s32 sequence = aki_packet_read_s32(packet); u64 pos = aki_packet_read_u64(packet); camu_list_toggle_pause(list, sequence, pos); break; } - case CAMU_LIST_SEEK: { + case CAMU_SEEK: { s32 sequence = aki_packet_read_s32(packet); f64 percent = aki_packet_read_f64(packet); camu_list_seek(list, sequence, percent); break; } - case CAMU_LIST_FINISHED: { + case CAMU_FINISHED: { s32 sequence = aki_packet_read_s32(packet); camu_list_finished(list, sequence); break; @@ -130,10 +186,8 @@ static bool list_command_callback(void *userdata, struct aki_rpc_connection *con } static struct aki_rpc_command commands[] = { - { .op = CAMU_SRV_IDENTIFY, .callback = identify_command_callback, .userdata = NULL }, - { .op = CAMU_SRV_LIST_ACTION, .callback = list_command_callback, .userdata = NULL }, - //{ .op = CAMU_SRV_CREATE_SEARCH, .callback = create_search_command_callback, .userdata = NULL }, - //{ .op = CAMU_SRV_GET_PAGE, .callback = get_page_command_callback, .userdata = NULL }, + { .op = CAMU_SERVER_IDENTIFY, .callback = identify_command_callback, .userdata = NULL }, + { .op = CAMU_SERVER_LIST_ACTION, .callback = list_action_command_callback, .userdata = NULL } }; static void connection_callback(void *userdata, struct aki_rpc_connection *conn) @@ -195,23 +249,6 @@ static void connection_closed_callback(void *userdata, struct aki_rpc_connection } } -static void list_entry_callback(void *entry, u8 op, void *opaque) -{ - struct camu_srv_resource *resource = (struct camu_srv_resource *)entry; - switch (op) { - case CAMU_LIST_ENTRY_IMPULSE: { - struct camu_list_timing *timing = (struct camu_list_timing *)opaque; - shrb_node_set_start(resource->node, timing->start); - break; - } - case CAMU_LIST_ENTRY_TOGGLE_PAUSE: - shrb_node_toggle_pause(resource->node, *(u64 *)opaque); - break; - case CAMU_LIST_ENTRY_SEEK: - shrb_node_seek(resource->node, *(u64 *)opaque); - break; - } -} #ifdef CAMU_LOCAL_SOCKET static u8 server_line_callback(void *userdata, str *line) @@ -226,20 +263,12 @@ static u8 server_line_callback(void *userdata, str *line) } else if (al_str_eq(line, al_str_c(";SHUFFLE"))) { camu_list_shuffle(list); } else if (al_str_eq(line, al_str_c(";SORT"))) { + camu_list_sort(list); + } else if (al_str_eq(line, al_str_c(";REVERSE"))) { + camu_list_reverse(list); } else if (al_str_eq(line, al_str_c(";CLEAR"))) { } else { - struct cch_entry *entry = entry_for_external_path(&srv->portal, line); - if (entry) { - struct shrb_node *node = shrb_server_create_node(&srv->resource, 0, entry); - if (node) { - struct camu_srv_resource *resource = al_alloc_object(struct camu_srv_resource); - al_array_push(srv->resources, resource); - al_str_clone(&resource->unique_id, line); - resource->entry = entry; - resource->node = node; - camu_list_add(list, resource, list_entry_callback, resource->node->duration, false); - } - } + list_add_local(srv, list, line); } return AKI_LINE_PROCESSOR_CONTINUE; } @@ -254,6 +283,22 @@ static void sigint_handler(s32 signum) static struct camu_server srv = { 0 }; +static bool multiplex_callback(void *userdata, u8 id, struct aki_socket *s) +{ + struct camu_server *srv = (struct camu_server *)userdata; + switch (id) { + case CAMU_MULTIPLEX_RPC: + aki_rpc_add_socket(&srv->server, s); + return true; + case CAMU_MULTIPLEX_SHRUB: + shrb_server_add_socket(&srv->resource, s); + return true; + default: + break; + } + return false; +} + s32 main(void) { aki_common_init(); @@ -275,15 +320,16 @@ s32 main(void) aki_event_loop_init(&srv.loop); - aki_rpc_init(&srv.server, AKI_SOCKET_TCP, connection_callback, connection_closed_callback, &srv); + aki_rpc_init(&srv.server, &srv.loop, connection_callback, connection_closed_callback, &srv); for (u32 i = 0; i < AL_ARRAY_SIZE(commands); i++) { commands[i].userdata = &srv; aki_rpc_add_command(&srv.server, &commands[i]); } - aki_rpc_listen(&srv.server, &srv.loop, al_str_c("0.0.0.0"), CAMU_PORT); - shrb_server_init(&srv.resource); - shrb_server_listen(&srv.resource, &srv.loop, al_str_c("0.0.0.0"), CAMU_RESOURCE_PORT); + shrb_server_init(&srv.resource, &srv.loop); + + aki_multiplex_socket_init(&srv.multi, AKI_SOCKET_TCP, multiplex_callback, &srv); + aki_multiplex_socket_listen(&srv.multi, &srv.loop, al_str_c("0.0.0.0"), CAMU_PORT); #ifdef CAMU_LOCAL_SOCKET srv.socket.type = AKI_SOCKET_UNIX; diff --git a/src/server/server.h b/src/server/server.h index 83e8523..9565bf6 100644 --- a/src/server/server.h +++ b/src/server/server.h @@ -2,6 +2,7 @@ #define CAMU_LOCAL_SOCKET +#include <aki/multiplex.h> #include <aki/rpc2.h> #ifdef CAMU_LOCAL_SOCKET #include <aki/line_processor.h> @@ -38,6 +39,7 @@ struct camu_srv_resource { struct camu_server { struct aki_event_loop loop; + struct aki_multiplex_socket multi; struct aki_rpc server; array(struct camu_srv_node *) nodes; array(struct camu_srv_client *) clients; diff --git a/src/shrub/client.c b/src/shrub/client.c index 015f47f..abe2d23 100644 --- a/src/shrub/client.c +++ b/src/shrub/client.c @@ -1,6 +1,7 @@ #include <al/log.h> #include "../codec/ffmpeg/packet_ext.h" +#include "../server/common.h" #include "handlers/codec.h" @@ -8,18 +9,6 @@ #include "handler.h" #include "common.h" -static void data_connection_callback(void *userdata, struct aki_packet_stream *stream) -{ - struct shrb_client *client = (struct shrb_client *)userdata; - (void)stream; - struct aki_packet *packet = aki_packet_create(); - aki_packet_write_u8(packet, SHRUB_CONNECTION_DATA); - aki_packet_write_u16(packet, client->connection_id); - aki_packet_write_u16(packet, client->node_id); - aki_packet_write_u64(packet, client->seek_pos); - aki_packet_stream_send_packet(&client->data, packet); -} - static void send_reconnect_packet(struct shrb_client *client) { struct aki_packet *packet = aki_packet_create(); @@ -33,86 +22,55 @@ static void cleanup_connection_internal(struct shrb_client *client) client->callback(client->userdata, SHRUB_CLIENT_CLOSED, NULL, NULL); } -static void data_packet_callback(void *userdata, struct aki_packet_stream *stream, - struct aki_packet *packet) -{ - struct shrb_client *client = (struct shrb_client *)userdata; - (void)stream; - shrb_vcr_push_packet(&client->vcr, packet); -} - static void packet_sent_callback(void *userdata, struct aki_packet *packet) { (void)userdata; aki_packet_free(packet); } -static void data_connection_closed_callback(void *userdata, struct aki_packet_stream *stream); - -void connect_data(struct shrb_client *client) -{ - if (!client->reconnect) { - client->reconnect = true; - } else { - aki_packet_stream_free(&client->data); - } - aki_packet_stream_init(&client->data, AKI_SOCKET_TCP, data_connection_callback, - data_connection_closed_callback, data_packet_callback, packet_sent_callback, client); - aki_packet_stream_connect(&client->data, client->loop, &client->addr, client->port); -} - -static void parse_pause_packet(struct shrb_client *client, struct aki_packet *packet) -{ - u64 delay = aki_packet_read_u64(packet); - u64 seek_pos = aki_packet_read_u64(packet); - struct shrb_seek_req req = { .base = seek_pos, .ts = UINT64_MAX, .delay = delay }; - // TODO: separare function. - client->callback(client->userdata, SHRUB_CLIENT_SET, NULL, &req); - client->seek_pos = req.base; -} - -void data_connection_closed_callback(void *userdata, struct aki_packet_stream *stream) +static void data_packet_callback(void *userdata, struct aki_packet_stream *stream, + struct aki_packet *packet) { struct shrb_client *client = (struct shrb_client *)userdata; (void)stream; - if (client->closed) { - if (!client->control.connected) cleanup_connection_internal(client); - return; - } - client->callback(client->userdata, SHRUB_CLIENT_REMOVE_BUFFERS, NULL, NULL); - shrb_vcr_flush(&client->vcr); - if (!client->paused) { - send_reconnect_packet(client); - } else { - parse_pause_packet(client, client->paused); - aki_packet_free(client->paused); - client->paused = NULL; - client->callback(client->userdata, SHRUB_CLIENT_PAUSE, NULL, NULL); - connect_data(client); - } + shrb_vcr_push_packet(&client->vcr, packet); } -static void control_connection_callback(void *userdata, struct aki_packet_stream *stream) +static void data_connection_callback(void *userdata, struct aki_packet_stream *stream) { struct shrb_client *client = (struct shrb_client *)userdata; (void)stream; + client->data.packet_callback = data_packet_callback; + client->data.packet_sent_callback = packet_sent_callback; struct aki_packet *packet = aki_packet_create(); - aki_packet_write_u8(packet, SHRUB_CONNECTION_CONTROL); - aki_packet_write_u16(packet, 0); + aki_packet_write_u8(packet, SHRUB_CONNECTION_DATA); + aki_packet_write_u16(packet, client->connection_id); aki_packet_write_u16(packet, client->node_id); - aki_packet_stream_send_packet(&client->control, packet); + aki_packet_write_u64(packet, client->seek_pos); + aki_packet_stream_send_packet(&client->data, packet); } -static void control_connection_closed_callback(void *userdata, struct aki_packet_stream *stream) +static void data_connection_closed_callback(void *userdata, struct aki_packet_stream *stream) { struct shrb_client *client = (struct shrb_client *)userdata; (void)stream; if (client->closed) { - if (!client->data.connected) cleanup_connection_internal(client); + if (!client->control.connected) cleanup_connection_internal(client); return; } - aki_packet_stream_free(&client->control); - // TODO: Reconnect. + client->callback(client->userdata, SHRUB_CLIENT_REMOVE_BUFFERS, NULL, NULL); + shrb_vcr_flush(&client->vcr); + send_reconnect_packet(client); +} + +void connect_data(struct shrb_client *client) +{ + if (!client->reconnect) client->reconnect = true; + else aki_packet_stream_free(&client->data); + aki_packet_stream_init(&client->data, AKI_SOCKET_TCP, data_connection_callback, + data_connection_closed_callback, client); + aki_packet_stream_set_multiplex(&client->data, CAMU_MULTIPLEX_SHRUB); + aki_packet_stream_connect(&client->data, client->loop, &client->addr, client->port); } static void parse_info_packet(struct shrb_client *client, struct aki_packet *packet) @@ -188,22 +146,18 @@ static void parse_start_packet(struct shrb_client *client, struct aki_packet *pa u64 delay = aki_packet_read_u64(packet); u64 seek_pos = aki_packet_read_u64(packet); u8 paused = aki_packet_read_u8(packet); - (void)paused; - struct shrb_seek_req req = { .base = seek_pos, .ts = ts, .delay = delay }; + u64 paused_at = aki_packet_read_u64(packet); + struct shrb_seek_req req = { + .base = seek_pos, + .ts = ts, + .delay = delay, + .paused = paused, + .paused_at = paused_at + }; client->callback(client->userdata, SHRUB_CLIENT_SET, NULL, &req); client->seek_pos = req.base; } -static void parse_resume_packet(struct shrb_client *client, struct aki_packet *packet) -{ - u64 ts = aki_packet_read_u64(packet); - s64 delay = aki_packet_read_s64(packet); - u64 seek_pos = aki_packet_read_u64(packet); - struct shrb_seek_req req = { .base = seek_pos, .ts = ts, .delay = delay }; - client->callback(client->userdata, SHRUB_CLIENT_RESUME, NULL, &req); - client->seek_pos = req.base; -} - static void control_packet_callback(void *userdata, struct aki_packet_stream *stream, struct aki_packet *packet) { @@ -221,17 +175,18 @@ static void control_packet_callback(void *userdata, struct aki_packet_stream *st connect_data(client); break; case SHRUB_CONTROL_PAUSE: { - al_assert(!client->paused); f64 pts = aki_packet_read_u64(packet) / 1000000.f; client->callback(client->userdata, SHRUB_CLIENT_PAUSE, NULL, &pts); - //client->paused = packet; - //disconnect_data_internal(client); break; } case SHRUB_CONTROL_RESUME: { - u64 ts = aki_packet_read_u64(packet); - client->callback(client->userdata, SHRUB_CLIENT_RESUME, NULL, &ts); - //parse_resume_packet(client, packet); + u64 start = aki_packet_read_u64(packet); + u64 offset = aki_packet_read_u64(packet); + struct shrb_resume_req req = { + .start = start, + .offset = offset + }; + client->callback(client->userdata, SHRUB_CLIENT_RESUME, NULL, &req); break; } case SHRUB_CONTROL_SEEK: @@ -245,7 +200,31 @@ static void control_packet_callback(void *userdata, struct aki_packet_stream *st aki_packet_free(packet); } -void shrb_client_connect(struct shrb_client *client, struct aki_event_loop *loop, str *addr, s32 port, u16 node_id) +static void control_connection_callback(void *userdata, struct aki_packet_stream *stream) +{ + struct shrb_client *client = (struct shrb_client *)userdata; + (void)stream; + client->control.packet_callback = control_packet_callback; + client->control.packet_sent_callback = packet_sent_callback; + struct aki_packet *packet = aki_packet_create(); + aki_packet_write_u8(packet, SHRUB_CONNECTION_CONTROL); + aki_packet_write_u16(packet, 0); + aki_packet_write_u16(packet, client->node_id); + aki_packet_stream_send_packet(&client->control, packet); +} + +static void control_connection_closed_callback(void *userdata, struct aki_packet_stream *stream) +{ + struct shrb_client *client = (struct shrb_client *)userdata; + (void)stream; + if (client->closed) { + if (!client->data.connected) cleanup_connection_internal(client); + return; + } + // TODO: Reconnect. +} + +bool shrb_client_connect(struct shrb_client *client, struct aki_event_loop *loop, str *addr, s32 port, u16 node_id) { client->loop = loop; client->node_id = node_id; @@ -255,21 +234,16 @@ void shrb_client_connect(struct shrb_client *client, struct aki_event_loop *loop client->reconnect = false; client->level = 0; client->corked = false; - client->paused = NULL; al_array_init(client->subscribed); shrb_vcr_init(&client->vcr, client->loop, &client->data); - aki_packet_stream_init(&client->control, AKI_SOCKET_TCP, control_connection_callback, - control_connection_closed_callback, control_packet_callback, packet_sent_callback, client); - aki_packet_stream_set_no_delay(&client->control, 1); + if (!aki_packet_stream_init(&client->control, AKI_SOCKET_TCP, control_connection_callback, + control_connection_closed_callback, client)) { + return false; + } + aki_packet_stream_set_nodelay(&client->control, 1); + aki_packet_stream_set_multiplex(&client->control, CAMU_MULTIPLEX_SHRUB); aki_packet_stream_connect(&client->control, client->loop, &client->addr, client->port); -} - -void shrb_client_toggle_pause(struct shrb_client *client, f64 pts) -{ - struct aki_packet *packet = aki_packet_create(); - aki_packet_write_u8(packet, SHRUB_CONTROL_TOGGLE_PAUSE); - aki_packet_write_u64(packet, pts * 1000000L); - aki_packet_stream_send_packet(&client->control, packet); + return true; } void shrb_client_reseek(struct shrb_client *client) diff --git a/src/shrub/client.h b/src/shrub/client.h index cb3250d..14f7da2 100644 --- a/src/shrub/client.h +++ b/src/shrub/client.h @@ -19,14 +19,13 @@ struct shrb_client { bool corked; u64 duration; u64 seek_pos; - struct aki_packet *paused; array(s32) subscribed; struct shrb_vcr vcr; void (*callback)(void *, u8, struct shrb_vcr_stream *stream, void *); void *userdata; }; -void shrb_client_connect(struct shrb_client *client, struct aki_event_loop *loop, str *addr, s32 port, u16 node_id); +bool shrb_client_connect(struct shrb_client *client, struct aki_event_loop *loop, str *addr, s32 port, u16 node_id); void shrb_client_toggle_pause(struct shrb_client *client, f64 pts); void shrb_client_reseek(struct shrb_client *client); void shrb_client_seek(struct shrb_client *client, f64 percent); diff --git a/src/shrub/common.h b/src/shrub/common.h index 0be00e4..882eb49 100644 --- a/src/shrub/common.h +++ b/src/shrub/common.h @@ -16,7 +16,6 @@ enum { SHRUB_CONTROL_SUBSCRIBE, SHRUB_CONTROL_RECONNECT, SHRUB_CONTROL_START, - SHRUB_CONTROL_TOGGLE_PAUSE, SHRUB_CONTROL_PAUSE, SHRUB_CONTROL_RESUME, SHRUB_CONTROL_SEEK diff --git a/src/shrub/handler.h b/src/shrub/handler.h index 4ff2f2d..7242130 100644 --- a/src/shrub/handler.h +++ b/src/shrub/handler.h @@ -34,10 +34,17 @@ enum { SHRUB_CLIENT_CLOSED }; +struct shrb_resume_req { + u64 start; + u64 offset; +}; + struct shrb_seek_req { u64 base; u64 ts; u64 delay; + bool paused; + u64 paused_at; }; struct shrb_client_stream; diff --git a/src/shrub/server.c b/src/shrub/server.c index ba7259c..31d8787 100644 --- a/src/shrub/server.c +++ b/src/shrub/server.c @@ -65,11 +65,12 @@ static u8 packet_pool_callback(void *userdata, struct aki_packet *packet) return AKI_PACKET_POOL_RETURN; } -static void send_resume_packet(struct shrb_node_connection *connection, u64 ts) +static void send_resume_packet(struct shrb_node_connection *connection, u64 start, u64 offset) { struct aki_packet *packet = aki_packet_create(); aki_packet_write_u8(packet, SHRUB_CONTROL_RESUME); - aki_packet_write_u64(packet, ts); + aki_packet_write_u64(packet, start); + aki_packet_write_u64(packet, offset); aki_packet_stream_send_packet(connection->control, packet); } @@ -107,16 +108,21 @@ static void send_start_packet(struct shrb_node_connection *connection) u64 seek_pos = node->seek_pos; u64 delay = SHRUB_BASE_DELAY * powl(3, connection->level); u64 start = node->start; - s64 diff = aki_get_timestamp() - start; - if (diff > (s64)SHRUB_BASE_DELAY) { - start += diff; - seek_pos += diff; + bool paused = node->paused != SHRUB_NOT_PAUSED; + if (!paused) { + s64 diff = aki_get_timestamp() - start; + if (diff > (s64)SHRUB_BASE_DELAY) { + start += diff; + seek_pos += diff; + } } - seek_pos += delay - SHRUB_BASE_DELAY; + // Start is not used if the node is paused. aki_packet_write_u64(packet, start); + seek_pos += delay - SHRUB_BASE_DELAY; aki_packet_write_u64(packet, delay); aki_packet_write_u64(packet, seek_pos); - aki_packet_write_u8(packet, node->paused ? 1 : 0); + aki_packet_write_u8(packet, paused); + aki_packet_write_u64(packet, node->paused_at); aki_packet_stream_send_packet(connection->control, packet); } @@ -147,46 +153,6 @@ static void control_packet_callback(void *userdata, struct aki_packet_stream *st send_start_packet(connection); break; } - case SHRUB_CONTROL_TOGGLE_PAUSE: { - struct shrb_node *node = connection->node; - /* - u64 pos = aki_packet_read_u64(packet); - u8 paused = connection->paused; - if (paused == SHRUB_PLAYING) { - node->seek_pos = pos; - node->flags |= SHRUB_NODE_PAUSED; - // node->start won't be touched while SHRUB_NODE_PAUSED is set. - } else if (paused == SHRUB_PAUSED_AND_RUNNING) { - node->flags &= ~SHRUB_NODE_PAUSED; - // TODO: - node->start = aki_get_timestamp() + DELAY; - } - struct shrb_node_connection *rconnection; - al_array_foreach(node->connections, i, rconnection) { - switch (rconnection->paused) { - case SHRUB_PLAYING: - al_assert(paused == SHRUB_PLAYING || paused == SHRUB_NOT_PAUSED); - rconnection->paused = SHRUB_PAUSED; - send_pause_packet(rconnection, node->seek_pos); - break; - case SHRUB_NOT_PAUSED: - al_assert(paused != SHRUB_PAUSED); - rconnection->paused = SHRUB_PAUSED; - break; - case SHRUB_PAUSED: - al_assert(paused != SHRUB_NOT_PAUSED); - rconnection->paused = SHRUB_NOT_PAUSED; - break; - case SHRUB_PAUSED_AND_RUNNING: - al_assert(paused != SHRUB_NOT_PAUSED); - send_resume_packet(rconnection, node->seek_pos, node->start); - rconnection->paused = SHRUB_PLAYING; - break; - } - } - */ - break; - } case SHRUB_CONTROL_SEEK: { u64 pos = aki_packet_read_u64(packet); connection->node->seek_pos = pos; @@ -207,17 +173,18 @@ static struct shrb_node_connection *get_connection_by_id(struct shrb_node *node, return NULL; } -static void send_control_info_packet(struct shrb_node_connection *connection) +static bool send_control_info_packet(struct shrb_node_connection *connection) { - struct aki_packet *packet = aki_packet_create(); - aki_packet_write_u8(packet, SHRUB_CONTROL_INFO); - aki_packet_write_u16(packet, connection->id); connection->handler = shrb_codec_server_create(); if (!connection->handler->init(connection->handler, &connection->handle, &connection->pool)) { - al_assert(false); + return false; } + struct aki_packet *packet = aki_packet_create(); + aki_packet_write_u8(packet, SHRUB_CONTROL_INFO); + aki_packet_write_u16(packet, connection->id); connection->handler->write_info(connection->handler, packet); aki_packet_stream_send_packet(connection->control, packet); + return true; } static aki_thread_result AKI_THREADCALL handler_thread(void *userdata) @@ -230,19 +197,12 @@ static aki_thread_result AKI_THREADCALL handler_thread(void *userdata) static void maybe_start_handler(struct shrb_node_connection *connection) { al_assert(!connection->running); - // Don't seek to 0 unless NODE_SEEKED is set. + // Don't seek to 0 unless node->seeked is set. if (connection->seek_pos > 0 || connection->node->seeked) { connection->handler->seek(connection->handler, connection->seek_pos); } connection->running = 1; aki_thread_create(&connection->thread, handler_thread, connection); - //struct shrb_node *node = connection->node; - //if (connection->paused == SHRUB_NOT_PAUSED) { - // send_resume_packet(connection, connection->seek_pos, node->start); - // connection->paused = SHRUB_PLAYING; - //} else if (connection->paused == SHRUB_PAUSED) { - // connection->paused = SHRUB_PAUSED_AND_RUNNING; - //} } static void identify_packet_callback(void *userdata, struct aki_packet_stream *stream, struct aki_packet *packet) @@ -254,6 +214,12 @@ static void identify_packet_callback(void *userdata, struct aki_packet_stream *s u16 connection_id = aki_packet_read_u16(packet); u16 node_id = aki_packet_read_u16(packet); + // Before this point: + // - We get socket from accept + // - Create packet_stream + // - connection_closed_callback could have been called before this point + // - the socket exists only to wait for an identify packet + // Disconnecting a stream at any point before setting the new callbacks should // be valid and handled properly by the client. // Generalize erronious connection checking/handling. @@ -276,26 +242,28 @@ static void identify_packet_callback(void *userdata, struct aki_packet_stream *s connection->data = NULL; aki_packet_pool_init(&connection->pool, 16, server->loop, packet_pool_callback, connection); cch_entry_get_handle(connection->node->entry, &connection->handle); - //connection->paused = SHRUB_PLAYING; connection->level = 0; connection->running = 0; } switch (type) { case SHRUB_CONNECTION_CONTROL: - aki_packet_stream_set_no_delay(stream, 1); + aki_packet_stream_set_nodelay(stream, 1); connection->control = stream; stream->packet_callback = control_packet_callback; stream->packet_sent_callback = default_packet_sent_callback; stream->connection_closed_callback = control_connection_closed_callback; stream->userdata = connection; - send_control_info_packet(connection); + if (!send_control_info_packet(connection)) { + aki_packet_stream_disconnect(stream); + } break; case SHRUB_CONNECTION_DATA: // TODO: Account for control connection being in a reconnect state. // TODO: is it possible for data->data to be non-NULL during normal operation // (disconnect and reconnect very fast?). // - Call maybe_cleanup_data in this function? + // ^^^^ Literally any connection could call itself data, this needs to be handled better. al_assert(!connection->data); connection->seek_pos = aki_packet_read_u64(packet); connection->data = stream; @@ -310,32 +278,34 @@ static void identify_packet_callback(void *userdata, struct aki_packet_stream *s aki_packet_free(packet); } -static void connection_closed_callback(void *userdata, struct aki_packet_stream *stream) -{ - (void)userdata; - (void)stream; -} - static void connection_callback(void *userdata, struct aki_packet_stream *stream) { struct shrb_server *server = (struct shrb_server *)userdata; - stream->connection_closed_callback = connection_closed_callback; stream->packet_callback = identify_packet_callback; stream->packet_sent_callback = default_packet_sent_callback; - stream->flushed_callback = NULL; stream->userdata = server; } -bool shrb_server_init(struct shrb_server *server) +bool shrb_server_init(struct shrb_server *server, struct aki_event_loop *loop) { + server->loop = loop; al_array_init(server->nodes); - return aki_packet_stream_init(&server->server, AKI_SOCKET_TCP, connection_callback, NULL, NULL, NULL, server); + return true; } -void shrb_server_listen(struct shrb_server *server, struct aki_event_loop *loop, str *addr, s32 port) +static void connection_closed_callback(void *userdata, struct aki_packet_stream *stream) { - server->loop = loop; - aki_packet_stream_listen(&server->server, loop, addr, port); + (void)userdata; + (void)stream; +} + +void shrb_server_add_socket(struct shrb_server *server, struct aki_socket *s) +{ + struct aki_packet_stream *stream = al_alloc_object(struct aki_packet_stream); + stream->connection_callback = connection_callback; + stream->connection_closed_callback = connection_closed_callback; + stream->userdata = server; + aki_packet_stream_from_socket(stream, server->loop, s); } struct shrb_node *shrb_server_create_node(struct shrb_server *server, u16 prev, struct cch_entry *entry) @@ -344,14 +314,6 @@ struct shrb_node *shrb_server_create_node(struct shrb_server *server, u16 prev, node->id = al_rand_u16(); node->prev = prev; node->entry = entry; - node->paused = false; - node->seeked = false; - node->start = 0; - node->seek_pos = 0; - node->offset = 0; - node->connection_id = 1; - al_array_init(node->connections); - node->server = server; struct cch_handle handle; if (!cch_entry_get_handle(node->entry, &handle)) { al_free(node); @@ -365,47 +327,67 @@ struct shrb_node *shrb_server_create_node(struct shrb_server *server, u16 prev, node->duration = handler->get_duration(handler); handler->free(&handler); cch_entry_return_handle(node->entry, &handle); + node->seeked = false; + node->start = 0; + node->paused = SHRUB_NOT_PAUSED; + node->paused_at = 0; + node->seek_pos = 0; + node->connection_id = 1; + al_array_init(node->connections); + node->server = server; al_array_push(server->nodes, node); return node; } void shrb_node_set_start(struct shrb_node *node, u64 start) { + if (node->paused == SHRUB_PAUSED) { + node->paused = SHRUB_NOT_PAUSED; + } node->start = start; } -void shrb_node_set_offset(struct shrb_node *node, u64 offset) +void shrb_node_user_pause(struct shrb_node *node, u64 pos) { - node->offset = offset; + al_assert(node->paused != SHRUB_USER_PAUSED); + pos += SHRUB_PAUSE_DELAY; + node->seek_pos = pos; + node->paused_at = pos; + struct shrb_node_connection *connection; + al_array_foreach(node->connections, i, connection) { + send_pause_packet(connection, node->seek_pos); + } + node->paused = SHRUB_USER_PAUSED; } -void shrb_node_toggle_pause(struct shrb_node *node, u64 pos) +void shrb_node_user_resume(struct shrb_node *node, struct shrb_resume_req *req) { - if (!node->paused) { - node->seek_pos = pos; - node->seek_pos += SHRUB_PAUSE_DELAY; - } else { - node->start = aki_get_timestamp() + SHRUB_PAUSE_DELAY; - } + al_assert(node->paused == SHRUB_USER_PAUSED); + node->start = req->start + SHRUB_PAUSE_DELAY; struct shrb_node_connection *connection; al_array_foreach(node->connections, i, connection) { - if (!node->paused) { - send_pause_packet(connection, node->seek_pos); - } else { - send_resume_packet(connection, node->start); - } + send_resume_packet(connection, node->start, req->offset); } - if (node->paused) { - node->start -= SHRUB_BASE_DELAY; - node->offset = 0; - } - node->paused = !node->paused; + node->start -= SHRUB_BASE_DELAY; + node->paused = SHRUB_NOT_PAUSED; +} + +void shrb_node_pause(struct shrb_node *node) +{ + return; + if (node->paused != SHRUB_NOT_PAUSED) return; + u64 start = node->start; + s64 pos = node->seek_pos + (aki_get_timestamp() - (start + SHRUB_BASE_DELAY)); + pos += SHRUB_PAUSE_DELAY; + al_assert(pos >= 0); + node->seek_pos = (u64)pos; + node->paused_at = pos; + node->paused = SHRUB_PAUSED; } void shrb_node_seek(struct shrb_node *node, u64 pos) { node->seek_pos = pos; - node->start = aki_get_timestamp(); node->seeked = true; struct shrb_node_connection *connection; al_array_foreach(node->connections, i, connection) { diff --git a/src/shrub/server.h b/src/shrub/server.h index 2bce369..cb901c5 100644 --- a/src/shrub/server.h +++ b/src/shrub/server.h @@ -1,6 +1,7 @@ #pragma once #include <al/array.h> +#include <aki/multiplex.h> #include <aki/packet_stream.h> #include <aki/packet_pool.h> @@ -9,13 +10,6 @@ #include "handler.h" -enum { - SHRUB_PLAYING = 0, - SHRUB_NOT_PAUSED, - SHRUB_PAUSED, - SHRUB_PAUSED_AND_RUNNING -}; - struct shrb_node_connection { u16 id; struct shrb_node *node; @@ -23,7 +17,6 @@ struct shrb_node_connection { struct aki_packet_stream *data; struct aki_packet_pool pool; s32 running; - //u8 paused; u32 level; u64 seek_pos; struct cch_handle handle; @@ -31,16 +24,22 @@ struct shrb_node_connection { struct aki_thread thread; }; +enum { + SHRUB_NOT_PAUSED = 0, + SHRUB_PAUSED, + SHRUB_USER_PAUSED +}; + struct shrb_node { u16 id; u16 prev; struct cch_entry *entry; - bool paused; + u64 duration; bool seeked; - u64 start; + u64 start; // Exact start of the stream minus BASE_DELAY. + u8 paused; + u64 paused_at; u64 seek_pos; - u64 offset; - u64 duration; u16 connection_id; array(struct shrb_node_connection *) connections; struct shrb_server *server; @@ -48,15 +47,16 @@ struct shrb_node { struct shrb_server { struct aki_event_loop *loop; - struct aki_packet_stream server; array(struct shrb_node *) nodes; }; -bool shrb_server_init(struct shrb_server *server); -void shrb_server_listen(struct shrb_server *server, struct aki_event_loop *loop, str *addr, s32 port); +bool shrb_server_init(struct shrb_server *server, struct aki_event_loop *loop); +void shrb_server_add_socket(struct shrb_server *server, struct aki_socket *s); struct shrb_node *shrb_server_create_node(struct shrb_server *server, u16 prev, struct cch_entry *entry); void shrb_node_set_start(struct shrb_node *node, u64 start); -void shrb_node_set_offset(struct shrb_node *node, u64 offset); -void shrb_node_toggle_pause(struct shrb_node *node, u64 pos); +void shrb_node_user_pause(struct shrb_node *node, u64 pos); +void shrb_node_user_resume(struct shrb_node *node, struct shrb_resume_req *req); +void shrb_node_pause(struct shrb_node *node); +void shrb_node_resume(struct shrb_node *node); void shrb_node_seek(struct shrb_node *node, u64 pos); void shrb_server_close(struct shrb_server *server); diff --git a/src/shrub/vcr.c b/src/shrub/vcr.c index 1816142..7f2b058 100644 --- a/src/shrub/vcr.c +++ b/src/shrub/vcr.c @@ -149,6 +149,10 @@ void shrb_vcr_stream_cork(struct shrb_vcr_stream *stream) void shrb_vcr_stream_uncork(struct shrb_vcr_stream *stream) { + if (al_atomic_s32_load(&stream->state, AL_ATOMIC_RELAXED) == SHRUB_STREAM_RUNNING) { + // DEBUG: This will be reached during normal operation. + return; + } al_atomic_s32_store(&stream->state, SHRUB_STREAM_RUNNING, AL_ATOMIC_RELAXED); aki_mutex_lock(&stream->mutex); if (aki_cond_is_waiting(&stream->cond)) { |