IPTV: add pipe:// handler to read MPEG-TS stream from an external program

Makefile

@@ -280,6 +280,7 @@ SRCS-${CONFIG_IPTV} += \
         src/input/mpegts/iptv/iptv_service.c \
         src/input/mpegts/iptv/iptv_http.c \
         src/input/mpegts/iptv/iptv_udp.c \
+        src/input/mpegts/iptv/iptv_pipe.c
 
 # TSfile
 SRCS-$(CONFIG_TSFILE) += \

docs/html/config_muxes.html

@@ -73,6 +73,31 @@
 		<dt>URL
 		<dd>Mux URL.
 
+			<dl>
+				<dt>udp://
+				<dd>Raw MPEG-TS UDP packets
+
+				<dt>rtp://
+				<dd>MPEG-TS UDP packets with RTP header
+
+				<dt>http://
+				<dd>HTTP stream (MPEG-TS)
+
+				<dt>https://
+				<dd>Secure HTTP stream (MPEG-TS)
+
+				<dt>pipe://
+				<dd>Read standard output from an external program.
+				    If the program name does not have the first
+				    character '/', the PATH environment variable
+				    is used to find the program name in all
+				    directories specified by PATH.
+				    Additional arguments may be separated
+				    using spaces. A raw MPEG-TS stream is
+				    expected.
+
+			</dl>
+                
 		<dt># Services
 		<dd>The number of services found on this mux.
 

src/input/mpegts/iptv/iptv.c

@@ -220,6 +220,7 @@ iptv_input_start_mux ( mpegts_input_t *mi, mpegts_mux_instance_t *mmi )
   iptv_mux_t *im = (iptv_mux_t*)mmi->mmi_mux;
   iptv_handler_t *ih;
   char buf[256];
+  const char *scheme;
   url_t url;
 
   /* Already active */
@@ -229,15 +230,25 @@ iptv_input_start_mux ( mpegts_input_t *mi, mpegts_mux_instance_t *mmi )
   /* Parse URL */
   mpegts_mux_nice_name((mpegts_mux_t*)im, buf, sizeof(buf));
   memset(&url, 0, sizeof(url));
-  if (urlparse(im->mm_iptv_url ?: "", &url)) {
-    tvherror("iptv", "%s - invalid URL [%s]", buf, im->mm_iptv_url);
-    return ret;
+
+  if (im->mm_iptv_url && !strncmp(im->mm_iptv_url, "pipe://", 7)) {
+
+    scheme = "pipe";
+
+  } else {
+
+    if (urlparse(im->mm_iptv_url ?: "", &url)) {
+      tvherror("iptv", "%s - invalid URL [%s]", buf, im->mm_iptv_url);
+      return ret;
+    }
+    scheme = url.scheme;
+
   }
 
   /* Find scheme handler */
-  ih = iptv_handler_find(url.scheme ?: "");
+  ih = iptv_handler_find(scheme ?: "");
   if (!ih) {
-    tvherror("iptv", "%s - unsupported scheme [%s]", buf, url.scheme ?: "none");
+    tvherror("iptv", "%s - unsupported scheme [%s]", buf, scheme ?: "none");
     return ret;
   }
 
@@ -245,7 +256,7 @@ iptv_input_start_mux ( mpegts_input_t *mi, mpegts_mux_instance_t *mmi )
   pthread_mutex_lock(&iptv_lock);
   im->mm_active = mmi; // Note: must set here else mux_started call
                        // will not realise we're ready to accept pid open calls
-  ret            = ih->start(im, &url);
+  ret            = ih->start(im, im->mm_iptv_url, &url);
   if (!ret)
     im->im_handler = ih;
   else
@@ -577,6 +588,7 @@ void iptv_init ( void )
   /* Register handlers */
   iptv_http_init();
   iptv_udp_init();
+  iptv_pipe_init();
 
   iptv_input = calloc(1, sizeof(iptv_input_t));
 

src/input/mpegts/iptv/iptv_http.c

@@ -62,7 +62,7 @@ iptv_http_data
  */
 static int
 iptv_http_start
-  ( iptv_mux_t *im, const url_t *u )
+  ( iptv_mux_t *im, const char *raw, const url_t *u )
 {
   http_client_t *hc;
   int r;

src/input/mpegts/iptv/iptv_mux.c

@@ -30,7 +30,7 @@ static int
 iptv_mux_url_set ( void *p, const void *v )
 {
   iptv_mux_t *im = p;
-  const char *str = v;
+  const char *str = v, *x;
   char *buf, port[16] = "";
   size_t len;
   url_t url;
@@ -43,6 +43,17 @@ iptv_mux_url_set ( void *p, const void *v )
       im->mm_iptv_url_sane = NULL;
       return 1;
     }
+    if (!strncmp(str, "pipe://", 7)) {
+      x = str + strlen(str);
+      while (x != str && *x <= ' ')
+        x--;
+      ((char *)x)[1] = '\0';
+      free(im->mm_iptv_url);
+      free(im->mm_iptv_url_sane);
+      im->mm_iptv_url = strdup(str);
+      im->mm_iptv_url_sane = strdup(str);
+      return 1;
+    }
     memset(&url, 0, sizeof(url));
     if (!urlparse(str, &url)) {
       free(im->mm_iptv_url);

src/input/mpegts/iptv/iptv_pipe.c

@@ -0,0 +1,131 @@
+/*
+ *  IPTV - pipe handler
+ *
+ *  Copyright (C) 2014 Jaroslav Kysela
+ *
+ *  This program is free software: you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation, either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "tvheadend.h"
+#include "iptv_private.h"
+#include "spawn.h"
+
+#include <sys/socket.h>
+#include <sys/types.h>
+#include <sys/ioctl.h>
+#include <fcntl.h>
+#include <assert.h>
+
+/*
+ * Connect UDP/RTP
+ */
+static int
+iptv_pipe_start ( iptv_mux_t *im, const char *_raw, const url_t *url )
+{
+  char *argv[32], *f, *raw, *s;
+  int i = 1, rd;
+
+  if (strncmp(_raw, "pipe://", 7))
+    return -1;
+
+  s = raw = tvh_strdupa(_raw + 7);
+
+  argv[0] = NULL;
+
+  while (*s && *s != ' ')
+    s++;
+  if (*s == ' ') {
+    *(char *)s = '\0';
+    s++;
+  }
+
+  while (*s && i < ARRAY_SIZE(argv) - 1) {
+    f = s;
+    while (*s && *s != ' ')
+      s++;
+    if (f != s) {
+      *(char *)s = '\0';
+      argv[i++] = f;
+      s++;
+    }
+  }
+  argv[i] = NULL;
+
+  if (spawn_and_give_stdout(raw, argv, &rd, 1)) {
+    tvherror("iptv", "Unable to start pipe '%s' (wrong executable?)", raw);
+    return -1;
+  }
+
+  fcntl(rd, F_SETFD, fcntl(rd, F_GETFD) | FD_CLOEXEC);
+  fcntl(rd, F_SETFL, fcntl(rd, F_GETFL) | O_NONBLOCK);
+
+  im->mm_iptv_fd = rd;
+
+  iptv_input_mux_started(im);
+  return 0;
+}
+
+static void
+iptv_pipe_stop
+  ( iptv_mux_t *im )
+{
+  int rd = im->mm_iptv_fd;
+  if (rd > 0)
+    close(rd);
+  im->mm_iptv_fd = -1;
+}
+
+static ssize_t
+iptv_pipe_read ( iptv_mux_t *im )
+{
+  int r, rd = im->mm_iptv_fd;
+  ssize_t res = 0;
+  char buf[64*1024];
+
+  while (rd > 0 && res < 1024*1024) {
+    r = read(rd, buf, sizeof(buf));
+    if (r < 0) {
+      if (ERRNO_AGAIN(errno))
+        continue;
+      break;
+    }
+    if (!r) {
+      close(rd);
+      im->mm_iptv_fd = -1;
+      break;
+    }
+    sbuf_append(&im->mm_iptv_buffer, buf, r);
+    res += r;
+  }
+
+  return res;
+}
+
+/*
+ * Initialise pipe handler
+ */
+
+void
+iptv_pipe_init ( void )
+{
+  static iptv_handler_t ih[] = {
+    {
+      .scheme = "pipe",
+      .start  = iptv_pipe_start,
+      .stop   = iptv_pipe_stop,
+      .read   = iptv_pipe_read,
+    },
+  };
+  iptv_handler_register(ih, ARRAY_SIZE(ih));
+}

src/input/mpegts/iptv/iptv_private.h

@@ -40,7 +40,7 @@ typedef struct iptv_handler iptv_handler_t;
 struct iptv_handler
 {
   const char *scheme;
-  int     (*start) ( iptv_mux_t *im, const url_t *url );
+  int     (*start) ( iptv_mux_t *im, const char *raw, const url_t *url );
   void    (*stop)  ( iptv_mux_t *im );
   ssize_t (*read)  ( iptv_mux_t *im );
   
@@ -118,6 +118,7 @@ void iptv_mux_load_all ( void );
 
 void iptv_http_init    ( void );
 void iptv_udp_init     ( void );
+void iptv_pipe_init     ( void );
 
 #endif /* __IPTV_PRIVATE_H__ */
 

src/input/mpegts/iptv/iptv_udp.c

@@ -32,7 +32,7 @@
  * Connect UDP/RTP
  */
 static int
-iptv_udp_start ( iptv_mux_t *im, const url_t *url )
+iptv_udp_start ( iptv_mux_t *im, const char *raw, const url_t *url )
 {
   char name[256];
   udp_connection_t *conn;

src/spawn.c

@@ -136,6 +136,13 @@ find_exec ( const char *name, char *out, size_t len )
   DIR *dir;
   struct dirent *de;
   struct stat st;
+  if (name[0] == '/') {
+    if (lstat(name, &st)) return 0;
+    if (!S_ISREG(st.st_mode) || !(st.st_mode & S_IEXEC)) return 0;
+    strncpy(out, name, len);
+    out[len-1] = '\0';
+    return 1;
+  }
   if (!(path = getenv("PATH"))) return 0;
   path = strdup(path);
   tmp  = strtok_r(path, ":", &tmp2);
@@ -147,6 +154,7 @@ find_exec ( const char *name, char *out, size_t len )
         if (lstat(bin, &st)) continue;
         if (!S_ISREG(st.st_mode) || !(st.st_mode & S_IEXEC)) continue;
         strncpy(out, bin, len);
+        out[len-1] = '\0';
         ret = 1;
         break;
       }