[byzanz] Make the GStreamer encoder pull data out of the stream manually

[Benjamin Otte <otte src gnome org>]

commit 468ccbb8f97876714d72318b599c29d25e2e43de
Author: Benjamin Otte <otte gnome org>
Date:   Thu Sep 3 12:28:17 2009 +0200

    Make the GStreamer encoder pull data out of the stream manually
    
    Gets around heavy memory usage when the encoder is too slow.

 src/byzanzencodergstreamer.c |  185 +++++++++++++++++++-----------------------
 1 files changed, 82 insertions(+), 103 deletions(-)
---
diff --git a/src/byzanzencodergstreamer.c b/src/byzanzencodergstreamer.c
index aedcab3..179dc6f 100644
--- a/src/byzanzencodergstreamer.c
+++ b/src/byzanzencodergstreamer.c
@@ -27,19 +27,86 @@
 #include <gst/app/gstappbuffer.h>
 #include <gst/video/video.h>
 
+#include "byzanzserialize.h"
+
 G_DEFINE_TYPE (ByzanzEncoderGStreamer, byzanz_encoder_gstreamer, BYZANZ_TYPE_ENCODER)
 
+static void
+byzanz_encoder_gstreamer_need_data (GstAppSrc *src, guint length, gpointer data)
+{
+  ByzanzEncoder *encoder = data;
+  ByzanzEncoderGStreamer *gst = data;
+  GstBuffer *buffer;
+  cairo_t *cr;
+  cairo_surface_t *surface;
+  GdkRegion *region;
+  GError *error = NULL;
+  guint64 msecs;
+
+  if (!byzanz_deserialize (encoder->input_stream, &msecs, &surface, &region, encoder->cancellable, &error)) {
+    gst_element_message_full (GST_ELEMENT (src), GST_MESSAGE_ERROR,
+        error->domain, error->code, g_strdup (error->message), NULL, __FILE__, GST_FUNCTION, __LINE__);
+    g_error_free (error);
+    return;
+  }
+
+  if (surface == NULL) {
+    gst_app_src_end_of_stream (gst->src);
+    return;
+  }
+
+  if (cairo_surface_get_reference_count (gst->surface) > 1) {
+    cairo_surface_t *copy = cairo_image_surface_create (CAIRO_FORMAT_RGB24,
+        cairo_image_surface_get_width (gst->surface), cairo_image_surface_get_height (gst->surface));
+    
+    cr = cairo_create (copy);
+    cairo_set_source_surface (cr, gst->surface, 0, 0);
+    cairo_paint (cr);
+    cairo_destroy (cr);
+
+    cairo_surface_destroy (gst->surface);
+    gst->surface = copy;
+  }
+  cr = cairo_create (gst->surface);
+  cairo_set_source_surface (cr, surface, 0, 0);
+  gdk_cairo_region (cr, region);
+  cairo_fill (cr);
+  cairo_destroy (cr);
+
+  /* create a buffer and send it */
+  /* FIXME: stride just works? */
+  cairo_surface_reference (gst->surface);
+  buffer = gst_app_buffer_new (cairo_image_surface_get_data (gst->surface),
+      cairo_image_surface_get_stride (gst->surface) * cairo_image_surface_get_height (gst->surface),
+      (GstAppBufferFinalizeFunc) cairo_surface_destroy, gst->surface);
+  GST_BUFFER_FLAG_SET (buffer, GST_BUFFER_FLAG_READONLY);
+  GST_BUFFER_TIMESTAMP (buffer) = msecs * GST_MSECOND;
+  gst_buffer_set_caps (buffer, gst->caps);
+  gst_app_src_push_buffer (gst->src, buffer);
+}
+
+static GstAppSrcCallbacks callbacks = {
+  byzanz_encoder_gstreamer_need_data, NULL, NULL
+};
+
 static gboolean
-byzanz_encoder_gstreamer_setup (ByzanzEncoder * encoder,
-                          GOutputStream * stream,
-                          guint           width,
-                          guint           height,
-                          GCancellable *  cancellable,
-                          GError **	  error)
+byzanz_encoder_gstreamer_run (ByzanzEncoder * encoder,
+                              GInputStream *  input,
+                              GOutputStream * output,
+                              GCancellable *  cancellable,
+                              GError **	      error)
 {
   ByzanzEncoderGStreamer *gstreamer = BYZANZ_ENCODER_GSTREAMER (encoder);
   ByzanzEncoderGStreamerClass *klass = BYZANZ_ENCODER_GSTREAMER_GET_CLASS (encoder);
   GstElement *sink;
+  guint width, height;
+  GstMessage *message;
+  GstBus *bus;
+
+  if (!byzanz_deserialize_header (input, &width, &height, cancellable, error))
+    return FALSE;
+
+  gstreamer->surface = cairo_image_surface_create (CAIRO_FORMAT_RGB24, width, height);
 
   g_assert (klass->pipeline_string);
   gstreamer->pipeline = gst_parse_launch (klass->pipeline_string, error);
@@ -51,7 +118,7 @@ byzanz_encoder_gstreamer_setup (ByzanzEncoder * encoder,
   g_assert (GST_IS_APP_SRC (gstreamer->src));
   sink = gst_bin_get_by_name (GST_BIN (gstreamer->pipeline), "sink");
   g_assert (sink);
-  g_object_set (sink, "stream", stream, NULL);
+  g_object_set (sink, "stream", output, NULL);
   g_object_unref (sink);
 
 #if G_BYTE_ORDER == G_LITTLE_ENDIAN
@@ -68,6 +135,7 @@ byzanz_encoder_gstreamer_setup (ByzanzEncoder * encoder,
   g_assert (gst_caps_is_fixed (gstreamer->caps));
 
   gst_app_src_set_caps (gstreamer->src, gstreamer->caps);
+  gst_app_src_set_callbacks (gstreamer->src, &callbacks, gstreamer, NULL);
   gst_app_src_set_stream_type (gstreamer->src, GST_APP_STREAM_TYPE_STREAM);
 
   if (!gst_element_set_state (gstreamer->pipeline, GST_STATE_PLAYING)) {
@@ -75,99 +143,11 @@ byzanz_encoder_gstreamer_setup (ByzanzEncoder * encoder,
     return FALSE;
   }
 
-  return TRUE;
-}
-
-static gboolean
-byzanz_encoder_gstreamer_got_error (ByzanzEncoderGStreamer *gstreamer, GError **error)
-{
-  GstBus *bus = gst_pipeline_get_bus (GST_PIPELINE (gstreamer->pipeline));
-  GstMessage *message = gst_bus_pop_filtered (bus, GST_MESSAGE_ERROR | GST_MESSAGE_EOS);
-
-  g_object_unref (bus);
-  if (message != NULL) {
-    if (GST_MESSAGE_TYPE (message) == GST_MESSAGE_ERROR) {
-      gst_message_parse_error (message, error, NULL);
-    } else {
-      g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_FAILED, _("Premature end of GStreamer pipeline"));
-    }
-    gst_message_unref (message);
-    return FALSE;
-  }
-
-  return TRUE;
-}
-
-static gboolean
-byzanz_encoder_gstreamer_process (ByzanzEncoder *   encoder,
-                                  GOutputStream *   stream,
-                                  guint64           msecs,
-                                  cairo_surface_t * surface,
-                                  const GdkRegion * region,
-                                  GCancellable *    cancellable,
-                                  GError **	    error)
-{
-  ByzanzEncoderGStreamer *gstreamer = BYZANZ_ENCODER_GSTREAMER (encoder);
-  GstBuffer *buffer;
-
-  if (!byzanz_encoder_gstreamer_got_error (gstreamer, error))
-    return FALSE;
-
-  /* update the surface */
-  if (gstreamer->surface == NULL) {
-    /* just assume that the size is right and pray */
-    gstreamer->surface = cairo_surface_reference (surface);
-  } else {
-    cairo_t *cr;
-
-    if (cairo_surface_get_reference_count (gstreamer->surface) > 1) {
-      cairo_surface_t *copy = cairo_image_surface_create (CAIRO_FORMAT_RGB24,
-          cairo_image_surface_get_width (gstreamer->surface), cairo_image_surface_get_height (gstreamer->surface));
-      
-      cr = cairo_create (copy);
-      cairo_set_source_surface (cr, gstreamer->surface, 0, 0);
-      cairo_paint (cr);
-      cairo_destroy (cr);
-
-      cairo_surface_destroy (gstreamer->surface);
-      gstreamer->surface = copy;
-    }
-    cr = cairo_create (gstreamer->surface);
-    cairo_set_source_surface (cr, surface, 0, 0);
-    gdk_cairo_region (cr, region);
-    cairo_fill (cr);
-    cairo_destroy (cr);
-  }
-
-  /* create a buffer and send it */
-  /* FIXME: stride just works? */
-  cairo_surface_reference (gstreamer->surface);
-  buffer = gst_app_buffer_new (cairo_image_surface_get_data (gstreamer->surface),
-      cairo_image_surface_get_stride (gstreamer->surface) * cairo_image_surface_get_height (gstreamer->surface),
-      (GstAppBufferFinalizeFunc) cairo_surface_destroy, gstreamer->surface);
-  GST_BUFFER_FLAG_SET (buffer, GST_BUFFER_FLAG_READONLY);
-  GST_BUFFER_TIMESTAMP (buffer) = msecs * GST_MSECOND;
-  gst_buffer_set_caps (buffer, gstreamer->caps);
-  gst_app_src_push_buffer (gstreamer->src, buffer);
-
-  return TRUE;
-}
-
-static gboolean
-byzanz_encoder_gstreamer_close (ByzanzEncoder *  encoder,
-                                GOutputStream *  stream,
-                                guint64          msecs,
-                                GCancellable *   cancellable,
-                                GError **	 error)
-{
-  ByzanzEncoderGStreamer *gstreamer = BYZANZ_ENCODER_GSTREAMER (encoder);
-  GstBus *bus;
-  GstMessage *message;
-
-  gst_app_src_end_of_stream (gstreamer->src);
-
   bus = gst_pipeline_get_bus (GST_PIPELINE (gstreamer->pipeline));
   message = gst_bus_timed_pop_filtered (bus, GST_CLOCK_TIME_NONE, GST_MESSAGE_ERROR | GST_MESSAGE_EOS);
+  g_object_unref (bus);
+
+  gst_element_set_state (gstreamer->pipeline, GST_STATE_NULL);
 
   if (GST_MESSAGE_TYPE (message) == GST_MESSAGE_ERROR) {
     gst_message_parse_error (message, error, NULL);
@@ -175,8 +155,6 @@ byzanz_encoder_gstreamer_close (ByzanzEncoder *  encoder,
     return FALSE;
   }
   gst_message_unref (message);
-  g_object_unref (bus);
-  gst_element_set_state (gstreamer->pipeline, GST_STATE_NULL);
 
   return TRUE;
 }
@@ -195,6 +173,9 @@ byzanz_encoder_gstreamer_finalize (GObject *object)
   if (gstreamer->caps)
     gst_caps_unref (gstreamer->caps);
 
+  if (gstreamer->surface)
+    cairo_surface_destroy (gstreamer->surface);
+
   G_OBJECT_CLASS (byzanz_encoder_gstreamer_parent_class)->finalize (object);
 }
 
@@ -208,9 +189,7 @@ byzanz_encoder_gstreamer_class_init (ByzanzEncoderGStreamerClass *klass)
 
   object_class->finalize = byzanz_encoder_gstreamer_finalize;
 
-  encoder_class->setup = byzanz_encoder_gstreamer_setup;
-  encoder_class->process = byzanz_encoder_gstreamer_process;
-  encoder_class->close = byzanz_encoder_gstreamer_close;
+  encoder_class->run = byzanz_encoder_gstreamer_run;
 }
 
 static void