tests: add example of custom taskpools

Add an example to demonstrate the use of a custom taskpool and how to configure
it on the task. Currently the taskpool does not do much yet but it'll create
some custom threads later on.

tests/examples/streams/Makefile.am

@@ -1,4 +1,9 @@
-noinst_PROGRAMS = stream-status
+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)

tests/examples/streams/rtpool-test.c

@@ -0,0 +1,149 @@
+#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:
+      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:
+          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;
+  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 */
+  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 */
+  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);
+}

tests/examples/streams/testrtpool.c

@@ -0,0 +1,93 @@
+/* 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 "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);
+
+G_DEFINE_TYPE (TestRTPool, test_rt_pool, GST_TYPE_TASK_POOL);
+
+static void
+default_prepare (GstTaskPool * pool, GFunc func, gpointer user_data,
+    GError ** error)
+{
+  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, gpointer data, GError ** error)
+{
+  g_message ("pushing Realtime pool %p", pool);
+
+  *error = g_error_new (1, 1, "not supported");
+
+  return NULL;
+}
+
+static void
+default_join (GstTaskPool * pool, gpointer id)
+{
+  g_message ("joining Realtime pool %p", pool);
+}
+
+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;
+}

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__ */