Index: ext/ffmpeg/gstffmpegprotocol.c
===================================================================
--- ext/ffmpeg/gstffmpegprotocol.c	(revision 302)
+++ ext/ffmpeg/gstffmpegprotocol.c	(working copy)
@@ -32,6 +32,7 @@
 #include <gst/gst.h>
 
 #include "gstffmpeg.h"
+#include "gstffmpegpipe.h"
 
 typedef struct _GstProtocolInfo GstProtocolInfo;
 
@@ -286,3 +287,85 @@ URLProtocol gstreamer_protocol = {
   /*.url_seek = */gst_ffmpegdata_seek,
   /*.url_close = */gst_ffmpegdata_close,
 };
+
+
+/* specialized protocol for cross-thread pushing,
+ * based on ffmpeg's pipe protocol */
+
+static int
+gst_ffmpeg_pipe_open (URLContext * h, const char * filename, int flags)
+{
+  GstFFMpegPipe *ffpipe;
+
+  GST_LOG ("Opening %s", filename);
+
+  /* we don't support W together */
+  if (flags != URL_RDONLY) {
+    GST_WARNING ("Only read-only is supported");
+    return -EINVAL;
+  }
+
+  if (sscanf (&filename[10], "%p", &ffpipe) != 1) {
+    GST_WARNING ("could not decode pipe info from %s", filename);
+    return -EIO;
+  }
+
+  /* sanity check */
+  g_return_val_if_fail (GST_IS_ADAPTER (ffpipe->adapter), -EINVAL);
+
+  h->priv_data = (void *) ffpipe;
+  h->is_streamed = TRUE;
+  h->max_packet_size = 0;
+
+  return 0;
+}
+
+static int
+gst_ffmpeg_pipe_read (URLContext * h, unsigned char * buf, int size)
+{
+  GstFFMpegPipe * ffpipe;
+  const guint8 *data;
+  guint available;
+
+  ffpipe = (GstFFMpegPipe *) h->priv_data;
+
+  GST_LOG ("requested size %d", size);
+
+  GST_FFMPEG_PIPE_MUTEX_LOCK (ffpipe);
+  while ((available = gst_adapter_available (ffpipe->adapter)) < size
+          && !ffpipe->eos) {
+    ffpipe->needed = size;
+    GST_FFMPEG_PIPE_SIGNAL (ffpipe);
+    GST_FFMPEG_PIPE_WAIT (ffpipe);
+  }
+
+  size = MIN (available, size);
+  if (size) {
+    data = gst_adapter_peek (ffpipe->adapter, size);
+    memcpy (buf, data, size);
+    gst_adapter_flush (ffpipe->adapter, size);
+    ffpipe->needed = 0;
+  }
+  GST_FFMPEG_PIPE_MUTEX_UNLOCK (ffpipe);
+
+  return size;
+}
+
+static int
+gst_ffmpeg_pipe_close (URLContext * h)
+{
+  GST_LOG ("Closing pipe");
+
+  h->priv_data = NULL;
+
+  return 0;
+}
+
+URLProtocol gstpipe_protocol = {
+  "gstpipe",
+  gst_ffmpeg_pipe_open,
+  gst_ffmpeg_pipe_read,
+  NULL,
+  NULL,
+  gst_ffmpeg_pipe_close,
+};
Index: ext/ffmpeg/gstffmpegcodecmap.c
===================================================================
--- ext/ffmpeg/gstffmpegcodecmap.c	(revision 302)
+++ ext/ffmpeg/gstffmpegcodecmap.c	(working copy)
@@ -1695,7 +1695,7 @@ gst_ffmpeg_formatid_to_caps (const gchar
     caps = gst_caps_from_string ("application/gxf");
   } else if (!strcmp (format_name, "yuv4mpegpipe")) {
     caps = gst_caps_new_simple ("application/x-yuv4mpeg",
-				"y4mversion", G_TYPE_INT, 1,
+				"y4mversion", G_TYPE_INT, 2,
 				NULL);
   } else {
     gchar *name;
Index: ext/ffmpeg/gstffmpeg.c
===================================================================
--- ext/ffmpeg/gstffmpeg.c	(revision 302)
+++ ext/ffmpeg/gstffmpeg.c	(working copy)
@@ -122,6 +122,7 @@ plugin_init (GstPlugin * plugin)
 #endif
 
   register_protocol (&gstreamer_protocol);
+  register_protocol (&gstpipe_protocol);
 
   /* Now we can return the pointer to the newly created Plugin object. */
   return TRUE;
Index: ext/ffmpeg/gstffmpegdemux.c
===================================================================
--- ext/ffmpeg/gstffmpegdemux.c	(revision 302)
+++ ext/ffmpeg/gstffmpegdemux.c	(working copy)
@@ -36,6 +36,7 @@
 
 #include "gstffmpeg.h"
 #include "gstffmpegcodecmap.h"
+#include "gstffmpegpipe.h"
 
 typedef struct _GstFFMpegDemux GstFFMpegDemux;
 typedef struct _GstFFStream GstFFStream;
@@ -50,6 +51,7 @@ struct _GstFFStream
   GstClockTime last_ts;
   gboolean discont;
   gboolean eos;
+  GstFlowReturn last_flow;
 };
 
 struct _GstFFMpegDemux
@@ -79,6 +81,11 @@ struct _GstFFMpegDemux
 
   /* cached seek in READY */
   GstEvent *seek_event;
+
+  /* push mode data */
+  GstFFMpegPipe ffpipe;
+  /* surrogate srcpad for task */
+  GstPad *dummypad;
 };
 
 typedef struct _GstFFMpegDemuxClassParams
@@ -105,8 +112,12 @@ static GHashTable *global_plugins;
 static void gst_ffmpegdemux_class_init (GstFFMpegDemuxClass * klass);
 static void gst_ffmpegdemux_base_init (GstFFMpegDemuxClass * klass);
 static void gst_ffmpegdemux_init (GstFFMpegDemux * demux);
+static void gst_ffmpegdemux_finalize (GObject * object);
+
+static gboolean gst_ffmpegdemux_sink_event (GstPad * sinkpad, GstEvent * event);
+static GstFlowReturn gst_ffmpegdemux_chain (GstPad * sinkpad, GstBuffer * buf);
 
-static void gst_ffmpegdemux_loop (GstPad * pad);
+static void gst_ffmpegdemux_loop (GstFFMpegDemux * demux);
 static gboolean gst_ffmpegdemux_sink_activate (GstPad * sinkpad);
 static gboolean
 gst_ffmpegdemux_sink_activate_pull (GstPad * sinkpad, gboolean active);
@@ -218,6 +229,8 @@ gst_ffmpegdemux_class_init (GstFFMpegDem
 
   parent_class = g_type_class_peek_parent (klass);
 
+  gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_ffmpegdemux_finalize);
+
   gstelement_class->change_state = gst_ffmpegdemux_change_state;
   gstelement_class->send_event = gst_ffmpegdemux_send_event;
 }
@@ -230,13 +243,24 @@ gst_ffmpegdemux_init (GstFFMpegDemux * d
   gint n;
 
   demux->sinkpad = gst_pad_new_from_template (oclass->sinktempl, "sink");
-  gst_pad_set_activate_function (demux->sinkpad, gst_ffmpegdemux_sink_activate);
+  gst_pad_set_activate_function (demux->sinkpad,
+      GST_DEBUG_FUNCPTR (gst_ffmpegdemux_sink_activate));
   gst_pad_set_activatepull_function (demux->sinkpad,
-      gst_ffmpegdemux_sink_activate_pull);
+      GST_DEBUG_FUNCPTR (gst_ffmpegdemux_sink_activate_pull));
   gst_pad_set_activatepush_function (demux->sinkpad,
-      gst_ffmpegdemux_sink_activate_push);
+      GST_DEBUG_FUNCPTR (gst_ffmpegdemux_sink_activate_push));
   gst_element_add_pad (GST_ELEMENT (demux), demux->sinkpad);
 
+  /* push based setup */
+  /* dummy pad for task, slight hack but convenient, otherwise
+   * probably minor duplication of pad_xxx_task task manipulation required */
+  demux->dummypad = gst_pad_new (NULL, GST_PAD_SRC);
+  /* the following are not used in pull-based mode, so safe to set anyway */
+  gst_pad_set_event_function (demux->sinkpad,
+      GST_DEBUG_FUNCPTR (gst_ffmpegdemux_sink_event));
+  gst_pad_set_chain_function (demux->sinkpad,
+      GST_DEBUG_FUNCPTR (gst_ffmpegdemux_chain));
+
   demux->opened = FALSE;
   demux->context = NULL;
 
@@ -248,6 +272,23 @@ gst_ffmpegdemux_init (GstFFMpegDemux * d
 
   demux->seek_event = NULL;
   gst_segment_init (&demux->segment, GST_FORMAT_TIME);
+
+  /* push based data */
+  demux->ffpipe.tlock = g_mutex_new ();
+  demux->ffpipe.cond = g_cond_new ();
+  demux->ffpipe.adapter = gst_adapter_new ();
+}
+
+static void
+gst_ffmpegdemux_finalize (GObject * object)
+{
+  GstFFMpegDemux *demux;
+
+  demux = (GstFFMpegDemux *) object;
+
+  g_mutex_free (demux->ffpipe.tlock);
+  g_cond_free (demux->ffpipe.cond);
+  gst_object_unref (demux->ffpipe.adapter);
 }
 
 static void
@@ -434,6 +475,11 @@ gst_ffmpegdemux_perform_seek (GstFFMpegD
   gboolean update;
   GstSegment seeksegment;
 
+  if (!demux->seekable) {
+    GST_DEBUG_OBJECT (demux, "in push mode; ignoring seek");
+    return FALSE;
+  }
+
   GST_DEBUG_OBJECT (demux, "starting seek");
 
   if (event) {
@@ -508,9 +554,14 @@ gst_ffmpegdemux_perform_seek (GstFFMpegD
 
   /* and prepare to continue streaming */
   if (flush) {
+    gint n;
     /* send flush stop, peer will accept data and events again. We
      * are not yet providing data as we still have the STREAM_LOCK. */
     gst_ffmpegdemux_push_event (demux, gst_event_new_flush_stop ());
+    for (n = 0; n < MAX_STREAMS; ++n) {
+      if (demux->streams[n])
+        demux->streams[n]->last_flow = GST_FLOW_OK;
+    }
   } else if (res && demux->running) {
     /* we are running the current segment and doing a non-flushing seek,
      * close the segment first based on the last_stop. */
@@ -812,6 +863,7 @@ gst_ffmpegdemux_get_stream (GstFFMpegDem
   stream->discont = TRUE;
   stream->avstream = avstream;
   stream->last_ts = GST_CLOCK_TIME_NONE;
+  stream->last_flow = GST_FLOW_OK;
 
   switch (ctx->codec_type) {
     case CODEC_TYPE_VIDEO:
@@ -986,7 +1038,10 @@ gst_ffmpegdemux_open (GstFFMpegDemux * d
   gst_ffmpegdemux_close (demux);
 
   /* open via our input protocol hack */
-  location = g_strdup_printf ("gstreamer://%p", demux->sinkpad);
+  if (demux->seekable)
+    location = g_strdup_printf ("gstreamer://%p", demux->sinkpad);
+  else
+    location = g_strdup_printf ("gstpipe://%p", &demux->ffpipe);
   GST_DEBUG_OBJECT (demux, "about to call av_open_input_file %s", location);
 
   res = av_open_input_file (&demux->context, location,
@@ -1095,11 +1150,38 @@ gst_ffmpegdemux_type_find (GstTypeFind *
   }
 }
 
+static GstFlowReturn
+gst_ffmpegdemux_aggregated_flow (GstFFMpegDemux * demux)
+{
+  gint n;
+  GstFlowReturn res = GST_FLOW_OK;
+  gboolean have_ok = FALSE;
+
+  for (n = 0; n < MAX_STREAMS; n++) {
+    GstFFStream *s = demux->streams[n];
+
+    if (s) {
+      res = MIN (res, s->last_flow);
+
+      if (GST_FLOW_IS_SUCCESS (s->last_flow))
+        have_ok = TRUE;
+    }
+  }
+
+  /* NOT_LINKED is OK, if at least one pad is linked */
+  if (res == GST_FLOW_NOT_LINKED && have_ok)
+    res = GST_FLOW_OK;
+
+  GST_DEBUG_OBJECT (demux, "Returning aggregated value of %s",
+      gst_flow_get_name (res));
+
+  return res;
+}
+
 /* Task */
 static void
-gst_ffmpegdemux_loop (GstPad * pad)
+gst_ffmpegdemux_loop (GstFFMpegDemux *demux)
 {
-  GstFFMpegDemux *demux;
   GstFlowReturn ret;
   gint res;
   AVPacket pkt;
@@ -1109,8 +1191,6 @@ gst_ffmpegdemux_loop (GstPad * pad)
   GstBuffer *outbuf;
   GstClockTime timestamp, duration;
 
-  demux = (GstFFMpegDemux *) (GST_PAD_PARENT (pad));
-
   /* open file if we didn't so already */
   if (!demux->opened)
     if (!gst_ffmpegdemux_open (demux))
@@ -1193,7 +1273,15 @@ gst_ffmpegdemux_loop (GstPad * pad)
       "Sending out buffer time:%" GST_TIME_FORMAT " size:%d",
       GST_TIME_ARGS (timestamp), GST_BUFFER_SIZE (outbuf));
 
-  ret = gst_pad_push (srcpad, outbuf);
+  ret = stream->last_flow = gst_pad_push (srcpad, outbuf);
+
+  /* if a pad is in e.g. WRONG_STATE, we want to pause to unlock the STREAM_LOCK */
+  if ((ret != GST_FLOW_OK)
+       && ((ret = gst_ffmpegdemux_aggregated_flow (demux)) != GST_FLOW_OK)) {
+    GST_WARNING_OBJECT (demux, "stream_movi flow: %s / %s",
+        gst_flow_get_name (stream->last_flow), gst_flow_get_name (ret));
+    goto pause;
+  }
 
 done:
   /* can destroy the packet now */
@@ -1207,7 +1295,16 @@ pause:
     GST_LOG_OBJECT (demux, "pausing task, reason %d (%s)", ret,
         gst_flow_get_name (ret));
     demux->running = FALSE;
-    gst_pad_pause_task (demux->sinkpad);
+    if (demux->seekable)
+      gst_pad_pause_task (demux->sinkpad);
+    else {
+      GstFFMpegPipe *ffpipe = &demux->ffpipe;
+
+      GST_FFMPEG_PIPE_MUTEX_LOCK (ffpipe);
+      gst_pad_pause_task (demux->dummypad);
+      demux->ffpipe.srcresult = ret;
+      GST_FFMPEG_PIPE_MUTEX_UNLOCK (ffpipe);
+    }
 
     if (ret == GST_FLOW_UNEXPECTED) {
       if (demux->segment.flags & GST_SEEK_FLAG_SEGMENT) {
@@ -1275,6 +1372,128 @@ no_buffer:
 
 
 static gboolean
+gst_ffmpegdemux_sink_event (GstPad * sinkpad, GstEvent * event)
+{
+  GstFFMpegDemux *demux;
+  GstFFMpegPipe *ffpipe;
+  gboolean result = TRUE;
+
+  demux = (GstFFMpegDemux *) (GST_PAD_PARENT (sinkpad));
+  ffpipe = &(demux->ffpipe);
+
+  switch (GST_EVENT_TYPE (event)) {
+    case GST_EVENT_FLUSH_START:
+      /* forward event */
+      gst_pad_event_default (sinkpad, event);
+
+      /* now unblock the chain function */
+      GST_FFMPEG_PIPE_MUTEX_LOCK (ffpipe);
+      ffpipe->srcresult = GST_FLOW_WRONG_STATE;
+      GST_FFMPEG_PIPE_SIGNAL (ffpipe);
+      GST_FFMPEG_PIPE_MUTEX_UNLOCK (ffpipe);
+
+      /* loop might run into WRONG_STATE and end itself,
+       * but may also be waiting in a ffmpeg read
+       * trying to break that would make ffmpeg believe eos,
+       * so no harm to have the loop 'pausing' there ... */
+      goto done;
+    case GST_EVENT_FLUSH_STOP:
+      /* forward event */
+      gst_pad_event_default (sinkpad, event);
+
+      GST_FFMPEG_PIPE_MUTEX_LOCK (ffpipe);
+      gst_adapter_clear (ffpipe->adapter);
+      ffpipe->srcresult = GST_FLOW_OK;
+      /* loop may have decided to end itself as a result of flush WRONG_STATE */
+      gst_pad_start_task (demux->dummypad, (GstTaskFunction) gst_ffmpegdemux_loop,
+          demux);
+      demux->running = TRUE;
+      demux->flushing = FALSE;
+      GST_LOG_OBJECT (demux, "loop started");
+      GST_FFMPEG_PIPE_MUTEX_UNLOCK (ffpipe);
+      goto done;
+    case GST_EVENT_EOS:
+      /* inform the src task that it can stop now */
+      GST_FFMPEG_PIPE_MUTEX_LOCK (ffpipe);
+      ffpipe->eos = TRUE;
+      GST_FFMPEG_PIPE_SIGNAL (ffpipe);
+      GST_FFMPEG_PIPE_MUTEX_UNLOCK (ffpipe);
+
+      /* eat this event for now, task will send eos when finished */
+      gst_event_unref (event);
+      goto done;
+    default:
+      /* for a serialized event, wait until an earlier data is gone,
+       * though this is no guarantee as to when task is done with it */
+      if (GST_EVENT_IS_SERIALIZED (event)) {
+        GST_FFMPEG_PIPE_MUTEX_LOCK (ffpipe);
+        while (!ffpipe->needed)
+          GST_FFMPEG_PIPE_WAIT (ffpipe);
+        GST_FFMPEG_PIPE_MUTEX_UNLOCK (ffpipe);
+      }
+      break;
+  }
+
+  result = gst_pad_event_default (sinkpad, event);
+
+done:
+
+  return result;
+}
+
+static GstFlowReturn
+gst_ffmpegdemux_chain (GstPad * sinkpad, GstBuffer * buffer)
+{
+  GstFFMpegDemux *demux;
+  GstFFMpegPipe *ffpipe;
+
+  demux = (GstFFMpegDemux *) (GST_PAD_PARENT (sinkpad));
+  ffpipe = &demux->ffpipe;
+
+  GST_FFMPEG_PIPE_MUTEX_LOCK (ffpipe);
+
+  if (G_UNLIKELY (ffpipe->eos))
+    goto eos;
+
+  if (G_UNLIKELY (ffpipe->srcresult != GST_FLOW_OK))
+    goto ignore;
+
+  gst_adapter_push (ffpipe->adapter, buffer);
+  buffer = NULL;
+  while (gst_adapter_available (ffpipe->adapter) >= ffpipe->needed) {
+    GST_FFMPEG_PIPE_SIGNAL (ffpipe);
+    GST_FFMPEG_PIPE_WAIT (ffpipe);
+    /* may have become flushing */
+    if (G_UNLIKELY (ffpipe->srcresult != GST_FLOW_OK))
+      goto ignore;
+  }
+
+  GST_FFMPEG_PIPE_MUTEX_UNLOCK (ffpipe);
+
+  return GST_FLOW_OK;
+
+/* special cases */
+eos:
+  {
+    GST_DEBUG_OBJECT (demux, "ignoring buffer at end-of-stream");
+    GST_FFMPEG_PIPE_MUTEX_UNLOCK (ffpipe);
+
+    gst_buffer_unref (buffer);
+    return GST_FLOW_UNEXPECTED;
+  }
+ignore:
+  {
+    GST_DEBUG_OBJECT (demux, "ignoring buffer because src task encountered %s",
+        gst_flow_get_name (ffpipe->srcresult));
+    GST_FFMPEG_PIPE_MUTEX_UNLOCK (ffpipe);
+
+    if (buffer)
+      gst_buffer_unref (buffer);
+    return GST_FLOW_WRONG_STATE;
+  }
+}
+
+static gboolean
 gst_ffmpegdemux_sink_activate (GstPad * sinkpad)
 {
   GstFFMpegDemux *demux;
@@ -1293,23 +1512,54 @@ gst_ffmpegdemux_sink_activate (GstPad * 
   return res;
 }
 
+/* push mode:
+ * - not seekable
+ * - use gstpipe protocol, like ffmpeg's pipe protocol
+ * - task driving ffmpeg based on dummy src pad
+ */
 static gboolean
 gst_ffmpegdemux_sink_activate_push (GstPad * sinkpad, gboolean active)
 {
   GstFFMpegDemux *demux;
+  gboolean res;
 
   demux = (GstFFMpegDemux *) (gst_pad_get_parent (sinkpad));
 
-  GST_ELEMENT_ERROR (demux, STREAM, NOT_IMPLEMENTED,
-      (NULL),
-      ("failed to activate sinkpad in pull mode, push mode not implemented yet"));
+  if (active) {
+    demux->ffpipe.eos = FALSE;
+    demux->ffpipe.srcresult = GST_FLOW_OK;
+    demux->ffpipe.needed = 0;
+    demux->running = TRUE;
+    demux->seekable = FALSE;
+    res = gst_pad_start_task (demux->dummypad,
+        (GstTaskFunction) gst_ffmpegdemux_loop, demux);
+  } else {
+    GstFFMpegPipe *ffpipe = &demux->ffpipe;
+
+    /* release chain and loop */
+    GST_FFMPEG_PIPE_MUTEX_LOCK (ffpipe);
+    demux->ffpipe.srcresult = GST_FLOW_WRONG_STATE;
+    /* end streaming by making ffmpeg believe eos */
+    demux->ffpipe.eos = TRUE;
+    GST_FFMPEG_PIPE_SIGNAL (ffpipe);
+    GST_FFMPEG_PIPE_MUTEX_UNLOCK (ffpipe);
+
+    /* make sure streaming ends */
+    res = gst_pad_stop_task (demux->dummypad);
+    demux->running = FALSE;
+    demux->seekable = FALSE;
+  }
 
-  demux->seekable = FALSE;
   gst_object_unref (demux);
 
-  return FALSE;
+  return res;
 }
 
+/* pull mode:
+ * - seekable
+ * - use gstreamer protocol, like ffmpeg's file protocol
+ * - task driving ffmpeg based on sink pad
+ */
 static gboolean
 gst_ffmpegdemux_sink_activate_pull (GstPad * sinkpad, gboolean active)
 {
@@ -1322,7 +1572,7 @@ gst_ffmpegdemux_sink_activate_pull (GstP
     demux->running = TRUE;
     demux->seekable = TRUE;
     res = gst_pad_start_task (sinkpad, (GstTaskFunction) gst_ffmpegdemux_loop,
-        sinkpad);
+        demux);
   } else {
     demux->running = FALSE;
     res = gst_pad_stop_task (sinkpad);
@@ -1359,6 +1609,7 @@ gst_ffmpegdemux_change_state (GstElement
   switch (transition) {
     case GST_STATE_CHANGE_PAUSED_TO_READY:
       gst_ffmpegdemux_close (demux);
+      gst_adapter_clear (demux->ffpipe.adapter);
       break;
     default:
       break;
Index: ext/ffmpeg/gstffmpeg.h
===================================================================
--- ext/ffmpeg/gstffmpeg.h	(revision 302)
+++ ext/ffmpeg/gstffmpeg.h	(working copy)
@@ -57,6 +57,7 @@ int gst_ffmpeg_avcodec_close (AVCodecCon
 G_END_DECLS
 
 extern URLProtocol gstreamer_protocol;
+extern URLProtocol gstpipe_protocol;
 
 /* use GST_FFMPEG URL_STREAMHEADER with URL_WRONLY if the first
  * buffer should be used as streamheader property on the pad's caps. */
Index: ext/ffmpeg/Makefile.am
===================================================================
--- ext/ffmpeg/Makefile.am	(revision 302)
+++ ext/ffmpeg/Makefile.am	(working copy)
@@ -24,4 +24,5 @@ noinst_HEADERS = \
 	gstffmpeg.h \
 	gstffmpegcodecmap.h \
 	gstffmpegenc.h \
-	gstffmpegcfg.h
+	gstffmpegcfg.h \
+	gstffmpegpipe.h
