Implement delivery of packets via streaming_pads

diff --git a/packet.c b/packet.c
index 7493af9cd27fbb5be9c957442da6ea06c346d575..583e7802a90b5ae9a82a9fe475b426aed99884e7 100644 (file)
--- a/packet.c
+++ b/packet.c
@@ -88,6 +88,16 @@ pkt_ref_inc(th_pkt_t *pkt)
   pthread_mutex_unlock(&refmutex);
 }
 
+/**
+ *
+ */
+void
+pkt_ref_inc_poly(th_pkt_t *pkt, int n)
+{
+  pthread_mutex_lock(&refmutex);
+  pkt->pkt_refcount += n;
+  pthread_mutex_unlock(&refmutex);
+}
 
 
 /**

diff --git a/packet.h b/packet.h
index 48c2df4ef1a2425e9ef7672ede5b726a2c9f9d4f..a4e4035b586ff5ae5a64ba55df27424ea897e83b 100644 (file)
--- a/packet.h
+++ b/packet.h
@@ -33,7 +33,7 @@ typedef struct th_pkt {
   int pkt_duration;
   int pkt_refcount;
 
-  uint8_t pkt_streamindex;
+  uint8_t pkt_componentindex;
   uint8_t pkt_frametype;
   uint8_t pkt_commercial;
 
@@ -59,6 +59,8 @@ void pkt_ref_dec(th_pkt_t *pkt);
 
 void pkt_ref_inc(th_pkt_t *pkt);
 
+void pkt_ref_inc_poly(th_pkt_t *pkt, int n);
+
 void pktref_clear_queue(struct th_pktref_queue *q);
 
 th_pkt_t *pkt_alloc(void *data, size_t datalen, int64_t pts, int64_t dts);

diff --git a/parsers.c b/parsers.c
index 848b4ec8c6610d2a7e23dd09972f722c2c62ff6d..22ee392cd6d606344bfe432e43db153205abfb4c 100644 (file)
--- a/parsers.c
+++ b/parsers.c
@@ -31,6 +31,7 @@
 #include "bitstream.h"
 #include "packet.h"
 #include "transports.h"
+#include "streaming.h"
 
 static const AVRational mpeg_tc = {1, 90000};
 
@@ -870,16 +871,11 @@ parser_deliver(th_transport_t *t, th_stream_t *st, th_pkt_t *pkt)
    */
   transport_signal_status(t, SUBSCRIPTION_VALID_PACKETS);
 
-  /* Alert all muxers tied to us that a new packet has arrived */
-
-  lock_assert(&t->tht_stream_mutex);
-#if 0
-  LIST_FOREACH(tm, &t->tht_muxers, tm_transport_link)
-    tm->tm_new_pkt(tm, st, pkt);
-#endif
-  /* Unref (and possibly free) the packet, downstream code is supposed
-     to increase refcount or copy packet if they need anything */
+  /* Forward packet */
+  pkt->pkt_componentindex = st->st_sc.sc_index;
+  streaming_pad_deliver_packet(&t->tht_streaming_pad, pkt);
 
+  /* Decrease our own reference to the packet */
   pkt_ref_dec(pkt);
 }
 

diff --git a/streaming.c b/streaming.c
index 616bccd985809f7ce248c86b64797ff195a47353..4c102efb141f9ed81cee75bb53d4d6815cd00178 100644 (file)
--- a/streaming.c
+++ b/streaming.c
@@ -20,6 +20,7 @@
 
 #include "tvhead.h"
 #include "streaming.h"
+#include "packet.h"
 
 void
 streaming_pad_init(streaming_pad_t *sp, pthread_mutex_t *mutex)
@@ -28,3 +29,45 @@ streaming_pad_init(streaming_pad_t *sp, pthread_mutex_t *mutex)
   LIST_INIT(&sp->sp_components);
   sp->sp_mutex = mutex;
 }
+
+/**
+ *
+ */
+void
+streaming_target_connect(streaming_pad_t *sp, streaming_target_t *st)
+{
+  lock_assert(sp->sp_mutex);
+
+  sp->sp_ntargets++;
+  st->st_pad = sp;
+  LIST_INSERT_HEAD(&sp->sp_targets, st, st_link);
+}
+
+/**
+ *
+ */
+void
+streaming_pad_deliver_packet(streaming_pad_t *sp, th_pkt_t *pkt)
+{
+  streaming_target_t *st;
+  th_pktref_t *pr;
+
+  lock_assert(sp->sp_mutex);
+
+  if(sp->sp_ntargets == 0)
+    return;
+
+  /* Increase multiple refcounts at once */
+  pkt_ref_inc_poly(pkt, sp->sp_ntargets);
+
+  LIST_FOREACH(st, &sp->sp_targets, st_link) {
+
+    pr = malloc(sizeof(th_pktref_t));
+    pr->pr_pkt = pkt;
+
+    pthread_mutex_lock(&st->st_mutex);
+    TAILQ_INSERT_TAIL(&st->st_queue, pr, pr_link);
+    pthread_cond_signal(&st->st_cond);
+    pthread_mutex_unlock(&st->st_mutex);
+  }
+}

diff --git a/streaming.h b/streaming.h
index 0ac171a6a20254be95ba39cd9cf76b8ca29f1e93..ee33c275ba53e86b9a70864bf484cdc2735700fc 100644 (file)
--- a/streaming.h
+++ b/streaming.h
@@ -20,7 +20,7 @@
 #define STREAMING_H_
 
 #include "tvhead.h"
-
+#include "packet.h"
 
 /**
  *
@@ -31,4 +31,6 @@ void streaming_target_init(streaming_target_t *st);
 
 void streaming_target_connect(streaming_pad_t *pd, streaming_target_t *st);
 
+void streaming_pad_deliver_packet(streaming_pad_t *sp, th_pkt_t *pkt);
+
 #endif /* STREAMING_H_ */

diff --git a/tvhead.h b/tvhead.h
index cb865cd76389aca1dc0b556c535759ee2a686695..222f1af1c72bdea473a1f496d91ffe153e1f9f33 100644 (file)
--- a/tvhead.h
+++ b/tvhead.h
@@ -147,6 +147,7 @@ typedef struct streaming_component {
  */
 typedef struct streaming_pad {
   struct streaming_target_list sp_targets;
+  int sp_ntargets;
   struct streaming_component_list sp_components;
 
   pthread_mutex_t *sp_mutex; /* Mutex for protecting modification of