webrtcbin: fix DTLS when receivebin is set to DROP

Regression introduced by b4bdcf15b7237eb5c5943c4f746701a477333000

This commit prevents the handshake from reaching dtlsdec when
the receive state of the receive bin is set to DROP (for example
when transceivers are sendonly).

This preserves the intent of the commit, by blocking the bin
at its sinks until the receive state is no longer BLOCK, but
makes sure the handshake still goes through, by only dropping
data at the src pads, as was the case before.

diff --git a/ext/webrtc/transportreceivebin.c b/ext/webrtc/transportreceivebin.c
index deeb9fb..d059fa2 100644 (file)
--- a/ext/webrtc/transportreceivebin.c
+++ b/ext/webrtc/transportreceivebin.c
@@ -56,7 +56,8 @@ GST_DEBUG_CATEGORY_STATIC (GST_CAT_DEFAULT);
 G_DEFINE_TYPE_WITH_CODE (TransportReceiveBin, transport_receive_bin,
     GST_TYPE_BIN,
     GST_DEBUG_CATEGORY_INIT (gst_webrtc_transport_receive_bin_debug,
-        "webrtctransportreceivebin", 0, "webrtctransportreceivebin"););
+        "webrtctransportreceivebin", 0, "webrtctransportreceivebin");
+    );
 
 static GstStaticPadTemplate rtp_sink_template =
 GST_STATIC_PAD_TEMPLATE ("rtp_src",
@@ -100,21 +101,31 @@ _receive_state_to_string (ReceiveState state)
 static GstPadProbeReturn
 pad_block (GstPad * pad, GstPadProbeInfo * info, TransportReceiveBin * receive)
 {
-  GstPadProbeReturn ret;
-
   g_mutex_lock (&receive->pad_block_lock);
   while (receive->receive_state == RECEIVE_STATE_BLOCK) {
     g_cond_wait (&receive->pad_block_cond, &receive->pad_block_lock);
     GST_DEBUG_OBJECT (pad, "probe waited. new state %s",
         _receive_state_to_string (receive->receive_state));
   }
-  ret = GST_PAD_PROBE_PASS;
 
-  if (receive->receive_state == RECEIVE_STATE_DROP) {
-    ret = GST_PAD_PROBE_DROP;
-  } else if (receive->receive_state == RECEIVE_STATE_PASS) {
-    ret = GST_PAD_PROBE_OK;
-  }
+  g_mutex_unlock (&receive->pad_block_lock);
+
+  return GST_PAD_PROBE_OK;
+}
+
+static GstPadProbeReturn
+src_probe_cb (GstPad * pad, GstPadProbeInfo * info,
+    TransportReceiveBin * receive)
+{
+  GstPadProbeReturn ret;
+
+  g_mutex_lock (&receive->pad_block_lock);
+
+  g_assert (receive->receive_state != RECEIVE_STATE_BLOCK);
+
+  ret =
+      receive->receive_state ==
+      RECEIVE_STATE_DROP ? GST_PAD_PROBE_DROP : GST_PAD_PROBE_OK;
 
   g_mutex_unlock (&receive->pad_block_lock);
 
@@ -208,6 +219,10 @@ transport_receive_bin_change_state (GstElement * element,
           (GstPadProbeCallback) pad_block, receive, NULL);
       gst_object_unref (pad);
 
+      receive->rtp_src_probe_id = gst_pad_add_probe (receive->rtp_src,
+          GST_PAD_PROBE_TYPE_ALL_BOTH, (GstPadProbeCallback) src_probe_cb,
+          receive, NULL);
+
       transport = receive->stream->rtcp_transport;
       dtlssrtpdec = transport->dtlssrtpdec;
       pad = gst_element_get_static_pad (dtlssrtpdec, "sink");
@@ -218,6 +233,10 @@ transport_receive_bin_change_state (GstElement * element,
           (GstPadProbeCallback) pad_block, receive, NULL);
       gst_object_unref (pad);
 
+      receive->rtcp_src_probe_id = gst_pad_add_probe (receive->rtcp_src,
+          GST_PAD_PROBE_TYPE_ALL_BOTH, (GstPadProbeCallback) src_probe_cb,
+          receive, NULL);
+
       /* XXX: because nice needs the nicesrc internal main loop running in order
        * correctly STUN... */
       /* FIXME: this races with the pad exposure later and may get not-linked */
@@ -251,9 +270,19 @@ transport_receive_bin_change_state (GstElement * element,
       if (receive->rtp_block)
         _free_pad_block (receive->rtp_block);
       receive->rtp_block = NULL;
+
+      if (receive->rtp_src_probe_id)
+        gst_pad_remove_probe (receive->rtp_src, receive->rtp_src_probe_id);
+      receive->rtp_src_probe_id = 0;
+
       if (receive->rtcp_block)
         _free_pad_block (receive->rtcp_block);
       receive->rtcp_block = NULL;
+
+      if (receive->rtcp_src_probe_id)
+        gst_pad_remove_probe (receive->rtcp_src, receive->rtcp_src_probe_id);
+      receive->rtcp_src_probe_id = 0;
+
       break;
     }
     default:
@@ -347,9 +376,9 @@ transport_receive_bin_constructed (GObject * object)
     g_warn_if_reached ();
 
   pad = gst_element_get_static_pad (funnel, "src");
-  ghost = gst_ghost_pad_new ("rtp_src", pad);
+  receive->rtp_src = gst_ghost_pad_new ("rtp_src", pad);
 
-  gst_element_add_pad (GST_ELEMENT (receive), ghost);
+  gst_element_add_pad (GST_ELEMENT (receive), receive->rtp_src);
   gst_object_unref (pad);
 
   /* create funnel for rtcp_src */
@@ -363,8 +392,8 @@ transport_receive_bin_constructed (GObject * object)
     g_warn_if_reached ();
 
   pad = gst_element_get_static_pad (funnel, "src");
-  ghost = gst_ghost_pad_new ("rtcp_src", pad);
-  gst_element_add_pad (GST_ELEMENT (receive), ghost);
+  receive->rtcp_src = gst_ghost_pad_new ("rtcp_src", pad);
+  gst_element_add_pad (GST_ELEMENT (receive), receive->rtcp_src);
   gst_object_unref (pad);
 
   /* create funnel for data_src */

diff --git a/ext/webrtc/transportreceivebin.h b/ext/webrtc/transportreceivebin.h
index 1d2586d..a4c1870 100644 (file)
--- a/ext/webrtc/transportreceivebin.h
+++ b/ext/webrtc/transportreceivebin.h
@@ -45,6 +45,10 @@ struct _TransportReceiveBin
   TransportStream           *stream;        /* parent transport stream */
   gboolean                   rtcp_mux;
 
+  GstPad                    *rtp_src;
+  gulong                     rtp_src_probe_id;
+  GstPad                    *rtcp_src;
+  gulong                     rtcp_src_probe_id;
   struct pad_block          *rtp_block;
   struct pad_block          *rtcp_block;
   GMutex                     pad_block_lock;