/**
 * gst_buffer_copy_metadata:
 * @dest: a destination #GstBuffer
 * @src: a source #GstBuffer
 * @flags: flags indicating what metadata fields should be copied.
 *
 * Copies the metadata from @src into @dest. The data, size and mallocdata
 * fields are not copied.
 *
 * @flags indicate which fields will be copied. Use #GST_BUFFER_COPY_ALL to copy
 * all the metadata fields.
 *
 * This function is typically called from a custom buffer copy function after
 * creating @dest and setting the data, size, mallocdata.
 *
 * Since: 0.10.13
 */
void
gst_buffer_copy_metadata (GstBuffer * dest, const GstBuffer * src,
    GstBufferCopyFlags flags)
{
  g_return_if_fail (dest != NULL);
  g_return_if_fail (src != NULL);

  GST_CAT_LOG (GST_CAT_BUFFER, "copy %p to %p", src, dest);

  if (flags & GST_BUFFER_COPY_FLAGS) {
    guint mask;

    /* copy relevant flags */
    mask = GST_BUFFER_FLAG_PREROLL | GST_BUFFER_FLAG_IN_CAPS |
        GST_BUFFER_FLAG_DELTA_UNIT | GST_BUFFER_FLAG_DISCONT |
        GST_BUFFER_FLAG_GAP;
    GST_MINI_OBJECT_FLAGS (dest) |= GST_MINI_OBJECT_FLAGS (src) & mask;
  }

  if (flags & GST_BUFFER_COPY_TIMESTAMPS) {
    GST_BUFFER_TIMESTAMP (dest) = GST_BUFFER_TIMESTAMP (src);
    GST_BUFFER_DURATION (dest) = GST_BUFFER_DURATION (src);
    GST_BUFFER_OFFSET (dest) = GST_BUFFER_OFFSET (src);
    GST_BUFFER_OFFSET_END (dest) = GST_BUFFER_OFFSET_END (src);
  }

  if (flags & GST_BUFFER_COPY_CAPS) {
    if (GST_BUFFER_CAPS (src))
      GST_BUFFER_CAPS (dest) = gst_caps_ref (GST_BUFFER_CAPS (src));
    else
      GST_BUFFER_CAPS (dest) = NULL;
  }
}

static void
copy_caps(GstBuffer *from, GstBuffer *to)
{
  GstCaps *caps;
  if ((caps = GST_BUFFER_CAPS (from)))
    gst_caps_ref (caps);
  GST_BUFFER_CAPS (to) = caps;
}

static gboolean
_rtpbin_pad_have_data_callback (GstPad *pad, GstMiniObject *miniobj,
    gpointer user_data)
{
  FsRtpSubStream *self = FS_RTP_SUB_STREAM (user_data);
  gboolean ret = TRUE;
  gboolean remove = FALSE;
  FsRtpSession *session;

  if (fs_rtp_session_has_disposed_enter (self->priv->session, NULL))
    return FALSE;

  if (fs_rtp_sub_stream_has_stopped_enter (self))
  {
    fs_rtp_session_has_disposed_exit (self->priv->session);
    return FALSE;
  }

  g_object_ref (self);
  session = g_object_ref (self->priv->session);

  FS_RTP_SESSION_LOCK (self->priv->session);

  if (!self->priv->codecbin || !self->codec || !self->priv->caps)
  {
    ret = FALSE;
  }
  else if (GST_IS_BUFFER (miniobj))
  {
    if (!gst_caps_is_equal_fixed (GST_BUFFER_CAPS (miniobj), self->priv->caps))
    {
      if (!gst_caps_can_intersect (GST_BUFFER_CAPS (miniobj),
              self->priv->caps))
        ret = FALSE;
    }
    else
    {
      remove = TRUE;
    }
  }

  if (remove && self->priv->blocking_id)
  {
    gst_pad_remove_data_probe (pad, self->priv->blocking_id);
    self->priv->blocking_id = 0;
  }

  FS_RTP_SESSION_UNLOCK (self->priv->session);

  fs_rtp_sub_stream_has_stopped_exit (self);

  fs_rtp_session_has_disposed_exit (self->priv->session);

  g_object_unref (self);
  g_object_unref (session);

  return ret;
}

static void
do_test (SubParseInputChunk * input, guint num, const gchar * media_type)
{
    guint n;

    setup_subparse ();

    for (n = 0; n < num; ++n) {
        GstBuffer *buf;

        buf = buffer_from_static_string (input[n].in);
        fail_unless_equals_int (gst_pad_push (mysrcpad, buf), GST_FLOW_OK);
    }

    gst_pad_push_event (mysrcpad, gst_event_new_eos ());

    fail_unless_equals_int (g_list_length (buffers), num);

    for (n = 0; n < num; ++n) {
        const GstStructure *buffer_caps_struct;
        GstBuffer *buf;
        gchar *out;
        guint out_size;

        buf = g_list_nth_data (buffers, n);
        fail_unless (buf != NULL);

        /* check timestamp */
        fail_unless (GST_BUFFER_TIMESTAMP_IS_VALID (buf), NULL);
        fail_unless_equals_uint64 (GST_BUFFER_TIMESTAMP (buf), input[n].from_ts);

        /* might not be able to put a duration on the last buffer */
        if (input[n].to_ts != GST_CLOCK_TIME_NONE) {
            /* check duration */
            fail_unless (GST_BUFFER_DURATION_IS_VALID (buf), NULL);
            fail_unless_equals_uint64 (GST_BUFFER_DURATION (buf),
                                       input[n].to_ts - input[n].from_ts);
        }

        out = (gchar *) GST_BUFFER_DATA (buf);
        out_size = GST_BUFFER_SIZE (buf);
        /* shouldn't have trailing newline characters */
        fail_if (out_size > 0 && out[out_size - 1] == '\n');
        /* shouldn't include NUL-terminator in data size */
        fail_if (out_size > 0 && out[out_size - 1] == '\0');
        /* but should still have a  NUL-terminator behind the declared data */
        fail_unless_equals_int (out[out_size], '\0');
        /* make sure out string matches expected string */
        fail_unless_equals_string (out, input[n].out);
        /* check caps */
        fail_unless (GST_BUFFER_CAPS (buf) != NULL);
        buffer_caps_struct = gst_caps_get_structure (GST_BUFFER_CAPS (buf), 0);
        fail_unless_equals_string (gst_structure_get_name (buffer_caps_struct),
                                   media_type);
    }

    teardown_subparse ();
}

static void
test_srt_do_test (SubParseInputChunk * input, guint start_idx, guint num)
{
    guint n;

    GST_LOG ("srt test: start_idx = %u, num = %u", start_idx, num);

    setup_subparse ();

    for (n = start_idx; n < start_idx + num; ++n) {
        GstBuffer *buf;

        buf = buffer_from_static_string (input[n].in);
        fail_unless_equals_int (gst_pad_push (mysrcpad, buf), GST_FLOW_OK);
    }

    gst_pad_push_event (mysrcpad, gst_event_new_eos ());

    fail_unless_equals_int (g_list_length (buffers), num);

    for (n = start_idx; n < start_idx + num; ++n) {
        const GstStructure *buffer_caps_struct;
        GstBuffer *buf;
        gchar *out;
        guint out_size;

        buf = g_list_nth_data (buffers, n - start_idx);
        fail_unless (buf != NULL);
        fail_unless (GST_BUFFER_TIMESTAMP_IS_VALID (buf), NULL);
        fail_unless (GST_BUFFER_DURATION_IS_VALID (buf), NULL);
        fail_unless_equals_uint64 (GST_BUFFER_TIMESTAMP (buf), input[n].from_ts);
        fail_unless_equals_uint64 (GST_BUFFER_DURATION (buf),
                                   input[n].to_ts - input[n].from_ts);
        out = (gchar *) GST_BUFFER_DATA (buf);
        out_size = GST_BUFFER_SIZE (buf);
        /* shouldn't have trailing newline characters */
        fail_if (out_size > 0 && out[out_size - 1] == '\n');
        /* shouldn't include NUL-terminator in data size */
        fail_if (out_size > 0 && out[out_size - 1] == '\0');
        /* but should still have a  NUL-terminator behind the declared data */
        fail_unless_equals_int (out[out_size], '\0');
        /* make sure out string matches expected string */
        fail_unless_equals_string (out, input[n].out);
        /* check caps */
        fail_unless (GST_BUFFER_CAPS (buf) != NULL);
        buffer_caps_struct = gst_caps_get_structure (GST_BUFFER_CAPS (buf), 0);
        fail_unless_equals_string (gst_structure_get_name (buffer_caps_struct),
                                   "text/x-pango-markup");
    }

    teardown_subparse ();
}

static GstFlowReturn
rsn_parsetter_sink_bufferalloc (GstPad * pad, guint64 offset, guint size,
    GstCaps * caps, GstBuffer ** buf)
{
  RsnParSetter *parset = RSN_PARSETTER (gst_pad_get_parent (pad));
  GstFlowReturn ret;

  GST_LOG_OBJECT (parset, "Entering bufferalloc");

  if (rsn_parsetter_check_caps (parset, caps)) {
    ret = gst_pad_alloc_buffer (parset->srcpad, offset, size, caps, buf);
    GST_LOG_OBJECT (parset, "Not wrapping buf %p", *buf);
  } else {
    /* Allocate and wrap a downstream buffer */
    GstBuffer *orig_buf;
    GstBuffer *outbuf;
    GstCaps *override_caps = rsn_parsetter_convert_caps (parset, caps,
        parset->is_widescreen);

    ret = gst_pad_alloc_buffer (parset->srcpad, offset, size,
        override_caps, &orig_buf);
    gst_caps_unref (override_caps);

    if (ret != GST_FLOW_OK)
      return ret;

    outbuf = (GstBuffer *) rsn_wrapped_buffer_new (orig_buf);
    if (!outbuf) {
      /* FIXME: Throw error */
      return GST_FLOW_ERROR;
    }

    rsn_wrapped_buffer_set_owner (RSN_WRAPPEDBUFFER (outbuf),
        GST_ELEMENT (parset));

    gst_buffer_set_caps (outbuf, caps);

    GST_LOG_OBJECT (parset,
        "Wrapped ds buf %p with caps %" GST_PTR_FORMAT
        " into new buf %p with caps %" GST_PTR_FORMAT,
        orig_buf, GST_BUFFER_CAPS (orig_buf), outbuf, GST_BUFFER_CAPS (outbuf));

    *buf = outbuf;
  }

  gst_object_unref (GST_OBJECT (parset));

  return ret;
}

static void
got_buffer (GstElement * fakesink, GstBuffer * buf, GstPad * pad,
    gpointer user_data)
{
  GstStructure *s;

  /* Caps can be anything if we don't except icy caps */
  if (!icy_caps)
    return;

  /* Otherwise they _must_ be "application/x-icy" */
  fail_unless (GST_BUFFER_CAPS (buf) != NULL);
  s = gst_caps_get_structure (GST_BUFFER_CAPS (buf), 0);
  fail_unless_equals_string (gst_structure_get_name (s), "application/x-icy");
}

/* this is not made atomic because if the buffer were reffed from multiple
 * threads, it would have a refcount > 2 and thus be immutable.
 */
void
gst_buffer_set_caps (GstBuffer * buffer, GstCaps * caps)
{
  g_return_if_fail (buffer != NULL);

  gst_caps_replace (&GST_BUFFER_CAPS (buffer), caps);
}

static GstBusSyncReply
sync_bus_callback (GstBus * bus, GstMessage * message, gpointer data)
{
  const GstStructure *st;
  const GValue *image;
  GstBuffer *buf = NULL;
  guint8 *data_buf = NULL;
  gchar *caps_string;
  guint size = 0;
  gchar *preview_filename = NULL;
  FILE *f = NULL;
  size_t written;

  switch (GST_MESSAGE_TYPE (message)) {
    case GST_MESSAGE_ELEMENT:{
      st = gst_message_get_structure (message);
      if (st) {
        if (gst_structure_has_name (message->structure, "prepare-xwindow-id")) {
          if (!no_xwindow && window) {
            gst_x_overlay_set_window_handle (GST_X_OVERLAY (GST_MESSAGE_SRC
                    (message)), window);
            gst_message_unref (message);
            message = NULL;
            return GST_BUS_DROP;
          }
        } else if (gst_structure_has_name (st, "preview-image")) {
          GST_DEBUG ("preview-image");
          /* extract preview-image from msg */
          image = gst_structure_get_value (st, "buffer");
          if (image) {
            buf = gst_value_get_buffer (image);
            data_buf = GST_BUFFER_DATA (buf);
            size = GST_BUFFER_SIZE (buf);
            preview_filename = g_strdup_printf ("test_vga.rgb");
            caps_string = gst_caps_to_string (GST_BUFFER_CAPS (buf));
            g_print ("writing buffer to %s, elapsed: %.2fs, buffer caps: %s\n",
                preview_filename, g_timer_elapsed (timer, NULL), caps_string);
            g_free (caps_string);
            f = g_fopen (preview_filename, "w");
            if (f) {
              written = fwrite (data_buf, size, 1, f);
              if (!written) {
                g_print ("error writing file\n");
              }
              fclose (f);
            } else {
              g_print ("error opening file for raw image writing\n");
            }
            g_free (preview_filename);
          }
        }
      }
      break;
    }
    default:
      /* unhandled message */
      break;
  }
  return GST_BUS_PASS;
}

static void
gst_buffer_finalize (GstBuffer * buffer)
{
  g_return_if_fail (buffer != NULL);

  GST_CAT_LOG (GST_CAT_BUFFER, "finalize %p", buffer);

  /* free our data */
  if (G_LIKELY (buffer->malloc_data))
    buffer->free_func (buffer->malloc_data);

  gst_caps_replace (&GST_BUFFER_CAPS (buffer), NULL);

  if (buffer->parent)
    gst_buffer_unref (buffer->parent);

  if (G_UNLIKELY (buffer->priv != NULL)) {
    GstBufferPrivate *priv = buffer->priv;

    while (priv->qdata != NULL) {
      GstStructure *s = priv->qdata->data;

      gst_structure_set_parent_refcount (s, NULL);
      gst_structure_free (s);
      priv->qdata = g_list_delete_link (priv->qdata, priv->qdata);
    }
    priv->qdata = NULL;
  }

/*   ((GstMiniObjectClass *) */
/*       gst_buffer_parent_class)->finalize (GST_MINI_OBJECT_CAST (buffer)); */
}

/*
 * Verify that given buffer contains predefined ADTS frame.
 */
static void
buffer_verify_data (void *buffer, void *user_data)
{
  buffer_verify_data_s *vdata;

  if (!user_data) {
    return;
  }

  vdata = (buffer_verify_data_s *) user_data;

  GST_DEBUG ("discard: %d", vdata->discard);
  if (vdata->discard) {
    buffer_counter++;
    if (buffer_counter == vdata->discard) {
      buffer_counter = 0;
      vdata->discard = 0;
    }
    return;
  }

  fail_unless (GST_BUFFER_SIZE (buffer) == vdata->data_to_verify_size);
  fail_unless (memcmp (GST_BUFFER_DATA (buffer), vdata->data_to_verify,
          vdata->data_to_verify_size) == 0);

  if (vdata->buffers_before_offset_skip) {
    /* This is for skipping the garbage in some test cases */
    if (buffer_counter == vdata->buffers_before_offset_skip) {
      offset_counter += vdata->offset_skip_amount;
    }
  }
  if (!vdata->no_metadata) {
    fail_unless (GST_BUFFER_TIMESTAMP (buffer) == ts_counter);
    fail_unless (GST_BUFFER_DURATION (buffer) != 0);
    fail_unless (GST_BUFFER_OFFSET (buffer) == offset_counter);
  }

  if (vdata->caps) {
    GST_LOG ("%" GST_PTR_FORMAT " = %" GST_PTR_FORMAT " ?",
        GST_BUFFER_CAPS (buffer), vdata->caps);
    fail_unless (gst_caps_is_equal (GST_BUFFER_CAPS (buffer), vdata->caps));
  }

  ts_counter += GST_BUFFER_DURATION (buffer);
  offset_counter += GST_BUFFER_SIZE (buffer);
  buffer_counter++;
}

static GstFlowReturn
pad_chain(GstPad *pad,
	  GstBuffer *buf)
{
	GstDspDummy *self;
	GstFlowReturn ret;
	GstBuffer *out_buf;

	self = GST_DSP_DUMMY(GST_OBJECT_PARENT(pad));

	ret = gst_pad_alloc_buffer_and_set_caps(self->srcpad,
						GST_BUFFER_OFFSET_NONE,
						GST_BUFFER_SIZE(buf),
						GST_BUFFER_CAPS(buf),
						&out_buf);

	if (G_UNLIKELY(ret != GST_FLOW_OK)) {
		GST_ERROR_OBJECT(self, "couldn't allocate buffer");
		ret = GST_FLOW_ERROR;
		goto leave;
	}

	/* map dsp to gpp address */
	map_buffer(self, buf, self->in_buffer);
	map_buffer(self, out_buf, self->out_buffer);

	configure_dsp_node(self->dsp_handle, self->node, self->in_buffer, self->out_buffer);

	if (self->in_buffer->need_copy) {
		memcpy(self->in_buffer->data, GST_BUFFER_DATA(buf), GST_BUFFER_SIZE(buf));
		self->in_buffer->need_copy = false;
	}

	{
		dsp_msg_t msg;

		dmm_buffer_clean(self->in_buffer, self->in_buffer->size);
		msg.cmd = 1;
		msg.arg_1 = self->in_buffer->size;
		dsp_node_put_message(self->dsp_handle, self->node, &msg, -1);
		dsp_node_get_message(self->dsp_handle, self->node, &msg, -1);
		dmm_buffer_invalidate(self->out_buffer, self->out_buffer->size);
	}

	if (self->out_buffer->need_copy) {
		memcpy(GST_BUFFER_DATA(out_buf), self->out_buffer->data, GST_BUFFER_SIZE(out_buf));
		self->out_buffer->need_copy = false;
	}

	GST_BUFFER_TIMESTAMP(out_buf) = GST_BUFFER_TIMESTAMP(buf);

	ret = gst_pad_push(self->srcpad, out_buf);

leave:
	gst_buffer_unref(buf);

	return ret;
}

/**
 * gst_buffer_copy_metadata:
 * @dest: a destination #GstBuffer
 * @src: a source #GstBuffer
 * @flags: flags indicating what metadata fields should be copied.
 *
 * Copies the metadata from @src into @dest. The data, size and mallocdata
 * fields are not copied.
 *
 * @flags indicate which fields will be copied. Use #GST_BUFFER_COPY_ALL to copy
 * all the metadata fields.
 *
 * This function is typically called from a custom buffer copy function after
 * creating @dest and setting the data, size, mallocdata.
 *
 * Since: 0.10.13
 */
void
gst_buffer_copy_metadata (GstBuffer * dest, const GstBuffer * src,
    GstBufferCopyFlags flags)
{
  g_return_if_fail (dest != NULL);
  g_return_if_fail (src != NULL);

  /* nothing to copy if the buffers are the same */
  if (G_UNLIKELY (dest == src))
    return;

#if GST_VERSION_NANO == 1
  /* we enable this extra debugging in git versions only for now */
  g_warn_if_fail (gst_buffer_is_metadata_writable (dest));
#endif

  GST_CAT_LOG (GST_CAT_BUFFER, "copy %p to %p", src, dest);

  if (flags & GST_BUFFER_COPY_FLAGS) {
    guint mask;

    /* copy relevant flags */
    mask = GST_BUFFER_FLAG_PREROLL | GST_BUFFER_FLAG_IN_CAPS |
        GST_BUFFER_FLAG_DELTA_UNIT | GST_BUFFER_FLAG_DISCONT |
        GST_BUFFER_FLAG_GAP | GST_BUFFER_FLAG_MEDIA1 |
        GST_BUFFER_FLAG_MEDIA2 | GST_BUFFER_FLAG_MEDIA3;
    GST_MINI_OBJECT_FLAGS (dest) |= GST_MINI_OBJECT_FLAGS (src) & mask;
  }

  if (flags & GST_BUFFER_COPY_TIMESTAMPS) {
    GST_BUFFER_TIMESTAMP (dest) = GST_BUFFER_TIMESTAMP (src);
    GST_BUFFER_DURATION (dest) = GST_BUFFER_DURATION (src);
    GST_BUFFER_OFFSET (dest) = GST_BUFFER_OFFSET (src);
    GST_BUFFER_OFFSET_END (dest) = GST_BUFFER_OFFSET_END (src);
  }

  if (flags & GST_BUFFER_COPY_CAPS) {
    gst_caps_replace (&GST_BUFFER_CAPS (dest), GST_BUFFER_CAPS (src));
  }

  if ((flags & GST_BUFFER_COPY_QDATA)) {
    GST_CAT_TRACE (GST_CAT_BUFFER, "copying qdata from %p to %p", src, dest);
    gst_buffer_copy_qdata (dest, src);
  }
}

void MediaPlayerPrivateGStreamerBase::updateTexture(GstBuffer* buffer)
{
    if (!m_texture)
        return;

    if (!client())
        return;

    const void* srcData = 0;
#ifdef GST_API_VERSION_1
    GRefPtr<GstCaps> caps = currentVideoSinkCaps();
#else
    GRefPtr<GstCaps> caps = GST_BUFFER_CAPS(buffer);
#endif
    if (!caps)
        return;

    IntSize size;
    GstVideoFormat format;
    int pixelAspectRatioNumerator, pixelAspectRatioDenominator, stride;
    if (!getVideoSizeAndFormatFromCaps(caps.get(), size, format, pixelAspectRatioNumerator, pixelAspectRatioDenominator, stride))
        return;

    if (m_texture->size() != size)
        m_texture->reset(size);

#if GST_CHECK_VERSION(1, 1, 0)
    GstVideoGLTextureUploadMeta* meta;
    if ((meta = gst_buffer_get_video_gl_texture_upload_meta(buffer))) {
        if (meta->n_textures == 1) { // BRGx & BGRA formats use only one texture.
            const BitmapTextureGL* textureGL = static_cast<const BitmapTextureGL*>(m_texture.get());
            guint ids[4] = { textureGL->id(), 0, 0, 0 };

            if (gst_video_gl_texture_upload_meta_upload(meta, ids)) {
                client()->setPlatformLayerNeedsDisplay();
                return;
            }
        }
    }
#endif

#ifdef GST_API_VERSION_1
    GstMapInfo srcInfo;
    gst_buffer_map(buffer, &srcInfo, GST_MAP_READ);
    srcData = srcInfo.data;
#else
    srcData = GST_BUFFER_DATA(buffer);
#endif

    m_texture->updateContents(srcData, WebCore::IntRect(WebCore::IntPoint(0, 0), size), WebCore::IntPoint(0, 0), stride, BitmapTexture::UpdateCannotModifyOriginalImageData);

#ifdef GST_API_VERSION_1
    gst_buffer_unmap(buffer, &srcInfo);
#endif

    client()->setPlatformLayerNeedsDisplay();
}

static void
handoff_cb (GstElement * element, GstBuffer * buf, GstPad * pad,
            gint * p_counter)
{
    *p_counter += 1;
    GST_LOG ("counter = %d", *p_counter);

    fail_unless (GST_BUFFER_CAPS (buf) != NULL);
}

static void
feed_fakesrc (GstElement * src, GstBuffer * buf, GstPad * pad, gpointer data)
{
  GstBuffer *in_buf = GST_BUFFER (data);

  g_assert (GST_BUFFER_SIZE (buf) >= GST_BUFFER_SIZE (in_buf));
  g_assert (!GST_BUFFER_FLAG_IS_SET (buf, GST_BUFFER_FLAG_READONLY));

  gst_buffer_set_caps (buf, GST_BUFFER_CAPS (in_buf));

  memcpy (GST_BUFFER_DATA (buf), GST_BUFFER_DATA (in_buf),
      GST_BUFFER_SIZE (in_buf));

  GST_BUFFER_SIZE (buf) = GST_BUFFER_SIZE (in_buf);

  GST_DEBUG ("feeding buffer %p, size %u, caps %" GST_PTR_FORMAT,
      buf, GST_BUFFER_SIZE (buf), GST_BUFFER_CAPS (buf));
}

static void
live_switch_push (int rate, GstCaps * caps)
{
  GstBuffer *inbuffer;
  GstCaps *desired;
  GList *l;

  desired = gst_caps_copy (caps);
  gst_caps_set_simple (desired, "rate", G_TYPE_INT, rate, NULL);
  gst_pad_set_caps (mysrcpad, desired);

  fail_unless (gst_pad_alloc_buffer_and_set_caps (mysrcpad,
          GST_BUFFER_OFFSET_NONE, rate * 4, desired, &inbuffer) == GST_FLOW_OK);

  /* When the basetransform hits the non-configured case it always
   * returns a buffer with exactly the same caps as we requested so the actual
   * renegotiation (if needed) will be done in the _chain*/
  fail_unless (inbuffer != NULL);
  GST_DEBUG ("desired: %" GST_PTR_FORMAT ".... got: %" GST_PTR_FORMAT,
      desired, GST_BUFFER_CAPS (inbuffer));
  fail_unless (gst_caps_is_equal (desired, GST_BUFFER_CAPS (inbuffer)));

  memset (GST_BUFFER_DATA (inbuffer), 0, GST_BUFFER_SIZE (inbuffer));
  GST_BUFFER_DURATION (inbuffer) = GST_SECOND;
  GST_BUFFER_TIMESTAMP (inbuffer) = 0;
  GST_BUFFER_OFFSET (inbuffer) = 0;

  /* pushing gives away my reference ... */
  fail_unless (gst_pad_push (mysrcpad, inbuffer) == GST_FLOW_OK);

  /* ... but it ends up being collected on the global buffer list */
  fail_unless_equals_int (g_list_length (buffers), 1);

  for (l = buffers; l; l = l->next) {
    GstBuffer *buffer = GST_BUFFER (l->data);

    gst_buffer_unref (buffer);
  }

  g_list_free (buffers);
  buffers = NULL;

  gst_caps_unref (desired);
}

static void
shmdata_any_reader_on_new_buffer_from_source (GstElement * elt,
					      gpointer user_data)
{
  shmdata_any_reader_t *me = (shmdata_any_reader_t *) user_data;

  GstBuffer *buf;

  /* pull the next item, this can return NULL when there is no more data and
   * EOS has been received */
  buf = gst_app_sink_pull_buffer (GST_APP_SINK (me->sink_));

  if (me->on_data_ != NULL)
    {

      if (me->type_ == NULL
	  || gst_caps_is_always_compatible (me->data_caps_,
					    GST_BUFFER_CAPS (buf)))
	{
	  gchar *caps_string = gst_caps_to_string (GST_BUFFER_CAPS (buf)); 
	  me->on_data_ (me,
			(void *) buf,
			(void *) GST_BUFFER_DATA (buf),
			(int) GST_BUFFER_SIZE (buf),
			(unsigned long long)
			GST_TIME_AS_NSECONDS (GST_BUFFER_TIMESTAMP (buf)),
			(const char *)
			caps_string,
			(void *) me->on_data_user_data_);
	  g_free (caps_string);
	}
      else
	{
	  g_debug
	    ("incompatible data frame retrieved, data %p, data size %d, timestamp %llu, caps %"GST_PTR_FORMAT,
	     GST_BUFFER_DATA (buf), GST_BUFFER_SIZE (buf),
	     GST_TIME_AS_MSECONDS (GST_BUFFER_TIMESTAMP (buf)),
	     GST_BUFFER_CAPS (buf));
	}
    }
  /* if (buf) */
  /*  gst_buffer_unref (buf); */
}

static gboolean
_rtpbin_pad_have_data_callback (GstPad *pad, GstMiniObject *miniobj,
    gpointer user_data)
{
  FsRtpSubStream *self = FS_RTP_SUB_STREAM (user_data);
  gboolean ret = TRUE;
  gboolean remove = FALSE;

  FS_RTP_SESSION_LOCK (self->priv->session);

  if (!self->priv->codecbin || !self->codec || !self->priv->caps)
  {
    ret = FALSE;
  }
  else if (GST_IS_BUFFER (miniobj))
  {
    if (!gst_caps_is_equal_fixed (GST_BUFFER_CAPS (miniobj), self->priv->caps))
    {
      GstCaps *intersect = gst_caps_intersect (GST_BUFFER_CAPS (miniobj),
          self->priv->caps);

      if (gst_caps_is_empty (intersect))
        ret = FALSE;
      else
        gst_buffer_set_caps (GST_BUFFER (miniobj), self->priv->caps);
      gst_caps_unref (intersect);
    }
    else
    {
      remove = TRUE;
    }
  }

  if (remove && self->priv->blocking_id)
  {
    gst_pad_remove_data_probe (pad, self->priv->blocking_id);
    self->priv->blocking_id = 0;
  }

  FS_RTP_SESSION_UNLOCK (self->priv->session);

  return ret;
}

static GstFlowReturn
gst_caps_debug_bufferalloc (GstPad * pad, guint64 offset, guint size,
    GstCaps * caps, GstBuffer ** buf)
{
  GstCapsDebug *capsdebug;
  gchar *s;
  gchar *t;
  GstFlowReturn ret;
  GstPad *otherpad;
  gboolean newcaps;

  capsdebug = GST_CAPS_DEBUG (gst_pad_get_parent (pad));
  otherpad =
      (pad == capsdebug->srcpad) ? capsdebug->sinkpad : capsdebug->srcpad;

  newcaps = (caps != GST_PAD_CAPS (pad));

  if (newcaps) {
    s = gst_caps_to_string (caps);
    GST_INFO ("%s called bufferalloc with new caps, offset=%" G_GUINT64_FORMAT
        " size=%d caps=%s", THISPAD, offset, size, s);
    g_free (s);
  }

  ret = gst_pad_alloc_buffer_and_set_caps (otherpad, offset, size, caps, buf);

  if (newcaps) {
    GST_INFO ("%s returned %s", OTHERPAD, gst_flow_get_name (ret));
  }
  if (caps != GST_BUFFER_CAPS (*buf)) {
    s = gst_caps_to_string (caps);
    t = gst_caps_to_string (GST_BUFFER_CAPS (*buf));
    GST_INFO
        ("%s returned from bufferalloc with different caps, requested=%s returned=%s",
        OTHERPAD, s, t);
    g_free (s);
    g_free (t);
  }

  gst_object_unref (capsdebug);

  return ret;
}