Make stream push packets to sinks

Now that decoder and recorder implement the packet sink trait, make
stream push packets to the sinks without depending on the concrete sink
types.

app/src/decoder.c

@@ -41,8 +41,14 @@ decoder_close(struct decoder *decoder) {
     avcodec_free_context(&decoder->codec_ctx);
 }
 
-bool
+static bool
 decoder_push(struct decoder *decoder, const AVPacket *packet) {
+    bool is_config = packet->pts == AV_NOPTS_VALUE;
+    if (is_config) {
+        // nothing to do
+        return true;
+    }
+
     int ret;
     if ((ret = avcodec_send_packet(decoder->codec_ctx, packet)) < 0) {
         LOGE("Could not send video packet: %d", ret);

app/src/decoder.h

@@ -28,7 +28,4 @@ decoder_open(struct decoder *decoder, const AVCodec *codec);
 void
 decoder_close(struct decoder *decoder);
 
-bool
-decoder_push(struct decoder *decoder, const AVPacket *packet);
-
 #endif

app/src/recorder.c

@@ -216,7 +216,7 @@ run_recorder(void *data) {
     return 0;
 }
 
-bool
+static bool
 recorder_open(struct recorder *recorder, const AVCodec *input_codec) {
     const char *format_name = recorder_get_format_name(recorder->format);
     assert(format_name);
@@ -277,7 +277,7 @@ recorder_open(struct recorder *recorder, const AVCodec *input_codec) {
     return true;
 }
 
-void
+static void
 recorder_close(struct recorder *recorder) {
     sc_mutex_lock(&recorder->mutex);
     recorder->stopped = true;
@@ -290,7 +290,7 @@ recorder_close(struct recorder *recorder) {
     avformat_free_context(recorder->ctx);
 }
 
-bool
+static bool
 recorder_push(struct recorder *recorder, const AVPacket *packet) {
     sc_mutex_lock(&recorder->mutex);
     assert(!recorder->stopped);

app/src/recorder.h

@@ -49,13 +49,4 @@ recorder_init(struct recorder *recorder, const char *filename,
 void
 recorder_destroy(struct recorder *recorder);
 
-bool
-recorder_open(struct recorder *recorder, const AVCodec *input_codec);
-
-void
-recorder_close(struct recorder *recorder);
-
-bool
-recorder_push(struct recorder *recorder, const AVPacket *packet);
-
 #endif

app/src/scrcpy.c

@@ -336,7 +336,15 @@ scrcpy(const struct scrcpy_options *options) {
 
     av_log_set_callback(av_log_callback);
 
-    stream_init(&stream, server.video_socket, dec, rec);
+    stream_init(&stream, server.video_socket);
+
+    if (dec) {
+        stream_add_sink(&stream, &dec->packet_sink);
+    }
+
+    if (rec) {
+        stream_add_sink(&stream, &rec->packet_sink);
+    }
 
     if (options->display) {
         if (options->control) {

app/src/stream.c

@@ -66,25 +66,11 @@ notify_stopped(void) {
 }
 
 static bool
-process_config_packet(struct stream *stream, AVPacket *packet) {
+push_packet_to_sinks(struct stream *stream, const AVPacket *packet) {
-    if (stream->recorder && !recorder_push(stream->recorder, packet)) {
+    for (unsigned i = 0; i < stream->sink_count; ++i) {
-        LOGE("Could not send config packet to recorder");
+        struct sc_packet_sink *sink = stream->sinks[i];
-        return false;
+        if (!sink->ops->push(sink, packet)) {
-    }
+            LOGE("Could not send config packet to sink %d", i);
-    return true;
-}
-
-static bool
-process_frame(struct stream *stream, AVPacket *packet) {
-    if (stream->decoder && !decoder_push(stream->decoder, packet)) {
-        return false;
-    }
-
-    if (stream->recorder) {
-        packet->dts = packet->pts;
-
-        if (!recorder_push(stream->recorder, packet)) {
-            LOGE("Could not send packet to recorder");
             return false;
         }
     }
@@ -111,9 +97,11 @@ stream_parse(struct stream *stream, AVPacket *packet) {
         packet->flags |= AV_PKT_FLAG_KEY;
     }
 
-    bool ok = process_frame(stream, packet);
+    packet->dts = packet->pts;
+
+    bool ok = push_packet_to_sinks(stream, packet);
     if (!ok) {
-        LOGE("Could not process frame");
+        LOGE("Could not process packet");
         return false;
     }
 
@@ -156,7 +144,7 @@ stream_push_packet(struct stream *stream, AVPacket *packet) {
 
     if (is_config) {
         // config packet
-        bool ok = process_config_packet(stream, packet);
+        bool ok = push_packet_to_sinks(stream, packet);
         if (!ok) {
             return false;
         }
@@ -177,6 +165,33 @@ stream_push_packet(struct stream *stream, AVPacket *packet) {
     return true;
 }
 
+static void
+stream_close_first_sinks(struct stream *stream, unsigned count) {
+    while (count) {
+        struct sc_packet_sink *sink = stream->sinks[--count];
+        sink->ops->close(sink);
+    }
+}
+
+static inline void
+stream_close_sinks(struct stream *stream) {
+    stream_close_first_sinks(stream, stream->sink_count);
+}
+
+static bool
+stream_open_sinks(struct stream *stream, const AVCodec *codec) {
+    for (unsigned i = 0; i < stream->sink_count; ++i) {
+        struct sc_packet_sink *sink = stream->sinks[i];
+        if (!sink->ops->open(sink, codec)) {
+            LOGE("Could not open packet sink %d", i);
+            stream_close_first_sinks(stream, i);
+            return false;
+        }
+    }
+
+    return true;
+}
+
 static int
 run_stream(void *data) {
     struct stream *stream = data;
@@ -193,22 +208,15 @@ run_stream(void *data) {
         goto end;
     }
 
-    if (stream->decoder && !decoder_open(stream->decoder, codec)) {
+    if (!stream_open_sinks(stream, codec)) {
-        LOGE("Could not open decoder");
+        LOGE("Could not open stream sinks");
         goto finally_free_codec_ctx;
     }
 
-    if (stream->recorder) {
-        if (!recorder_open(stream->recorder, codec)) {
-            LOGE("Could not open recorder");
-            goto finally_close_decoder;
-        }
-    }
-
     stream->parser = av_parser_init(AV_CODEC_ID_H264);
     if (!stream->parser) {
         LOGE("Could not initialize parser");
-        goto finally_close_recorder;
+        goto finally_close_sinks;
     }
 
     // We must only pass complete frames to av_parser_parse2()!
@@ -238,14 +246,8 @@ run_stream(void *data) {
     }
 
     av_parser_close(stream->parser);
-finally_close_recorder:
+finally_close_sinks:
-    if (stream->recorder) {
+    stream_close_sinks(stream);
-        recorder_close(stream->recorder);
-    }
-finally_close_decoder:
-    if (stream->decoder) {
-        decoder_close(stream->decoder);
-    }
 finally_free_codec_ctx:
     avcodec_free_context(&stream->codec_ctx);
 end:
@@ -254,12 +256,18 @@ end:
 }
 
 void
-stream_init(struct stream *stream, socket_t socket,
+stream_init(struct stream *stream, socket_t socket) {
-            struct decoder *decoder, struct recorder *recorder) {
     stream->socket = socket;
-    stream->decoder = decoder,
-    stream->recorder = recorder;
     stream->has_pending = false;
+    stream->sink_count = 0;
+}
+
+void
+stream_add_sink(struct stream *stream, struct sc_packet_sink *sink) {
+    assert(stream->sink_count < STREAM_MAX_SINKS);
+    assert(sink);
+    assert(sink->ops);
+    stream->sinks[stream->sink_count++] = sink;
 }
 
 bool

app/src/stream.h

@@ -8,14 +8,19 @@
 #include <libavformat/avformat.h>
 #include <SDL2/SDL_atomic.h>
 
+#include "trait/packet_sink.h"
 #include "util/net.h"
 #include "util/thread.h"
 
+#define STREAM_MAX_SINKS 2
+
 struct stream {
     socket_t socket;
     sc_thread thread;
-    struct decoder *decoder;
+
-    struct recorder *recorder;
+    struct sc_packet_sink *sinks[STREAM_MAX_SINKS];
+    unsigned sink_count;
+
     AVCodecContext *codec_ctx;
     AVCodecParserContext *parser;
     // successive packets may need to be concatenated, until a non-config
@@ -25,8 +30,10 @@ struct stream {
 };
 
 void
-stream_init(struct stream *stream, socket_t socket,
+stream_init(struct stream *stream, socket_t socket);
-            struct decoder *decoder, struct recorder *recorder);
+
+void
+stream_add_sink(struct stream *stream, struct sc_packet_sink *sink);
 
 bool
 stream_start(struct stream *stream);