Changeset 2845 for pjproject

Timestamp:

Jul 29, 2009 12:19:25 PM (15 years ago)

Author:

bennylp

Message:

Ticket #924: Loop media transport now allows more than one streams to receive the reflected packets

• This ticket allows the same loop media transport instance to be attached to more than one streams, and allow application to control which stream(s) receives the reflected packets.

Location:

pjproject/trunk/pjmedia

Files:

• include/pjmedia/transport_loop.h (modified) (2 diffs)
• src/pjmedia/transport_loop.c (modified) (12 diffs)

pjproject/trunk/pjmedia/include/pjmedia/transport_loop.h

@@ -36 +36 @@
  * @{
  *
- * This is the loopback media transport, where packet sent to this transport
- * will immediately be sent back to the callback.
+ * This is the loopback media transport, where packets sent to this transport
+ * will be sent back to the streams attached to this transport. Unlike the
+ * other PJMEDIA transports, the loop transport may be attached to multiple
+ * streams (in other words, application should specify the same loop transport
+ * instance when calling #pjmedia_stream_create()). Any RTP or RTCP packets
+ * sent by one stream to this transport by default will be sent back to all 
+ * streams that are attached to this transport, including to the stream that
+ * sends the packet. Application may individually select which stream to
+ * receive packets by calling #pjmedia_transport_loop_disable_rx().
  */
 
@@ -55 +62 @@
 
 
+/**
+ * Set this stream as the receiver of incoming packets.
+ */
+PJ_DECL(pj_status_t) pjmedia_transport_loop_disable_rx(pjmedia_transport *tp,
+                                                       void *user,
+                                                       pj_bool_t disabled);
 
 

pjproject/trunk/pjmedia/src/pjmedia/transport_loop.c

@@ -19 +19 @@
  */
 #include <pjmedia/transport_loop.h>
+#include <pj/array.h>
 #include <pj/assert.h>
 #include <pj/errno.h>
@@ -28 +29 @@
 
 
-struct transport_loop
-{
-    pjmedia_transport   base;           /**< Base transport.                */
-
-    pj_pool_t          *pool;           /**< Memory pool                    */
+struct user
+{
+    pj_bool_t           rx_disabled;    /**< Doesn't want to receive pkt?   */
     void               *user_data;      /**< Only valid when attached       */
-    pj_bool_t           attached;       /**< Has attachment?                */
-    pj_sockaddr         rem_rtp_addr;   /**< Remote RTP address             */
-    pj_sockaddr         rem_rtcp_addr;  /**< Remote RTCP address            */
-    int                 addr_len;       /**< Length of addresses.           */
     void  (*rtp_cb)(    void*,          /**< To report incoming RTP.        */
                         void*,
@@ -44 +39 @@
                         void*,
                         pj_ssize_t);
+};
+
+struct transport_loop
+{
+    pjmedia_transport   base;           /**< Base transport.                */
+
+    pj_pool_t          *pool;           /**< Memory pool                    */
+    unsigned            user_cnt;       /**< Number of attachments          */
+    struct user         users[4];       /**< Array of users.                */
 
     unsigned            tx_drop_pct;    /**< Percent of tx pkts to drop.    */
@@ -149 +153 @@
 
 
+PJ_DEF(pj_status_t) pjmedia_transport_loop_disable_rx( pjmedia_transport *tp,
+                                                       void *user,
+                                                       pj_bool_t disabled)
+{
+    struct transport_loop *loop = (struct transport_loop*) tp;
+    unsigned i;
+
+    for (i=0; i<loop->user_cnt; ++i) {
+        if (loop->users[i].user_data == user) {
+            loop->users[i].rx_disabled = disabled;
+            return PJ_SUCCESS;
+        }
+    }
+    pj_assert(!"Invalid stream user");
+    return PJ_ENOTFOUND;
+}
+
 /**
  * Close loopback transport.
@@ -194 +215 @@
 {
     struct transport_loop *loop = (struct transport_loop*) tp;
+    unsigned i;
     const pj_sockaddr *rtcp_addr;
 
@@ -199 +221 @@
     PJ_ASSERT_RETURN(tp && rem_addr && addr_len, PJ_EINVAL);
 
-    /* Must not be "attached" to existing application */
-    PJ_ASSERT_RETURN(!loop->attached, PJ_EINVALIDOP);
+    /* Must not be "attached" to same user */
+    for (i=0; i<loop->user_cnt; ++i) {
+        PJ_ASSERT_RETURN(loop->users[i].user_data != user_data,
+                         PJ_EINVALIDOP);
+    }
+    PJ_ASSERT_RETURN(loop->user_cnt != PJ_ARRAY_SIZE(loop->users), 
+                     PJ_ETOOMANY);
 
     PJ_UNUSED_ARG(rem_rtcp);
@@ -207 +234 @@
     /* "Attach" the application: */
 
-    /* Save the callbacks */
-    loop->rtp_cb = rtp_cb;
-    loop->rtcp_cb = rtcp_cb;
-    loop->user_data = user_data;
-
-    /* Save address length */
-    loop->addr_len = addr_len;
-
-    /* Last, mark transport as attached */
-    loop->attached = PJ_TRUE;
+    /* Save the new user */
+    loop->users[loop->user_cnt].rtp_cb = rtp_cb;
+    loop->users[loop->user_cnt].rtcp_cb = rtcp_cb;
+    loop->users[loop->user_cnt].user_data = user_data;
+    ++loop->user_cnt;
 
     return PJ_SUCCESS;
@@ -227 +249 @@
 {
     struct transport_loop *loop = (struct transport_loop*) tp;
+    unsigned i;
 
     pj_assert(tp);
 
-    if (loop->attached) {
-        /* User data is unreferenced on Release build */
-        PJ_UNUSED_ARG(user_data);
-
-        /* As additional checking, check if the same user data is specified */
-        pj_assert(user_data == loop->user_data);
-
-        /* First, mark transport as unattached */
-        loop->attached = PJ_FALSE;
-
-        /* Clear up application infos from transport */
-        loop->rtp_cb = NULL;
-        loop->rtcp_cb = NULL;
-        loop->user_data = NULL;
-    }
+    for (i=0; i<loop->user_cnt; ++i) {
+        if (loop->users[i].user_data == user_data)
+            break;
+    }
+    PJ_ASSERT_ON_FAIL(i != loop->user_cnt, return);
+
+    /* Remove this user */
+    pj_array_erase(loop->users, sizeof(loop->users[0]),
+                   loop->user_cnt, i);
+    --loop->user_cnt;
 }
 
@@ -254 +272 @@
 {
     struct transport_loop *loop = (struct transport_loop*)tp;
-    void (*cb)(void*,void*,pj_ssize_t);
-    void *user_data;
-
-    /* Must be attached */
-    PJ_ASSERT_RETURN(loop->attached, PJ_EINVALIDOP);
+    unsigned i;
 
     /* Simulate packet lost on TX direction */
@@ -270 +284 @@
     }
 
-    cb = loop->rtp_cb;
-    user_data = loop->user_data;
-
     /* Simulate packet lost on RX direction */
     if (loop->rx_drop_pct) {
@@ -283 +294 @@
     }
 
-    if (loop->attached && cb)
-        (*cb)(user_data, (void*)pkt, size);
+    /* Distribute to users */
+    for (i=0; i<loop->user_cnt; ++i) {
+        if (!loop->users[i].rx_disabled && loop->users[i].rtp_cb)
+            (*loop->users[i].rtp_cb)(loop->users[i].user_data, (void*)pkt, 
+                                     size);
+    }
 
     return PJ_SUCCESS;
@@ -306 +321 @@
 {
     struct transport_loop *loop = (struct transport_loop*)tp;
-    void (*cb)(void*,void*,pj_ssize_t);
-    void *user_data;
-
-    PJ_ASSERT_RETURN(loop->attached, PJ_EINVALIDOP);
+    unsigned i;
 
     PJ_UNUSED_ARG(addr_len);
     PJ_UNUSED_ARG(addr);
 
-    cb = loop->rtcp_cb;
-    user_data = loop->user_data;
-
-    if (loop->attached && cb)
-        (*cb)(user_data, (void*)pkt, size);
+    /* Distribute to users */
+    for (i=0; i<loop->user_cnt; ++i) {
+        if (!loop->users[i].rx_disabled && loop->users[i].rtcp_cb)
+            (*loop->users[i].rtcp_cb)(loop->users[i].user_data, (void*)pkt,
+                                      size);
+    }
 
     return PJ_SUCCESS;