listening for incoming data from udp sink worsk now

1 files changed, 80 insertions, 19 deletions

diff --git a/src/sydra.c b/src/sydra.c
index e751cbe..41716de 100644
--- a/src/sydra.c
+++ b/src/sydra.c
@@ -38,8 +38,9 @@
 #include <stdio.h>
 #include <errno.h>
 #include <string.h>
-#include <sys/select.h>
-#include <time.h>
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <netdb.h>
 
 #include <glib.h>
 #include <glib-unix.h>
@@ -338,22 +339,6 @@ static gboolean create_udp_elements(options_t* opt, GstElement* pipeline, GstEle
   return TRUE;
 }
 
-static void get_udp_descriptors(struct udp_elements *udp)
-{
-  GSocket *sock;
-  g_object_get(G_OBJECT(udp->rtp_video_), "used-socket", &sock, NULL);
-  int rtp_video = g_socket_get_fd(sock);
-  g_object_get(G_OBJECT(udp->rtcp_video_), "used-socket", &sock, NULL);
-  int rtcp_video = g_socket_get_fd(sock);
-  g_object_get(G_OBJECT(udp->rtp_audio_), "used-socket", &sock, NULL);
-  int rtp_audio = g_socket_get_fd(sock);
-  g_object_get(G_OBJECT(udp->rtcp_audio_), "used-socket", &sock, NULL);
-  int rtcp_audio = g_socket_get_fd(sock);
-
-  log_printf(DEBUG, "UDP file descriptors: video RTP=%d RTCP=%d, audio RTP=%d RTCP=%d",
-             rtp_video, rtcp_video, rtp_audio, rtcp_audio);
-}
-
 static gboolean create_preview_elements(const char* preview_bin_desc, GstElement* pipeline, GstElement* tee)
 {
   GstElement *qr = sydra_create_element("queue", NULL);
@@ -481,6 +466,81 @@ static GstElement* create_pipeline(options_t* opt, struct udp_elements *udp)
 }
 
 
+static char* sockaddr_to_string(struct sockaddr_storage *addr, socklen_t addrlen, char* buf, int buflen)
+{
+  if(!buf)
+    return NULL;
+
+  if(!addr)
+    return strncpy(buf, "<null>", buflen);
+
+  char addrstr[INET6_ADDRSTRLEN + 1], portstr[6];
+  char addrport_sep = ':';
+
+  switch(addr->ss_family)
+  {
+  case AF_INET: addrport_sep = ':'; break;
+  case AF_INET6: addrport_sep = '.'; break;
+  default: return strncpy(buf, "<unknown address type>", buflen);
+  }
+
+  int errcode  = getnameinfo((struct sockaddr *)addr, addrlen, addrstr, sizeof(addrstr), portstr, sizeof(portstr), NI_NUMERICHOST | NI_NUMERICSERV);
+  if (errcode != 0) strncpy(buf, "<getnameinfo error>", buflen);
+  int len = snprintf(buf, buflen, "%s%c%s", addrstr, addrport_sep ,portstr);
+  if(len == -1) return strncpy(buf, "<snprintf error>", buflen);
+
+  return buf;
+}
+
+static gboolean on_udp_desc_ready(gint fd, GIOCondition cond, gpointer user_data)
+{
+  log_printf(DEBUG, "%d is ready for reading", fd);
+
+  struct sockaddr_storage addr;
+  socklen_t addrlen = sizeof(addr);
+  char buf[2048];
+  ssize_t bytes = recvfrom(fd, buf, sizeof(buf), 0, (struct sockaddr *)&addr, &addrlen);
+  if(bytes < 1) {
+    if(errno == EINTR)
+      return TRUE;
+
+    log_printf(WARNING, "Error while receiving UDP data on fd %d, will remove callback", fd);
+    return FALSE;
+  }
+
+  log_printf(DEBUG, "received %d bytes from %s\n", bytes, sockaddr_to_string(&addr, addrlen, buf, sizeof(buf)));
+
+  return TRUE;
+}
+
+static gboolean attach_udp_descriptor(GstElement* udp, const char* name)
+{
+  GSocket *sock;
+  g_object_get(G_OBJECT(udp), "used-socket", &sock, NULL);
+  int fd = g_socket_get_fd(sock);
+
+  guint id = g_unix_fd_add(fd, G_IO_IN, on_udp_desc_ready, udp);
+  if(id <= 0) {
+    log_printf(ERROR, "Error while adding %s file descriptor (%d) to main loop", name, fd);
+    return FALSE;
+  }
+  log_printf(DEBUG, "successfully installed callback for %s (fd: %d) for reading (id: %d)", name, fd, id);
+  return TRUE;
+}
+
+
+static gboolean attach_udp_descriptors(struct udp_elements *udp)
+{
+  if(!attach_udp_descriptor(udp->rtp_video_, "RTP(video)") ||
+     !attach_udp_descriptor(udp->rtcp_video_, "RTCP(video)") ||
+     !attach_udp_descriptor(udp->rtp_audio_, "RTP(audio)") ||
+     !attach_udp_descriptor(udp->rtcp_audio_, "RTCP(audio)")) {
+    return FALSE;
+  }
+  return TRUE;
+}
+
+
 int main_loop(options_t* opt)
 {
   log_printf(INFO, "entering main loop");
@@ -507,7 +567,8 @@ int main_loop(options_t* opt)
   log_printf(INFO, "Set State: Paused");
   gst_element_set_state(pipeline, GST_STATE_PAUSED);
 
-  get_udp_descriptors(&udp);
+  if(!attach_udp_descriptors(&udp))
+    return -1;
 
   log_printf(INFO, "Set State: Playing");
   gst_element_set_state(pipeline, GST_STATE_PLAYING);