gst/librfb/: Some much needed hackage. Fixed #171659, but then went on to actually make it work with Vino, get timin...

Original commit message from CVS:
* gst/librfb/Makefile.am:
* gst/librfb/gstrfbsrc.c:
* gst/librfb/rfb.h:
* gst/librfb/rfbbuffer.c:
* gst/librfb/rfbbuffer.h:
* gst/librfb/rfbbytestream.c:
* gst/librfb/rfbbytestream.h:
* gst/librfb/rfbdecoder.c:
* gst/librfb/rfbdecoder.h:
Some much needed hackage.  Fixed #171659, but then went on
to actually make it work with Vino, get timing correct,
make sure it doesn't leak, etc.

9 files changed, 599 insertions, 356 deletions

diff --git a/gst/librfb/Makefile.am b/gst/librfb/Makefile.am
index 720e5cc2..2f4bb970 100644
--- a/gst/librfb/Makefile.am
+++ b/gst/librfb/Makefile.am
@@ -13,17 +13,14 @@ libgstrfbsrc_la_LDFLAGS = $(GST_PLUGIN_LDFLAGS)
 noinst_HEADERS = \
 	rfb.h \
 	rfbdecoder.h \
-	rfbbuffer.h \
-	rfbbytestream.h
+	rfbbuffer.h
 
 librfb_la_SOURCES = \
 	rfb.h \
 	rfbdecoder.c \
 	rfbdecoder.h \
 	rfbbuffer.c \
-	rfbbuffer.h \
-	rfbbytestream.c \
-	rfbbytestream.h
+	rfbbuffer.h
 
 librfb_la_CFLAGS = $(GST_CFLAGS) -I$(srcdir)/..
 librfb_la_LIBADD = $(GLIB_LIBS)
diff --git a/gst/librfb/gstrfbsrc.c b/gst/librfb/gstrfbsrc.c
index 467f1ca6..22480280 100644
--- a/gst/librfb/gstrfbsrc.c
+++ b/gst/librfb/gstrfbsrc.c
@@ -27,9 +27,12 @@
 
 #include <string.h>
 #include <stdlib.h>
+#include <unistd.h>
 
 #include <librfb/rfb.h>
 
+GST_DEBUG_CATEGORY (gst_debug_rfbsrc);
+#define GST_CAT_DEFAULT gst_debug_rfbsrc
 
 #define GST_TYPE_RFBSRC \
   (gst_rfbsrc_get_type())
@@ -63,6 +66,9 @@ struct _GstRfbsrc
   unsigned int button_mask;
 
   double framerate;
+
+  GstClock *clock;
+  GstClockTime timestamp;
 };
 
 struct _GstRfbsrcClass
@@ -110,6 +116,10 @@ gst_rfbsrc_get_type (void)
 
      static GstData *gst_rfbsrc_get (GstPad * pad);
 
+     static gboolean gst_rfbsrc_connect_to_server (GstRfbsrc * rfbsrc);
+     static gboolean gst_rfbsrc_handle_input (GstRfbsrc * rfbsrc, int timeout);
+     static void gst_rfbsrc_set_clock (GstElement * element, GstClock * clock);
+
      static const GstQueryType *gst_rfbsrc_get_query_types (GstPad * pad);
      static gboolean gst_rfbsrc_src_query (GstPad * pad,
     GstQueryType type, GstFormat * format, gint64 * value);
@@ -138,7 +148,7 @@ gst_rfbsrc_get_type (void)
         "green_mask = (int) 0x00ff0000, "
         "blue_mask = (int) 0xff000000, "
         "width = [ 16, 4096 ], "
-        "height = [ 16, 4096 ], " "framerate = [ 1.0, 60.0] ")
+        "height = [ 16, 4096 ], " "framerate = [ 1.0, 10.0] ")
     );
 
 GType
@@ -204,21 +214,27 @@ gst_rfbsrc_class_init (GstRfbsrcClass * klass)
   gobject_class->get_property = gst_rfbsrc_get_property;
 
   gstelement_class->change_state = gst_rfbsrc_change_state;
+  gstelement_class->set_clock = gst_rfbsrc_set_clock;
 }
 
 static GstElementStateReturn
 gst_rfbsrc_change_state (GstElement * element)
 {
   GstRfbsrc *rfbsrc;
+  gboolean ret;
+
+  GST_DEBUG ("enter");
 
   rfbsrc = GST_RFBSRC (element);
 
   switch (GST_STATE_TRANSITION (element)) {
     case GST_STATE_NULL_TO_READY:
-      rfbsrc->decoder = rfb_decoder_new ();
-      rfb_decoder_connect_tcp (rfbsrc->decoder, rfbsrc->server, rfbsrc->port);
-      rfbsrc->decoder->paint_rect = gst_rfbsrc_paint_rect;
-      rfbsrc->decoder->decoder_private = rfbsrc;
+      ret = gst_rfbsrc_connect_to_server (rfbsrc);
+      if (!ret) {
+        GST_ELEMENT_ERROR (element, STREAM, TOO_LAZY,
+            ("failed to connect to server"), (NULL));
+        return GST_STATE_FAILURE;
+      }
       break;
     case GST_STATE_READY_TO_PAUSED:
       break;
@@ -294,7 +310,7 @@ gst_rfbsrc_fixate (GstPad * pad, const GstCaps * caps)
   structure = gst_caps_get_structure (newcaps, 0);
 
   if (gst_caps_structure_fixate_field_nearest_double (structure, "framerate",
-          30.0)) {
+          10.0)) {
     return newcaps;
   }
 
@@ -408,7 +424,6 @@ gst_rfbsrc_paint_rect (RfbDecoder * decoder, int x, int y, int w, int h,
 {
   int i, j;
   guint8 *frame;
-  guint8 color;
   GstRfbsrc *rfbsrc;
   int width;
   int offset;
@@ -418,19 +433,45 @@ gst_rfbsrc_paint_rect (RfbDecoder * decoder, int x, int y, int w, int h,
 
   frame = rfbsrc->frame;
   width = decoder->width;
-  for (j = 0; j < h; j++) {
-    for (i = 0; i < w; i++) {
-      color = data[j * w + i];
+  switch (decoder->image_format) {
+    case RFB_DECODER_IMAGE_RGB332:
+      for (j = 0; j < h; j++) {
+        for (i = 0; i < w; i++) {
+          guint8 color;
+
+          color = data[j * w + i];
 
 #define RGB332_R(x)  ((((x)&0x07) * 0x124)>>3)
 #define RGB332_G(x)  ((((x)&0x38) * 0x124)>>6)
 #define RGB332_B(x)  ((((x)&0xc0) * 0x149)>>8)
-      offset = ((j + x) * width + (i + x)) * 4;
-      frame[offset + 0] = RGB332_B (color);
-      frame[offset + 1] = RGB332_G (color);
-      frame[offset + 2] = RGB332_R (color);
-      frame[offset + 3] = 0;
-    }
+          offset = ((j + y) * width + (i + x)) * 4;
+          frame[offset + 0] = RGB332_B (color);
+          frame[offset + 1] = RGB332_G (color);
+          frame[offset + 2] = RGB332_R (color);
+          frame[offset + 3] = 0;
+        }
+      }
+      break;
+    case RFB_DECODER_IMAGE_xRGB:
+      for (j = 0; j < h; j++) {
+        for (i = 0; i < w; i++) {
+          guint32 color;
+
+          color = ((guint32 *) data)[j * w + i];
+
+#define xRGB_R(x)  (((x)&0xff0000)>>16)
+#define xRGB_G(x)  (((x)&0x00ff00)>>8)
+#define xRGB_B(x)  ((x)&0x0000ff)
+          offset = ((j + y) * width + (i + x)) * 4;
+          frame[offset + 0] = xRGB_B (color);
+          frame[offset + 1] = xRGB_G (color);
+          frame[offset + 2] = xRGB_R (color);
+          frame[offset + 3] = 0;
+        }
+      }
+      break;
+    default:
+      g_assert_not_reached ();
   }
 
   rfbsrc->go = FALSE;
@@ -444,6 +485,8 @@ gst_rfbsrc_get (GstPad * pad)
   GstBuffer *buf;
   RfbDecoder *decoder;
   int ret;
+  GstClockTime the_time = 0;
+  GstClockTime delay;
 
   GST_DEBUG ("gst_rfbsrc_get");
 
@@ -453,39 +496,32 @@ gst_rfbsrc_get (GstPad * pad)
   rfbsrc = GST_RFBSRC (gst_pad_get_parent (pad));
   decoder = rfbsrc->decoder;
 
-  if (!decoder->inited) {
-    while (!decoder->inited) {
-      ret = rfb_decoder_iterate (decoder);
-      if (!ret) {
-        /* error */
-      }
-    }
-
-    gst_pad_renegotiate (rfbsrc->srcpad);
-
-    if (rfbsrc->frame)
-      g_free (rfbsrc->frame);
-    rfbsrc->frame = g_malloc (decoder->width * decoder->height * 4);
-
-    GST_DEBUG ("red_mask = %08x\n", decoder->red_max << decoder->red_shift);
-    GST_DEBUG ("green_mask = %08x\n",
-        decoder->green_max << decoder->green_shift);
-    GST_DEBUG ("blue_mask = %08x\n", decoder->blue_max << decoder->blue_shift);
-    rfbsrc->inter = FALSE;
-  }
-
   rfb_decoder_send_update_request (decoder, rfbsrc->inter, 0, 0, decoder->width,
       decoder->height);
-  //rfbsrc->inter = TRUE;
+  rfbsrc->inter = TRUE;
+
+  delay = GST_SECOND / rfbsrc->framerate / 2;
+  do {
+    ret = gst_rfbsrc_handle_input (rfbsrc, 10000);
 
-  rfbsrc->go = TRUE;
-  while (rfbsrc->go) {
     ret = rfb_decoder_iterate (decoder);
-    if (!ret) {
-      return GST_DATA (gst_event_new (GST_EVENT_EOS));
+    if (decoder->error_msg) {
+      char *msg;
+
+      msg = decoder->error_msg ? decoder->error_msg : "unknown";
+      GST_ELEMENT_ERROR (rfbsrc, STREAM, TOO_LAZY,
+          ("error in RFB decoder: %s", msg), (NULL));
+      //return GST_DATA (gst_event_new (GST_EVENT_EOS));
+      return NULL;
     }
-    GST_DEBUG ("iterate...\n");
-  }
+
+    if (rfbsrc->clock) {
+      the_time = gst_clock_get_time (rfbsrc->clock);
+      if (rfbsrc->timestamp == GST_CLOCK_TIME_NONE) {
+        rfbsrc->timestamp = the_time;
+      }
+    }
+  } while (the_time < rfbsrc->timestamp + delay);
 
   newsize = decoder->width * decoder->height * 4;
   g_return_val_if_fail (newsize > 0, NULL);
@@ -494,12 +530,104 @@ gst_rfbsrc_get (GstPad * pad)
 
   buf = gst_buffer_new_and_alloc (newsize);
   g_return_val_if_fail (GST_BUFFER_DATA (buf) != NULL, NULL);
+  GST_BUFFER_TIMESTAMP (buf) = rfbsrc->timestamp;
+  GST_BUFFER_DURATION (buf) = GST_SECOND / rfbsrc->framerate;
 
   memcpy (GST_BUFFER_DATA (buf), rfbsrc->frame, newsize);
 
+  rfbsrc->timestamp += GST_SECOND / rfbsrc->framerate;
+
+  GST_ERROR ("pushing");
   return GST_DATA (buf);
 }
 
+static gboolean
+gst_rfbsrc_handle_input (GstRfbsrc * rfbsrc, int timeout)
+{
+  fd_set readfds;
+  struct timeval tv;
+  int ret;
+
+  GST_DEBUG ("enter");
+
+  FD_ZERO (&readfds);
+  FD_SET (rfbsrc->decoder->fd, &readfds);
+  tv.tv_usec = timeout;
+  tv.tv_sec = 0;
+  ret = select (rfbsrc->decoder->fd + 1, &readfds, NULL, NULL, &tv);
+  GST_DEBUG ("select returned %d", ret);
+  if (ret > 0) {
+    RfbBuffer *buffer;
+
+    buffer = rfb_buffer_new_and_alloc (65536);
+    ret = read (rfbsrc->decoder->fd, buffer->data, 65536);
+    if (ret < 0) {
+      g_warning ("FIXME read error");
+    }
+    buffer->length = ret;
+    GST_DEBUG ("pushing buffer length %d", ret);
+    rfb_buffer_queue_push (rfbsrc->decoder->queue, buffer);
+
+    return TRUE;
+  } else {
+    GST_DEBUG ("timeout");
+    return FALSE;
+  }
+}
+
+static gboolean
+gst_rfbsrc_connect_to_server (GstRfbsrc * rfbsrc)
+{
+  int n_timeouts;
+
+  GST_DEBUG ("enter");
+
+  rfbsrc->decoder = rfb_decoder_new ();
+  rfb_decoder_connect_tcp (rfbsrc->decoder, rfbsrc->server, rfbsrc->port);
+  rfbsrc->decoder->paint_rect = gst_rfbsrc_paint_rect;
+  rfbsrc->decoder->decoder_private = rfbsrc;
+
+  n_timeouts = 10;
+  while (!rfbsrc->decoder->inited) {
+    gboolean ret;
+
+    ret = gst_rfbsrc_handle_input (rfbsrc, 10000);
+    if (!ret) {
+      n_timeouts--;
+      if (n_timeouts == 0)
+        return FALSE;
+    }
+
+    ret = rfb_decoder_iterate (rfbsrc->decoder);
+    if (rfbsrc->decoder->error_msg) {
+      char *msg;
+
+      msg = rfbsrc->decoder->error_msg ? rfbsrc->decoder->error_msg : "unknown";
+      GST_ELEMENT_ERROR (rfbsrc, STREAM, TOO_LAZY,
+          ("error in RFB decoder: %s", msg), (NULL));
+      return FALSE;
+    }
+  }
+
+  if (rfbsrc->frame)
+    g_free (rfbsrc->frame);
+
+  rfbsrc->frame =
+      g_malloc (rfbsrc->decoder->width * rfbsrc->decoder->height * 4);
+  rfbsrc->inter = FALSE;
+  rfbsrc->timestamp = GST_CLOCK_TIME_NONE;
+
+  return TRUE;
+}
+
+static void
+gst_rfbsrc_set_clock (GstElement * element, GstClock * clock)
+{
+  GstRfbsrc *rfbsrc = GST_RFBSRC (element);
+
+  gst_object_replace ((GstObject **) & rfbsrc->clock, (GstObject *) clock);
+}
+
 static void
 gst_rfbsrc_set_property (GObject * object, guint prop_id, const GValue * value,
     GParamSpec * pspec)
@@ -550,8 +678,11 @@ gst_rfbsrc_get_property (GObject * object, guint prop_id, GValue * value,
 static gboolean
 plugin_init (GstPlugin * plugin)
 {
-  return gst_element_register (plugin, "rfbsrc", GST_RANK_NONE,
-      GST_TYPE_RFBSRC);
+  gst_element_register (plugin, "rfbsrc", GST_RANK_NONE, GST_TYPE_RFBSRC);
+
+  GST_DEBUG_CATEGORY_INIT (gst_debug_rfbsrc, "rfbsrc", 0, "rfbsrc element");
+
+  return TRUE;
 }
 
 GST_PLUGIN_DEFINE (GST_VERSION_MAJOR,
diff --git a/gst/librfb/rfb.h b/gst/librfb/rfb.h
index 27543140..452d6bca 100644
--- a/gst/librfb/rfb.h
+++ b/gst/librfb/rfb.h
@@ -3,8 +3,9 @@
 #define _RFB_RFB_H_
 
 #include <librfb/rfbdecoder.h>
-#include <librfb/rfbbytestream.h>
 #include <librfb/rfbbuffer.h>
 
+#include <gstrfbsrc.h>
+
 #endif
 
diff --git a/gst/librfb/rfbbuffer.c b/gst/librfb/rfbbuffer.c
index e38def36..7b9c3c47 100644
--- a/gst/librfb/rfbbuffer.c
+++ b/gst/librfb/rfbbuffer.c
@@ -1,27 +1,233 @@
 
+#ifndef HAVE_CONFIG_H
+#include "config.h"
+#endif
+
 #include <rfbbuffer.h>
+#include <glib.h>
+#include <string.h>
+#include <gst/gst.h>
+
+static void rfb_buffer_free_mem (RfbBuffer * buffer, void *);
+static void rfb_buffer_free_subbuffer (RfbBuffer * buffer, void *priv);
+
+#define oil_copy_u8(a,b,c) memcpy(a,b,c)
 
 RfbBuffer *
 rfb_buffer_new (void)
 {
-  return g_new0 (RfbBuffer, 1);
+  RfbBuffer *buffer;
 
+  buffer = g_new0 (RfbBuffer, 1);
+  buffer->ref_count = 1;
+  return buffer;
 }
 
 RfbBuffer *
-rfb_buffer_new_and_alloc (int len)
+rfb_buffer_new_and_alloc (int size)
 {
-  RfbBuffer *buffer = g_new0 (RfbBuffer, 1);
+  RfbBuffer *buffer = rfb_buffer_new ();
 
-  buffer->data = g_malloc (len);
-  buffer->free_data = (void *) g_free;
+  buffer->data = g_malloc (size);
+  buffer->length = size;
+  buffer->free = rfb_buffer_free_mem;
 
   return buffer;
 }
 
+RfbBuffer *
+rfb_buffer_new_with_data (void *data, int size)
+{
+  RfbBuffer *buffer = rfb_buffer_new ();
+
+  buffer->data = data;
+  buffer->length = size;
+  buffer->free = rfb_buffer_free_mem;
+
+  return buffer;
+}
+
+RfbBuffer *
+rfb_buffer_new_subbuffer (RfbBuffer * buffer, int offset, int length)
+{
+  RfbBuffer *subbuffer = rfb_buffer_new ();
+
+  if (buffer->parent) {
+    rfb_buffer_ref (buffer->parent);
+    subbuffer->parent = buffer->parent;
+  } else {
+    rfb_buffer_ref (buffer);
+    subbuffer->parent = buffer;
+  }
+  subbuffer->data = buffer->data + offset;
+  subbuffer->length = length;
+  subbuffer->free = rfb_buffer_free_subbuffer;
+
+  return subbuffer;
+}
+
 void
-rfb_buffer_free (RfbBuffer * buffer)
+rfb_buffer_ref (RfbBuffer * buffer)
+{
+  buffer->ref_count++;
+}
+
+void
+rfb_buffer_unref (RfbBuffer * buffer)
+{
+  buffer->ref_count--;
+  if (buffer->ref_count == 0) {
+    if (buffer->free)
+      buffer->free (buffer, buffer->priv);
+    g_free (buffer);
+  }
+}
+
+static void
+rfb_buffer_free_mem (RfbBuffer * buffer, void *priv)
+{
+  g_free (buffer->data);
+}
+
+static void
+rfb_buffer_free_subbuffer (RfbBuffer * buffer, void *priv)
+{
+  rfb_buffer_unref (buffer->parent);
+}
+
+
+RfbBufferQueue *
+rfb_buffer_queue_new (void)
+{
+  return g_new0 (RfbBufferQueue, 1);
+}
+
+int
+rfb_buffer_queue_get_depth (RfbBufferQueue * queue)
+{
+  return queue->depth;
+}
+
+int
+rfb_buffer_queue_get_offset (RfbBufferQueue * queue)
 {
-  buffer->free_data (buffer->data, buffer->buffer_private);
+  return queue->offset;
+}
+
+void
+rfb_buffer_queue_free (RfbBufferQueue * queue)
+{
+  GList *g;
+
+  for (g = g_list_first (queue->buffers); g; g = g_list_next (g)) {
+    rfb_buffer_unref ((RfbBuffer *) g->data);
+  }
+  g_list_free (queue->buffers);
+  g_free (queue);
+}
+
+void
+rfb_buffer_queue_push (RfbBufferQueue * queue, RfbBuffer * buffer)
+{
+  queue->buffers = g_list_append (queue->buffers, buffer);
+  queue->depth += buffer->length;
+}
+
+RfbBuffer *
+rfb_buffer_queue_pull (RfbBufferQueue * queue, int length)
+{
+  GList *g;
+  RfbBuffer *newbuffer;
+  RfbBuffer *buffer;
+  RfbBuffer *subbuffer;
+
+  g_return_val_if_fail (length > 0, NULL);
+
+  if (queue->depth < length) {
+    return NULL;
+  }
+
+  GST_LOG ("pulling %d, %d available", length, queue->depth);
+
+  g = g_list_first (queue->buffers);
+  buffer = g->data;
+
+  if (buffer->length > length) {
+    newbuffer = rfb_buffer_new_subbuffer (buffer, 0, length);
+
+    subbuffer = rfb_buffer_new_subbuffer (buffer, length,
+        buffer->length - length);
+    g->data = subbuffer;
+    rfb_buffer_unref (buffer);
+  } else {
+    int offset = 0;
+
+    newbuffer = rfb_buffer_new_and_alloc (length);
+
+    while (offset < length) {
+      g = g_list_first (queue->buffers);
+      buffer = g->data;
+
+      if (buffer->length > length - offset) {
+        int n = length - offset;
+
+        oil_copy_u8 (newbuffer->data + offset, buffer->data, n);
+        subbuffer = rfb_buffer_new_subbuffer (buffer, n, buffer->length - n);
+        g->data = subbuffer;
+        rfb_buffer_unref (buffer);
+        offset += n;
+      } else {
+        oil_copy_u8 (newbuffer->data + offset, buffer->data, buffer->length);
+
+        queue->buffers = g_list_delete_link (queue->buffers, g);
+        offset += buffer->length;
+      }
+    }
+  }
+
+  queue->depth -= length;
+  queue->offset += length;
+
+  return newbuffer;
+}
+
+RfbBuffer *
+rfb_buffer_queue_peek (RfbBufferQueue * queue, int length)
+{
+  GList *g;
+  RfbBuffer *newbuffer;
+  RfbBuffer *buffer;
+  int offset = 0;
+
+  g_return_val_if_fail (length > 0, NULL);
+
+  if (queue->depth < length) {
+    return NULL;
+  }
+
+  GST_LOG ("peeking %d, %d available", length, queue->depth);
+
+  g = g_list_first (queue->buffers);
+  buffer = g->data;
+  if (buffer->length > length) {
+    newbuffer = rfb_buffer_new_subbuffer (buffer, 0, length);
+  } else {
+    newbuffer = rfb_buffer_new_and_alloc (length);
+    while (offset < length) {
+      buffer = g->data;
+
+      if (buffer->length > length - offset) {
+        int n = length - offset;
+
+        oil_copy_u8 (newbuffer->data + offset, buffer->data, n);
+        offset += n;
+      } else {
+        oil_copy_u8 (newbuffer->data + offset, buffer->data, buffer->length);
+        offset += buffer->length;
+      }
+      g = g_list_next (g);
+    }
+  }
 
+  return newbuffer;
 }
diff --git a/gst/librfb/rfbbuffer.h b/gst/librfb/rfbbuffer.h
index 7c4d40b5..7ef1ef6f 100644
--- a/gst/librfb/rfbbuffer.h
+++ b/gst/librfb/rfbbuffer.h
@@ -1,26 +1,47 @@
 
-#ifndef _LIBRFB_BUFFER_H_
-#define _LIBRFB_BUFFER_H_
+#ifndef __RFB_BUFFER_H__
+#define __RFB_BUFFER_H__
 
 #include <glib.h>
 
-G_BEGIN_DECLS
-
 typedef struct _RfbBuffer RfbBuffer;
+typedef struct _RfbBufferQueue RfbBufferQueue;
 
 struct _RfbBuffer
 {
-  guint8 *data;
+  unsigned char *data;
   int length;
 
-  void (*free_data) (guint8 *data, gpointer priv);
-  gpointer buffer_private;
+  int ref_count;
+
+  RfbBuffer *parent;
+
+  void (*free) (RfbBuffer *, void *);
+  void *priv;
 };
 
-RfbBuffer *rfb_buffer_new (void);
-RfbBuffer *rfb_buffer_new_and_alloc (int len);
-void rfb_buffer_free (RfbBuffer *buffer);
+struct _RfbBufferQueue
+{
+  GList *buffers;
+  int depth;
+  int offset;
+};
 
-G_END_DECLS
+RfbBuffer *rfb_buffer_new (void);
+RfbBuffer *rfb_buffer_new_and_alloc (int size);
+RfbBuffer *rfb_buffer_new_with_data (void *data, int size);
+RfbBuffer *rfb_buffer_new_subbuffer (RfbBuffer * buffer, int offset,
+    int length);
+void rfb_buffer_ref (RfbBuffer * buffer);
+void rfb_buffer_unref (RfbBuffer * buffer);
+
+RfbBufferQueue *rfb_buffer_queue_new (void);
+void rfb_buffer_queue_free (RfbBufferQueue * queue);
+int rfb_buffer_queue_get_depth (RfbBufferQueue * queue);
+int rfb_buffer_queue_get_offset (RfbBufferQueue * queue);
+void rfb_buffer_queue_push (RfbBufferQueue * queue,
+    RfbBuffer * buffer);
+RfbBuffer *rfb_buffer_queue_pull (RfbBufferQueue * queue, int len);
+RfbBuffer *rfb_buffer_queue_peek (RfbBufferQueue * queue, int len);
 
 #endif
diff --git a/gst/librfb/rfbbytestream.c b/gst/librfb/rfbbytestream.c
deleted file mode 100644
index 729851b0..00000000
--- a/gst/librfb/rfbbytestream.c
+++ /dev/null
@@ -1,138 +0,0 @@
-
-#include <rfbbytestream.h>
-#include <string.h>
-
-#include <gst/gst.h>
-
-
-RfbBytestream *
-rfb_bytestream_new (void)
-{
-  return g_new0 (RfbBytestream, 1);
-}
-
-int
-rfb_bytestream_get (RfbBytestream * bs, int len)
-{
-  RfbBuffer *buffer;
-
-  buffer = bs->get_buffer (len, bs->user_data);
-
-  if (buffer) {
-    GST_DEBUG ("got buffer (%d bytes)", buffer->length);
-    bs->buffer_list = g_list_append (bs->buffer_list, buffer);
-
-    bs->length += buffer->length;
-
-    return len;
-  } else {
-    bs->disconnected = TRUE;
-  }
-
-  return 0;
-}
-
-gboolean
-rfb_bytestream_check (RfbBytestream * bs, int len)
-{
-  while (bs->length < len) {
-    rfb_bytestream_get (bs, len - bs->length);
-    if (bs->disconnected)
-      return FALSE;
-  }
-  return TRUE;
-}
-
-static int
-rfb_bytestream_copy_nocheck (RfbBytestream * bs, RfbBuffer * buffer, int len)
-{
-  GList *item;
-  int offset;
-  int first_offset;
-  RfbBuffer *frombuf;
-  int n;
-
-  offset = 0;
-  first_offset = bs->offset;
-  for (item = bs->buffer_list; item; item = g_list_next (item)) {
-    frombuf = (RfbBuffer *) item->data;
-    n = MIN (len, frombuf->length - first_offset);
-    GST_DEBUG ("copying %d bytes from %p", n, frombuf);
-    memcpy (buffer->data + offset, frombuf->data + first_offset, n);
-    first_offset = 0;
-    len -= n;
-    offset += n;
-    if (len == 0)
-      return len;
-  }
-
-  g_assert_not_reached ();
-  return 0;
-}
-
-int
-rfb_bytestream_read (RfbBytestream * bs, RfbBuffer ** buffer, int len)
-{
-  RfbBuffer *buf;
-  int ret;
-
-  if (bs->disconnected)
-    return 0;
-
-  ret = rfb_bytestream_check (bs, len);
-  if (!ret)
-    return 0;
-
-  buf = rfb_buffer_new_and_alloc (len);
-  rfb_bytestream_copy_nocheck (bs, buf, len);
-
-  rfb_bytestream_flush (bs, len);
-
-  *buffer = buf;
-  return len;
-}
-
-int
-rfb_bytestream_peek (RfbBytestream * bs, RfbBuffer ** buffer, int len)
-{
-  RfbBuffer *buf;
-
-  if (bs->disconnected)
-    return 0;
-
-  rfb_bytestream_check (bs, len);
-
-  buf = rfb_buffer_new_and_alloc (len);
-  rfb_bytestream_copy_nocheck (bs, buf, len);
-
-  *buffer = buf;
-  return len;
-}
-
-int
-rfb_bytestream_flush (RfbBytestream * bs, int len)
-{
-  GList *item;
-  RfbBuffer *buf;
-  int n;
-
-  while ((item = bs->buffer_list)) {
-    buf = (RfbBuffer *) item->data;
-
-    n = MIN (buf->length - bs->offset, len);
-    if (n <= len) {
-      bs->offset = 0;
-      bs->buffer_list = g_list_delete_link (bs->buffer_list, item);
-      rfb_buffer_free (buf);
-    } else {
-      bs->offset = bs->offset + len;
-    }
-    bs->length -= n;
-    len -= n;
-    if (len == 0)
-      return 0;
-  }
-
-  g_assert_not_reached ();
-  return 0;
-}
diff --git a/gst/librfb/rfbbytestream.h b/gst/librfb/rfbbytestream.h
deleted file mode 100644
index 9f121948..00000000
--- a/gst/librfb/rfbbytestream.h
+++ /dev/null
@@ -1,35 +0,0 @@
-
-#ifndef _LIBRFB_BYTESTREAM_H_
-#define _LIBRFB_BYTESTREAM_H_
-
-#include <glib.h>
-
-#include <librfb/rfbbuffer.h>
-
-G_BEGIN_DECLS
-
-typedef struct _RfbBytestream RfbBytestream;
-
-struct _RfbBytestream
-{
-  RfbBuffer * (*get_buffer) (int length, gpointer user_data);
-  gpointer user_data;
-  
-  GList *buffer_list;
-  int length;
-  int offset;
-
-  int disconnected;
-};
-
-
-RfbBytestream * rfb_bytestream_new (void);
-
-int rfb_bytestream_read (RfbBytestream *bs, RfbBuffer **buffer, int len);
-int rfb_bytestream_peek (RfbBytestream *bs, RfbBuffer **buffer, int len);
-int rfb_bytestream_flush (RfbBytestream *bs, int len);
-
-
-G_END_DECLS
-
-#endif
diff --git a/gst/librfb/rfbdecoder.c b/gst/librfb/rfbdecoder.c
index 45d93782..3841c448 100644
--- a/gst/librfb/rfbdecoder.c
+++ b/gst/librfb/rfbdecoder.c
@@ -10,84 +10,17 @@
 #include <gst/gst.h>
 
 
-#if 0
-struct _RfbSocketPrivate
-{
-  int fd;
-  sockaddr sa;
-}
-#endif
-
-
-static RfbBuffer *
-rfb_socket_get_buffer (int length, gpointer user_data)
-{
-  RfbBuffer *buffer;
-  int fd = GPOINTER_TO_INT (user_data);
-  int ret;
-
-  buffer = rfb_buffer_new ();
-
-  buffer->data = g_malloc (length);
-  buffer->free_data = (void *) g_free;
-
-  GST_DEBUG ("calling read(%d, %p, %d)", fd, buffer->data, length);
-  ret = read (fd, buffer->data, length);
-  if (ret <= 0) {
-    g_critical ("read: %s", strerror (errno));
-    rfb_buffer_free (buffer);
-    return NULL;
-  }
-
-  buffer->length = ret;
-
-  return buffer;
-}
-
-static int
-rfb_socket_send_buffer (guint8 * buffer, int length, gpointer user_data)
-{
-  int fd = GPOINTER_TO_INT (user_data);
-  int ret;
-
-  GST_DEBUG ("calling write(%d, %p, %d)", fd, buffer, length);
-  ret = write (fd, buffer, length);
-  if (ret < 0) {
-    g_critical ("write: %s", strerror (errno));
-    return 0;
-  }
-
-  g_assert (ret == length);
-
-  return ret;
-}
-
-
 RfbDecoder *
 rfb_decoder_new (void)
 {
   RfbDecoder *decoder = g_new0 (RfbDecoder, 1);
 
-  decoder->bytestream = rfb_bytestream_new ();
+  decoder->queue = rfb_buffer_queue_new ();
 
   return decoder;
 }
 
 void
-rfb_decoder_use_file_descriptor (RfbDecoder * decoder, int fd)
-{
-  g_return_if_fail (decoder != NULL);
-  g_return_if_fail (!decoder->inited);
-  g_return_if_fail (fd >= 0);
-
-  decoder->bytestream->get_buffer = rfb_socket_get_buffer;
-  decoder->bytestream->user_data = GINT_TO_POINTER (fd);
-
-  decoder->send_data = rfb_socket_send_buffer;
-  decoder->buffer_handler_data = GINT_TO_POINTER (fd);
-}
-
-void
 rfb_decoder_connect_tcp (RfbDecoder * decoder, char *addr, unsigned int port)
 {
   int fd;
@@ -100,7 +33,7 @@ rfb_decoder_connect_tcp (RfbDecoder * decoder, char *addr, unsigned int port)
   inet_pton (AF_INET, addr, &sa.sin_addr);
   connect (fd, (struct sockaddr *) &sa, sizeof (struct sockaddr));
 
-  rfb_decoder_use_file_descriptor (decoder, fd);
+  decoder->fd = fd;
 }
 
 
@@ -145,19 +78,33 @@ rfb_decoder_state_wait_for_protocol_version (RfbDecoder * decoder)
 {
   RfbBuffer *buffer;
   guint8 *data;
-  int ret;
 
-  ret = rfb_bytestream_read (decoder->bytestream, &buffer, 12);
-  if (ret < 12)
+  GST_DEBUG ("enter");
+
+  buffer = rfb_buffer_queue_pull (decoder->queue, 12);
+  if (!buffer)
     return FALSE;
 
   data = buffer->data;
 
-  g_assert (memcmp (buffer->data, "RFB 003.00", 10) == 0);
   GST_DEBUG ("\"%.11s\"", buffer->data);
-  rfb_buffer_free (buffer);
+  if (memcmp (buffer->data, "RFB 003.00", 10) != 0) {
+    decoder->error_msg = g_strdup ("bad version string from server");
+    return FALSE;
+  }
 
-  rfb_decoder_send (decoder, (guchar *) "RFB 003.003\n", 12);
+  decoder->protocol_minor = RFB_GET_UINT8 (buffer->data + 10) - '0';
+  if (decoder->protocol_minor != 3 && decoder->protocol_minor != 7) {
+    decoder->error_msg = g_strdup ("bad version number from server");
+    return FALSE;
+  }
+  rfb_buffer_unref (buffer);
+
+  if (decoder->protocol_minor == 3) {
+    rfb_decoder_send (decoder, (guchar *) "RFB 003.003\n", 12);
+  } else {
+    rfb_decoder_send (decoder, (guchar *) "RFB 003.007\n", 12);
+  }
 
   decoder->state = rfb_decoder_state_wait_for_security;
 
@@ -168,19 +115,73 @@ static gboolean
 rfb_decoder_state_wait_for_security (RfbDecoder * decoder)
 {
   RfbBuffer *buffer;
-  int ret;
+  int n;
+  int i;
 
-  ret = rfb_bytestream_read (decoder->bytestream, &buffer, 4);
-  if (ret < 4)
-    return FALSE;
+  GST_DEBUG ("enter");
 
-  decoder->security_type = RFB_GET_UINT32 (buffer->data);
-  GST_DEBUG ("security = %d", decoder->security_type);
+  if (decoder->protocol_minor == 3) {
+    buffer = rfb_buffer_queue_pull (decoder->queue, 4);
+    if (!buffer)
+      return FALSE;
 
-  rfb_buffer_free (buffer);
+    decoder->security_type = RFB_GET_UINT32 (buffer->data);
+    GST_DEBUG ("security = %d", decoder->security_type);
 
-  decoder->state = rfb_decoder_state_send_client_initialisation;
-  return TRUE;
+    if (decoder->security_type == 0) {
+      decoder->error_msg = g_strdup ("connection failed");
+    } else if (decoder->security_type == 2) {
+      decoder->error_msg =
+          g_strdup ("server asked for authentication, which is unsupported");
+    }
+
+    rfb_buffer_unref (buffer);
+
+    decoder->state = rfb_decoder_state_send_client_initialisation;
+    return TRUE;
+  } else {
+    guint8 reply;
+
+    buffer = rfb_buffer_queue_peek (decoder->queue, 1);
+    if (!buffer)
+      return FALSE;
+
+    n = RFB_GET_UINT8 (buffer->data);
+    rfb_buffer_unref (buffer);
+
+    GST_DEBUG ("n = %d", n);
+
+    if (n > 0) {
+      gboolean have_none = FALSE;
+
+      buffer = rfb_buffer_queue_pull (decoder->queue, n + 1);
+
+      for (i = 0; i < n; i++) {
+        GST_DEBUG ("security = %d", RFB_GET_UINT8 (buffer->data + 1 + i));
+        if (RFB_GET_UINT8 (buffer->data + 1 + i) == 1) {
+          /* does the server allow no authentication? */
+          have_none = TRUE;
+        }
+      }
+
+      rfb_buffer_unref (buffer);
+
+      if (!have_none) {
+        decoder->error_msg =
+            g_strdup ("server asked for authentication, which is unsupported");
+        return FALSE;
+      }
+
+      reply = 1;
+      rfb_decoder_send (decoder, &reply, 1);
+    } else {
+      g_critical ("FIXME");
+      return FALSE;
+    }
+
+    decoder->state = rfb_decoder_state_send_client_initialisation;
+    return TRUE;
+  }
 }
 
 static gboolean
@@ -188,6 +189,8 @@ rfb_decoder_state_send_client_initialisation (RfbDecoder * decoder)
 {
   guint8 shared_flag;
 
+  GST_DEBUG ("enter");
+
   shared_flag = decoder->shared_flag;
   rfb_decoder_send (decoder, &shared_flag, 1);
 
@@ -200,11 +203,12 @@ rfb_decoder_state_wait_for_server_initialisation (RfbDecoder * decoder)
 {
   RfbBuffer *buffer;
   guint8 *data;
-  int ret;
   guint32 name_length;
 
-  ret = rfb_bytestream_peek (decoder->bytestream, &buffer, 24);
-  if (ret < 24)
+  GST_DEBUG ("enter");
+
+  buffer = rfb_buffer_queue_peek (decoder->queue, 24);
+  if (!buffer)
     return FALSE;
 
   data = buffer->data;
@@ -224,21 +228,50 @@ rfb_decoder_state_wait_for_server_initialisation (RfbDecoder * decoder)
 
   GST_DEBUG ("width: %d", decoder->width);
   GST_DEBUG ("height: %d", decoder->height);
+  GST_DEBUG ("bpp: %d", decoder->bpp);
+  GST_DEBUG ("depth: %d", decoder->depth);
+  GST_DEBUG ("true color: %d", decoder->true_colour);
+  GST_DEBUG ("big endian: %d", decoder->big_endian);
+
+  GST_DEBUG ("red shift: %d", decoder->red_shift);
+  GST_DEBUG ("red max: %d", decoder->red_max);
+  GST_DEBUG ("blue shift: %d", decoder->blue_shift);
+  GST_DEBUG ("blue max: %d", decoder->blue_max);
+  GST_DEBUG ("green shift: %d", decoder->green_shift);
+  GST_DEBUG ("green max: %d", decoder->green_max);
 
   name_length = RFB_GET_UINT32 (data + 20);
-  rfb_buffer_free (buffer);
+  rfb_buffer_unref (buffer);
 
-  ret = rfb_bytestream_read (decoder->bytestream, &buffer, 24 + name_length);
-  if (ret < 24 + name_length)
+  buffer = rfb_buffer_queue_pull (decoder->queue, 24 + name_length);
+  if (!buffer)
     return FALSE;
 
   decoder->name = g_strndup ((char *) (buffer->data) + 24, name_length);
   GST_DEBUG ("name: %s", decoder->name);
-  rfb_buffer_free (buffer);
+  rfb_buffer_unref (buffer);
 
   decoder->state = rfb_decoder_state_normal;
+  decoder->busy = FALSE;
   decoder->inited = TRUE;
 
+  if (decoder->bpp == 8 && decoder->depth == 8 &&
+      decoder->true_colour &&
+      decoder->red_shift == 0 && decoder->red_max == 0x07 &&
+      decoder->green_shift == 3 && decoder->green_max == 0x07 &&
+      decoder->blue_shift == 6 && decoder->blue_max == 0x03) {
+    decoder->image_format = RFB_DECODER_IMAGE_RGB332;
+  } else if (decoder->bpp == 32 && decoder->depth == 24 &&
+      decoder->true_colour && decoder->big_endian == FALSE &&
+      decoder->red_shift == 16 && decoder->red_max == 0xff &&
+      decoder->green_shift == 8 && decoder->green_max == 0xff &&
+      decoder->blue_shift == 0 && decoder->blue_max == 0xff) {
+    decoder->image_format = RFB_DECODER_IMAGE_xRGB;
+  } else {
+    decoder->error_msg = g_strdup_printf ("unsupported server image format");
+    return FALSE;
+  }
+
   return TRUE;
 }
 
@@ -246,14 +279,17 @@ static gboolean
 rfb_decoder_state_normal (RfbDecoder * decoder)
 {
   RfbBuffer *buffer;
-  int ret;
   int message_type;
 
-  ret = rfb_bytestream_read (decoder->bytestream, &buffer, 1);
-  if (ret < 1)
+  GST_DEBUG ("enter");
+
+  buffer = rfb_buffer_queue_pull (decoder->queue, 1);
+  if (!buffer)
     return FALSE;
   message_type = RFB_GET_UINT8 (buffer->data);
 
+  decoder->busy = TRUE;
+
   switch (message_type) {
     case 0:
       decoder->state = rfb_decoder_state_framebuffer_update;
@@ -263,6 +299,7 @@ rfb_decoder_state_normal (RfbDecoder * decoder)
       break;
     case 2:
       /* bell, ignored */
+      decoder->busy = FALSE;
       decoder->state = rfb_decoder_state_normal;
       break;
     case 3:
@@ -272,7 +309,7 @@ rfb_decoder_state_normal (RfbDecoder * decoder)
       g_critical ("unknown message type %d", message_type);
   }
 
-  rfb_buffer_free (buffer);
+  rfb_buffer_unref (buffer);
 
   return TRUE;
 }
@@ -281,10 +318,11 @@ static gboolean
 rfb_decoder_state_framebuffer_update (RfbDecoder * decoder)
 {
   RfbBuffer *buffer;
-  int ret;
 
-  ret = rfb_bytestream_read (decoder->bytestream, &buffer, 3);
-  if (ret < 3)
+  GST_DEBUG ("enter");
+
+  buffer = rfb_buffer_queue_pull (decoder->queue, 3);
+  if (!buffer)
     return FALSE;
 
   decoder->n_rects = RFB_GET_UINT16 (buffer->data + 1);
@@ -297,13 +335,14 @@ static gboolean
 rfb_decoder_state_framebuffer_update_rectangle (RfbDecoder * decoder)
 {
   RfbBuffer *buffer;
-  int ret;
   int x, y, w, h;
   int encoding;
   int size;
 
-  ret = rfb_bytestream_peek (decoder->bytestream, &buffer, 12);
-  if (ret < 12)
+  GST_DEBUG ("enter");
+
+  buffer = rfb_buffer_queue_peek (decoder->queue, 12);
+  if (!buffer)
     return FALSE;
 
   x = RFB_GET_UINT16 (buffer->data + 0);
@@ -315,21 +354,22 @@ rfb_decoder_state_framebuffer_update_rectangle (RfbDecoder * decoder)
   if (encoding != 0)
     g_critical ("unimplemented encoding\n");
 
-  rfb_buffer_free (buffer);
+  rfb_buffer_unref (buffer);
 
-  size = w * h;
-  ret = rfb_bytestream_read (decoder->bytestream, &buffer, size + 12);
-  if (ret < size)
+  size = w * h * (decoder->bpp / 8);
+  buffer = rfb_buffer_queue_pull (decoder->queue, size + 12);
+  if (!buffer)
     return FALSE;
 
   if (decoder->paint_rect) {
     decoder->paint_rect (decoder, x, y, w, h, buffer->data + 12);
   }
 
-  rfb_buffer_free (buffer);
+  rfb_buffer_unref (buffer);
 
   decoder->n_rects--;
   if (decoder->n_rects == 0) {
+    decoder->busy = FALSE;
     decoder->state = rfb_decoder_state_normal;
   }
   return TRUE;
@@ -397,7 +437,18 @@ rfb_decoder_send_pointer_event (RfbDecoder * decoder,
 }
 
 int
-rfb_decoder_send (RfbDecoder * decoder, guint8 * buffer, int len)
+rfb_decoder_send (RfbDecoder * decoder, guint8 * buffer, int length)
 {
-  return decoder->send_data (buffer, len, decoder->buffer_handler_data);
+  int ret;
+
+  GST_DEBUG ("calling write(%d, %p, %d)", decoder->fd, buffer, length);
+  ret = write (decoder->fd, buffer, length);
+  if (ret < 0) {
+    decoder->error_msg = g_strdup_printf ("write: %s", strerror (errno));
+    return 0;
+  }
+
+  g_assert (ret == length);
+
+  return ret;
 }
diff --git a/gst/librfb/rfbdecoder.h b/gst/librfb/rfbdecoder.h
index db16cb2c..dd73ac44 100644
--- a/gst/librfb/rfbdecoder.h
+++ b/gst/librfb/rfbdecoder.h
@@ -3,18 +3,25 @@
 #define _LIBRFB_DECODER_H_
 
 #include <glib.h>
-#include <librfb/rfbbytestream.h>
+#include <rfbbuffer.h>
 
 G_BEGIN_DECLS
 
+#define GST_CAT_DEFAULT gst_debug_rfbsrc
+
 typedef struct _RfbDecoder RfbDecoder;
 
+typedef enum {
+  RFB_DECODER_IMAGE_UNKNOWN = 0,
+  RFB_DECODER_IMAGE_RGB332,
+  RFB_DECODER_IMAGE_xRGB
+} RfbDecoderImageFormat;
+
 struct _RfbDecoder
 {
-  int (*send_data) (guint8 *buffer, int length, gpointer user_data);
-  gpointer buffer_handler_data;
+  int fd;
 
-  RfbBytestream *bytestream;
+  RfbBufferQueue *queue;
 
   gpointer decoder_private;
 
@@ -28,6 +35,7 @@ struct _RfbDecoder
 
   /* readable properties */
   gboolean inited;
+  gboolean busy;
 
   int protocol_major;
   int protocol_minor;
@@ -45,8 +53,10 @@ struct _RfbDecoder
   unsigned int red_shift;
   unsigned int green_shift;
   unsigned int blue_shift;
+  RfbDecoderImageFormat image_format;
 
   char *name;
+  char *error_msg;
 
   /* state information */
   gboolean (*state) (RfbDecoder *decoder);
@@ -70,7 +80,6 @@ typedef struct _RfbRect
 
 
 RfbDecoder *rfb_decoder_new (void);
-void rfb_decoder_use_file_descriptor (RfbDecoder * decoder, int fd);
 void rfb_decoder_connect_tcp (RfbDecoder *decoder, char * addr, unsigned int port);
 void rfb_decoder_set_peer (RfbDecoder * decoder);
 gboolean rfb_decoder_iterate (RfbDecoder * decoder);