streaming: change streaming_pad_deliver() semantics

diff --git a/src/input/mpegts/linuxdvb/linuxdvb_frontend.c b/src/input/mpegts/linuxdvb/linuxdvb_frontend.c
index 572d01bc43a9f0881ddb1ee42bda01e0431e6bfb..b8c51ad9bc75183e6113587bd81f223e55126613 100644 (file)
--- a/src/input/mpegts/linuxdvb/linuxdvb_frontend.c
+++ b/src/input/mpegts/linuxdvb/linuxdvb_frontend.c
@@ -793,7 +793,7 @@ linuxdvb_frontend_monitor ( void *aux )
   sm.sm_data = &sigstat;
   LIST_FOREACH(s, &lfe->mi_transports, s_active_link) {
     pthread_mutex_lock(&s->s_stream_mutex);
-    streaming_pad_deliver(&s->s_streaming_pad, &sm);
+    streaming_pad_deliver(&s->s_streaming_pad, streaming_msg_clone(&sm));
     pthread_mutex_unlock(&s->s_stream_mutex);
   }
 }

diff --git a/src/input/mpegts/mpegts_input.c b/src/input/mpegts/mpegts_input.c
index cb52b91977887a2c2a8f3588abaaf9632350cbba..bba6539134d945af29bdfbe5350bc3499cf5fe50 100644 (file)
--- a/src/input/mpegts/mpegts_input.c
+++ b/src/input/mpegts/mpegts_input.c
@@ -814,7 +814,7 @@ done:
     memset(&sm, 0, sizeof(sm));
     sm.sm_type = SMT_MPEGTS;
     sm.sm_data = pb;
-    streaming_pad_deliver(&mmi->mmi_streaming_pad, &sm);
+    streaming_pad_deliver(&mmi->mmi_streaming_pad, streaming_msg_clone(&sm));
     pktbuf_ref_dec(pb);
   }
 

diff --git a/src/input/mpegts/satip/satip_frontend.c b/src/input/mpegts/satip/satip_frontend.c
index 686ca576400742feb026944cc71b95bfea4dc987..56bab05d30237f10ce15e85a83eebe9db72e82f3 100644 (file)
--- a/src/input/mpegts/satip/satip_frontend.c
+++ b/src/input/mpegts/satip/satip_frontend.c
@@ -1345,7 +1345,7 @@ satip_frontend_signal_cb( void *aux )
   sm.sm_data = &sigstat;
   LIST_FOREACH(svc, &lfe->mi_transports, s_active_link) {
     pthread_mutex_lock(&svc->s_stream_mutex);
-    streaming_pad_deliver(&svc->s_streaming_pad, &sm);
+    streaming_pad_deliver(&svc->s_streaming_pad, streaming_msg_clone(&sm));
     pthread_mutex_unlock(&svc->s_stream_mutex);
   }
   gtimer_arm_ms(&lfe->sf_monitor_timer, satip_frontend_signal_cb, lfe, 250);

diff --git a/src/input/mpegts/tsdemux.c b/src/input/mpegts/tsdemux.c
index 57883dad0e324365a32216091cdc9cce1eed9b74..2b4f7e07212856f5a85e499fff736a31c510cdc6 100644 (file)
--- a/src/input/mpegts/tsdemux.c
+++ b/src/input/mpegts/tsdemux.c
@@ -283,7 +283,7 @@ ts_remux(mpegts_service_t *t, const uint8_t *src)
 
   sm.sm_type = SMT_MPEGTS;
   sm.sm_data = pb;
-  streaming_pad_deliver(&t->s_streaming_pad, &sm);
+  streaming_pad_deliver(&t->s_streaming_pad, streaming_msg_clone(&sm));
 
   pktbuf_ref_dec(pb);
 

diff --git a/src/input/mpegts/tvhdhomerun/tvhdhomerun_frontend.c b/src/input/mpegts/tvhdhomerun/tvhdhomerun_frontend.c
index ace952346f6dea8ed6d184d15bce5b3d86baf390..07fcc89436bd76569b4200b97c08a160a83bced7 100644 (file)
--- a/src/input/mpegts/tvhdhomerun/tvhdhomerun_frontend.c
+++ b/src/input/mpegts/tvhdhomerun/tvhdhomerun_frontend.c
@@ -333,7 +333,7 @@ tvhdhomerun_frontend_monitor_cb( void *aux )
 
   LIST_FOREACH(svc, &hfe->mi_transports, s_active_link) {
     pthread_mutex_lock(&svc->s_stream_mutex);
-    streaming_pad_deliver(&svc->s_streaming_pad, &sm);
+    streaming_pad_deliver(&svc->s_streaming_pad, streaming_msg_clone(&sm));
     pthread_mutex_unlock(&svc->s_stream_mutex);
   }
 }

diff --git a/src/parsers/parser_teletext.c b/src/parsers/parser_teletext.c
index a492ecc27ec9ef27e3e7cc66fc51b3305f3a7241..54ce573ba78e38c94146885042e0d337ed36824c 100644 (file)
--- a/src/parsers/parser_teletext.c
+++ b/src/parsers/parser_teletext.c
@@ -801,9 +801,7 @@ extract_subtitle(mpegts_service_t *t, elementary_stream_t *st,
   th_pkt_t *pkt = pkt_alloc(sub, off, pts, pts);
   pkt->pkt_componentindex = st->es_index;
 
-  streaming_message_t *sm = streaming_msg_create_pkt(pkt);
-  streaming_pad_deliver(&t->s_streaming_pad, sm);
-  streaming_msg_free(sm);
+  streaming_pad_deliver(&t->s_streaming_pad, streaming_msg_create_pkt(pkt));
 
   /* Decrease our own reference to the packet */
   pkt_ref_dec(pkt);

diff --git a/src/parsers/parsers.c b/src/parsers/parsers.c
index 77ed6e61de62910d24b5d5ed0956da415ed3a316..1a590f9876a55e89ec27d1aa1d045296e15786c9 100644 (file)
--- a/src/parsers/parsers.c
+++ b/src/parsers/parsers.c
@@ -1408,10 +1408,7 @@ parser_deliver(service_t *t, elementary_stream_t *st, th_pkt_t *pkt, int error)
   /* Forward packet */
   pkt->pkt_componentindex = st->es_index;
 
-  streaming_message_t *sm = streaming_msg_create_pkt(pkt);
-
-  streaming_pad_deliver(&t->s_streaming_pad, sm);
-  streaming_msg_free(sm);
+  streaming_pad_deliver(&t->s_streaming_pad, streaming_msg_create_pkt(pkt));
 
   /* Decrease our own reference to the packet */
   pkt_ref_dec(pkt);

diff --git a/src/service.c b/src/service.c
index 7c51c8112d65a50fbde2091701fc82e2de1bb2de..0783cc785e5f89e6ea79e50cde8ae0f6b060a422 100644 (file)
--- a/src/service.c
+++ b/src/service.c
@@ -1117,7 +1117,6 @@ service_servicetype_txt ( service_t *s )
 void
 service_set_streaming_status_flags_(service_t *t, int set)
 {
-  streaming_message_t *sm;
   lock_assert(&t->s_stream_mutex);
 
   if(set == t->s_streaming_status)
@@ -1136,10 +1135,9 @@ service_set_streaming_status_flags_(service_t *t, int set)
         set & TSS_GRACEPERIOD    ? "[Graceperiod expired] " : "",
         set & TSS_TIMEOUT        ? "[Data timeout] " : "");
 
-  sm = streaming_msg_create_code(SMT_SERVICE_STATUS,
-                                t->s_streaming_status);
-  streaming_pad_deliver(&t->s_streaming_pad, sm);
-  streaming_msg_free(sm);
+  streaming_pad_deliver(&t->s_streaming_pad,
+                        streaming_msg_create_code(SMT_SERVICE_STATUS,
+                                                  t->s_streaming_status));
 
   pthread_cond_broadcast(&t->s_tss_cond);
 }
@@ -1153,22 +1151,20 @@ service_set_streaming_status_flags_(service_t *t, int set)
 void
 service_restart(service_t *t, int had_components)
 {
-  streaming_message_t *sm;
   pthread_mutex_lock(&t->s_stream_mutex);
 
   if(had_components) {
-    sm = streaming_msg_create_code(SMT_STOP, SM_CODE_SOURCE_RECONFIGURED);
-    streaming_pad_deliver(&t->s_streaming_pad, sm);
-    streaming_msg_free(sm);
+    streaming_pad_deliver(&t->s_streaming_pad,
+                          streaming_msg_create_code(SMT_STOP,
+                                                    SM_CODE_SOURCE_RECONFIGURED));
   }
 
   service_build_filter(t);
 
   if(TAILQ_FIRST(&t->s_filt_components) != NULL) {
-    sm = streaming_msg_create_data(SMT_START, 
-                                  service_build_stream_start(t));
-    streaming_pad_deliver(&t->s_streaming_pad, sm);
-    streaming_msg_free(sm);
+    streaming_pad_deliver(&t->s_streaming_pad,
+                          streaming_msg_create_data(SMT_START,
+                                                    service_build_stream_start(t)));
   }
 
   pthread_mutex_unlock(&t->s_stream_mutex);

diff --git a/src/streaming.c b/src/streaming.c
index 3cabff9cfac3844c37423bbd622ec8f745c9f9bb..9204689f69cb6693643461ed7e77ea4a49aaf61f 100644 (file)
--- a/src/streaming.c
+++ b/src/streaming.c
@@ -316,10 +316,10 @@ streaming_msg_free(streaming_message_t *sm)
 void
 streaming_target_deliver2(streaming_target_t *st, streaming_message_t *sm)
 {
-  if(st->st_reject_filter & SMT_TO_MASK(sm->sm_type))
+  if (st->st_reject_filter & SMT_TO_MASK(sm->sm_type))
     streaming_msg_free(sm);
   else
-    st->st_cb(st->st_opaque, sm);
+    streaming_target_deliver(st, sm);
 }
 
 /**
@@ -328,18 +328,23 @@ streaming_target_deliver2(streaming_target_t *st, streaming_message_t *sm)
 void
 streaming_pad_deliver(streaming_pad_t *sp, streaming_message_t *sm)
 {
-  streaming_target_t *st, *next;
+  streaming_target_t *st, *next, *run = NULL;
 
-  for(st = LIST_FIRST(&sp->sp_targets);st; st = next) {
+  for (st = LIST_FIRST(&sp->sp_targets); st; st = next) {
     next = LIST_NEXT(st, st_link);
     assert(next != st);
-    if(st->st_reject_filter & SMT_TO_MASK(sm->sm_type))
+    if (st->st_reject_filter & SMT_TO_MASK(sm->sm_type))
       continue;
-    st->st_cb(st->st_opaque, streaming_msg_clone(sm));
+    if (run)
+      streaming_target_deliver(run, streaming_msg_clone(sm));
+    run = st;
   }
+  if (run)
+    streaming_target_deliver(run, sm);
+  else
+    streaming_msg_free(sm);
 }
 
-
 /**
  *
  */

diff --git a/src/streaming.h b/src/streaming.h
index 8f1cde8091d1f22160db6d242f4b43ba3667f032..c7c5f0e2a069b72ed5a1cab8b4671481fbf8b701 100644 (file)
--- a/src/streaming.h
+++ b/src/streaming.h
@@ -101,7 +101,9 @@ streaming_message_t *streaming_msg_create_code(streaming_message_type_t type,
 
 streaming_message_t *streaming_msg_create_pkt(th_pkt_t *pkt);
 
-#define streaming_target_deliver(st, sm) ((st)->st_cb((st)->st_opaque, (sm)))
+static inline void
+streaming_target_deliver(streaming_target_t *st, streaming_message_t *sm)
+  { st->st_cb(st->st_opaque, sm); }
 
 void streaming_target_deliver2(streaming_target_t *st, streaming_message_t *sm);
 

diff --git a/src/subscriptions.c b/src/subscriptions.c
index 99bfa48ac978b2a398359999a7239059a7a297f6..23ee942f65040149d892458b183aa8e918b426a2 100644 (file)
--- a/src/subscriptions.c
+++ b/src/subscriptions.c
@@ -89,9 +89,10 @@ subscription_link_service(th_subscription_t *s, service_t *t)
   // Link to service output
   streaming_target_connect(&t->s_streaming_pad, &s->ths_input);
 
-  sm = streaming_msg_create_code(SMT_GRACE, s->ths_postpone + t->s_grace_delay);
-  streaming_pad_deliver(&t->s_streaming_pad, sm);
-  streaming_msg_free(sm);
+  streaming_pad_deliver(&t->s_streaming_pad,
+                        streaming_msg_create_code(SMT_GRACE,
+                                                  s->ths_postpone +
+                                                    t->s_grace_delay));
 
   if(s->ths_start_message != NULL && t->s_streaming_status & TSS_PACKETS) {