streaming: Stop streaming if frames cannot be decoded

Report the stream as invalid if the frames cannot be decoded. This will
force the server to send regular screen updates instead.

Signed-off-by: Francois Gouget <fgouget@codeweavers.com>
Acked-by: Victor Toso <victortoso@redhat.com>

4 files changed, 51 insertions, 10 deletions

diff --git a/src/channel-display-gst.c b/src/channel-display-gst.c
index f52299f..2b4ef7f 100644
--- a/src/channel-display-gst.c
+++ b/src/channel-display-gst.c
@@ -280,6 +280,28 @@ static void free_pipeline(SpiceGstDecoder *decoder)
     decoder->pipeline = NULL;
 }
 
+static gboolean handle_pipeline_message(GstBus *bus, GstMessage *msg, gpointer video_decoder)
+{
+    SpiceGstDecoder *decoder = video_decoder;
+
+    if (GST_MESSAGE_TYPE(msg) == GST_MESSAGE_ERROR) {
+        GError *err = NULL;
+        gchar *debug_info = NULL;
+        gst_message_parse_error(msg, &err, &debug_info);
+        spice_warning("GStreamer error from element %s: %s",
+                      GST_OBJECT_NAME(msg->src), err->message);
+        if (debug_info) {
+            SPICE_DEBUG("debug information: %s", debug_info);
+            g_free(debug_info);
+        }
+        g_clear_error(&err);
+
+        /* We won't be able to process any more frame anyway */
+        free_pipeline(decoder);
+    }
+    return TRUE;
+}
+
 static gboolean create_pipeline(SpiceGstDecoder *decoder)
 {
     gchar *desc;
@@ -287,6 +309,7 @@ static gboolean create_pipeline(SpiceGstDecoder *decoder)
     guint opt;
     GstAppSinkCallbacks appsink_cbs = { NULL };
     GError *err = NULL;
+    GstBus *bus;
 
     auto_enabled = (g_getenv("SPICE_GSTVIDEO_AUTO") != NULL);
     if (auto_enabled || !VALID_VIDEO_CODEC_TYPE(decoder->base.codec_type)) {
@@ -324,6 +347,9 @@ static gboolean create_pipeline(SpiceGstDecoder *decoder)
 
     appsink_cbs.new_sample = new_sample;
     gst_app_sink_set_callbacks(decoder->appsink, &appsink_cbs, decoder, NULL);
+    bus = gst_pipeline_get_bus(GST_PIPELINE(decoder->pipeline));
+    gst_bus_add_watch(bus, handle_pipeline_message, decoder);
+    gst_object_unref(bus);
 
     decoder->clock = gst_pipeline_get_clock(GST_PIPELINE(decoder->pipeline));
 
@@ -390,9 +416,9 @@ static void release_buffer_data(gpointer data)
     spice_msg_in_unref(frame_msg);
 }
 
-static void spice_gst_decoder_queue_frame(VideoDecoder *video_decoder,
-                                          SpiceMsgIn *frame_msg,
-                                          int32_t latency)
+static gboolean spice_gst_decoder_queue_frame(VideoDecoder *video_decoder,
+                                              SpiceMsgIn *frame_msg,
+                                              int32_t latency)
 {
     SpiceGstDecoder *decoder = (SpiceGstDecoder*)video_decoder;
 
@@ -400,7 +426,7 @@ static void spice_gst_decoder_queue_frame(VideoDecoder *video_decoder,
     uint32_t size = spice_msg_in_frame_data(frame_msg, &data);
     if (size == 0) {
         SPICE_DEBUG("got an empty frame buffer!");
-        return;
+        return TRUE;
     }
 
     SpiceStreamDataHeader *frame_op = spice_msg_in_parsed(frame_msg);
@@ -419,7 +445,13 @@ static void spice_gst_decoder_queue_frame(VideoDecoder *video_decoder,
          * saves CPU so do it.
          */
         SPICE_DEBUG("dropping a late MJPEG frame");
-        return;
+        return TRUE;
+    }
+
+    if (decoder->pipeline == NULL) {
+        /* An error occurred, causing the GStreamer pipeline to be freed */
+        spice_warning("An error occurred, stopping the video stream");
+        return FALSE;
     }
 
     /* ref() the frame_msg for the buffer */
@@ -440,6 +472,7 @@ static void spice_gst_decoder_queue_frame(VideoDecoder *video_decoder,
         SPICE_DEBUG("GStreamer error: unable to push frame of size %u", size);
         stream_dropped_frame_on_playback(decoder->base.stream);
     }
+    return TRUE;
 }
 
 static gboolean gstvideo_init(void)
diff --git a/src/channel-display-mjpeg.c b/src/channel-display-mjpeg.c
index 4976d53..67746c3 100644
--- a/src/channel-display-mjpeg.c
+++ b/src/channel-display-mjpeg.c
@@ -235,8 +235,9 @@ static void mjpeg_decoder_drop_queue(MJpegDecoder *decoder)
 
 /* ---------- VideoDecoder's public API ---------- */
 
-static void mjpeg_decoder_queue_frame(VideoDecoder *video_decoder,
-                                      SpiceMsgIn *frame_msg, int32_t latency)
+static gboolean mjpeg_decoder_queue_frame(VideoDecoder *video_decoder,
+                                          SpiceMsgIn *frame_msg,
+                                          int32_t latency)
 {
     MJpegDecoder *decoder = (MJpegDecoder*)video_decoder;
     SpiceMsgIn *last_msg;
@@ -262,12 +263,13 @@ static void mjpeg_decoder_queue_frame(VideoDecoder *video_decoder,
      * So drop late frames as early as possible to save on processing time.
      */
     if (latency < 0) {
-        return;
+        return TRUE;
     }
 
     spice_msg_in_ref(frame_msg);
     g_queue_push_tail(decoder->msgq, frame_msg);
     mjpeg_decoder_schedule(decoder);
+    return TRUE;
 }
 
 static void mjpeg_decoder_reschedule(VideoDecoder *video_decoder)
diff --git a/src/channel-display-priv.h b/src/channel-display-priv.h
index aa57f95..b9c08a3 100644
--- a/src/channel-display-priv.h
+++ b/src/channel-display-priv.h
@@ -50,8 +50,10 @@ struct VideoDecoder {
      * @frame_msg: The Spice message containing the compressed frame.
      * @latency:   How long in milliseconds until the frame should be
      *             displayed. Negative values mean the frame is late.
+     * @return:    False if the decoder can no longer decode frames,
+     *             True otherwise.
      */
-    void (*queue_frame)(VideoDecoder *decoder, SpiceMsgIn *frame_msg, int32_t latency);
+    gboolean (*queue_frame)(VideoDecoder *decoder, SpiceMsgIn *frame_msg, int32_t latency);
 
     /* The format of the encoded video. */
     int codec_type;
diff --git a/src/channel-display.c b/src/channel-display.c
index 500ddfb..ca56cd1 100644
--- a/src/channel-display.c
+++ b/src/channel-display.c
@@ -1424,7 +1424,11 @@ static void display_handle_stream_data(SpiceChannel *channel, SpiceMsgIn *in)
      * decoding and best decide if/when to drop them when they are late,
      * taking into account the impact on later frames.
      */
-    st->video_decoder->queue_frame(st->video_decoder, in,  latency);
+    if (!st->video_decoder->queue_frame(st->video_decoder, in, latency)) {
+        destroy_stream(channel, op->id);
+        report_invalid_stream(channel, op->id);
+        return;
+    }
     if (c->enable_adaptive_streaming) {
         display_update_stream_report(SPICE_DISPLAY_CHANNEL(channel), op->id,
                                      op->multi_media_time, latency);