clapper: Add messaging between app, reactables and player

In some cases, simple properties in reactable enhancers are not enough.
This allows enhancers to do something on a signal from user (e.g. button/shortcut press).

It can also be used to communicate with the player for things that were
already done externally (like playlist parsing in Media Scanner enhancer).

src/lib/clapper/clapper-enums.h

@@ -54,6 +54,21 @@ typedef enum
   CLAPPER_PLAYER_SEEK_METHOD_FAST,
 } ClapperPlayerSeekMethod;
 
+/**
+ * ClapperPlayerMessageDestination:
+ * @CLAPPER_PLAYER_MESSAGE_DESTINATION_PLAYER: Messaging from application or reactable enhancers to the player itself.
+ * @CLAPPER_PLAYER_MESSAGE_DESTINATION_REACTABLES: Messaging from application to the reactable enhancers.
+ * @CLAPPER_PLAYER_MESSAGE_DESTINATION_APPLICATION: Messaging from reactable enhancers to the application.
+ *
+ * Since: 0.10
+ */
+typedef enum
+{
+  CLAPPER_PLAYER_MESSAGE_DESTINATION_PLAYER = 0,
+  CLAPPER_PLAYER_MESSAGE_DESTINATION_REACTABLES,
+  CLAPPER_PLAYER_MESSAGE_DESTINATION_APPLICATION,
+} ClapperPlayerMessageDestination;
+
 /**
  * ClapperQueueProgressionMode:
  * @CLAPPER_QUEUE_PROGRESSION_NONE: Queue will not change current item after playback finishes.

src/lib/clapper/clapper-playbin-bus-private.h

@@ -49,4 +49,6 @@ void clapper_playbin_bus_post_current_item_change (GstBus *bus, ClapperMediaItem
 
 void clapper_playbin_bus_post_item_suburi_change (GstBus *bus, ClapperMediaItem *item);
 
+void clapper_playbin_bus_post_user_message (GstBus *bus, GstMessage *msg);
+
 G_END_DECLS

src/lib/clapper/clapper-playbin-bus.c

@@ -42,7 +42,8 @@ enum
   CLAPPER_PLAYBIN_BUS_STRUCTURE_RATE_CHANGE,
   CLAPPER_PLAYBIN_BUS_STRUCTURE_STREAM_CHANGE,
   CLAPPER_PLAYBIN_BUS_STRUCTURE_CURRENT_ITEM_CHANGE,
-  CLAPPER_PLAYBIN_BUS_STRUCTURE_ITEM_SUBURI_CHANGE
+  CLAPPER_PLAYBIN_BUS_STRUCTURE_ITEM_SUBURI_CHANGE,
+  CLAPPER_PLAYBIN_BUS_STRUCTURE_USER_MESSAGE
 };
 
 static ClapperBusQuark _structure_quarks[] = {
@@ -54,6 +55,7 @@ static ClapperBusQuark _structure_quarks[] = {
   {"stream-change", 0},
   {"current-item-change", 0},
   {"item-suburi-change", 0},
+  {"user-message", 0},
   {NULL, 0}
 };
 
@@ -793,6 +795,106 @@ _handle_stream_change_msg (GstMessage *msg,
   }
 }
 
+void
+clapper_playbin_bus_post_user_message (GstBus *bus, GstMessage *msg)
+{
+  GstStructure *structure = gst_structure_new_id_empty (_STRUCTURE_QUARK (USER_MESSAGE));
+  GValue value = G_VALUE_INIT;
+
+  g_value_init (&value, GST_TYPE_MESSAGE);
+  g_value_take_boxed (&value, msg);
+
+  gst_structure_id_take_value (structure, _FIELD_QUARK (VALUE), &value);
+
+  gst_bus_post (bus, gst_message_new_application (NULL, structure));
+}
+
+static inline void
+_on_playlist_parsed_msg (GstMessage *msg, ClapperPlayer *player)
+{
+  ClapperMediaItem *playlist_item = NULL;
+  GListStore *playlist = NULL;
+  const GstStructure *structure = gst_message_get_structure (msg);
+  guint n_items;
+
+  /* If message contains item, use that.
+   * Otherwise assume pending item was parsed. */
+  if (gst_structure_has_field (structure, "item")) {
+    gst_structure_get (structure,
+        "item", CLAPPER_TYPE_MEDIA_ITEM, &playlist_item, NULL);
+  } else {
+    GST_OBJECT_LOCK (player);
+
+    /* Playlist is always parsed before playback starts */
+    if (player->pending_item)
+      playlist_item = gst_object_ref (player->pending_item);
+
+    GST_OBJECT_UNLOCK (player);
+  }
+
+  if (G_UNLIKELY (playlist_item == NULL)) {
+    GST_WARNING_OBJECT (player, "Playlist parsed without media item set");
+    return;
+  }
+
+  GST_INFO_OBJECT (player, "Received parsed playlist of %" GST_PTR_FORMAT
+      "(%s)", playlist_item, clapper_media_item_get_uri (playlist_item));
+
+  gst_structure_get (structure,
+      "playlist", G_TYPE_LIST_STORE, &playlist, NULL);
+
+  n_items = g_list_model_get_n_items (G_LIST_MODEL (playlist));
+
+  if (G_LIKELY (n_items > 0)) {
+    ClapperMediaItem *active_item = g_list_model_get_item (G_LIST_MODEL (playlist), 0);
+    gboolean updated;
+
+    /* Update redirect URI (must be done from player thread) */
+    updated = clapper_media_item_update_from_item (playlist_item, active_item, player);
+    gst_object_unref (active_item);
+
+    if (!updated) {
+      GstMessage *msg;
+      GError *error;
+
+      error = g_error_new (GST_RESOURCE_ERROR, GST_RESOURCE_ERROR_FAILED,
+          "Detected infinite redirection in playlist");
+      msg = gst_message_new_error (GST_OBJECT (player), error, NULL);
+
+      _handle_error_msg (msg, player);
+
+      g_error_free (error);
+      gst_message_unref (msg);
+    } else if (n_items > 1) {
+      /* Forward to append remaining items (must be done from main thread) */
+      clapper_app_bus_post_insert_playlist (player->app_bus,
+          GST_OBJECT_CAST (player),
+          GST_OBJECT_CAST (playlist_item),
+          G_OBJECT (playlist));
+    }
+  }
+
+  gst_object_unref (playlist_item);
+  g_object_unref (playlist);
+}
+
+static inline void
+_handle_user_message_msg (GstMessage *msg, const GstStructure *structure, ClapperPlayer *player)
+{
+  GstMessage *user_message = NULL;
+
+  gst_structure_id_get (structure,
+      _FIELD_QUARK (VALUE), GST_TYPE_MESSAGE, &user_message,
+      NULL);
+
+  GST_DEBUG_OBJECT (player, "Received user message: %" GST_PTR_FORMAT, user_message);
+
+  if (gst_message_has_name (user_message, "ClapperPlaylistParsed"))
+    _on_playlist_parsed_msg (user_message, player);
+
+  gst_message_unref (user_message);
+}
+
 static inline void
 _handle_app_msg (GstMessage *msg, ClapperPlayer *player)
 {
@@ -813,6 +915,8 @@ _handle_app_msg (GstMessage *msg, ClapperPlayer *player)
     _handle_current_item_change_msg (msg, structure, player);
   else if (quark == _STRUCTURE_QUARK (ITEM_SUBURI_CHANGE))
     _handle_item_suburi_change_msg (msg, structure, player);
+  else if (quark == _STRUCTURE_QUARK (USER_MESSAGE))
+    _handle_user_message_msg (msg, structure, player);
 }
 
 static inline void
@@ -832,70 +936,7 @@ _handle_element_msg (GstMessage *msg, ClapperPlayer *player)
     g_free (name);
     g_free (details);
   } else if (gst_message_has_name (msg, "ClapperPlaylistParsed")) {
-    ClapperMediaItem *playlist_item = NULL;
-    GListStore *playlist = NULL;
-    const GstStructure *structure = gst_message_get_structure (msg);
-    guint n_items;
-
-    /* If message contains item, use that.
-     * Otherwise assume pending item was parsed. */
-    if (gst_structure_has_field (structure, "item")) {
-      gst_structure_get (structure,
-          "item", CLAPPER_TYPE_MEDIA_ITEM, &playlist_item, NULL);
-    } else {
-      GST_OBJECT_LOCK (player);
-
-      /* Playlist is always parsed before playback starts */
-      if (player->pending_item)
-        playlist_item = gst_object_ref (player->pending_item);
-
-      GST_OBJECT_UNLOCK (player);
-    }
-
-    if (G_UNLIKELY (playlist_item == NULL)) {
-      GST_WARNING_OBJECT (player, "Playlist parsed without media item set");
-      return;
-    }
-
-    GST_INFO_OBJECT (player, "Received parsed playlist of %" GST_PTR_FORMAT
-        "(%s)", playlist_item, clapper_media_item_get_uri (playlist_item));
-
-    gst_structure_get (structure,
-        "playlist", G_TYPE_LIST_STORE, &playlist, NULL);
-
-    n_items = g_list_model_get_n_items (G_LIST_MODEL (playlist));
-
-    if (G_LIKELY (n_items > 0)) {
-      ClapperMediaItem *active_item = g_list_model_get_item (G_LIST_MODEL (playlist), 0);
-      gboolean updated;
-
-      /* Update redirect URI (must be done from player thread) */
-      updated = clapper_media_item_update_from_item (playlist_item, active_item, player);
-      gst_object_unref (active_item);
-
-      if (!updated) {
-        GstMessage *msg;
-        GError *error;
-
-        error = g_error_new (GST_RESOURCE_ERROR, GST_RESOURCE_ERROR_FAILED,
-            "Detected infinite redirection in playlist");
-        msg = gst_message_new_error (GST_OBJECT (player), error, NULL);
-
-        _handle_error_msg (msg, player);
-
-        g_error_free (error);
-        gst_message_unref (msg);
-      } else if (n_items > 1) {
-        /* Forward to append remaining items (must be done from main thread) */
-        clapper_app_bus_post_insert_playlist (player->app_bus,
-            GST_OBJECT_CAST (player),
-            GST_OBJECT_CAST (playlist_item),
-            G_OBJECT (playlist));
-      }
-    }
-
-    gst_object_unref (playlist_item);
-    g_object_unref (playlist);
+    _on_playlist_parsed_msg (msg, player);
   } else if (gst_message_has_name (msg, "GstCacheDownloadComplete")) {
     ClapperMediaItem *downloaded_item = NULL;
     const GstStructure *structure;

src/lib/clapper/clapper-player.c

@@ -2245,6 +2245,56 @@ clapper_player_make_pipeline_graph (ClapperPlayer *self, GstDebugGraphDetails de
   return gst_debug_bin_to_dot_data (GST_BIN (self->playbin), details);
 }
 
+/**
+ * clapper_player_post_message:
+ * @player: a #ClapperPlayer
+ * @msg: (transfer full): a #GstMessage
+ * @destination: a #ClapperPlayerMessageDestination
+ *
+ * Allows sending custom messages to the desired @destination.
+ *
+ * This functionality can be used for communication with enhancers implementing
+ * [iface@Clapper.Reactable] interface. Useful for applications to send custom messages
+ * to enhacers that can react to them and/or for enhancers development to send events
+ * from them to the applications. It can also be used for sending specific messages
+ * from application or enhancers to the player itself.
+ *
+ * Messages send to the application can be received by connecting a
+ * [signal@Clapper.Player::message] signal handler. Inspection of message source
+ * object can be done to determine who send given message.
+ *
+ * Since: 0.10
+ */
+void
+clapper_player_post_message (ClapperPlayer *self, GstMessage *msg,
+    ClapperPlayerMessageDestination destination)
+{
+  g_return_if_fail (CLAPPER_IS_PLAYER (self));
+  g_return_if_fail (GST_IS_MESSAGE (msg));
+
+  switch (destination) {
+    case CLAPPER_PLAYER_MESSAGE_DESTINATION_PLAYER:
+      clapper_playbin_bus_post_user_message (self->bus, msg);
+      return; // Message taken
+    case CLAPPER_PLAYER_MESSAGE_DESTINATION_REACTABLES:
+      if (self->reactables_manager) {
+        clapper_reactables_manager_post_message (self->reactables_manager, msg);
+        return; // Message taken
+      }
+      break;
+    case CLAPPER_PLAYER_MESSAGE_DESTINATION_APPLICATION:
+      clapper_app_bus_post_message_signal (self->app_bus,
+          GST_OBJECT_CAST (self), signals[SIGNAL_MESSAGE], msg);
+      break; // Above function does not take message (unref needed)
+    default:
+      g_assert_not_reached ();
+      break;
+  }
+
+  /* Unref when not taken */
+  gst_message_unref (msg);
+}
+
 static void
 clapper_player_thread_start (ClapperThreadedObject *threaded_object)
 {

src/lib/clapper/clapper-player.h

@@ -210,4 +210,7 @@ void clapper_player_add_feature (ClapperPlayer *player, ClapperFeature *feature)
 CLAPPER_API
 gchar * clapper_player_make_pipeline_graph (ClapperPlayer *player, GstDebugGraphDetails details);
 
+CLAPPER_API
+void clapper_player_post_message (ClapperPlayer *player, GstMessage *msg, ClapperPlayerMessageDestination destination);
+
 G_END_DECLS

src/lib/clapper/clapper-reactable.h

@@ -206,6 +206,17 @@ struct _ClapperReactableInterface
    */
   void (* queue_progression_changed) (ClapperReactable *reactable, ClapperQueueProgressionMode mode);
 
+  /**
+   * ClapperReactableInterface::message_received:
+   * @reactable: a #ClapperReactable
+   * @msg: a #GstMessage
+   *
+   * Custom message from user was received on reactables bus.
+   *
+   * Since: 0.10
+   */
+  void (* message_received) (ClapperReactable *reactable, GstMessage *msg);
+
   /*< private >*/
   gpointer padding[8];
 };

src/lib/clapper/clapper-reactables-manager-private.h

@@ -18,6 +18,8 @@
 
 #pragma once
 
+#include <gst/gst.h>
+
 #include "clapper-enums.h"
 #include "clapper-threaded-object.h"
 #include "clapper-enhancer-proxy.h"
@@ -36,6 +38,9 @@ void clapper_reactables_manager_initialize (void);
 G_GNUC_INTERNAL
 ClapperReactablesManager * clapper_reactables_manager_new (void);
 
+G_GNUC_INTERNAL
+void clapper_reactables_manager_post_message (ClapperReactablesManager *manager, GstMessage *msg);
+
 G_GNUC_INTERNAL
 void clapper_reactables_manager_trigger_configure_take_config (ClapperReactablesManager *manager, ClapperEnhancerProxy *proxy, GstStructure *config);
 

src/lib/clapper/clapper-reactables-manager.c

@@ -78,6 +78,7 @@ enum
 {
   CLAPPER_REACTABLES_MANAGER_QUARK_CONFIGURE = 0,
   CLAPPER_REACTABLES_MANAGER_QUARK_EVENT,
+  CLAPPER_REACTABLES_MANAGER_QUARK_USER_MESSAGE,
   CLAPPER_REACTABLES_MANAGER_QUARK_VALUE,
   CLAPPER_REACTABLES_MANAGER_QUARK_EXTRA_VALUE
 };
@@ -85,6 +86,7 @@ enum
 static ClapperBusQuark _quarks[] = {
   {"configure", 0},
   {"event", 0},
+  {"user-message", 0},
   {"value", 0},
   {"extra-value", 0},
   {NULL, 0}
@@ -308,6 +310,23 @@ clapper_reactables_manager_handle_event (ClapperReactablesManager *self, const G
   }
 }
 
+static inline void
+clapper_reactables_manager_handle_user_message (ClapperReactablesManager *self, const GstStructure *structure)
+{
+  const GValue *value = gst_structure_id_get_value (structure, _QUARK (VALUE));
+  guint i;
+
+  for (i = 0; i < self->array->len; ++i) {
+    ClapperReactableManagerData *data = g_ptr_array_index (self->array, i);
+    ClapperReactableInterface *reactable_iface = CLAPPER_REACTABLE_GET_IFACE (data->reactable);
+
+    if (reactable_iface->message_received) {
+      reactable_iface->message_received (data->reactable,
+          GST_MESSAGE_CAST (g_value_get_boxed (value)));
+    }
+  }
+}
+
 static gboolean
 _bus_message_func (GstBus *bus, GstMessage *msg, gpointer user_data G_GNUC_UNUSED)
 {
@@ -324,6 +343,8 @@ _bus_message_func (GstBus *bus, GstMessage *msg, gpointer user_data G_GNUC_UNUSE
       clapper_reactables_manager_handle_event (self, structure);
     } else if (quark == _QUARK (CONFIGURE)) {
       clapper_reactables_manager_handle_configure (self, structure);
+    } else if (quark == _QUARK (USER_MESSAGE)) {
+      clapper_reactables_manager_handle_user_message (self, structure);
     } else {
       GST_ERROR_OBJECT (self, "Received invalid quark on reactables bus!");
     }
@@ -365,6 +386,21 @@ clapper_reactables_manager_new (void)
   return reactables_manager;
 }
 
+void
+clapper_reactables_manager_post_message (ClapperReactablesManager *self, GstMessage *msg)
+{
+  GstStructure *structure = gst_structure_new_id_empty (_QUARK (USER_MESSAGE));
+  GValue value = G_VALUE_INIT;
+
+  g_value_init (&value, GST_TYPE_MESSAGE);
+  g_value_take_boxed (&value, msg);
+
+  gst_structure_id_take_value (structure, _QUARK (VALUE), &value);
+
+  gst_bus_post (self->bus, gst_message_new_application (
+      GST_OBJECT_CAST (self), structure));
+}
+
 void
 clapper_reactables_manager_trigger_configure_take_config (ClapperReactablesManager *self,
     ClapperEnhancerProxy *proxy, GstStructure *config)