socketsrc: handle GstNetworkMessage eventsmultisocketsink-receive

1 files changed, 53 insertions, 0 deletions

diff --git a/gst/tcp/gstsocketsrc.c b/gst/tcp/gstsocketsrc.c
index 149ced185..42380c3dc 100644
--- a/gst/tcp/gstsocketsrc.c
+++ b/gst/tcp/gstsocketsrc.c
@@ -87,6 +87,7 @@ G_DEFINE_TYPE (GstSocketSrc, gst_socket_src, GST_TYPE_PUSH_SRC);
 static void gst_socket_src_finalize (GObject * gobject);
 
 static GstCaps *gst_socketsrc_getcaps (GstBaseSrc * src, GstCaps * filter);
+static gboolean gst_socketsrc_event (GstBaseSrc * src, GstEvent * event);
 static GstFlowReturn gst_socket_src_fill (GstPushSrc * psrc,
     GstBuffer * outbuf);
 static gboolean gst_socket_src_unlock (GstBaseSrc * bsrc);
@@ -140,6 +141,7 @@ gst_socket_src_class_init (GstSocketSrcClass * klass)
       "Thomas Vander Stichele <thomas at apestaart dot org>, "
       "William Manley <will@williammanley.net>");
 
+  gstbasesrc_class->event = gst_socketsrc_event;
   gstbasesrc_class->get_caps = gst_socketsrc_getcaps;
   gstbasesrc_class->unlock = gst_socket_src_unlock;
   gstbasesrc_class->unlock_stop = gst_socket_src_unlock_stop;
@@ -169,6 +171,57 @@ gst_socket_src_finalize (GObject * gobject)
   G_OBJECT_CLASS (parent_class)->finalize (gobject);
 }
 
+static gboolean
+gst_socketsrc_event (GstBaseSrc * bsrc, GstEvent * event)
+{
+  GstSocketSrc *src;
+  gboolean res = FALSE;
+
+  src = GST_SOCKET_SRC (bsrc);
+
+  switch (GST_EVENT_TYPE (event)) {
+    case GST_EVENT_CUSTOM_UPSTREAM:
+      if (gst_event_has_name (event, "GstNetworkMessage")) {
+        const GstStructure *str = gst_event_get_structure (event);
+        GSocket *socket;
+
+        GST_OBJECT_LOCK (src);
+        if ((socket = src->socket))
+          g_object_ref (socket);
+        GST_OBJECT_UNLOCK (src);
+
+        if (socket) {
+          GstBuffer *buf;
+          GstMapInfo map;
+          GError *err = NULL;
+          gssize ret;
+
+          gst_structure_get (str, "buffer", GST_TYPE_BUFFER, &buf, NULL);
+
+          gst_buffer_map (buf, &map, GST_MAP_READ);
+          GST_LOG ("sending buffer of size %" G_GSIZE_FORMAT, map.size);
+          ret =
+              g_socket_send_with_blocking (socket, (gchar *) map.data, map.size,
+              FALSE, src->cancellable, &err);
+          gst_buffer_unmap (buf, &map);
+
+          if (ret == -1) {
+            GST_WARNING ("could not send message: %s", err->message);
+            g_clear_error (&err);
+          }
+          gst_buffer_unref (buf);
+          g_object_unref (socket);
+        }
+      }
+      res = TRUE;
+      break;
+    default:
+      res = GST_BASE_SRC_CLASS (parent_class)->event (bsrc, event);
+      break;
+  }
+  return res;
+}
+
 static GstCaps *
 gst_socketsrc_getcaps (GstBaseSrc * src, GstCaps * filter)
 {