2 * Copyright (C) 2004 Wim Taymans <wim@fluendo.com>
4 * gstbus.c: GstBus subsystem
6 * This library is free software; you can redistribute it and/or
7 * modify it under the terms of the GNU Library General Public
8 * License as published by the Free Software Foundation; either
9 * version 2 of the License, or (at your option) any later version.
11 * This library is distributed in the hope that it will be useful,
12 * but WITHOUT ANY WARRANTY; without even the implied warranty of
13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
14 * Library General Public License for more details.
16 * You should have received a copy of the GNU Library General Public
17 * License along with this library; if not, write to the
18 * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
19 * Boston, MA 02111-1307, USA.
25 #include <sys/types.h>
26 #include <sys/socket.h>
28 #include "gst_private.h"
38 static void gst_bus_class_init (GstBusClass * klass);
39 static void gst_bus_init (GstBus * bus);
40 static void gst_bus_dispose (GObject * object);
42 static void gst_bus_set_property (GObject * object, guint prop_id,
43 const GValue * value, GParamSpec * pspec);
44 static void gst_bus_get_property (GObject * object, guint prop_id,
45 GValue * value, GParamSpec * pspec);
47 static GstObjectClass *parent_class = NULL;
49 /* static guint gst_bus_signals[LAST_SIGNAL] = { 0 }; */
52 gst_bus_get_type (void)
54 static GType bus_type = 0;
57 static const GTypeInfo bus_info = {
61 (GClassInitFunc) gst_bus_class_init,
66 (GInstanceInitFunc) gst_bus_init,
70 bus_type = g_type_register_static (GST_TYPE_OBJECT, "GstBus", &bus_info, 0);
76 gst_bus_class_init (GstBusClass * klass)
78 GObjectClass *gobject_class;
79 GstObjectClass *gstobject_class;
81 gobject_class = (GObjectClass *) klass;
82 gstobject_class = (GstObjectClass *) klass;
84 parent_class = g_type_class_ref (GST_TYPE_OBJECT);
86 if (!g_thread_supported ())
89 gobject_class->dispose = GST_DEBUG_FUNCPTR (gst_bus_dispose);
90 gobject_class->set_property = GST_DEBUG_FUNCPTR (gst_bus_set_property);
91 gobject_class->get_property = GST_DEBUG_FUNCPTR (gst_bus_get_property);
95 gst_bus_init (GstBus * bus)
97 bus->queue = g_queue_new ();
98 bus->queue_lock = g_mutex_new ();
100 if (socketpair (PF_UNIX, SOCK_STREAM, 0, bus->control_socket) < 0)
103 bus->io_channel = g_io_channel_unix_new (bus->control_socket[0]);
110 g_warning ("cannot create io channel");
111 bus->io_channel = NULL;
116 gst_bus_dispose (GObject * object)
120 bus = GST_BUS (object);
122 if (bus->io_channel) {
123 g_io_channel_shutdown (bus->io_channel, TRUE, NULL);
124 g_io_channel_unref (bus->io_channel);
125 bus->io_channel = NULL;
127 close (bus->control_socket[0]);
128 close (bus->control_socket[1]);
131 g_mutex_lock (bus->queue_lock);
132 g_queue_free (bus->queue);
134 g_mutex_unlock (bus->queue_lock);
135 g_mutex_free (bus->queue_lock);
136 bus->queue_lock = NULL;
139 G_OBJECT_CLASS (parent_class)->dispose (object);
143 gst_bus_set_property (GObject * object, guint prop_id,
144 const GValue * value, GParamSpec * pspec)
148 bus = GST_BUS (object);
152 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
158 gst_bus_get_property (GObject * object, guint prop_id,
159 GValue * value, GParamSpec * pspec)
163 bus = GST_BUS (object);
167 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
177 result = g_object_new (gst_bus_get_type (), NULL);
184 * @bus: a #GstBus to post on
185 * @message: The #GstMessage to post
187 * Post a message on the given bus.
189 * Returns: TRUE if the message could be posted.
194 gst_bus_post (GstBus * bus, GstMessage * message)
197 GstBusSyncReply reply = GST_BUS_PASS;
198 GstBusSyncHandler handler;
199 gpointer handler_data;
200 gboolean need_write = FALSE;
201 ssize_t write_ret = -1;
203 g_return_val_if_fail (GST_IS_BUS (bus), FALSE);
204 g_return_val_if_fail (GST_IS_MESSAGE (message), FALSE);
206 //g_print ("posting message on bus, type %d\n", GST_MESSAGE_TYPE (message));
207 GST_DEBUG_OBJECT (bus, "posting message on bus");
210 handler = bus->sync_handler;
211 handler_data = bus->sync_handler_data;
214 /* first call the sync handler if it is installed */
216 reply = handler (bus, message, handler_data);
219 /* now see what we should do with the message */
222 /* drop the message */
225 /* pass the message to the async queue */
226 g_mutex_lock (bus->queue_lock);
227 if (g_queue_get_length (bus->queue) == 0)
229 g_queue_push_tail (bus->queue, message);
230 g_mutex_unlock (bus->queue_lock);
232 if (g_queue_get_length (bus->queue) == 0)
237 while (write_ret == -1) {
243 perror ("gst_bus_post: could not write to fd");
246 write_ret = write (bus->control_socket[1], &c, 1);
252 /* async delivery, we need a mutex and a cond to block
254 GMutex *lock = g_mutex_new ();
255 GCond *cond = g_cond_new ();
257 GST_MESSAGE_COND (message) = cond;
258 GST_MESSAGE_GET_LOCK (message) = lock;
260 GST_DEBUG ("waiting for async delivery of message %p", message);
262 /* now we lock the message mutex, send the message to the async
263 * queue. When the message is handled by the app and destroyed,
264 * the cond will be signalled and we can continue */
266 g_mutex_lock (bus->queue_lock);
267 if (g_queue_get_length (bus->queue) == 0)
269 g_queue_push_tail (bus->queue, message);
270 g_mutex_unlock (bus->queue_lock);
275 while (write_ret == -1) {
281 perror ("gst_bus_post: could not write to fd");
284 write_ret = write (bus->control_socket[1], &c, 1);
288 /* now block till the message is freed */
289 g_cond_wait (cond, lock);
290 g_mutex_unlock (lock);
292 GST_DEBUG ("message %p delivered asynchronously", message);
304 * gst_bus_have_pending:
305 * @bus: a #GstBus to check
307 * Check if there are pending messages on the bus that should be
310 * Returns: TRUE if there are messages on the bus to be handled.
315 gst_bus_have_pending (GstBus * bus)
319 g_return_val_if_fail (GST_IS_BUS (bus), FALSE);
321 g_mutex_lock (bus->queue_lock);
322 length = g_queue_get_length (bus->queue);
323 g_mutex_unlock (bus->queue_lock);
330 * @bus: a #GstBus to pop
332 * Get a message from the bus.
334 * Returns: The #GstMessage that is on the bus, or NULL if the bus is empty.
339 gst_bus_pop (GstBus * bus)
342 gboolean needs_read = FALSE;
344 g_return_val_if_fail (GST_IS_BUS (bus), NULL);
346 g_mutex_lock (bus->queue_lock);
347 message = g_queue_pop_head (bus->queue);
348 if (message && g_queue_get_length (bus->queue) == 0)
350 g_mutex_unlock (bus->queue_lock);
354 ssize_t read_ret = -1;
356 /* the char in the fd is essentially just a way to wake us up. read it off so
357 we're not woken up again. */
359 while (read_ret == -1) {
365 perror ("gst_bus_pop: could not read from fd");
368 read_ret = read (bus->control_socket[0], &c, 1);
379 * Peek the message on the top of the bus' queue. The bus maintains ownership of
380 * the message, and the message will remain on the bus' message queue.
382 * Returns: The #GstMessage that is on the bus, or NULL if the bus is empty.
387 gst_bus_peek (GstBus * bus)
391 g_return_val_if_fail (GST_IS_BUS (bus), NULL);
393 g_mutex_lock (bus->queue_lock);
394 message = g_queue_peek_head (bus->queue);
395 g_mutex_unlock (bus->queue_lock);
401 * gst_bus_set_sync_handler:
402 * @bus: a #GstBus to install the handler on
403 * @func: The handler function to install
404 * @data: User data that will be sent to the handler function.
406 * Install a synchronous handler on the bus. The function will be called
407 * every time a new message is posted on the bus. Note that the function
408 * will be called in the same thread context as the posting object.
411 gst_bus_set_sync_handler (GstBus * bus, GstBusSyncHandler func, gpointer data)
413 g_return_if_fail (GST_IS_BUS (bus));
416 bus->sync_handler = func;
417 bus->sync_handler_data = data;
422 * gst_bus_create_watch:
423 * @bus: a #GstBus to create the watch for
425 * Create watch for this bus.
427 * Returns: A #GSource that can be added to a mainloop.
430 gst_bus_create_watch (GstBus * bus)
434 g_return_val_if_fail (GST_IS_BUS (bus), NULL);
436 /* FIXME, we need to ref the bus and unref it when the source
438 source = g_io_create_watch (bus->io_channel, G_IO_IN);
448 GstBusHandler handler;
450 GDestroyNotify notify;
454 bus_watch_callback (GIOChannel * channel, GIOCondition cond,
458 gboolean needs_pop = TRUE;
460 g_return_val_if_fail (GST_IS_BUS (watch->bus), FALSE);
462 message = gst_bus_peek (watch->bus);
464 g_return_val_if_fail (message != NULL, TRUE);
467 needs_pop = watch->handler (watch->bus, message, watch->user_data);
470 gst_message_unref (gst_bus_pop (watch->bus));
476 bus_watch_destroy (GstBusWatch * watch)
479 watch->notify (watch->user_data);
481 gst_object_unref (GST_OBJECT_CAST (watch->bus));
486 * gst_bus_add_watch_full:
487 * @bus: a #GstBus to create the watch for
488 * @handler: A function to call when a message is received.
490 * Adds the bus to the mainloop with the given priority. If the handler returns
491 * TRUE, the message will then be popped off the queue. When the handler is
492 * called, the message belongs to the caller; if you want to keep a copy of it,
493 * call gst_message_ref before leaving the handler.
495 * Returns: The event source id.
500 gst_bus_add_watch_full (GstBus * bus, gint priority,
501 GstBusHandler handler, gpointer user_data, GDestroyNotify notify)
506 g_return_val_if_fail (GST_IS_BUS (bus), 0);
508 watch = g_new (GstBusWatch, 1);
510 gst_object_ref (GST_OBJECT_CAST (bus));
511 watch->source = gst_bus_create_watch (bus);
513 watch->priority = priority;
514 watch->handler = handler;
515 watch->user_data = user_data;
516 watch->notify = notify;
518 if (priority != G_PRIORITY_DEFAULT)
519 g_source_set_priority (watch->source, priority);
521 g_source_set_callback (watch->source, (GSourceFunc) bus_watch_callback,
522 watch, (GDestroyNotify) bus_watch_destroy);
524 id = g_source_attach (watch->source, NULL);
525 g_source_unref (watch->source);
532 * @bus: a #GstBus to create the watch for
534 * Adds the bus to the mainloop with the default priority.
536 * Returns: The event source id.
541 gst_bus_add_watch (GstBus * bus, GstBusHandler handler, gpointer user_data)
543 return gst_bus_add_watch_full (bus, G_PRIORITY_DEFAULT, handler, user_data,
551 GstMessageType events;
552 GstMessageType revent;
556 poll_handler (GstBus * bus, GstMessage * message, GstBusPollData * poll_data)
558 if (GST_MESSAGE_TYPE (message) & poll_data->events) {
559 poll_data->revent = GST_MESSAGE_TYPE (message);
560 if (g_main_loop_is_running (poll_data->loop))
561 g_main_loop_quit (poll_data->loop);
562 /* keep the message on the queue */
565 /* pop and unref the message */
571 poll_timeout (GstBusPollData * poll_data)
573 poll_data->timeout_id = 0;
574 g_main_loop_quit (poll_data->loop);
575 /* returning FALSE will remove the source id */
582 * @events: a mask of #GstMessageType, representing the set of message types to
584 * @timeout: the poll timeout, as a #GstClockTimeDiff, or -1 to poll indefinitely.
586 * Poll the bus for events. Will block while waiting for events to come. You can
587 * specify a maximum time to poll with the @timeout parameter. If @timeout is
588 * negative, this function will block indefinitely.
590 * Returns: The type of the message that was received, or GST_MESSAGE_UNKNOWN if
591 * the poll timed out. The message will remain in the bus queue; you will need
592 * to gst_bus_pop() it off before entering gst_bus_poll() again.
595 gst_bus_poll (GstBus * bus, GstMessageType events, GstClockTimeDiff timeout)
597 GstBusPollData *poll_data;
601 poll_data = g_new0 (GstBusPollData, 1);
603 poll_data->timeout_id = g_timeout_add (timeout / GST_MSECOND,
604 (GSourceFunc) poll_timeout, poll_data);
605 poll_data->loop = g_main_loop_new (NULL, FALSE);
606 poll_data->events = events;
607 poll_data->revent = GST_MESSAGE_UNKNOWN;
609 id = gst_bus_add_watch (bus, (GstBusHandler) poll_handler, poll_data);
610 g_main_loop_run (poll_data->loop);
611 g_source_remove (id);
613 ret = poll_data->revent;
615 if (poll_data->timeout_id)
616 g_source_remove (poll_data->timeout_id);
617 g_main_loop_unref (poll_data->loop);