SAT>IP Server: more work on the descrambling

src/input/mpegts.h

@@ -87,8 +87,10 @@ struct mpegts_apids {
   int all;
 };
 
-int mpegts_pid_init ( mpegts_apids_t *pids, mpegts_apid_t *vals, int count );
+int mpegts_pid_init ( mpegts_apids_t *pids );
 void mpegts_pid_done ( mpegts_apids_t *pids );
+mpegts_apids_t *mpegts_pid_alloc ( void );
+void mpegts_pid_destroy ( mpegts_apids_t **pids );
 void mpegts_pid_reset ( mpegts_apids_t *pids );
 int mpegts_pid_add ( mpegts_apids_t *pids, mpegts_apid_t pid );
 int mpegts_pid_add_group ( mpegts_apids_t *pids, mpegts_apids_t *vals );
@@ -504,8 +506,19 @@ struct mpegts_service
 {
   service_t; // Parent
 
+  int (*s_update_pids)(mpegts_service_t *t, struct mpegts_apids *pids);
+  int (*s_link)(mpegts_service_t *master, mpegts_service_t *slave);
+  int (*s_unlink)(mpegts_service_t *master, mpegts_service_t *slave);
+
   int      s_dvb_subscription_flags;
 
+  mpegts_apids_t             *s_pids;
+  LIST_HEAD(, mpegts_service) s_masters;
+  LIST_ENTRY(mpegts_service)  s_masters_link;
+  LIST_HEAD(, mpegts_service) s_slaves;
+  LIST_ENTRY(mpegts_service)  s_slaves_link;
+  mpegts_apids_t             *s_slaves_pids;
+
   /*
    * Fields defined by DVB standard EN 300 468
    */

src/input/mpegts/mpegts_input.c

@@ -533,6 +533,7 @@ mpegts_input_open_service ( mpegts_input_t *mi, mpegts_service_t *s, int flags,
 {
   elementary_stream_t *st;
   mpegts_apids_t *pids;
+  mpegts_service_t *s2;
   int i;
 
   /* Add to list */
@@ -545,15 +546,30 @@ mpegts_input_open_service ( mpegts_input_t *mi, mpegts_service_t *s, int flags,
   /* Register PIDs */
   pthread_mutex_lock(&s->s_stream_mutex);
   if (s->s_type == STYPE_STD) {
+
+    pids = mpegts_pid_alloc();
+
     mi->mi_open_pid(mi, s->s_dvb_mux, s->s_pmt_pid, MPS_SERVICE, s);
     mi->mi_open_pid(mi, s->s_dvb_mux, s->s_pcr_pid, MPS_SERVICE, s);
+    mpegts_pid_add(pids, s->s_pmt_pid);
+    mpegts_pid_add(pids, s->s_pcr_pid);
     /* Open only filtered components here */
     TAILQ_FOREACH(st, &s->s_filt_components, es_filt_link) {
       if (st->es_type != SCT_CA) {
         st->es_pid_opened = 1;
         mi->mi_open_pid(mi, s->s_dvb_mux, st->es_pid, MPS_SERVICE, s);
+        mpegts_pid_add(pids, st->es_pid);
       }
     }
+
+    LIST_FOREACH(s2, &s->s_masters, s_masters_link) {
+      pthread_mutex_lock(&s2->s_stream_mutex);
+      mpegts_pid_add_group(s2->s_slaves_pids, pids);
+      pthread_mutex_unlock(&s2->s_stream_mutex);
+    }
+
+    mpegts_pid_destroy(&pids);
+
   } else {
     if ((pids = s->s_pids) != NULL) {
       if (pids->all) {
@@ -583,6 +599,8 @@ void
 mpegts_input_close_service ( mpegts_input_t *mi, mpegts_service_t *s )
 {
   elementary_stream_t *st;
+  mpegts_apids_t *pids;
+  mpegts_service_t *s2;
 
   /* Close PMT table */
   if (s->s_type == STYPE_STD && s->s_pmt_mon)
@@ -599,15 +617,30 @@ mpegts_input_close_service ( mpegts_input_t *mi, mpegts_service_t *s )
   /* Close PID */
   pthread_mutex_lock(&s->s_stream_mutex);
   if (s->s_type == STYPE_STD) {
+
+    pids = mpegts_pid_alloc();
+
     mi->mi_close_pid(mi, s->s_dvb_mux, s->s_pmt_pid, MPS_SERVICE, s);
     mi->mi_close_pid(mi, s->s_dvb_mux, s->s_pcr_pid, MPS_SERVICE, s);
+    mpegts_pid_del(pids, s->s_pmt_pid);
+    mpegts_pid_del(pids, s->s_pcr_pid);
     /* Close all opened PIDs (the component filter may be changed at runtime) */
     TAILQ_FOREACH(st, &s->s_components, es_link) {
       if (st->es_pid_opened) {
         st->es_pid_opened = 0;
         mi->mi_close_pid(mi, s->s_dvb_mux, st->es_pid, MPS_SERVICE, s);
+        mpegts_pid_del(pids, st->es_pid);
       }
     }
+
+    LIST_FOREACH(s2, &s->s_masters, s_masters_link) {
+      pthread_mutex_lock(&s2->s_stream_mutex);
+      mpegts_pid_del_group(s2->s_slaves_pids, pids);
+      pthread_mutex_unlock(&s2->s_stream_mutex);
+    }
+
+    mpegts_pid_destroy(&pids);
+
   } else {
     mpegts_input_close_pids(mi, s->s_dvb_mux, s);
   }

src/input/mpegts/mpegts_pid.c

@@ -20,21 +20,10 @@
 #include "input.h"
 
 int
-mpegts_pid_init(mpegts_apids_t *pids, mpegts_apid_t *vals, int count)
+mpegts_pid_init(mpegts_apids_t *pids)
 {
-  int alloc = count + 32;
-  mpegts_apid_t *p = calloc(alloc, sizeof(*pids));
-
-  if (p == NULL)
-    return -1;
-  pids->pids = p;
-  pids->alloc = alloc;
-  pids->count = 0;
-  pids->all = 0;
-  if (vals) {
-    memcpy(p, vals, count * sizeof(*pids));
-    pids->count = count;
-  }
+  assert(pids);
+  memset(pids, 0, sizeof(*pids));
   return 0;
 }
 
@@ -48,6 +37,22 @@ mpegts_pid_done(mpegts_apids_t *pids)
   pids->alloc = pids->count = 0;
 }
 
+mpegts_apids_t *
+mpegts_pid_alloc(void)
+{
+  return calloc(1, sizeof(mpegts_apids_t));
+}
+
+void
+mpegts_pid_destroy(mpegts_apids_t **pids)
+{
+  if (pids) {
+    mpegts_pid_done(*pids);
+    free(*pids);
+    *pids = NULL;
+  }
+}
+
 void
 mpegts_pid_reset(mpegts_apids_t *pids)
 {
@@ -165,8 +170,7 @@ mpegts_pid_compare(mpegts_apids_t *dst, mpegts_apids_t *src,
   assert(dst);
   assert(add);
   assert(del);
-  if (mpegts_pid_init(add, NULL, 0) ||
-      mpegts_pid_init(del, NULL, 0))
+  if (mpegts_pid_init(add) || mpegts_pid_init(del))
     return -1;
   if (src == NULL) {
     mpegts_pid_copy(add, dst);

src/input/mpegts/mpegts_service.c

@@ -515,6 +515,14 @@ mpegts_service_delete ( service_t *t, int delconf )
     LIST_REMOVE(ms, s_dvb_mux_link);
   sbuf_free(&ms->s_tsbuf);
 
+  /* Remove master/slave linking */
+  LIST_SAFE_REMOVE(ms, s_masters_link);
+  LIST_SAFE_REMOVE(ms, s_slaves_link);
+
+  /* Remove PID lists */
+  mpegts_pid_destroy(&ms->s_pids);
+  mpegts_pid_destroy(&ms->s_slaves_pids);
+
   // Note: the ultimate deletion and removal from the idnode list
   //       is done in service_destroy
 }
@@ -647,19 +655,18 @@ mpegts_service_raw_setsourceinfo(service_t *t, source_info_t *si)
 }
 
 static int
-mpegts_service_raw_update_pids(service_t *t, mpegts_apids_t *pids)
+mpegts_service_raw_update_pids(mpegts_service_t *t, mpegts_apids_t *pids)
 {
-  mpegts_service_t *ms = (mpegts_service_t *)t;
-  mpegts_input_t *mi = ms->s_dvb_active_input;
-  mpegts_mux_t *mm = ms->s_dvb_mux;
+  mpegts_input_t *mi = t->s_dvb_active_input;
+  mpegts_mux_t *mm = t->s_dvb_mux;
   mpegts_apids_t *p, *x;
   mpegts_apids_t add, del;
   int i;
 
   lock_assert(&global_lock);
   if (pids) {
     p = calloc(1, sizeof(*p));
-    mpegts_pid_init(p, NULL, 0);
+    mpegts_pid_init(p);
     mpegts_pid_copy(p, pids);
   } else
     p = NULL;
@@ -702,6 +709,28 @@ mpegts_service_raw_update_pids(service_t *t, mpegts_apids_t *pids)
   return 0;
 }
 
+static int
+mpegts_service_link ( mpegts_service_t *master, mpegts_service_t *slave )
+{
+  pthread_mutex_lock(&master->s_stream_mutex);
+  assert(slave->s_status == SERVICE_IDLE);
+  LIST_INSERT_HEAD(&slave->s_masters, master, s_masters_link);
+  LIST_INSERT_HEAD(&master->s_slaves, slave, s_slaves_link);
+  pthread_mutex_unlock(&master->s_stream_mutex);
+  return 0;
+}
+
+static int
+mpegts_service_unlink ( mpegts_service_t *master, mpegts_service_t *slave )
+{
+  pthread_mutex_lock(&master->s_stream_mutex);
+  assert(slave->s_status == SERVICE_IDLE);
+  LIST_SAFE_REMOVE(master, s_masters_link);
+  LIST_SAFE_REMOVE(slave, s_slaves_link);
+  pthread_mutex_unlock(&master->s_stream_mutex);
+  return 0;
+}
+
 mpegts_service_t *
 mpegts_service_create_raw ( mpegts_mux_t *mm )
 {
@@ -736,6 +765,8 @@ mpegts_service_create_raw ( mpegts_mux_t *mm )
   s->s_channel_icon   = mpegts_service_channel_icon;
   s->s_mapped         = mpegts_service_mapped;
   s->s_update_pids    = mpegts_service_raw_update_pids;
+  s->s_link           = mpegts_service_link;
+  s->s_unlink         = mpegts_service_unlink;
 
   pthread_mutex_lock(&s->s_stream_mutex);
   free(s->s_nicename);

src/input/mpegts/tsdemux.c

@@ -51,6 +51,7 @@ static void
 ts_recv_packet0
   (mpegts_service_t *t, elementary_stream_t *st, const uint8_t *tsb)
 {
+  mpegts_service_t *m;
   int off, pusi, cc, error;
 
   service_set_streaming_status_flags((service_t*)t, TSS_MUX_PACKETS);
@@ -84,6 +85,13 @@ ts_recv_packet0
   if(streaming_pad_probe_type(&t->s_streaming_pad, SMT_MPEGTS))
     ts_remux(t, tsb, error);
 
+  LIST_FOREACH(m, &t->s_masters, s_masters_link) {
+    pthread_mutex_lock(&t->s_stream_mutex);
+    if(streaming_pad_probe_type(&t->s_streaming_pad, SMT_MPEGTS))
+      ts_remux(t, tsb, error);
+    pthread_mutex_unlock(&t->s_stream_mutex);
+  }
+
   off = tsb[3] & 0x20 ? tsb[4] + 5 : 4;
 
   switch(st->es_type) {
@@ -220,19 +228,20 @@ ts_recv_packet2(mpegts_service_t *t, const uint8_t *tsb)
 void
 ts_recv_raw(mpegts_service_t *t, const uint8_t *tsb)
 {
-  elementary_stream_t *st = NULL;
-  int pid;
+  int pid, parent = 0;
 
   pthread_mutex_lock(&t->s_stream_mutex);
   service_set_streaming_status_flags((service_t*)t, TSS_MUX_PACKETS);
-  if (t->s_parent) {
-    /* If PID is owned by parent, let parent service to
+  if (!LIST_EMPTY(&t->s_slaves)) {
+    /* If PID is owned by a slave service, let parent service to
      * deliver this PID (decrambling)
      */
     pid = (tsb[1] & 0x1f) << 8 | tsb[2];
-    st = service_stream_find(t->s_parent, pid);
+    parent = mpegts_pid_exists(t->s_slaves_pids, pid);
+    service_set_streaming_status_flags((service_t*)t, TSS_PACKETS);
+    t->s_streaming_live |= TSS_LIVE;
   }
-  if(st == NULL) {
+  if(!parent) {
     if (streaming_pad_probe_type(&t->s_streaming_pad, SMT_MPEGTS))
       ts_remux(t, tsb, 0);
     else {

src/profile.c

@@ -729,19 +729,21 @@ profile_chain_open(profile_chain_t *prch,
  *
  */
 int
-profile_chain_raw_open(profile_chain_t *prch, void *id, size_t qsize)
+profile_chain_raw_open(profile_chain_t *prch, void *id, size_t qsize, int muxer)
 {
   muxer_config_t c;
 
-  memset(&c, 0, sizeof(c));
-  c.m_type = MC_RAW;
   memset(prch, 0, sizeof(*prch));
   prch->prch_id    = id;
   prch->prch_flags = SUBSCRIPTION_RAW_MPEGTS;
   streaming_queue_init(&prch->prch_sq, SMT_PACKET, qsize);
   prch->prch_sq_used = 1;
   prch->prch_st    = &prch->prch_sq.sq_st;
-  prch->prch_muxer = muxer_create(&c);
+  if (muxer) {
+    memset(&c, 0, sizeof(c));
+    c.m_type = MC_RAW;
+    prch->prch_muxer = muxer_create(&c);
+  }
   return 0;
 }
 

src/profile.h

@@ -144,7 +144,7 @@ int
 profile_chain_open(profile_chain_t *prch,
                    muxer_config_t *m_cfg, int flags, size_t qsize);
 void profile_chain_init(profile_chain_t *prch, profile_t *pro, void *id);
-int  profile_chain_raw_open(profile_chain_t *prch, void *id, size_t qsize);
+int  profile_chain_raw_open(profile_chain_t *prch, void *id, size_t qsize, int muxer);
 void profile_chain_close(profile_chain_t *prch);
 
 static inline profile_t *profile_find_by_uuid(const char *uuid)

src/queue.h

@@ -30,6 +30,12 @@
 #define	LIST_NEXT(elm, field)		((elm)->field.le_next)
 #endif
 
+#ifndef LIST_SAFE_REMOVE
+#define LIST_SAFE_REMOVE(elm, field) \
+        if ((elm)->field.le_next != NULL || (elm)->field.le_prev != NULL) \
+                LIST_REMOVE(elm, field)
+#endif
+
 #ifndef LIST_INSERT_BEFORE
 #define	LIST_INSERT_BEFORE(listelm, elm, field) do {			\
 	(elm)->field.le_prev = (listelm)->field.le_prev;		\

src/satip/rtp.c

@@ -260,7 +260,7 @@ void satip_rtp_queue(void *id, th_subscription_t *subs,
   rtp->fd_rtcp = fd_rtcp;
   rtp->subs = subs;
   rtp->sq = sq;
-  mpegts_pid_init(&rtp->pids, NULL, pids->count);
+  mpegts_pid_init(&rtp->pids);
   mpegts_pid_copy(&rtp->pids, pids);
   udp_multisend_init(&rtp->um, RTP_PACKETS, RTP_PAYLOAD, &rtp->um_iovec);
   satip_rtp_header(rtp);