diff --git a/configure.ac b/configure.ac
index 10ff728..8c0a1a5 100644
--- a/configure.ac
+++ b/configure.ac
@@ -680,6 +680,7 @@ tests/examples/launch/Makefile
 tests/examples/manual/Makefile
 tests/examples/metadata/Makefile
 tests/examples/queue/Makefile
+tests/examples/streams/Makefile
 tests/examples/typefind/Makefile
 tests/examples/xml/Makefile
 tools/Makefile
diff --git a/docs/design/draft-stream-status.txt b/docs/design/draft-stream-status.txt
new file mode 100644
index 0000000..a435a22
--- /dev/null
+++ b/docs/design/draft-stream-status.txt
@@ -0,0 +1,109 @@
+Stream Status
+-------------
+
+This document describes the design and use cases for the stream status
+messages.
+
+STREAM_STATUS messages are posted on the bus when the state of a streaming
+thread changes. The purpose of this message is to allow the application to
+interact with the streaming thread properties, such as the thread priority or
+the threadpool to use.
+
+We accomodate for the following requirements:
+
+ - Application is informed when a streaming thread is about to be created. It
+   should be possible for the application to suggest a custom GstTask.
+ - Application is informed when the status of a streaming thread is changed.
+   This can be interesting for GUI application that want to visualize the status
+   of the streaming threads (playing/paused/stopped)
+ - Application is informed when a streaming thread is destroyed.
+
+We allow for the following scenarios:
+
+ - Elements require a specific (internal) streaming thread to operate or the
+   application can create/specify a thread for the element.
+ - Elements allow the application to configure a priority on the threads.
+
+
+Use cases
+---------
+
+ * boost the priority of the udp receiver streaming thread
+
+    .--------.    .-------.    .------.    .-------.
+    | udpsrc |    | depay |    | adec |    | asink |
+    |       src->sink    src->sink   src->sink     |
+    '--------'    '-------'    '------'    '-------'
+
+  - when going from READY to PAUSED state, udpsrc will require a streaming
+    thread for pushing data into the depayloader. It will post a STREAM_STATUS
+    message indicating its requirement for a streaming thread.
+
+  - The application will usually react to the STREAM_STATUS messages with a sync
+    bus handler.
+
+  - The application can create and configure a custom GstTask to manage the
+    streaming thread or it can ignore the message which will make the element
+    use its default GstTask.
+
+  - The application can react to the ENTER/LEAVE stream status message to
+    configure the thread right before it is started/stopped. This can be used to
+    configure the thread priority.
+
+  - Before the GstTask is changed state (start/pause/stop) a STREAM_STATUS
+    message is posted that can be used by the application to keep track of
+    the running streaming threads.
+
+
+Messages
+--------
+
+ The existing STREAM_STATUS message will be further defined and implemented in
+ (selected) elements. The following fields will be contained in the message:
+
+  - "type", GST_TYPE_STREAM_STATUS_TYPE
+
+      - a set of types to control the lifecycle of the thread:
+       GST_STREAM_STATUS_TYPE_CREATE: a new streaming thread is going to be
+          created. The application has the chance to configure a custom thread.
+       GST_STREAM_STATUS_TYPE_ENTER: the streaming thread is about to enter its
+          loop function for the first time.
+       GST_STREAM_STATUS_TYPE_LEAVE: the streaming thread is about to leave its
+          loop.
+       GST_STREAM_STATUS_TYPE_DESTROY: a streaming thread is destroyed
+
+      - A set of types to control the state of the threads: 
+       GST_STREAM_STATUS_TYPE_START: a streaming thread is started
+       GST_STREAM_STATUS_TYPE_PAUSE: a streaming thread is paused
+       GST_STREAM_STATUS_TYPE_STOP: a streaming thread is stopped
+
+  - "owner", GST_TYPE_ELEMENT
+      The owner element of the thread. The message source will contain the pad
+      (or one of the pads) that will produce data by this thread. If this thread
+      does not produce data on a pad, the message source will contain the owner
+      as well. The idea is that the application should be able to see from the
+      element/pad what function this thread has in the context of the
+      application and configure the thread appropriatly.
+
+  - "object", G_TYPE, GstTask/GThread
+      A GstTask/GThread controlling this streaming thread. This can be NULL when
+      the object controlling the streaming thread is not yet created.
+
+  - "flow-return", GstFlowReturn
+      A status code for why the thread state changed. when threads are created
+      and started, this is usually GST_FLOW_OK but when they are stopping it
+      contains the reason code why it stopped.
+
+  - "reason", G_TYPE_STRING
+      A string describing the reason why the thread started/stopped/paused.
+      Can be NULL if no reason is given.
+
+
+
+Events
+------
+
+
+
+
+
diff --git a/docs/gst/gstreamer-docs.sgml b/docs/gst/gstreamer-docs.sgml
index 76e9378..c0051e5 100644
--- a/docs/gst/gstreamer-docs.sgml
+++ b/docs/gst/gstreamer-docs.sgml
@@ -86,6 +86,7 @@ Windows.  It is released under the GNU Library General Public License
     <xi:include href="xml/gsttaglist.xml" />
     <xi:include href="xml/gsttagsetter.xml" />
     <xi:include href="xml/gsttask.xml" />
+    <xi:include href="xml/gsttaskpool.xml" />
     <xi:include href="xml/gsttypefind.xml" />
     <xi:include href="xml/gsttypefindfactory.xml" />
     <xi:include href="xml/gsturihandler.xml" />
diff --git a/docs/gst/gstreamer-sections.txt b/docs/gst/gstreamer-sections.txt
index eb28680..2b40173 100644
--- a/docs/gst/gstreamer-sections.txt
+++ b/docs/gst/gstreamer-sections.txt
@@ -1043,7 +1043,6 @@ gst_iterator_result_get_type
 <TITLE>GstMessage</TITLE>
 GstMessage
 GstMessageType
-GstStructureChangeType
 GST_MESSAGE_SRC
 GST_MESSAGE_TIMESTAMP
 GST_MESSAGE_TYPE
@@ -1094,10 +1093,19 @@ gst_message_new_latency
 gst_message_new_async_start
 gst_message_parse_async_start
 gst_message_new_async_done
+
+GstStructureChangeType
 gst_message_new_structure_change
 gst_message_parse_structure_change
 gst_message_new_request_state
 gst_message_parse_request_state
+
+GstStreamStatusType
+gst_message_new_stream_status
+gst_message_parse_stream_status
+gst_message_set_stream_status_object
+gst_message_get_stream_status_object
+
 <SUBSECTION Standard>
 GstMessageClass
 GST_MESSAGE
@@ -1109,10 +1117,12 @@ GST_IS_MESSAGE_CLASS
 GST_MESSAGE_GET_CLASS
 GST_TYPE_MESSAGE_TYPE
 GST_TYPE_STRUCTURE_CHANGE_TYPE
+GST_TYPE_STREAM_STATUS_TYPE
 <SUBSECTION Private>
 gst_message_get_type
 gst_message_type_get_type
 gst_structure_change_type_get_type
+gst_stream_status_type_get_type
 GST_MESSAGE_COND
 GST_MESSAGE_GET_LOCK
 GST_MESSAGE_LOCK
@@ -2157,6 +2167,29 @@ GST_TYPE_TAG_SETTER
 gst_tag_setter_get_type
 </SECTION>
 
+<SECTION>
+<FILE>gsttaskpool</FILE>
+<TITLE>GstTaskPool</TITLE>
+GstTaskPool
+GstTaskPoolClass
+GstTaskPoolFunction
+gst_task_pool_new
+gst_task_pool_prepare
+gst_task_pool_push
+gst_task_pool_join
+gst_task_pool_cleanup
+<SUBSECTION Standard>
+GST_IS_TASK_POOL
+GST_IS_TASK_POOL_CLASS
+GST_TASK_POOL
+GST_TASK_POOL_CAST
+GST_TASK_POOL_CLASS
+GST_TASK_POOL_GET_CLASS
+GST_TYPE_TASK_POOL
+<SUBSECTION Private>
+gst_task_pool_get_type
+</SECTION>
+
 
 <SECTION>
 <FILE>gsttask</FILE>
@@ -2172,17 +2205,29 @@ GST_TASK_SIGNAL
 GST_TASK_STATE
 GST_TASK_WAIT
 
-gst_task_cleanup_all
 gst_task_create
+gst_task_set_lock
+gst_task_set_priority
+
+gst_task_set_pool
+gst_task_get_pool
+
+GstTaskThreadCallbacks
+gst_task_set_thread_callbacks
+
 gst_task_get_state
-gst_task_join
+gst_task_set_state
+
 gst_task_pause
-gst_task_set_lock
 gst_task_start
 gst_task_stop
+gst_task_join
+
+gst_task_cleanup_all
 
 <SUBSECTION Standard>
 GstTaskClass
+GstTaskPrivate
 GST_TASK
 GST_IS_TASK
 GST_TYPE_TASK
diff --git a/gst/Makefile.am b/gst/Makefile.am
index d67ea80..ebf64d4 100644
--- a/gst/Makefile.am
+++ b/gst/Makefile.am
@@ -96,6 +96,7 @@ libgstreamer_@GST_MAJORMINOR@_la_SOURCES = \
 	gsttaglist.c		\
 	gsttagsetter.c		\
 	gsttask.c		\
+	gsttaskpool.c		\
 	$(GST_TRACE_SRC)	\
 	gsttypefind.c		\
 	gsttypefindfactory.c	\
@@ -181,6 +182,7 @@ gst_headers =			\
 	gsttaglist.h		\
 	gsttagsetter.h		\
 	gsttask.h		\
+	gsttaskpool.h		\
 	gsttrace.h		\
 	gsttypefind.h		\
 	gsttypefindfactory.h	\
diff --git a/gst/gst.h b/gst/gst.h
index 9c5200a..8920de9 100644
--- a/gst/gst.h
+++ b/gst/gst.h
@@ -64,6 +64,7 @@
 #include <gst/gsttaglist.h>
 #include <gst/gsttagsetter.h>
 #include <gst/gsttask.h>
+#include <gst/gsttaskpool.h>
 #include <gst/gsttrace.h>
 #include <gst/gsttypefind.h>
 #include <gst/gsttypefindfactory.h>
diff --git a/gst/gstmessage.c b/gst/gstmessage.c
index 3078ccf..1775877 100644
--- a/gst/gstmessage.c
+++ b/gst/gstmessage.c
@@ -1458,3 +1458,114 @@ gst_message_parse_request_state (GstMessage * message, GstState * state)
     *state = g_value_get_enum (gst_structure_id_get_value (message->structure,
             GST_QUARK (NEW_STATE)));
 }
+
+/**
+ * gst_message_new_stream_status:
+ * @src: The object originating the message.
+ * @type: The stream status type.
+ * @owner: The owner element of @src.
+ *
+ * Create a new stream status message. This message is posted when a streaming
+ * thread is created/destroyed or when the state changed.
+ * 
+ * Returns: The new stream status message.
+ *
+ * MT safe.
+ *
+ * Since: 0.10.24.
+ */
+GstMessage *
+gst_message_new_stream_status (GstObject * src, GstStreamStatusType type,
+    GstElement * owner)
+{
+  GstMessage *message;
+  GstStructure *structure;
+
+  structure = gst_structure_empty_new ("GstMessageStreamStatus");
+  gst_structure_id_set (structure,
+      GST_QUARK (TYPE), GST_TYPE_STREAM_STATUS_TYPE, (gint) type,
+      GST_QUARK (OWNER), GST_TYPE_ELEMENT, owner, NULL);
+  message = gst_message_new_custom (GST_MESSAGE_STREAM_STATUS, src, structure);
+
+  return message;
+}
+
+/**
+ * gst_message_parse_stream_status:
+ * @message: A valid #GstMessage of type GST_MESSAGE_STREAM_STATUS.
+ * @type: A pointer to hold the status type
+ * @owner: The owner element of the message source
+ *
+ * Extracts the stream status type and owner the GstMessage. The returned
+ * owner remains valid for as long as the reference to @message is valid and
+ * should thus not be unreffed.
+ *
+ * Since: 0.10.24.
+ *
+ * MT safe.
+ */
+void
+gst_message_parse_stream_status (GstMessage * message,
+    GstStreamStatusType * type, GstElement ** owner)
+{
+  const GValue *owner_gvalue;
+
+  g_return_if_fail (GST_IS_MESSAGE (message));
+  g_return_if_fail (GST_MESSAGE_TYPE (message) == GST_MESSAGE_STREAM_STATUS);
+
+  owner_gvalue =
+      gst_structure_id_get_value (message->structure, GST_QUARK (OWNER));
+  g_return_if_fail (owner_gvalue != NULL);
+
+  if (type)
+    *type = g_value_get_enum (gst_structure_id_get_value (message->structure,
+            GST_QUARK (TYPE)));
+  if (owner)
+    *owner = (GstElement *) g_value_get_object (owner_gvalue);
+}
+
+/**
+ * gst_message_set_stream_status_object:
+ * @message: A valid #GstMessage of type GST_MESSAGE_STREAM_STATUS.
+ * @object: the object controlling the streaming
+ *
+ * Configures the object handling the streaming thread. This is usually a
+ * GstTask object but other objects might be added in the future.
+ *
+ * Since: 0.10.24
+ */
+void
+gst_message_set_stream_status_object (GstMessage * message,
+    const GValue * object)
+{
+  g_return_if_fail (GST_IS_MESSAGE (message));
+  g_return_if_fail (GST_MESSAGE_TYPE (message) == GST_MESSAGE_STREAM_STATUS);
+
+  gst_structure_id_set_value (message->structure, GST_QUARK (OBJECT), object);
+}
+
+/**
+ * gst_message_get_stream_status_object:
+ * @message: A valid #GstMessage of type GST_MESSAGE_STREAM_STATUS.
+ *
+ * Extracts the object managing the streaming thread from @message.
+ *
+ * Returns: a GValue containing the object that manages the streaming thread.
+ * This object is usually of type GstTask but other types can be added in the
+ * future. The object remains valid as long as @message is valid.
+ *
+ * Since: 0.10.24
+ */
+const GValue *
+gst_message_get_stream_status_object (GstMessage * message)
+{
+  const GValue *result;
+
+  g_return_val_if_fail (GST_IS_MESSAGE (message), NULL);
+  g_return_val_if_fail (GST_MESSAGE_TYPE (message) == GST_MESSAGE_STREAM_STATUS,
+      NULL);
+
+  result = gst_structure_id_get_value (message->structure, GST_QUARK (OBJECT));
+
+  return result;
+}
diff --git a/gst/gstmessage.h b/gst/gstmessage.h
index 9a06c9d..6ed1790 100644
--- a/gst/gstmessage.h
+++ b/gst/gstmessage.h
@@ -198,6 +198,32 @@ typedef enum {
 } GstStructureChangeType;
 
 /**
+ * GstStreamStatusType:
+ * @GST_STREAM_STATUS_TYPE_CREATE: A new thread need to be created.
+ * @GST_STREAM_STATUS_TYPE_ENTER: a thread entered its loop function
+ * @GST_STREAM_STATUS_TYPE_LEAVE: a thread left its loop function
+ * @GST_STREAM_STATUS_TYPE_DESTROY: a thread is destroyed
+ * @GST_STREAM_STATUS_TYPE_START: a thread is started
+ * @GST_STREAM_STATUS_TYPE_PAUSE: a thread is paused
+ * @GST_STREAM_STATUS_TYPE_STOP: a thread is stopped
+ *
+ * The type of a #GstMessageStreamStatus. The stream status messages inform the
+ * application of new streaming threads and their status.
+ *
+ * Since: 0.10.24
+ */
+typedef enum {
+  GST_STREAM_STATUS_TYPE_CREATE   = 0,
+  GST_STREAM_STATUS_TYPE_ENTER    = 1,
+  GST_STREAM_STATUS_TYPE_LEAVE    = 2,
+  GST_STREAM_STATUS_TYPE_DESTROY  = 3,
+
+  GST_STREAM_STATUS_TYPE_START    = 8,
+  GST_STREAM_STATUS_TYPE_PAUSE    = 9,
+  GST_STREAM_STATUS_TYPE_STOP     = 10
+} GstStreamStatusType;
+
+/**
  * GstMessage:
  * @mini_object: the parent structure
  * @type: the #GstMessageType of the message
@@ -337,11 +363,11 @@ GstMessage *	gst_message_new_tag		(GstObject * src, GstTagList * tag_list);
 void		gst_message_parse_tag		(GstMessage *message, GstTagList **tag_list);
 
 /* BUFFERING */
-GstMessage *	gst_message_new_buffering	(GstObject * src, gint percent);
-void 		gst_message_parse_buffering	(GstMessage *message, gint *percent);
-void            gst_message_set_buffering_stats (GstMessage *message, GstBufferingMode mode,
-		                                 gint avg_in, gint avg_out,
-						 gint64 buffering_left);
+GstMessage *	gst_message_new_buffering	  (GstObject * src, gint percent);
+void 		gst_message_parse_buffering	  (GstMessage *message, gint *percent);
+void            gst_message_set_buffering_stats   (GstMessage *message, GstBufferingMode mode,
+                                                   gint avg_in, gint avg_out,
+                                                   gint64 buffering_left);
 void            gst_message_parse_buffering_stats (GstMessage *message, GstBufferingMode *mode,
                                                    gint *avg_in, gint *avg_out,
                                                    gint64 *buffering_left);
@@ -405,6 +431,14 @@ GstMessage *	gst_message_new_structure_change   (GstObject * src, GstStructureCh
 void		gst_message_parse_structure_change (GstMessage *message, GstStructureChangeType *type,
                                                     GstElement **owner, gboolean *busy);
 
+/* STREAM STATUS */
+GstMessage *	gst_message_new_stream_status        (GstObject * src, GstStreamStatusType type,
+                                                      GstElement *owner);
+void		gst_message_parse_stream_status      (GstMessage *message, GstStreamStatusType *type,
+                                                      GstElement **owner);
+void            gst_message_set_stream_status_object (GstMessage *message, const GValue *object);
+const GValue *  gst_message_get_stream_status_object (GstMessage *message);
+
 /* REQUEST_STATE */
 GstMessage *    gst_message_new_request_state   (GstObject * src, GstState state);
 void            gst_message_parse_request_state (GstMessage * message, GstState *state);
diff --git a/gst/gstpad.c b/gst/gstpad.c
index 5a0361c..d010d48 100644
--- a/gst/gstpad.c
+++ b/gst/gstpad.c
@@ -4801,6 +4801,53 @@ gst_pad_get_element_private (GstPad * pad)
   return pad->element_private;
 }
 
+static void
+do_stream_status (GstPad * pad, GstStreamStatusType type,
+    GThread * thread, GstTask * task)
+{
+  GstElement *parent;
+
+  GST_DEBUG_OBJECT (pad, "doing stream-status %d", type);
+
+  if ((parent = GST_ELEMENT_CAST (gst_pad_get_parent (pad)))) {
+    if (GST_IS_ELEMENT (parent)) {
+      GstMessage *message;
+      GValue value = { 0 };
+
+      message = gst_message_new_stream_status (GST_OBJECT_CAST (pad),
+          type, parent);
+
+      g_value_init (&value, GST_TYPE_TASK);
+      g_value_set_object (&value, task);
+      gst_message_set_stream_status_object (message, &value);
+      g_value_unset (&value);
+
+      GST_DEBUG_OBJECT (pad, "posting stream-status %d", type);
+      gst_element_post_message (parent, message);
+    }
+    gst_object_unref (parent);
+  }
+}
+
+static void
+pad_enter_thread (GstTask * task, GThread * thread, gpointer user_data)
+{
+  do_stream_status (GST_PAD_CAST (user_data), GST_STREAM_STATUS_TYPE_ENTER,
+      thread, task);
+}
+
+static void
+pad_leave_thread (GstTask * task, GThread * thread, gpointer user_data)
+{
+  do_stream_status (GST_PAD_CAST (user_data), GST_STREAM_STATUS_TYPE_LEAVE,
+      thread, task);
+}
+
+static GstTaskThreadCallbacks thr_callbacks = {
+  pad_enter_thread,
+  pad_leave_thread
+};
+
 /**
  * gst_pad_start_task:
  * @pad: the #GstPad to start the task of
@@ -4818,6 +4865,7 @@ gboolean
 gst_pad_start_task (GstPad * pad, GstTaskFunction func, gpointer data)
 {
   GstTask *task;
+  gboolean res;
 
   g_return_val_if_fail (GST_IS_PAD (pad), FALSE);
   g_return_val_if_fail (func != NULL, FALSE);
@@ -4829,13 +4877,33 @@ gst_pad_start_task (GstPad * pad, GstTaskFunction func, gpointer data)
   if (task == NULL) {
     task = gst_task_create (func, data);
     gst_task_set_lock (task, GST_PAD_GET_STREAM_LOCK (pad));
-    GST_PAD_TASK (pad) = task;
+    gst_task_set_thread_callbacks (task, &thr_callbacks, pad, NULL);
     GST_DEBUG_OBJECT (pad, "created task");
+    /* release lock to post the message */
+    GST_OBJECT_UNLOCK (pad);
+
+    do_stream_status (pad, GST_STREAM_STATUS_TYPE_CREATE, NULL, task);
+
+    GST_OBJECT_LOCK (pad);
+    /* nobody else is supposed to have changed the pad now */
+    if (GST_PAD_TASK (pad) != NULL)
+      goto concurrent_start;
+    GST_PAD_TASK (pad) = task;
   }
-  gst_task_start (task);
+  res = gst_task_set_state (task, GST_TASK_STARTED);
   GST_OBJECT_UNLOCK (pad);
 
-  return TRUE;
+  return res;
+
+  /* ERRORS */
+concurrent_start:
+  {
+    g_warning ("two threads started pad %s:%s at the same time",
+        GST_DEBUG_PAD_NAME (pad));
+    GST_OBJECT_UNLOCK (pad);
+    gst_object_unref (task);
+    return FALSE;
+  }
 }
 
 /**
@@ -4853,6 +4921,7 @@ gboolean
 gst_pad_pause_task (GstPad * pad)
 {
   GstTask *task;
+  gboolean res;
 
   g_return_val_if_fail (GST_IS_PAD (pad), FALSE);
 
@@ -4862,7 +4931,7 @@ gst_pad_pause_task (GstPad * pad)
   task = GST_PAD_TASK (pad);
   if (task == NULL)
     goto no_task;
-  gst_task_pause (task);
+  res = gst_task_set_state (task, GST_TASK_PAUSED);
   GST_OBJECT_UNLOCK (pad);
 
   /* wait for task function to finish, this lock is recursive so it does nothing
@@ -4870,7 +4939,7 @@ gst_pad_pause_task (GstPad * pad)
   GST_PAD_STREAM_LOCK (pad);
   GST_PAD_STREAM_UNLOCK (pad);
 
-  return TRUE;
+  return res;
 
 no_task:
   {
@@ -4900,6 +4969,7 @@ gboolean
 gst_pad_stop_task (GstPad * pad)
 {
   GstTask *task;
+  gboolean res;
 
   g_return_val_if_fail (GST_IS_PAD (pad), FALSE);
 
@@ -4910,7 +4980,7 @@ gst_pad_stop_task (GstPad * pad)
   if (task == NULL)
     goto no_task;
   GST_PAD_TASK (pad) = NULL;
-  gst_task_stop (task);
+  res = gst_task_set_state (task, GST_TASK_STOPPED);
   GST_OBJECT_UNLOCK (pad);
 
   GST_PAD_STREAM_LOCK (pad);
@@ -4921,7 +4991,7 @@ gst_pad_stop_task (GstPad * pad)
 
   gst_object_unref (task);
 
-  return TRUE;
+  return res;
 
 no_task:
   {
diff --git a/gst/gstquark.c b/gst/gstquark.c
index 9e2887d..d1d69fb 100644
--- a/gst/gstquark.c
+++ b/gst/gstquark.c
@@ -37,7 +37,7 @@ static const gchar *_quark_strings[] = {
   "max-latency", "busy", "type", "owner", "update", "applied-rate",
   "start", "stop", "minsize", "maxsize", "async", "proportion",
   "diff", "timestamp", "flags", "cur-type", "cur", "stop-type",
-  "latency", "uri"
+  "latency", "uri", "object"
 };
 
 GQuark _priv_gst_quark_table[GST_QUARK_MAX];
diff --git a/gst/gstquark.h b/gst/gstquark.h
index b4e78e8..0f29119 100644
--- a/gst/gstquark.h
+++ b/gst/gstquark.h
@@ -78,8 +78,9 @@ typedef enum _GstQuarkId
   GST_QUARK_STOP_TYPE = 49,
   GST_QUARK_LATENCY = 50,
   GST_QUARK_URI = 51,
+  GST_QUARK_OBJECT = 52,
 
-  GST_QUARK_MAX = 52
+  GST_QUARK_MAX = 53
 } GstQuarkId;
 
 extern GQuark _priv_gst_quark_table[GST_QUARK_MAX];
diff --git a/gst/gsttask.c b/gst/gsttask.c
index 0f88dfe..2a6dd1a 100644
--- a/gst/gsttask.c
+++ b/gst/gsttask.c
@@ -42,11 +42,12 @@
  * gst_task_set_lock().
  *
  * The task can be started, paused and stopped with gst_task_start(), gst_task_pause()
- * and gst_task_stop() respectively.
+ * and gst_task_stop() respectively or with the gst_task_set_state() function.
  *
  * A #GstTask will repeatedly call the #GstTaskFunction with the user data
- * that was provided when creating the task with gst_task_create(). Before calling
- * the function it will acquire the provided lock.
+ * that was provided when creating the task with gst_task_create(). While calling
+ * the function it will acquire the provided lock. The provided lock is released
+ * when the task pauses or stops.
  *
  * Stopping a task with gst_task_stop() will not immediately make sure the task is
  * not running anymore. Use gst_task_join() to make sure the task is completely
@@ -66,13 +67,32 @@
 GST_DEBUG_CATEGORY_STATIC (task_debug);
 #define GST_CAT_DEFAULT (task_debug)
 
+#define GST_TASK_GET_PRIVATE(obj)  \
+   (G_TYPE_INSTANCE_GET_PRIVATE ((obj), GST_TYPE_TASK, GstTaskPrivate))
+
+struct _GstTaskPrivate
+{
+  /* callbacks for managing the thread of this task */
+  GstTaskThreadCallbacks thr_callbacks;
+  gpointer thr_user_data;
+  GDestroyNotify thr_notify;
+
+  gboolean prio_set;
+  GThreadPriority priority;
+
+  /* configured pool */
+  GstTaskPool *pool;
+
+  /* remember the pool and id that is currently running. */
+  gpointer id;
+  GstTaskPool *pool_id;
+};
+
 static void gst_task_class_init (GstTaskClass * klass);
 static void gst_task_init (GstTask * task);
 static void gst_task_finalize (GObject * object);
 
-static void gst_task_func (GstTask * task, GstTaskClass * tclass);
-
-static GstObjectClass *parent_class = NULL;
+static void gst_task_func (GstTask * task);
 
 static GStaticMutex pool_lock = G_STATIC_MUTEX_INIT;
 
@@ -84,50 +104,85 @@ static GStaticMutex pool_lock = G_STATIC_MUTEX_INIT;
 G_DEFINE_TYPE_WITH_CODE (GstTask, gst_task, GST_TYPE_OBJECT, _do_init);
 
 static void
+init_klass_pool (GstTaskClass * klass)
+{
+  g_static_mutex_lock (&pool_lock);
+  if (klass->pool) {
+    gst_task_pool_cleanup (klass->pool);
+    gst_object_unref (klass->pool);
+  }
+  klass->pool = gst_task_pool_new ();
+  gst_task_pool_prepare (klass->pool, NULL);
+  g_static_mutex_unlock (&pool_lock);
+}
+
+static void
 gst_task_class_init (GstTaskClass * klass)
 {
   GObjectClass *gobject_class;
 
   gobject_class = (GObjectClass *) klass;
 
-  parent_class = g_type_class_peek_parent (klass);
+  g_type_class_add_private (klass, sizeof (GstTaskPrivate));
 
   gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_task_finalize);
 
-  klass->pool = g_thread_pool_new (
-      (GFunc) gst_task_func, klass, -1, FALSE, NULL);
+  init_klass_pool (klass);
 }
 
 static void
 gst_task_init (GstTask * task)
 {
+  GstTaskClass *klass;
+
+  klass = GST_TASK_GET_CLASS (task);
+
+  task->priv = GST_TASK_GET_PRIVATE (task);
   task->running = FALSE;
   task->abidata.ABI.thread = NULL;
   task->lock = NULL;
   task->cond = g_cond_new ();
   task->state = GST_TASK_STOPPED;
+  task->priv->prio_set = FALSE;
+
+  /* use the default klass pool for this task, users can
+   * override this later */
+  g_static_mutex_lock (&pool_lock);
+  task->priv->pool = gst_object_ref (klass->pool);
+  g_static_mutex_unlock (&pool_lock);
 }
 
 static void
 gst_task_finalize (GObject * object)
 {
   GstTask *task = GST_TASK (object);
+  GstTaskPrivate *priv = task->priv;
 
   GST_DEBUG ("task %p finalize", task);
 
+  if (priv->thr_notify)
+    priv->thr_notify (priv->thr_user_data);
+  priv->thr_notify = NULL;
+  priv->thr_user_data = NULL;
+
+  gst_object_unref (priv->pool);
+
   /* task thread cannot be running here since it holds a ref
    * to the task so that the finalize could not have happened */
   g_cond_free (task->cond);
   task->cond = NULL;
 
-  G_OBJECT_CLASS (parent_class)->finalize (object);
+  G_OBJECT_CLASS (gst_task_parent_class)->finalize (object);
 }
 
 static void
-gst_task_func (GstTask * task, GstTaskClass * tclass)
+gst_task_func (GstTask * task)
 {
   GStaticRecMutex *lock;
   GThread *tself;
+  GstTaskPrivate *priv;
+
+  priv = task->priv;
 
   tself = g_thread_self ();
 
@@ -143,8 +198,15 @@ gst_task_func (GstTask * task, GstTaskClass * tclass)
   if (G_UNLIKELY (lock == NULL))
     goto no_lock;
   task->abidata.ABI.thread = tself;
+  /* only update the priority when it was changed */
+  if (priv->prio_set)
+    g_thread_set_priority (tself, priv->priority);
   GST_OBJECT_UNLOCK (task);
 
+  /* fire the enter_thread callback when we need to */
+  if (priv->thr_callbacks.enter_thread)
+    priv->thr_callbacks.enter_thread (task, tself, priv->thr_user_data);
+
   /* locking order is TASK_LOCK, LOCK */
   g_static_rec_mutex_lock (lock);
   GST_OBJECT_LOCK (task);
@@ -189,6 +251,17 @@ exit:
    * before releasing the lock as we can be sure that a ref is held by the
    * caller of the join(). */
   task->running = FALSE;
+  if (priv->thr_callbacks.leave_thread) {
+    /* fire the leave_thread callback when we need to. We need to do this before
+     * we signal the task and with the task lock released. */
+    GST_OBJECT_UNLOCK (task);
+    priv->thr_callbacks.leave_thread (task, tself, priv->thr_user_data);
+    GST_OBJECT_LOCK (task);
+  } else {
+    /* restore normal priority when releasing back into the pool, we will not
+     * touch the priority when a custom callback has been installed. */
+    g_thread_set_priority (tself, G_THREAD_PRIORITY_NORMAL);
+  }
   GST_TASK_SIGNAL (task);
   GST_OBJECT_UNLOCK (task);
 
@@ -218,18 +291,7 @@ gst_task_cleanup_all (void)
   GstTaskClass *klass;
 
   if ((klass = g_type_class_peek (GST_TYPE_TASK))) {
-    g_static_mutex_lock (&pool_lock);
-    if (klass->pool) {
-      /* Shut down all the threads, we still process the ones scheduled
-       * because the unref happens in the thread function.
-       * Also wait for currently running ones to finish. */
-      g_thread_pool_free (klass->pool, FALSE, TRUE);
-      /* create new pool, so we can still do something after this
-       * call. */
-      klass->pool = g_thread_pool_new (
-          (GFunc) gst_task_func, klass, -1, FALSE, NULL);
-    }
-    g_static_mutex_unlock (&pool_lock);
+    init_klass_pool (klass);
   }
 }
 
@@ -245,6 +307,13 @@ gst_task_cleanup_all (void)
  * The function cannot be changed after the task has been created. You
  * must create a new #GstTask to change the function.
  *
+ * This function will not yet create and start a thread. Use gst_task_start() or
+ * gst_task_pause() to create and start the GThread.
+ *
+ * Before the task can be used, a #GStaticRecMutex must be configured using the
+ * gst_task_set_lock() function. This lock will always be acquired while
+ * @func is called.
+ *
  * Returns: A new #GstTask.
  *
  * MT safe.
@@ -295,6 +364,159 @@ is_running:
   }
 }
 
+/**
+ * gst_task_set_priority:
+ * @task: a #GstTask
+ * @priority: a new priority for @task
+ *
+ * Changes the priority of @task to @priority.
+ *
+ * Note: try not to depend on task priorities.
+ *
+ * MT safe.
+ *
+ * Since: 0.10.24
+ */
+void
+gst_task_set_priority (GstTask * task, GThreadPriority priority)
+{
+  GstTaskPrivate *priv;
+  GThread *thread;
+
+  g_return_if_fail (GST_IS_TASK (task));
+
+  priv = task->priv;
+
+  GST_OBJECT_LOCK (task);
+  priv->prio_set = TRUE;
+  priv->priority = priority;
+  thread = task->abidata.ABI.thread;
+  if (thread != NULL) {
+    /* if this task already has a thread, we can configure the priority right
+     * away, else we do that when we assign a thread to the task. */
+    g_thread_set_priority (thread, priority);
+  }
+  GST_OBJECT_UNLOCK (task);
+}
+
+/**
+ * gst_task_get_pool:
+ * @task: a #GstTask
+ *
+ * Get the #GstTaskPool that this task will use for its streaming
+ * threads.
+ *
+ * MT safe.
+ *
+ * Returns: the #GstTaskPool used by @task. gst_object_unref()
+ * after usage.
+ *
+ * Since: 0.10.24
+ */
+GstTaskPool *
+gst_task_get_pool (GstTask * task)
+{
+  GstTaskPool *result;
+  GstTaskPrivate *priv;
+
+  g_return_val_if_fail (GST_IS_TASK (task), NULL);
+
+  priv = task->priv;
+
+  GST_OBJECT_LOCK (task);
+  result = gst_object_ref (priv->pool);
+  GST_OBJECT_UNLOCK (task);
+
+  return result;
+}
+
+/**
+ * gst_task_set_pool:
+ * @task: a #GstTask
+ * @pool: a #GstTaskPool
+ *
+ * Set @pool as the new GstTaskPool for @task. Any new streaming threads that
+ * will be created by @task will now use @pool.
+ *
+ * MT safe.
+ *
+ * Since: 0.10.24
+ */
+void
+gst_task_set_pool (GstTask * task, GstTaskPool * pool)
+{
+  GstTaskPool *old;
+  GstTaskPrivate *priv;
+
+  g_return_if_fail (GST_IS_TASK (task));
+  g_return_if_fail (GST_IS_TASK_POOL (pool));
+
+  priv = task->priv;
+
+  GST_OBJECT_LOCK (task);
+  if (priv->pool != pool) {
+    old = priv->pool;
+    priv->pool = gst_object_ref (pool);
+  } else
+    old = NULL;
+  GST_OBJECT_UNLOCK (task);
+
+  if (old)
+    gst_object_unref (old);
+}
+
+
+/**
+ * gst_task_set_thread_callbacks:
+ * @task: The #GstTask to use
+ * @callbacks: a #GstTaskThreadCallbacks pointer
+ * @user_data: user data passed to the callbacks
+ * @notify: called when @user_data is no longer referenced
+ *
+ * Set callbacks which will be executed when a new thread is needed, the thread
+ * function is entered and left and when the thread is joined.
+ *
+ * By default a thread for @task will be created from a default thread pool.
+ *
+ * Objects can use custom GThreads or can perform additional configuration of
+ * the threads (such as changing the thread priority) by installing callbacks.
+ *
+ * Since: 0.10.24
+ *
+ * MT safe.
+ */
+void
+gst_task_set_thread_callbacks (GstTask * task,
+    GstTaskThreadCallbacks * callbacks, gpointer user_data,
+    GDestroyNotify notify)
+{
+  GDestroyNotify old_notify;
+
+  g_return_if_fail (task != NULL);
+  g_return_if_fail (GST_IS_TASK (task));
+  g_return_if_fail (callbacks != NULL);
+
+  GST_OBJECT_LOCK (task);
+  old_notify = task->priv->thr_notify;
+
+  if (old_notify) {
+    gpointer old_data;
+
+    old_data = task->priv->thr_user_data;
+
+    task->priv->thr_user_data = NULL;
+    task->priv->thr_notify = NULL;
+    GST_OBJECT_UNLOCK (task);
+
+    old_notify (old_data);
+
+    GST_OBJECT_LOCK (task);
+  }
+  task->priv->thr_callbacks = *callbacks;
+  task->priv->thr_user_data = user_data;
+  task->priv->thr_notify = notify;
+  GST_OBJECT_UNLOCK (task);
+}
 
 /**
  * gst_task_get_state:
@@ -320,79 +542,127 @@ gst_task_get_state (GstTask * task)
   return result;
 }
 
+/* make sure the task is running and start a thread if it's not.
+ * This function must be called with the task LOCK. */
+static gboolean
+start_task (GstTask * task)
+{
+  gboolean res = TRUE;
+  GstTaskClass *tclass;
+  GstTaskPrivate *priv;
+  GError *error = NULL;
+
+  priv = task->priv;
+
+  /* new task, We ref before so that it remains alive while
+   * the thread is running. */
+  gst_object_ref (task);
+  /* mark task as running so that a join will wait until we schedule
+   * and exit the task function. */
+  task->running = TRUE;
+
+  tclass = GST_TASK_GET_CLASS (task);
+
+  /* push on the thread pool, we remember the original pool because the user
+   * could change it later on and then we join to the wrong pool. */
+  priv->pool_id = gst_object_ref (priv->pool);
+  priv->id =
+      gst_task_pool_push (priv->pool_id, (GstTaskPoolFunction) gst_task_func,
+      task, &error);
+
+  if (error != NULL) {
+    g_warning ("failed to create thread: %s", error->message);
+    g_error_free (error);
+    res = FALSE;
+  }
+  return res;
+}
+
 /**
- * gst_task_start:
- * @task: The #GstTask to start
+ * gst_task_set_state:
+ * @task: a #GstTask
+ * @state: the new task state
  *
- * Starts @task. The @task must have a lock associated with it using
- * gst_task_set_lock() or this function will return %FALSE.
+ * Sets the state of @task to @state.
  *
- * Returns: %TRUE if the task could be started.
+ * The @task must have a lock associated with it using
+ * gst_task_set_lock() when going to GST_TASK_STARTED or GST_TASK_PAUSED or
+ * this function will return %FALSE.
+ *
+ * Returns: %TRUE if the state could be changed.
+ *
+ * Since: 0.10.24
  *
  * MT safe.
  */
 gboolean
-gst_task_start (GstTask * task)
+gst_task_set_state (GstTask * task, GstTaskState state)
 {
+  gboolean res = TRUE;
   GstTaskState old;
 
   g_return_val_if_fail (GST_IS_TASK (task), FALSE);
 
-  GST_DEBUG_OBJECT (task, "Starting task %p", task);
+  GST_DEBUG_OBJECT (task, "Changing task %p to state %d", task, state);
 
   GST_OBJECT_LOCK (task);
-  if (G_UNLIKELY (GST_TASK_GET_LOCK (task) == NULL))
-    goto no_lock;
+  if (state != GST_TASK_STOPPED)
+    if (G_UNLIKELY (GST_TASK_GET_LOCK (task) == NULL))
+      goto no_lock;
 
+  /* if the state changed, do our thing */
   old = task->state;
-  task->state = GST_TASK_STARTED;
-  switch (old) {
-    case GST_TASK_STOPPED:
-    {
-      GstTaskClass *tclass;
-
-      /* If the task already has a thread scheduled we don't have to do
-       * anything. */
-      if (task->running)
+  if (old != state) {
+    task->state = state;
+    switch (old) {
+      case GST_TASK_STOPPED:
+        /* If the task already has a thread scheduled we don't have to do
+         * anything. */
+        if (G_UNLIKELY (!task->running))
+          res = start_task (task);
+        break;
+      case GST_TASK_PAUSED:
+        /* when we are paused, signal to go to the new state */
+        GST_TASK_SIGNAL (task);
+        break;
+      case GST_TASK_STARTED:
+        /* if we were started, we'll go to the new state after the next
+         * iteration. */
         break;
-
-      /* new task, push on thread pool. We ref before so
-       * that it remains alive while on the thread pool. */
-      gst_object_ref (task);
-      /* mark task as running so that a join will wait until we schedule
-       * and exit the task function. */
-      task->running = TRUE;
-
-      tclass = GST_TASK_GET_CLASS (task);
-
-      g_static_mutex_lock (&pool_lock);
-      g_thread_pool_push (tclass->pool, task, NULL);
-      g_static_mutex_unlock (&pool_lock);
-      break;
     }
-    case GST_TASK_PAUSED:
-      /* PAUSE to PLAY, signal */
-      GST_TASK_SIGNAL (task);
-      break;
-    case GST_TASK_STARTED:
-      /* was OK */
-      break;
   }
   GST_OBJECT_UNLOCK (task);
 
-  return TRUE;
+  return res;
 
   /* ERRORS */
 no_lock:
   {
-    GST_WARNING_OBJECT (task, "starting task without a lock");
+    GST_WARNING_OBJECT (task, "state %d set on task without a lock", state);
     GST_OBJECT_UNLOCK (task);
-    g_warning ("starting task without a lock");
+    g_warning ("task without a lock can't be set to state %d", state);
     return FALSE;
   }
 }
 
 /**
+ * gst_task_start:
+ * @task: The #GstTask to start
+ *
+ * Starts @task. The @task must have a lock associated with it using
+ * gst_task_set_lock() or this function will return %FALSE.
+ *
+ * Returns: %TRUE if the task could be started.
+ *
+ * MT safe.
+ */
+gboolean
+gst_task_start (GstTask * task)
+{
+  return gst_task_set_state (task, GST_TASK_STARTED);
+}
+
+/**
  * gst_task_stop:
  * @task: The #GstTask to stop
  *
@@ -407,27 +677,7 @@ no_lock:
 gboolean
 gst_task_stop (GstTask * task)
 {
-  GstTaskState old;
-
-  g_return_val_if_fail (GST_IS_TASK (task), FALSE);
-
-  GST_DEBUG_OBJECT (task, "Stopping task %p", task);
-
-  GST_OBJECT_LOCK (task);
-  old = task->state;
-  task->state = GST_TASK_STOPPED;
-  switch (old) {
-    case GST_TASK_STOPPED:
-      break;
-    case GST_TASK_PAUSED:
-      GST_TASK_SIGNAL (task);
-      break;
-    case GST_TASK_STARTED:
-      break;
-  }
-  GST_OBJECT_UNLOCK (task);
-
-  return TRUE;
+  return gst_task_set_state (task, GST_TASK_STOPPED);
 }
 
 /**
@@ -446,53 +696,7 @@ gst_task_stop (GstTask * task)
 gboolean
 gst_task_pause (GstTask * task)
 {
-  GstTaskState old;
-
-  g_return_val_if_fail (GST_IS_TASK (task), FALSE);
-
-  GST_DEBUG_OBJECT (task, "Pausing task %p", task);
-
-  GST_OBJECT_LOCK (task);
-  if (G_UNLIKELY (GST_TASK_GET_LOCK (task) == NULL))
-    goto no_lock;
-
-  old = task->state;
-  task->state = GST_TASK_PAUSED;
-  switch (old) {
-    case GST_TASK_STOPPED:
-    {
-      GstTaskClass *tclass;
-
-      if (task->running)
-        break;
-
-      gst_object_ref (task);
-      task->running = TRUE;
-
-      tclass = GST_TASK_GET_CLASS (task);
-
-      g_static_mutex_lock (&pool_lock);
-      g_thread_pool_push (tclass->pool, task, NULL);
-      g_static_mutex_unlock (&pool_lock);
-      break;
-    }
-    case GST_TASK_PAUSED:
-      break;
-    case GST_TASK_STARTED:
-      break;
-  }
-  GST_OBJECT_UNLOCK (task);
-
-  return TRUE;
-
-  /* ERRORS */
-no_lock:
-  {
-    GST_WARNING_OBJECT (task, "pausing task without a lock");
-    GST_OBJECT_UNLOCK (task);
-    g_warning ("pausing task without a lock");
-    return FALSE;
-  }
+  return gst_task_set_state (task, GST_TASK_PAUSED);
 }
 
 /**
@@ -515,7 +719,12 @@ no_lock:
 gboolean
 gst_task_join (GstTask * task)
 {
-  GThread *tself;
+  GThread *tself, *thread;
+  GstTaskPrivate *priv;
+  gpointer id;
+  GstTaskPool *pool = NULL;
+
+  priv = task->priv;
 
   g_return_val_if_fail (GST_IS_TASK (task), FALSE);
 
@@ -526,7 +735,8 @@ gst_task_join (GstTask * task)
   /* we don't use a real thread join here because we are using
    * thread pools */
   GST_OBJECT_LOCK (task);
-  if (G_UNLIKELY (tself == task->abidata.ABI.thread))
+  thread = task->abidata.ABI.thread;
+  if (G_UNLIKELY (tself == thread))
     goto joining_self;
   task->state = GST_TASK_STOPPED;
   /* signal the state change for when it was blocked in PAUSED. */
@@ -536,8 +746,22 @@ gst_task_join (GstTask * task)
    * to join it here. */
   while (G_LIKELY (task->running))
     GST_TASK_WAIT (task);
+  /* clean the thread */
+  task->abidata.ABI.thread = NULL;
+  /* get the id and pool to join */
+  if ((id = priv->id)) {
+    if ((pool = priv->pool_id))
+      gst_object_ref (pool);
+    priv->pool_id = NULL;
+    priv->id = NULL;
+  }
   GST_OBJECT_UNLOCK (task);
 
+  if (pool) {
+    gst_task_pool_join (pool, id);
+    gst_object_unref (pool);
+  }
+
   GST_DEBUG_OBJECT (task, "Joined task %p", task);
 
   return TRUE;
diff --git a/gst/gsttask.h b/gst/gsttask.h
index 70c4ab4..b0b7c30 100644
--- a/gst/gsttask.h
+++ b/gst/gsttask.h
@@ -24,6 +24,7 @@
 #define __GST_TASK_H__
 
 #include <gst/gstobject.h>
+#include <gst/gsttaskpool.h>
 
 G_BEGIN_DECLS
 
@@ -47,6 +48,7 @@ typedef void         (*GstTaskFunction)          (void *data);
 
 typedef struct _GstTask GstTask;
 typedef struct _GstTaskClass GstTaskClass;
+typedef struct _GstTaskPrivate GstTaskPrivate;
 
 /**
  * GstTaskState:
@@ -108,6 +110,25 @@ typedef enum {
 #define GST_TASK_GET_LOCK(task)		(GST_TASK_CAST(task)->lock)
 
 /**
+ * GstTaskThreadCallbacks:
+ * @enter_thread: a thread is entered, this callback is called when the new
+ *   thread enters its function.
+ * @leave_thread: a thread is exiting, this is called when the thread is about
+ *   to leave its function
+ *
+ * Custom GstTask thread callback functions that can be installed. 
+ *
+ * Since: 0.10.24
+ */
+typedef struct {
+  /* manage the lifetime of the thread */
+  void      (*enter_thread)     (GstTask *task, GThread *thread, gpointer user_data);
+  void      (*leave_thread)     (GstTask *task, GThread *thread, gpointer user_data);
+  /*< private >*/
+  gpointer     _gst_reserved[GST_PADDING];
+} GstTaskThreadCallbacks;
+
+/**
  * GstTask:
  * @state: the state of the task
  * @cond: used to pause/resume the task
@@ -138,17 +159,17 @@ struct _GstTask {
       /* thread this task is currently running in */
       GThread  *thread;
     } ABI;
-    /* adding + 0 to mark ABI change to be undone later */
-    gpointer _gst_reserved[GST_PADDING + 0];
+    gpointer _gst_reserved[GST_PADDING - 1];
   } abidata;
 
+  GstTaskPrivate *priv;
 };
 
 struct _GstTaskClass {
   GstObjectClass parent_class;
 
   /*< private >*/
-  GThreadPool *pool;
+  GstTaskPool *pool;
 
   /*< private >*/
   gpointer _gst_reserved[GST_PADDING];
@@ -160,8 +181,18 @@ GType           gst_task_get_type       (void);
 
 GstTask*	gst_task_create		(GstTaskFunction func, gpointer data);
 void		gst_task_set_lock	(GstTask *task, GStaticRecMutex *mutex);
+void		gst_task_set_priority	(GstTask *task, GThreadPriority priority);
+
+GstTaskPool *   gst_task_get_pool       (GstTask *task);
+void            gst_task_set_pool       (GstTask *task, GstTaskPool *pool);
+
+void            gst_task_set_thread_callbacks  (GstTask *task,
+                                                GstTaskThreadCallbacks *callbacks,
+						gpointer user_data,
+						GDestroyNotify notify);
 
 GstTaskState	gst_task_get_state	(GstTask *task);
+gboolean        gst_task_set_state      (GstTask *task, GstTaskState state);
 
 gboolean	gst_task_start		(GstTask *task);
 gboolean	gst_task_stop		(GstTask *task);
diff --git a/gst/gsttaskpool.c b/gst/gsttaskpool.c
new file mode 100644
index 0000000..e17af0b
--- /dev/null
+++ b/gst/gsttaskpool.c
@@ -0,0 +1,271 @@
+/* GStreamer
+ * Copyright (C) 2009 Wim Taymans <wim.taymans@gmail.com>
+ *
+ * gsttaskpool.c: Pool for streaming threads
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * You should have received a copy of the GNU Library General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
+ * Boston, MA 02111-1307, USA.
+ */
+
+/**
+ * SECTION:gsttaskpool
+ * @short_description: Pool of GStreamer streaming threads
+ * @see_also: #GstTask, #GstPad
+ *
+ * This object provides an abstraction for creating threads. The default
+ * implementation uses a regular GThreadPool to start tasks.
+ *
+ * Subclasses can be made to create custom threads.
+ *
+ * Last reviewed on 2009-04-23 (0.10.24)
+ */
+
+#include "gst_private.h"
+
+#include "gstinfo.h"
+#include "gsttaskpool.h"
+
+GST_DEBUG_CATEGORY_STATIC (taskpool_debug);
+#define GST_CAT_DEFAULT (taskpool_debug)
+
+static void gst_task_pool_class_init (GstTaskPoolClass * klass);
+static void gst_task_pool_init (GstTaskPool * pool);
+static void gst_task_pool_finalize (GObject * object);
+
+#define _do_init \
+{ \
+  GST_DEBUG_CATEGORY_INIT (taskpool_debug, "taskpool", 0, "Thread pool"); \
+}
+
+G_DEFINE_TYPE_WITH_CODE (GstTaskPool, gst_task_pool, GST_TYPE_OBJECT, _do_init);
+
+typedef struct
+{
+  GstTaskPoolFunction func;
+  gpointer user_data;
+} TaskData;
+
+static void
+default_func (TaskData * tdata, GstTaskPool * pool)
+{
+  GstTaskPoolFunction func;
+  gpointer user_data;
+
+  func = tdata->func;
+  user_data = tdata->user_data;
+  g_slice_free (TaskData, tdata);
+
+  func (user_data);
+}
+
+static void
+default_prepare (GstTaskPool * pool, GError ** error)
+{
+  GST_OBJECT_LOCK (pool);
+  pool->pool = g_thread_pool_new ((GFunc) default_func, pool, -1, FALSE, NULL);
+  GST_OBJECT_UNLOCK (pool);
+}
+
+static void
+default_cleanup (GstTaskPool * pool)
+{
+  GST_OBJECT_LOCK (pool);
+  if (pool->pool) {
+    /* Shut down all the threads, we still process the ones scheduled
+     * because the unref happens in the thread function.
+     * Also wait for currently running ones to finish. */
+    g_thread_pool_free (pool->pool, FALSE, TRUE);
+    pool->pool = NULL;
+  }
+  GST_OBJECT_UNLOCK (pool);
+}
+
+static gpointer
+default_push (GstTaskPool * pool, GstTaskPoolFunction func,
+    gpointer user_data, GError ** error)
+{
+  TaskData *tdata;
+
+  tdata = g_slice_new (TaskData);
+  tdata->func = func;
+  tdata->user_data = user_data;
+
+  GST_OBJECT_LOCK (pool);
+  if (pool->pool)
+    g_thread_pool_push (pool->pool, tdata, error);
+  else {
+    g_slice_free (TaskData, tdata);
+  }
+  GST_OBJECT_UNLOCK (pool);
+
+  return NULL;
+}
+
+static void
+default_join (GstTaskPool * pool, gpointer id)
+{
+  /* we do nothing here, we can't join from the pools */
+}
+
+static void
+gst_task_pool_class_init (GstTaskPoolClass * klass)
+{
+  GObjectClass *gobject_class;
+  GstTaskPoolClass *gsttaskpool_class;
+
+  gobject_class = (GObjectClass *) klass;
+  gsttaskpool_class = (GstTaskPoolClass *) klass;
+
+  gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_task_pool_finalize);
+
+  gsttaskpool_class->prepare = default_prepare;
+  gsttaskpool_class->cleanup = default_cleanup;
+  gsttaskpool_class->push = default_push;
+  gsttaskpool_class->join = default_join;
+}
+
+static void
+gst_task_pool_init (GstTaskPool * pool)
+{
+}
+
+static void
+gst_task_pool_finalize (GObject * object)
+{
+  GstTaskPool *pool = GST_TASK_POOL (object);
+
+  GST_DEBUG ("taskpool %p finalize", pool);
+
+  G_OBJECT_CLASS (gst_task_pool_parent_class)->finalize (object);
+}
+
+/**
+ * gst_task_pool_new:
+ *
+ * Create a new default task pool. The default task pool will use a regular
+ * GThreadPool for threads.
+ *
+ * Returns: a new #GstTaskPool. gst_object_unref() after usage.
+ */
+GstTaskPool *
+gst_task_pool_new (void)
+{
+  GstTaskPool *pool;
+
+  pool = g_object_new (GST_TYPE_TASK_POOL, NULL);
+
+  return pool;
+}
+
+/**
+ * gst_task_pool_prepare:
+ * @pool: a #GstTaskPool
+ * @error: an error return location
+ *
+ * Prepare the taskpool for accepting gst_task_pool_push() operations.
+ *
+ * MT safe.
+ */
+void
+gst_task_pool_prepare (GstTaskPool * pool, GError ** error)
+{
+  GstTaskPoolClass *klass;
+
+  g_return_if_fail (GST_IS_TASK_POOL (pool));
+
+  klass = GST_TASK_POOL_GET_CLASS (pool);
+
+  if (klass->prepare)
+    klass->prepare (pool, error);
+}
+
+/**
+ * gst_task_pool_cleanup:
+ * @pool: a #GstTaskPool
+ *
+ * Wait for all tasks to be stopped. This is mainly used internally
+ * to ensure proper cleanup of internal data structures in test suites.
+ *
+ * MT safe.
+ */
+void
+gst_task_pool_cleanup (GstTaskPool * pool)
+{
+  GstTaskPoolClass *klass;
+
+  g_return_if_fail (GST_IS_TASK_POOL (pool));
+
+  klass = GST_TASK_POOL_GET_CLASS (pool);
+
+  if (klass->cleanup)
+    klass->cleanup (pool);
+}
+
+/**
+ * gst_task_pool_push:
+ * @pool: a #GstTaskPool
+ * @func: the function to call
+ * @user_data: data to pass to @func
+ * @error: return location for an error
+ *
+ * Start the execution of a new thread from @pool.
+ *
+ * Returns: a pointer that should be used for the gst_task_pool_join
+ * function. This pointer can be NULL, you must check @error to detect
+ * errors.
+ */
+gpointer
+gst_task_pool_push (GstTaskPool * pool, GstTaskPoolFunction func,
+    gpointer user_data, GError ** error)
+{
+  GstTaskPoolClass *klass;
+
+  g_return_val_if_fail (GST_IS_TASK_POOL (pool), NULL);
+
+  klass = GST_TASK_POOL_GET_CLASS (pool);
+
+  if (klass->push == NULL)
+    goto not_supported;
+
+  return klass->push (pool, func, user_data, error);
+
+  /* ERRORS */
+not_supported:
+  {
+    g_warning ("pushing tasks on pool %p is not supported", pool);
+    return NULL;
+  }
+}
+
+/**
+ * gst_task_pool_join:
+ * @pool: a #GstTaskPool
+ * @id: the id
+ *
+ * Join a task and/or return it to the pool. @id is the id obtained from 
+ * gst_task_pool_push().
+ */
+void
+gst_task_pool_join (GstTaskPool * pool, gpointer id)
+{
+  GstTaskPoolClass *klass;
+
+  g_return_if_fail (GST_IS_TASK_POOL (pool));
+
+  klass = GST_TASK_POOL_GET_CLASS (pool);
+
+  if (klass->join)
+    klass->join (pool, id);
+}
diff --git a/gst/gsttaskpool.h b/gst/gsttaskpool.h
new file mode 100644
index 0000000..eda5c6c
--- /dev/null
+++ b/gst/gsttaskpool.h
@@ -0,0 +1,95 @@
+/* GStreamer
+ * Copyright (C) <2009> Wim Taymans <wim.taymans@gmail.com>
+ *
+ * gsttaskpool.h: Pool for creating streaming threads
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * You should have received a copy of the GNU Library General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
+ * Boston, MA 02111-1307, USA.
+ */
+
+#ifndef __GST_TASK_POOL_H__
+#define __GST_TASK_POOL_H__
+
+#include <gst/gstobject.h>
+
+G_BEGIN_DECLS
+
+/* --- standard type macros --- */
+#define GST_TYPE_TASK_POOL             (gst_task_pool_get_type ())
+#define GST_TASK_POOL(pool)            (G_TYPE_CHECK_INSTANCE_CAST ((pool), GST_TYPE_TASK_POOL, GstTaskPool))
+#define GST_IS_TASK_POOL(pool)         (G_TYPE_CHECK_INSTANCE_TYPE ((pool), GST_TYPE_TASK_POOL))
+#define GST_TASK_POOL_CLASS(pclass)    (G_TYPE_CHECK_CLASS_CAST ((pclass), GST_TYPE_TASK_POOL, GstTaskPoolClass))
+#define GST_IS_TASK_POOL_CLASS(pclass) (G_TYPE_CHECK_CLASS_TYPE ((pclass), GST_TYPE_TASK_POOL))
+#define GST_TASK_POOL_GET_CLASS(pool)  (G_TYPE_INSTANCE_GET_CLASS ((pool), GST_TYPE_TASK_POOL, GstTaskPoolClass))
+#define GST_TASK_POOL_CAST(pool)       ((GstTaskPool*)(pool))
+
+typedef struct _GstTaskPool GstTaskPool;
+typedef struct _GstTaskPoolClass GstTaskPoolClass;
+
+typedef void   (*GstTaskPoolFunction)          (void *data);
+
+/**
+ * GstTaskPool:
+ *
+ * The #GstTaskPool object.
+ */
+struct _GstTaskPool {
+  GstObject      object;
+
+  /*< private >*/
+  GThreadPool   *pool;
+
+  gpointer _gst_reserved[GST_PADDING];
+};
+
+/**
+ * GstTaskPoolClass:
+ * @parent_class: the parent class structure
+ * @prepare: prepare the threadpool
+ * @cleanup: make sure all threads are stopped
+ * @push: start a new thread
+ * @join: join a thread
+ *
+ * The #GstTaskPoolClass object.
+ */
+struct _GstTaskPoolClass {
+  GstObjectClass parent_class;
+
+  /*< public >*/
+  void      (*prepare)  (GstTaskPool *pool, GError **error);
+  void      (*cleanup)  (GstTaskPool *pool);
+
+  gpointer  (*push)     (GstTaskPool *pool, GstTaskPoolFunction func,
+                         gpointer user_data, GError **error);
+  void      (*join)     (GstTaskPool *pool, gpointer id);
+
+  /*< private >*/
+  gpointer _gst_reserved[GST_PADDING];
+};
+
+GType           gst_task_pool_get_type    (void);
+
+GstTaskPool *   gst_task_pool_new         (void);
+void            gst_task_pool_prepare     (GstTaskPool *pool, GError **error);
+
+gpointer        gst_task_pool_push        (GstTaskPool *pool, GstTaskPoolFunction func, 
+                                           gpointer user_data, GError **error);
+void            gst_task_pool_join        (GstTaskPool *pool, gpointer id);
+
+void		gst_task_pool_cleanup     (GstTaskPool *pool);
+
+G_END_DECLS
+
+#endif /* __GST_TASK_POOL_H__ */
diff --git a/libs/gst/base/gstbasesrc.c b/libs/gst/base/gstbasesrc.c
index 6fc4191..16b68ca 100644
--- a/libs/gst/base/gstbasesrc.c
+++ b/libs/gst/base/gstbasesrc.c
@@ -1147,7 +1147,7 @@ gst_base_src_prepare_seek_segment (GstBaseSrc * src, GstEvent * event,
 static gboolean
 gst_base_src_perform_seek (GstBaseSrc * src, GstEvent * event, gboolean unlock)
 {
-  gboolean res = TRUE;
+  gboolean res = TRUE, tres;
   gdouble rate;
   GstFormat seek_format, dest_format;
   GstSeekFlags flags;
@@ -1334,8 +1334,10 @@ gst_base_src_perform_seek (GstBaseSrc * src, GstEvent * event, gboolean unlock)
   src->data.ABI.running = TRUE;
   /* and restart the task in case it got paused explicitely or by
    * the FLUSH_START event we pushed out. */
-  gst_pad_start_task (src->srcpad, (GstTaskFunction) gst_base_src_loop,
+  tres = gst_pad_start_task (src->srcpad, (GstTaskFunction) gst_base_src_loop,
       src->srcpad);
+  if (res && !tres)
+    res = FALSE;
 
   /* and release the lock again so we can continue streaming */
   GST_PAD_STREAM_UNLOCK (src->srcpad);
diff --git a/tests/check/gst/gstbin.c b/tests/check/gst/gstbin.c
index 26558fb..d466a28 100644
--- a/tests/check/gst/gstbin.c
+++ b/tests/check/gst/gstbin.c
@@ -300,8 +300,9 @@ GST_START_TEST (test_message_state_changed_children)
   /* each object is referenced by a message;
    * base_src is blocked in the push and has an extra refcount.
    * base_sink_chain has taken a refcount on the sink, and is blocked on
-   * preroll */
-  ASSERT_OBJECT_REFCOUNT (src, "src", 2);
+   * preroll
+   * The stream-status messages holds 2 more refs to the element */
+  ASSERT_OBJECT_REFCOUNT (src, "src", 4);
   /* refcount can be 4 if the bin is still processing the async_done message of
    * the sink. */
   ASSERT_OBJECT_REFCOUNT_BETWEEN (sink, "sink", 2, 3);
@@ -352,8 +353,9 @@ GST_START_TEST (test_message_state_changed_children)
   ret = gst_element_set_state (GST_ELEMENT (pipeline), GST_STATE_READY);
   fail_unless (ret == GST_STATE_CHANGE_SUCCESS);
 
-  /* each object is referenced by two messages */
-  ASSERT_OBJECT_REFCOUNT (src, "src", 3);
+  /* each object is referenced by two messages, the source also has the
+   * stream-status message referencing it */
+  ASSERT_OBJECT_REFCOUNT (src, "src", 4);
   ASSERT_OBJECT_REFCOUNT (sink, "sink", 3);
   ASSERT_OBJECT_REFCOUNT (pipeline, "pipeline", 3);
 
diff --git a/tests/check/gst/gstmessage.c b/tests/check/gst/gstmessage.c
index 39bae40..dc9a571 100644
--- a/tests/check/gst/gstmessage.c
+++ b/tests/check/gst/gstmessage.c
@@ -186,6 +186,60 @@ GST_START_TEST (test_parsing)
      void            gst_message_parse_warning       (GstMessage *message, GError **gerror, gchar **debug);
    */
 
+  /* GST_MESSAGE_STREAM_STATUS   */
+  {
+    GstStreamStatusType type;
+    GstTask *task, *task2;
+    GValue value = { 0 };
+    const GValue *val;
+
+    message =
+        gst_message_new_stream_status (NULL, GST_STREAM_STATUS_TYPE_ENTER,
+        NULL);
+    fail_if (message == NULL);
+    fail_unless (GST_MESSAGE_TYPE (message) == GST_MESSAGE_STREAM_STATUS);
+    fail_unless (GST_MESSAGE_SRC (message) == NULL);
+
+    /* set some wrong values to check if the parse method overwrites them
+     * with the good values */
+    type = GST_STREAM_STATUS_TYPE_START;
+    gst_message_parse_stream_status (message, &type, NULL);
+    fail_unless (type == GST_STREAM_STATUS_TYPE_ENTER);
+
+    /* create a task with some dummy function, we're not actually going to run
+     * the task here */
+    task = gst_task_create ((GstTaskFunction) gst_object_unref, NULL);
+
+    ASSERT_OBJECT_REFCOUNT (task, "task", 1);
+
+    /* set the task */
+    g_value_init (&value, GST_TYPE_TASK);
+    g_value_set_object (&value, task);
+
+    ASSERT_OBJECT_REFCOUNT (task, "task", 2);
+
+    gst_message_set_stream_status_object (message, &value);
+    ASSERT_OBJECT_REFCOUNT (task, "task", 3);
+    g_value_unset (&value);
+    ASSERT_OBJECT_REFCOUNT (task, "task", 2);
+    gst_object_unref (task);
+    ASSERT_OBJECT_REFCOUNT (task, "task", 1);
+
+    /* get the object back, no refcount is changed */
+    val = gst_message_get_stream_status_object (message);
+    ASSERT_OBJECT_REFCOUNT (task, "task", 1);
+
+    task2 = g_value_get_object (val);
+
+    fail_unless (GST_IS_TASK (task2));
+    fail_unless (task2 == task);
+
+    ASSERT_OBJECT_REFCOUNT (task, "task", 1);
+    ASSERT_OBJECT_REFCOUNT (task2, "task", 1);
+
+    gst_message_unref (message);
+  }
+
   /* GST_MESSAGE_REQUEST_STATE   */
   {
     GstState state;
diff --git a/tests/check/pipelines/cleanup.c b/tests/check/pipelines/cleanup.c
index 1446888..473adde 100644
--- a/tests/check/pipelines/cleanup.c
+++ b/tests/check/pipelines/cleanup.c
@@ -91,7 +91,7 @@ GST_START_TEST (test_pipeline_unref)
 
   run_pipeline (pipeline, s,
       GST_MESSAGE_NEW_CLOCK | GST_MESSAGE_STATE_CHANGED |
-      GST_MESSAGE_ASYNC_DONE, GST_MESSAGE_EOS);
+      GST_MESSAGE_STREAM_STATUS | GST_MESSAGE_ASYNC_DONE, GST_MESSAGE_EOS);
   while (GST_OBJECT_REFCOUNT_VALUE (src) > 1)
     THREAD_SWITCH ();
   ASSERT_OBJECT_REFCOUNT (src, "src", 1);
diff --git a/tests/check/pipelines/simple-launch-lines.c b/tests/check/pipelines/simple-launch-lines.c
index 167e049..8575ad8 100644
--- a/tests/check/pipelines/simple-launch-lines.c
+++ b/tests/check/pipelines/simple-launch-lines.c
@@ -102,27 +102,28 @@ GST_START_TEST (test_2_elements)
   s = "fakesrc can-activate-push=false ! fakesink can-activate-pull=true";
   run_pipeline (setup_pipeline (s), s,
       GST_MESSAGE_NEW_CLOCK | GST_MESSAGE_STATE_CHANGED |
-      GST_MESSAGE_ASYNC_DONE, GST_MESSAGE_UNKNOWN);
+      GST_MESSAGE_STREAM_STATUS | GST_MESSAGE_ASYNC_DONE, GST_MESSAGE_UNKNOWN);
 
   s = "fakesrc can-activate-push=true ! fakesink can-activate-pull=false";
   run_pipeline (setup_pipeline (s), s,
       GST_MESSAGE_NEW_CLOCK | GST_MESSAGE_STATE_CHANGED |
-      GST_MESSAGE_ASYNC_DONE, GST_MESSAGE_UNKNOWN);
+      GST_MESSAGE_STREAM_STATUS | GST_MESSAGE_ASYNC_DONE, GST_MESSAGE_UNKNOWN);
 
   s = "fakesrc can-activate-push=false num-buffers=10 ! fakesink can-activate-pull=true";
   run_pipeline (setup_pipeline (s), s,
       GST_MESSAGE_NEW_CLOCK | GST_MESSAGE_STATE_CHANGED |
-      GST_MESSAGE_ASYNC_DONE, GST_MESSAGE_EOS);
+      GST_MESSAGE_STREAM_STATUS | GST_MESSAGE_ASYNC_DONE, GST_MESSAGE_EOS);
 
   s = "fakesrc can-activate-push=true num-buffers=10 ! fakesink can-activate-pull=false";
   run_pipeline (setup_pipeline (s), s,
       GST_MESSAGE_NEW_CLOCK | GST_MESSAGE_STATE_CHANGED |
-      GST_MESSAGE_ASYNC_DONE, GST_MESSAGE_EOS);
+      GST_MESSAGE_STREAM_STATUS | GST_MESSAGE_ASYNC_DONE, GST_MESSAGE_EOS);
 
   s = "fakesrc can-activate-push=false ! fakesink can-activate-pull=false";
   ASSERT_CRITICAL (run_pipeline (setup_pipeline (s), s,
           GST_MESSAGE_NEW_CLOCK | GST_MESSAGE_STATE_CHANGED |
-          GST_MESSAGE_ASYNC_DONE, GST_MESSAGE_UNKNOWN));
+          GST_MESSAGE_STREAM_STATUS | GST_MESSAGE_ASYNC_DONE,
+          GST_MESSAGE_UNKNOWN));
 }
 
 GST_END_TEST;
diff --git a/tests/examples/Makefile.am b/tests/examples/Makefile.am
index e688d8e..87c13fe 100644
--- a/tests/examples/Makefile.am
+++ b/tests/examples/Makefile.am
@@ -22,6 +22,7 @@ always_dirs = \
 	helloworld \
 	manual     \
 	metadata   \
+	streams    \
 	queue
 
 #appreader
diff --git a/tests/examples/streams/.gitignore b/tests/examples/streams/.gitignore
new file mode 100644
index 0000000..d8cfe30
--- /dev/null
+++ b/tests/examples/streams/.gitignore
@@ -0,0 +1,6 @@
+stream-status
+rtpool-test
+*.bb
+*.bbg
+*.da
+stream-status-stream-status.gcno
diff --git a/tests/examples/streams/Makefile.am b/tests/examples/streams/Makefile.am
new file mode 100644
index 0000000..c840996
--- /dev/null
+++ b/tests/examples/streams/Makefile.am
@@ -0,0 +1,9 @@
+noinst_PROGRAMS = stream-status rtpool-test
+
+stream_status_SOURCES = stream-status.c
+stream_status_LDADD = $(GST_OBJ_LIBS)
+stream_status_CFLAGS = $(GST_OBJ_CFLAGS)
+
+rtpool_test_SOURCES = rtpool-test.c testrtpool.c
+rtpool_test_LDADD = $(GST_OBJ_LIBS)
+rtpool_test_CFLAGS = $(GST_OBJ_CFLAGS)
diff --git a/tests/examples/streams/rtpool-test.c b/tests/examples/streams/rtpool-test.c
new file mode 100644
index 0000000..6ef4b3a
--- /dev/null
+++ b/tests/examples/streams/rtpool-test.c
@@ -0,0 +1,159 @@
+#include <stdlib.h>
+#include <gst/gst.h>
+
+#include "testrtpool.h"
+
+static GstTaskPool *pool;
+
+static void
+event_loop (GstBus * bus, GstElement * pipe)
+{
+  GstMessage *message = NULL;
+
+  while (TRUE) {
+    message = gst_bus_poll (bus, GST_MESSAGE_ANY, -1);
+
+    g_assert (message != NULL);
+
+    switch (message->type) {
+      case GST_MESSAGE_EOS:
+        g_message ("received EOS");
+        gst_message_unref (message);
+        return;
+      case GST_MESSAGE_WARNING:
+      {
+        GError *gerror;
+        gchar *debug;
+
+        gst_message_parse_warning (message, &gerror, &debug);
+        gst_object_default_error (GST_MESSAGE_SRC (message), gerror, debug);
+        gst_message_unref (message);
+        g_error_free (gerror);
+        g_free (debug);
+        return;
+      }
+      case GST_MESSAGE_ERROR:{
+        GError *gerror;
+        gchar *debug;
+
+        gst_message_parse_error (message, &gerror, &debug);
+        gst_object_default_error (GST_MESSAGE_SRC (message), gerror, debug);
+        gst_message_unref (message);
+        g_error_free (gerror);
+        g_free (debug);
+        return;
+      }
+      default:
+        gst_message_unref (message);
+        break;
+    }
+  }
+}
+
+static GstBusSyncReply
+sync_bus_handler (GstBus * bus, GstMessage * message, GstElement * bin)
+{
+  switch (GST_MESSAGE_TYPE (message)) {
+    case GST_MESSAGE_STREAM_STATUS:
+    {
+      GstStreamStatusType type;
+      GstElement *owner;
+      const GValue *val;
+      gchar *path;
+      GstTask *task = NULL;
+
+      g_message ("received STREAM_STATUS");
+      gst_message_parse_stream_status (message, &type, &owner);
+
+      val = gst_message_get_stream_status_object (message);
+
+      g_message ("type:   %d", type);
+      path = gst_object_get_path_string (GST_MESSAGE_SRC (message));
+      g_message ("source: %s", path);
+      g_free (path);
+      path = gst_object_get_path_string (GST_OBJECT (owner));
+      g_message ("owner:  %s", path);
+      g_free (path);
+      g_message ("object: type %s, value %p", G_VALUE_TYPE_NAME (val),
+          g_value_get_object (val));
+
+      /* see if we know how to deal with this object */
+      if (G_VALUE_TYPE (val) == GST_TYPE_TASK) {
+        task = g_value_get_object (val);
+      }
+
+      switch (type) {
+        case GST_STREAM_STATUS_TYPE_CREATE:
+          if (task) {
+            g_message ("created task %p, setting pool", task);
+            gst_task_set_pool (task, pool);
+          }
+          break;
+        case GST_STREAM_STATUS_TYPE_ENTER:
+          break;
+        case GST_STREAM_STATUS_TYPE_LEAVE:
+          break;
+        default:
+          break;
+      }
+      break;
+    }
+    default:
+      break;
+  }
+  /* pass all messages on the async queue */
+  return GST_BUS_PASS;
+}
+
+int
+main (int argc, char *argv[])
+{
+  GstElement *bin, *alsasrc, *alsasink;
+  GstBus *bus;
+  GstStateChangeReturn ret;
+
+  gst_init (&argc, &argv);
+
+  /* create a custom thread pool */
+  pool = test_rt_pool_new ();
+
+  /* create a new bin to hold the elements */
+  bin = gst_pipeline_new ("pipeline");
+  g_assert (bin);
+
+  /* create a source */
+  alsasrc = gst_element_factory_make ("alsasrc", "alsasrc");
+  g_assert (alsasrc);
+  g_object_set (alsasrc, "device", "hw:0", NULL);
+  g_object_set (alsasrc, "slave-method", 2, NULL);
+
+  /* and a sink */
+  alsasink = gst_element_factory_make ("alsasink", "alsasink");
+  g_assert (alsasink);
+  g_object_set (alsasink, "device", "hw:0", NULL);
+  g_object_set (alsasink, "buffer-time", (gint64) 10000, NULL);
+
+  /* add objects to the main pipeline */
+  gst_bin_add_many (GST_BIN (bin), alsasrc, alsasink, NULL);
+
+  /* link the elements */
+  gst_element_link (alsasrc, alsasink);
+
+  /* get the bus, we need to install a sync handler */
+  bus = gst_pipeline_get_bus (GST_PIPELINE (bin));
+  gst_bus_set_sync_handler (bus, (GstBusSyncHandler) sync_bus_handler, bin);
+
+  /* start playing */
+  ret = gst_element_set_state (bin, GST_STATE_PLAYING);
+  if (ret == GST_STATE_CHANGE_FAILURE)
+    return 0;
+
+  /* Run event loop listening for bus messages until EOS or ERROR */
+  event_loop (bus, bin);
+
+  /* stop the bin */
+  gst_element_set_state (bin, GST_STATE_NULL);
+  gst_object_unref (bus);
+
+  exit (0);
+}
diff --git a/tests/examples/streams/stream-status.c b/tests/examples/streams/stream-status.c
new file mode 100644
index 0000000..e8ebb67
--- /dev/null
+++ b/tests/examples/streams/stream-status.c
@@ -0,0 +1,136 @@
+#include <stdlib.h>
+#include <gst/gst.h>
+
+static void
+event_loop (GstBus * bus, GstElement * pipe)
+{
+  GstMessage *message = NULL;
+
+  while (TRUE) {
+    message = gst_bus_poll (bus, GST_MESSAGE_ANY, -1);
+
+    g_assert (message != NULL);
+
+    switch (message->type) {
+      case GST_MESSAGE_EOS:
+        g_message ("received EOS");
+        gst_message_unref (message);
+        return;
+      case GST_MESSAGE_WARNING:
+      case GST_MESSAGE_ERROR:{
+        GError *gerror;
+        gchar *debug;
+
+        gst_message_parse_error (message, &gerror, &debug);
+        gst_object_default_error (GST_MESSAGE_SRC (message), gerror, debug);
+        gst_message_unref (message);
+        g_error_free (gerror);
+        g_free (debug);
+        return;
+      }
+      default:
+        gst_message_unref (message);
+        break;
+    }
+  }
+}
+
+static GstBusSyncReply
+sync_bus_handler (GstBus * bus, GstMessage * message, GstElement * bin)
+{
+  switch (GST_MESSAGE_TYPE (message)) {
+    case GST_MESSAGE_STREAM_STATUS:
+    {
+      GstStreamStatusType type;
+      GstElement *owner;
+      const GValue *val;
+      gchar *path;
+      GstTask *task = NULL;
+
+      g_message ("received STREAM_STATUS");
+      gst_message_parse_stream_status (message, &type, &owner);
+
+      val = gst_message_get_stream_status_object (message);
+
+      g_message ("type:   %d", type);
+      path = gst_object_get_path_string (GST_MESSAGE_SRC (message));
+      g_message ("source: %s", path);
+      g_free (path);
+      path = gst_object_get_path_string (GST_OBJECT (owner));
+      g_message ("owner:  %s", path);
+      g_free (path);
+      g_message ("object: type %s, value %p", G_VALUE_TYPE_NAME (val),
+          g_value_get_object (val));
+
+      /* see if we know how to deal with this object */
+      if (G_VALUE_TYPE (val) == GST_TYPE_TASK) {
+        task = g_value_get_object (val);
+      }
+
+      switch (type) {
+        case GST_STREAM_STATUS_TYPE_CREATE:
+          g_message ("created task %p", task);
+          break;
+        case GST_STREAM_STATUS_TYPE_ENTER:
+          if (task) {
+            g_message ("raising task priority for %p", task);
+            gst_task_set_priority (task, G_THREAD_PRIORITY_HIGH);
+          }
+          break;
+        case GST_STREAM_STATUS_TYPE_LEAVE:
+          break;
+        default:
+          break;
+      }
+      break;
+    }
+    default:
+      break;
+  }
+  /* pass all messages on the async queue */
+  return GST_BUS_PASS;
+}
+
+int
+main (int argc, char *argv[])
+{
+  GstElement *bin, *fakesrc, *fakesink;
+  GstBus *bus;
+
+  gst_init (&argc, &argv);
+
+  /* create a new bin to hold the elements */
+  bin = gst_pipeline_new ("pipeline");
+  g_assert (bin);
+
+  /* create a source */
+  fakesrc = gst_element_factory_make ("fakesrc", "fakesrc");
+  g_assert (fakesrc);
+  g_object_set (fakesrc, "num-buffers", 50, NULL);
+
+  /* and a sink */
+  fakesink = gst_element_factory_make ("fakesink", "fakesink");
+  g_assert (fakesink);
+
+  /* add objects to the main pipeline */
+  gst_bin_add_many (GST_BIN (bin), fakesrc, fakesink, NULL);
+
+  /* link the elements */
+  gst_element_link (fakesrc, fakesink);
+
+  /* get the bus, we need to install a sync handler */
+  bus = gst_pipeline_get_bus (GST_PIPELINE (bin));
+  gst_bus_set_sync_handler (bus, (GstBusSyncHandler) sync_bus_handler, bin);
+
+  /* start playing */
+  gst_element_set_state (bin, GST_STATE_PLAYING);
+
+  /* Run event loop listening for bus messages until EOS or ERROR */
+  event_loop (bus, bin);
+
+  /* stop the bin */
+  gst_element_set_state (bin, GST_STATE_NULL);
+  gst_object_unref (bus);
+
+  exit (0);
+}
diff --git a/tests/examples/streams/testrtpool.c b/tests/examples/streams/testrtpool.c
new file mode 100644
index 0000000..995a2f5
--- /dev/null
+++ b/tests/examples/streams/testrtpool.c
@@ -0,0 +1,137 @@
+/* GStreamer
+ * Copyright (C) 2009 Wim Taymans <wim.taymans@gmail.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * You should have received a copy of the GNU Library General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
+ * Boston, MA 02111-1307, USA.
+ */
+
+#include <pthread.h>
+
+#include "testrtpool.h"
+
+static void test_rt_pool_class_init (TestRTPoolClass * klass);
+static void test_rt_pool_init (TestRTPool * pool);
+static void test_rt_pool_finalize (GObject * object);
+
+typedef struct
+{
+  pthread_t thread;
+} TestRTId;
+
+G_DEFINE_TYPE (TestRTPool, test_rt_pool, GST_TYPE_TASK_POOL);
+
+static void
+default_prepare (GstTaskPool * pool, GError ** error)
+{
+  /* we don't do anything here. We could construct a pool of threads here that
+   * we could reuse later but we don't */
+  g_message ("prepare Realtime pool %p", pool);
+}
+
+static void
+default_cleanup (GstTaskPool * pool)
+{
+  g_message ("cleanup Realtime pool %p", pool);
+}
+
+static gpointer
+default_push (GstTaskPool * pool, GstTaskPoolFunction func, gpointer data,
+    GError ** error)
+{
+  TestRTId *tid;
+  gint res;
+  pthread_attr_t attr;
+  struct sched_param param;
+
+  g_message ("pushing Realtime pool %p, %p", pool, func);
+
+  tid = g_slice_new0 (TestRTId);
+
+  g_message ("set policy");
+  pthread_attr_init (&attr);
+  if ((res = pthread_attr_setschedpolicy (&attr, SCHED_RR)) != 0)
+    g_warning ("setschedpolicy: failure: %p", g_strerror (res));
+
+  g_message ("set prio");
+  param.sched_priority = 50;
+  if ((res = pthread_attr_setschedparam (&attr, &param)) != 0)
+    g_warning ("setschedparam: failure: %p", g_strerror (res));
+
+  g_message ("set inherit");
+  if ((res = pthread_attr_setinheritsched (&attr, PTHREAD_EXPLICIT_SCHED)) != 0)
+    g_warning ("setinheritsched: failure: %p", g_strerror (res));
+
+  g_message ("create thread");
+  res = pthread_create (&tid->thread, &attr, (void *(*)(void *)) func, data);
+
+  if (res != 0) {
+    g_set_error (error, G_THREAD_ERROR, G_THREAD_ERROR_AGAIN,
+        "Error creating thread: %s", g_strerror (res));
+    g_slice_free (TestRTId, tid);
+    tid = NULL;
+  }
+
+  return tid;
+}
+
+static void
+default_join (GstTaskPool * pool, gpointer id)
+{
+  TestRTId *tid = (TestRTId *) id;
+
+  g_message ("joining Realtime pool %p", pool);
+
+  pthread_join (tid->thread, NULL);
+
+  g_slice_free (TestRTId, tid);
+}
+
+static void
+test_rt_pool_class_init (TestRTPoolClass * klass)
+{
+  GObjectClass *gobject_class;
+  GstTaskPoolClass *gsttaskpool_class;
+
+  gobject_class = (GObjectClass *) klass;
+  gsttaskpool_class = (GstTaskPoolClass *) klass;
+
+  gobject_class->finalize = GST_DEBUG_FUNCPTR (test_rt_pool_finalize);
+
+  gsttaskpool_class->prepare = default_prepare;
+  gsttaskpool_class->cleanup = default_cleanup;
+  gsttaskpool_class->push = default_push;
+  gsttaskpool_class->join = default_join;
+}
+
+static void
+test_rt_pool_init (TestRTPool * pool)
+{
+}
+
+static void
+test_rt_pool_finalize (GObject * object)
+{
+  G_OBJECT_CLASS (test_rt_pool_parent_class)->finalize (object);
+}
+
+GstTaskPool *
+test_rt_pool_new (void)
+{
+  GstTaskPool *pool;
+
+  pool = g_object_new (TEST_TYPE_RT_POOL, NULL);
+
+  return pool;
+}
diff --git a/tests/examples/streams/testrtpool.h b/tests/examples/streams/testrtpool.h
new file mode 100644
index 0000000..580cffa
--- /dev/null
+++ b/tests/examples/streams/testrtpool.h
@@ -0,0 +1,53 @@
+/* GStreamer
+ * Copyright (C) <2009> Wim Taymans <wim.taymans@gmail.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * You should have received a copy of the GNU Library General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
+ * Boston, MA 02111-1307, USA.
+ */
+
+#ifndef __TEST_RT_POOL_H__
+#define __TEST_RT_POOL_H__
+
+#include <gst/gst.h>
+
+G_BEGIN_DECLS
+
+/* --- standard type macros --- */
+#define TEST_TYPE_RT_POOL             (test_rt_pool_get_type ())
+#define TEST_RT_POOL(pool)            (G_TYPE_CHECK_INSTANCE_CAST ((pool), TEST_TYPE_RT_POOL, TestRTPool))
+#define TEST_IS_RT_POOL(pool)         (G_TYPE_CHECK_INSTANCE_TYPE ((pool), TEST_TYPE_RT_POOL))
+#define TEST_RT_POOL_CLASS(pclass)    (G_TYPE_CHECK_CLASS_CAST ((pclass), TEST_TYPE_RT_POOL, TestRTPoolClass))
+#define TEST_IS_RT_POOL_CLASS(pclass) (G_TYPE_CHECK_CLASS_TYPE ((pclass), TEST_TYPE_RT_POOL))
+#define TEST_RT_POOL_GET_CLASS(pool)  (G_TYPE_INSTANCE_GET_CLASS ((pool), TEST_TYPE_RT_POOL, TestRTPoolClass))
+#define TEST_RT_POOL_CAST(pool)       ((TestRTPool*)(pool))
+
+typedef struct _TestRTPool TestRTPool;
+typedef struct _TestRTPoolClass TestRTPoolClass;
+
+struct _TestRTPool {
+  GstTaskPool    object;
+};
+
+struct _TestRTPoolClass {
+  GstTaskPoolClass parent_class;
+};
+
+GType           test_rt_pool_get_type    (void);
+
+GstTaskPool *   test_rt_pool_new         (void);
+
+G_END_DECLS
+
+#endif /* __TEST_RT_POOL_H__ */
diff --git a/win32/common/libgstreamer.def b/win32/common/libgstreamer.def
index 0840222..9bab26d 100644
--- a/win32/common/libgstreamer.def
+++ b/win32/common/libgstreamer.def
@@ -446,6 +446,7 @@ EXPORTS
 	gst_marshal_VOID__POINTER_OBJECT
 	gst_marshal_VOID__UINT_BOXED
 	gst_message_get_seqnum
+	gst_message_get_stream_status_object
 	gst_message_get_structure
 	gst_message_get_type
 	gst_message_new_application
@@ -467,6 +468,7 @@ EXPORTS
 	gst_message_new_segment_start
 	gst_message_new_state_changed
 	gst_message_new_state_dirty
+	gst_message_new_stream_status
 	gst_message_new_structure_change
 	gst_message_new_tag
 	gst_message_new_warning
@@ -483,11 +485,13 @@ EXPORTS
 	gst_message_parse_segment_done
 	gst_message_parse_segment_start
 	gst_message_parse_state_changed
+	gst_message_parse_stream_status
 	gst_message_parse_structure_change
 	gst_message_parse_tag
 	gst_message_parse_warning
 	gst_message_set_buffering_stats
 	gst_message_set_seqnum
+	gst_message_set_stream_status_object
 	gst_message_type_get_name
 	gst_message_type_get_type
 	gst_message_type_to_quark
@@ -814,6 +818,7 @@ EXPORTS
 	gst_static_pad_template_get_type
 	gst_stream_error_get_type
 	gst_stream_error_quark
+	gst_stream_status_type_get_type
 	gst_structure_change_type_get_type
 	gst_structure_copy
 	gst_structure_empty_new
@@ -933,11 +938,22 @@ EXPORTS
 	gst_tag_setter_set_tag_merge_mode
 	gst_task_cleanup_all
 	gst_task_create
+	gst_task_get_pool
 	gst_task_get_state
 	gst_task_get_type
 	gst_task_join
 	gst_task_pause
+	gst_task_pool_cleanup
+	gst_task_pool_get_type
+	gst_task_pool_join
+	gst_task_pool_new
+	gst_task_pool_prepare
+	gst_task_pool_push
 	gst_task_set_lock
+	gst_task_set_pool
+	gst_task_set_priority
+	gst_task_set_state
+	gst_task_set_thread_callbacks
 	gst_task_start
 	gst_task_state_get_type
 	gst_task_stop
