diff --git a/gst/gstbus.c b/gst/gstbus.c
index 2cc41af..e742b56 100644
--- a/gst/gstbus.c
+++ b/gst/gstbus.c
@@ -97,6 +97,7 @@ enum
   PROP_ENABLE_ASYNC
 };
 
+static void gst_bus_set_main_context (GstBus * bus, GMainContext * ctx);
 static void gst_bus_dispose (GObject * object);
 
 static GstObjectClass *parent_class = NULL;
@@ -106,6 +107,7 @@ struct _GstBusPrivate
 {
   guint num_sync_message_emitters;
   GSource *watch_id;
+  GMainContext *main_context;
 
   gboolean enable_async;
   GstPoll *poll;
@@ -271,11 +273,52 @@ gst_bus_dispose (GObject * object)
     if (bus->priv->poll)
       gst_poll_free (bus->priv->poll);
     bus->priv->poll = NULL;
+
+    if (bus->priv->main_context) {
+      g_main_context_unref (bus->priv->main_context);
+      bus->priv->main_context = NULL;
+    }
   }
 
   G_OBJECT_CLASS (parent_class)->dispose (object);
 }
 
+static void
+gst_bus_wakeup_main_context (GstBus * bus)
+{
+  GMainContext *ctx;
+
+  GST_OBJECT_LOCK (bus);
+  if ((ctx = bus->priv->main_context))
+    g_main_context_ref (ctx);
+  GST_OBJECT_UNLOCK (bus);
+
+  g_main_context_wakeup (ctx);
+
+ if (ctx)
+    g_main_context_unref (ctx);
+}
+
+static void
+gst_bus_set_main_context (GstBus * bus, GMainContext * ctx)
+{
+  GST_OBJECT_LOCK (bus);
+
+  if (bus->priv->main_context != NULL) {
+    g_main_context_unref (bus->priv->main_context);
+    bus->priv->main_context = NULL;
+  }
+
+  if (ctx != NULL) {
+    bus->priv->main_context = g_main_context_ref (ctx);
+  }
+
+  GST_DEBUG_OBJECT (bus, "setting main context to %p, GLib default context: %p",
+      ctx, g_main_context_default ());
+
+  GST_OBJECT_UNLOCK (bus);
+}
+
 /**
  * gst_bus_new:
  *
@@ -358,6 +401,7 @@ gst_bus_post (GstBus * bus, GstMessage * message)
       GST_DEBUG_OBJECT (bus, "[msg %p] pushing on async queue", message);
       gst_atomic_queue_push (bus->queue, message);
       gst_poll_write_control (bus->priv->poll);
+      gst_bus_wakeup_main_context (bus);
       GST_DEBUG_OBJECT (bus, "[msg %p] pushed on async queue", message);
 
       break;
@@ -380,6 +424,7 @@ gst_bus_post (GstBus * bus, GstMessage * message)
 
       gst_atomic_queue_push (bus->queue, message);
       gst_poll_write_control (bus->priv->poll);
+      gst_bus_wakeup_main_context (bus);
 
       /* now block till the message is freed */
       g_cond_wait (cond, lock);
@@ -719,13 +764,24 @@ typedef struct
 {
   GSource source;
   GstBus *bus;
+  gboolean inited;
 } GstBusSource;
 
 static gboolean
 gst_bus_source_prepare (GSource * source, gint * timeout)
 {
+  GstBusSource *bsrc = (GstBusSource *) source;
+
+  /* we do this here now that we know that we're attached to a main context
+   * (we don't support detaching a source from a main context and then
+   * re-attaching it to a different main context) */
+  if (G_UNLIKELY (!bsrc->inited)) {
+    gst_bus_set_main_context (bsrc->bus, g_source_get_context (source));
+    bsrc->inited = TRUE;
+  }
+
   *timeout = -1;
-  return FALSE;
+  return gst_bus_have_pending (bsrc->bus);
 }
 
 static gboolean
@@ -733,7 +789,7 @@ gst_bus_source_check (GSource * source)
 {
   GstBusSource *bsrc = (GstBusSource *) source;
 
-  return bsrc->bus->priv->pollfd.revents & (G_IO_IN | G_IO_HUP | G_IO_ERR);
+  return gst_bus_have_pending (bsrc->bus);
 }
 
 static gboolean
@@ -795,6 +851,7 @@ gst_bus_source_finalize (GSource * source)
     bus->priv->watch_id = NULL;
   GST_OBJECT_UNLOCK (bus);
 
+  gst_bus_set_main_context (bsource->bus, NULL);
   gst_object_unref (bsource->bus);
   bsource->bus = NULL;
 }
@@ -822,12 +879,11 @@ gst_bus_create_watch (GstBus * bus)
   GstBusSource *source;
 
   g_return_val_if_fail (GST_IS_BUS (bus), NULL);
-  g_return_val_if_fail (bus->priv->poll != NULL, NULL);
 
   source = (GstBusSource *) g_source_new (&gst_bus_source_funcs,
       sizeof (GstBusSource));
   source->bus = gst_object_ref (bus);
-  g_source_add_poll ((GSource *) source, &bus->priv->pollfd);
+  source->inited = FALSE;
 
   return (GSource *) source;
 }
@@ -837,9 +893,9 @@ static guint
 gst_bus_add_watch_full_unlocked (GstBus * bus, gint priority,
     GstBusFunc func, gpointer user_data, GDestroyNotify notify)
 {
-  GMainContext *ctx;
   guint id;
   GSource *source;
+  GMainContext *ctx;
 
   if (bus->priv->watch_id) {
     GST_ERROR_OBJECT (bus,
