checking if client is already part of list before adding

1 files changed, 79 insertions, 56 deletions

diff --git a/src/sydra.c b/src/sydra.c
index f80d38f..bb55c97 100644
--- a/src/sydra.c
+++ b/src/sydra.c
@@ -301,51 +301,57 @@ static gboolean create_media_elements(struct media_elements *me, GstElement* pip
   return TRUE;
 }
 
-struct udp_elements {
-  GstElement *rtp_video_;
-  GstElement *rtcp_video_;
-  GstElement *rtp_audio_;
-  GstElement *rtcp_audio_;
+struct udp_sink {
+  GstElement *udp_;
+  GList* clients_;
+};
+
+struct udp_sinks {
+  struct udp_sink rtp_video_;
+  struct udp_sink rtcp_video_;
+  struct udp_sink rtp_audio_;
+  struct udp_sink rtcp_audio_;
 };
 
-static gboolean create_udp_elements(options_t* opt, GstElement* pipeline, GstElement* rtp, struct udp_elements *udp)
+static gboolean create_udp_sinks(options_t* opt, GstElement* pipeline, GstElement* rtp, struct udp_sinks *sinks)
 {
-  udp->rtp_video_ = sydra_create_element("multiudpsink", "udprtpv");
-  udp->rtcp_video_ = sydra_create_element("multiudpsink", "udprtcpv");
-  udp->rtp_audio_ = sydra_create_element("multiudpsink", "udprtpa");
-  udp->rtcp_audio_ = sydra_create_element("multiudpsink", "udprtcpa");
+  sinks->rtp_video_.udp_ = sydra_create_element("multiudpsink", "udprtpv");
+  sinks->rtcp_video_.udp_ = sydra_create_element("multiudpsink", "udprtcpv");
+  sinks->rtp_audio_.udp_ = sydra_create_element("multiudpsink", "udprtpa");
+  sinks->rtcp_audio_.udp_ = sydra_create_element("multiudpsink", "udprtcpa");
 
-  if(!udp->rtp_video_ || !udp->rtcp_video_ || !udp->rtp_audio_ || !udp->rtcp_audio_)
+  if(!(sinks->rtp_video_.udp_) || !(sinks->rtcp_video_.udp_) || !(sinks->rtp_audio_.udp_) || !(sinks->rtcp_audio_.udp_))
     return FALSE;
 
   log_printf(DEBUG, "udp elements created successfully!");
 
   int rtp_port_local = opt->rtp_port_base_local_;
-  g_object_set(G_OBJECT(udp->rtp_video_), "bind-port", rtp_port_local++, NULL);
-  g_object_set(G_OBJECT(udp->rtcp_video_), "bind-port", rtp_port_local++, "sync", FALSE, "async", FALSE, NULL);
-  g_object_set(G_OBJECT(udp->rtp_audio_), "bind-port", rtp_port_local++, NULL);
-  g_object_set(G_OBJECT(udp->rtcp_audio_), "bind-port", rtp_port_local++, "sync", FALSE, "async", FALSE, NULL);
+  g_object_set(G_OBJECT(sinks->rtp_video_.udp_), "bind-port", rtp_port_local++, NULL);
+  g_object_set(G_OBJECT(sinks->rtcp_video_.udp_), "bind-port", rtp_port_local++, "sync", FALSE, "async", FALSE, NULL);
+  g_object_set(G_OBJECT(sinks->rtp_audio_.udp_), "bind-port", rtp_port_local++, NULL);
+  g_object_set(G_OBJECT(sinks->rtcp_audio_.udp_), "bind-port", rtp_port_local++, "sync", FALSE, "async", FALSE, NULL);
 
   if(opt->rtp_host_) {
     int rtp_port = opt->rtp_port_base_;
-    g_signal_emit_by_name(G_OBJECT(udp->rtp_video_), "add", opt->rtp_host_, rtp_port++, NULL);
-    g_signal_emit_by_name(G_OBJECT(udp->rtcp_video_), "add", opt->rtp_host_, rtp_port++, NULL);
-    g_signal_emit_by_name(G_OBJECT(udp->rtp_audio_), "add", opt->rtp_host_, rtp_port++, NULL);
-    g_signal_emit_by_name(G_OBJECT(udp->rtcp_audio_), "add", opt->rtp_host_, rtp_port++, NULL);
+    g_signal_emit_by_name(G_OBJECT(sinks->rtp_video_.udp_), "add", opt->rtp_host_, rtp_port++, NULL);
+    g_signal_emit_by_name(G_OBJECT(sinks->rtcp_video_.udp_), "add", opt->rtp_host_, rtp_port++, NULL);
+    g_signal_emit_by_name(G_OBJECT(sinks->rtp_audio_.udp_), "add", opt->rtp_host_, rtp_port++, NULL);
+    g_signal_emit_by_name(G_OBJECT(sinks->rtcp_audio_.udp_), "add", opt->rtp_host_, rtp_port++, NULL);
+        // TODO: add this to the clients_ list
   }
   if(opt->rtp_addr_local_) {
-    g_object_set(G_OBJECT(udp->rtp_video_), "bind-address", opt->rtp_addr_local_, NULL);
-    g_object_set(G_OBJECT(udp->rtcp_video_), "bind-address", opt->rtp_addr_local_, NULL);
-    g_object_set(G_OBJECT(udp->rtp_audio_), "bind-address", opt->rtp_addr_local_, NULL);
-    g_object_set(G_OBJECT(udp->rtcp_audio_), "bind-address", opt->rtp_addr_local_, NULL);
+    g_object_set(G_OBJECT(sinks->rtp_video_.udp_), "bind-address", opt->rtp_addr_local_, NULL);
+    g_object_set(G_OBJECT(sinks->rtcp_video_.udp_), "bind-address", opt->rtp_addr_local_, NULL);
+    g_object_set(G_OBJECT(sinks->rtp_audio_.udp_), "bind-address", opt->rtp_addr_local_, NULL);
+    g_object_set(G_OBJECT(sinks->rtcp_audio_.udp_), "bind-address", opt->rtp_addr_local_, NULL);
   }
 
-  gst_bin_add_many(GST_BIN (pipeline), udp->rtp_video_, udp->rtcp_video_, udp->rtp_audio_, udp->rtcp_audio_, NULL);
+  gst_bin_add_many(GST_BIN (pipeline), sinks->rtp_video_.udp_, sinks->rtcp_video_.udp_, sinks->rtp_audio_.udp_, sinks->rtcp_audio_.udp_, NULL);
 
-  if(!sydra_link_static_static(rtp, "send_rtp_src_0", udp->rtp_video_, "sink") ||
-     !sydra_link_request_static(rtp, "send_rtcp_src_0", udp->rtcp_video_, "sink") ||
-     !sydra_link_static_static(rtp, "send_rtp_src_1", udp->rtp_audio_, "sink") ||
-     !sydra_link_request_static(rtp, "send_rtcp_src_1", udp->rtcp_audio_, "sink"))
+  if(!sydra_link_static_static(rtp, "send_rtp_src_0", sinks->rtp_video_.udp_, "sink") ||
+     !sydra_link_request_static(rtp, "send_rtcp_src_0", sinks->rtcp_video_.udp_, "sink") ||
+     !sydra_link_static_static(rtp, "send_rtp_src_1", sinks->rtp_audio_.udp_, "sink") ||
+     !sydra_link_request_static(rtp, "send_rtcp_src_1", sinks->rtcp_audio_.udp_, "sink"))
     return FALSE;
 
   log_printf(DEBUG, "udp elements linked successfully!");
@@ -440,7 +446,7 @@ static gboolean create_recorder_elements(options_t* opt, GstElement* pipeline, s
   return TRUE;
 }
 
-static GstElement* create_pipeline(options_t* opt, struct udp_elements *udp)
+static GstElement* create_pipeline(options_t* opt, struct udp_sinks *udp)
 {
   GstElement *pipeline = gst_pipeline_new ("sydra");
   if(!pipeline) {
@@ -461,7 +467,7 @@ static GstElement* create_pipeline(options_t* opt, struct udp_elements *udp)
                                   opt->audio_payloader_, NULL };
   if(!create_media_elements(&video, pipeline, rtp, 0) ||
      !create_media_elements(&audio, pipeline, rtp, 1) ||
-     !create_udp_elements(opt, pipeline, rtp, udp)) {
+     !create_udp_sinks(opt, pipeline, rtp, udp)) {
     return NULL;
   }
 
@@ -486,14 +492,17 @@ struct addr_port{
   guint16 port_;
 };
 
-int cmp_addr_port(struct addr_port *a, struct addr_port *b)
+gint cmp_addr_port(gconstpointer A, gconstpointer B)
 {
+  const struct addr_port *a = (const struct addr_port *)A;
+  const struct addr_port *b = (const struct addr_port *)B;
+
   if(a->type_ != b->type_ || a->port_ != b->port_)
     return -1;
   return strncmp(a->addr_, b->addr_, sizeof(a->addr_));
 }
 
-static void udp_add_client(struct sockaddr_storage *addr, socklen_t addrlen, GstElement* udp)
+static void udp_add_client(struct sockaddr_storage *addr, socklen_t addrlen, struct udp_sink* sink)
 {
   struct addr_port* client = g_new(struct addr_port, 1);
   if(!client) return;
@@ -510,22 +519,30 @@ static void udp_add_client(struct sockaddr_storage *addr, socklen_t addrlen, Gst
     log_printf(WARNING, "getnameinfo() error: %s", gai_strerror(errcode));
     return;
   }
-  gchar* name = gst_element_get_name(udp);
-  log_printf(DEBUG, "adding host %s%c%d to list of receivers to element %s", client->addr_, client->type_ == IPv4 ? ':' : '.', client->port_, name);
+  gchar* name = gst_element_get_name(sink->udp_);
+  if(!g_list_find_custom(sink->clients_, client, cmp_addr_port)) {
+    log_printf(INFO, "adding host %s%c%d to list of receivers for element %s", client->addr_, client->type_ == IPv4 ? ':' : '.', client->port_, name);
+    g_signal_emit_by_name(G_OBJECT(sink->udp_), "add", client->addr_, client->port_, NULL);
+    sink->clients_ = g_list_append(sink->clients_, client);
+  } else {
+    log_printf(DEBUG, "not adding host %s%c%d to list of receivers for element %s - already added", client->addr_, client->type_ == IPv4 ? ':' : '.', client->port_, name);
+    g_free(client);
+  }
   g_free(name);
-  g_signal_emit_by_name(G_OBJECT(udp), "add", client->addr_, client->port_, NULL);
-  g_free(client);
 }
 
-struct on_udp_desc_read_param {
-  GstElement* udp_;
-  GList* clients_;
-};
+static void udp_remove_client(struct sockaddr_storage *addr, socklen_t addrlen, struct udp_sink* sink)
+{
+      // TODO: implement this!
+  gchar* name = gst_element_get_name(sink->udp_);
+  log_printf(ERROR, "removing client from %s failed! - removing not yet implemented!", name);
+  g_free(name);
+}
 
 static gboolean on_udp_desc_ready(gint fd, GIOCondition cond, gpointer user_data)
 {
-  GstElement* udp = (GstElement*)user_data;
-  if(!udp) {
+  struct udp_sink* sink = (struct udp_sink*)user_data;
+  if(!sink || !(sink->udp_)) {
     log_printf(WARNING, "File descriptor %d is ready for reading - but supplied element is NULL removing callback", fd);
     return FALSE;
   }
@@ -542,17 +559,22 @@ static gboolean on_udp_desc_ready(gint fd, GIOCondition cond, gpointer user_data
     return FALSE;
   }
 
-  udp_add_client(&addr, addrlen, udp);
+      // TODO: parse real packet format
+  if(buf[0] == 'a') {
+    udp_add_client(&addr, addrlen, sink);
+  } else {
+    udp_remove_client(&addr, addrlen, sink);
+  }
   return TRUE;
 }
 
-static gboolean attach_udp_descriptor(GstElement* udp, const char* name, const char* prop)
+static gboolean attach_udp_descriptor(struct udp_sink* sink, const char* name, const char* prop)
 {
   GSocket *sock;
-  g_object_get(G_OBJECT(udp), prop, &sock, NULL);
+  g_object_get(G_OBJECT(sink->udp_), prop, &sock, NULL);
   int fd = g_socket_get_fd(sock);
 
-  guint id = g_unix_fd_add(fd, G_IO_IN, on_udp_desc_ready, udp);
+  guint id = g_unix_fd_add(fd, G_IO_IN, on_udp_desc_ready, sink);
   if(id <= 0) {
     log_printf(ERROR, "Error while adding %s file descriptor (%d) to main loop", name, fd);
     return FALSE;
@@ -562,16 +584,16 @@ static gboolean attach_udp_descriptor(GstElement* udp, const char* name, const c
 }
 
 
-static gboolean attach_udp_descriptors(struct udp_elements *udp)
+static gboolean attach_udp_descriptors(struct udp_sinks *sinks)
 {
-  if(!attach_udp_descriptor(udp->rtp_video_, "RTP(video) IPv4", "used-socket") ||
-     !attach_udp_descriptor(udp->rtp_video_, "RTP(video) IPv6", "used-socket-v6") ||
-     !attach_udp_descriptor(udp->rtcp_video_, "RTCP(video) IPv4", "used-socket") ||
-     !attach_udp_descriptor(udp->rtcp_video_, "RTCP(video) IPv6", "used-socket-v6") ||
-     !attach_udp_descriptor(udp->rtp_audio_, "RTP(audio) IPv4", "used-socket") ||
-     !attach_udp_descriptor(udp->rtp_audio_, "RTP(audio) IPv6", "used-socket-v6") ||
-     !attach_udp_descriptor(udp->rtcp_audio_, "RTCP(audio) IPv4", "used-socket") ||
-     !attach_udp_descriptor(udp->rtcp_audio_, "RTCP(audio) IPv6", "used-socket-v6")) {
+  if(!attach_udp_descriptor(&(sinks->rtp_video_), "RTP(video) IPv4", "used-socket") ||
+     !attach_udp_descriptor(&(sinks->rtp_video_), "RTP(video) IPv6", "used-socket-v6") ||
+     !attach_udp_descriptor(&(sinks->rtcp_video_), "RTCP(video) IPv4", "used-socket") ||
+     !attach_udp_descriptor(&(sinks->rtcp_video_), "RTCP(video) IPv6", "used-socket-v6") ||
+     !attach_udp_descriptor(&(sinks->rtp_audio_), "RTP(audio) IPv4", "used-socket") ||
+     !attach_udp_descriptor(&(sinks->rtp_audio_), "RTP(audio) IPv6", "used-socket-v6") ||
+     !attach_udp_descriptor(&(sinks->rtcp_audio_), "RTCP(audio) IPv4", "used-socket") ||
+     !attach_udp_descriptor(&(sinks->rtcp_audio_), "RTCP(audio) IPv6", "used-socket-v6")) {
     return FALSE;
   }
   return TRUE;
@@ -583,7 +605,8 @@ int main_loop(options_t* opt)
 {
   log_printf(INFO, "entering main loop");
 
-  struct udp_elements udp;
+  struct udp_sinks udp = { { NULL, NULL }, { NULL, NULL },
+                           { NULL, NULL }, { NULL, NULL } };
   GstElement *pipeline = create_pipeline(opt, &udp);
   if(!pipeline) {
     log_printf(ERROR, "creating pipeline failed");