Changeset 452

Timestamp:

May 17, 2006 5:17:39 PM (18 years ago)

Author:

bennylp

Message:

Major modification in pjmedia to split stream transport into separate functionality, to allow using custom transports with streams

Location:

pjproject/trunk

Files:

• pjmedia/build/Makefile (modified) (1 diff)
• pjmedia/build/os-win32.mak (modified) (1 diff)
• pjmedia/build/pjmedia.dsp (modified) (2 diffs)
• pjmedia/include/pjmedia.h (modified) (1 diff)
• pjmedia/include/pjmedia/session.h (modified) (6 diffs)
• pjmedia/include/pjmedia/stream.h (modified) (6 diffs)
• pjmedia/include/pjmedia/transport_udp.h (added)
• pjmedia/src/pjmedia/session.c (modified) (6 diffs)
• pjmedia/src/pjmedia/stream.c (modified) (12 diffs)
• pjmedia/src/pjmedia/transport_udp.c (added)
• pjsip-apps/src/samples/simpleua.c (modified) (4 diffs)
• pjsip-apps/src/samples/siprtp.c (modified) (1 diff)
• pjsip-apps/src/samples/streamutil.c (modified) (5 diffs)
• pjsip/include/pjsua-lib/pjsua.h (modified) (1 diff)
• pjsip/src/pjsua-lib/pjsua_call.c (modified) (4 diffs)
• pjsip/src/pjsua-lib/pjsua_core.c (modified) (4 diffs)

pjproject/trunk/pjmedia/build/Makefile

@@ -70 +70 @@
                         port.o resample.o \
                         resample_port.o rtcp.o rtp.o sdp.o sdp_cmp.o sdp_neg.o \
-                        session.o silencedet.o sound_port.o stream.o wave.o \
+                        session.o silencedet.o sound_port.o stream.o \
+                        transport_udp.o wave.o \
                         $(SOUND_OBJS) $(NULLSOUND_OBJS)
 

pjproject/trunk/pjmedia/build/os-win32.mak

@@ -11 +11 @@
                        $(PA_DIR)/pa_win_wmme.o
 
-export OS_CFLAGS += -DPA_NO_ASIO -DPA_NO_DS
+export OS_CFLAGS += -DPA_NO_ASIO -DPA_NO_DS \
+                    -DPJMEDIA_SOUND_IMPLEMENTATION=PJMEDIA_SOUND_PORTAUDIO_SOUND
 
 # Example:

pjproject/trunk/pjmedia/build/pjmedia.dsp

@@ -192 +192 @@
 # Begin Source File
 
+SOURCE=..\src\pjmedia\transport_udp.c
+# End Source File
+# Begin Source File
+
 SOURCE=..\src\pjmedia\wav_player.c
 # End Source File
@@ -301 +305 @@
 
 SOURCE=..\include\pjmedia\stream.h
+# End Source File
+# Begin Source File
+
+SOURCE=..\include\pjmedia\transport_udp.h
 # End Source File
 # Begin Source File

pjproject/trunk/pjmedia/include/pjmedia.h

@@ -44 +44 @@
 #include <pjmedia/silencedet.h>
 #include <pjmedia/session.h>
+#include <pjmedia/transport_udp.h>
 #include <pjmedia/sound.h>
 #include <pjmedia/sound_port.h>

pjproject/trunk/pjmedia/include/pjmedia/session.h

@@ -89 +89 @@
  * @param max_streams   Maximum number of stream infos to be created.
  * @param si            Session info structure to be initialized.
- * @param skinfo        Optional array of media socket info to be copied
- *                      to the stream info. If this argument is specified,
- *                      the array must contain sufficient elements for
- *                      each stream to be initialized.
  * @param local         Local SDP session descriptor.
  * @param remote        Remote SDP session descriptor.
@@ -104 +100 @@
                                unsigned max_streams,
                                pjmedia_session_info *si,
-                               const pjmedia_sock_info skinfo[],
                                const pjmedia_sdp_session *local,
                                const pjmedia_sdp_session *remote);
@@ -119 +114 @@
  * @param pool          Pool to allocate memory.
  * @param endpt         PJMEDIA endpoint instance.
- * @param skinfo        Optional socket info to be copied to the stream info.
  * @param local         Local SDP session descriptor.
  * @param remote        Remote SDP session descriptor.
@@ -130 +124 @@
                               pj_pool_t *pool,
                               pjmedia_endpt *endpt,
-                              const pjmedia_sock_info *skinfo,
                               const pjmedia_sdp_session *local,
                               const pjmedia_sdp_session *remote,
@@ -146 +139 @@
  * 
  * @param endpt         The PJMEDIA endpoint instance.
- * @param stream_cnt    Maximum number of streams to be created. This
- *                      also denotes the number of elements in the
- *                      socket information.
- * @param local_sdp     The SDP describing local capability.
- * @param rem_sdp       The SDP describing remote capability.
+ * @param si            Session info containing stream count and array of
+ *                      stream info. The stream count indicates how many
+ *                      streams to be created in the session.
+ * @param transports    Array of media stream transports, with 
+ *                      sufficient number of elements (one for each stream).
  * @param user_data     Arbitrary user data to be kept in the session.
  * @param p_session     Pointer to receive the media session.
@@ -160 +153 @@
 pjmedia_session_create( pjmedia_endpt *endpt, 
                         const pjmedia_session_info *si,
+                        pjmedia_transport *transports[],
                         void *user_data,
                         pjmedia_session **p_session );

pjproject/trunk/pjmedia/include/pjmedia/stream.h

@@ -70 +70 @@
     pjmedia_type        type;       /**< Media type (audio, video)          */
     pjmedia_dir         dir;        /**< Media direction.                   */
-    pjmedia_sock_info   sock_info;  /**< Media transport (RTP/RTCP sockets) */
     pj_sockaddr_in      rem_addr;   /**< Remote RTP address                 */
     pjmedia_codec_info  fmt;        /**< Incoming codec format info.        */
@@ -101 +100 @@
 
 /**
+ * @see pjmedia_transport_op.
+ */
+typedef struct pjmedia_transport pjmedia_transport;
+
+
+/**
+ * This structure describes the operations for the stream transport.
+ */
+struct pjmedia_transport_op
+{
+    /**
+     * This function is called by the stream when the transport is about
+     * to be used by the stream for the first time, and it tells the transport
+     * about remote RTP address to send the packet and some callbacks to be 
+     * called for incoming packets.
+     */
+    pj_status_t (*attach)(pjmedia_transport *tp,
+                          pjmedia_stream *strm,
+                          const pj_sockaddr_t *rem_addr,
+                          unsigned addr_len,
+                          void (*rtp_cb)(pjmedia_stream*,
+                                         const void*,
+                                         pj_ssize_t),
+                          void (*rtcp_cb)(pjmedia_stream*,
+                                          const void*,
+                                          pj_ssize_t));
+
+    /**
+     * This function is called by the stream when the stream is no longer
+     * need the transport (normally when the stream is about to be closed).
+     */
+    void (*detach)(pjmedia_transport *tp,
+                   pjmedia_stream *strm);
+
+    /**
+     * This function is called by the stream to send RTP packet using the 
+     * transport.
+     */
+    pj_status_t (*send_rtp)(pjmedia_transport *tp,
+                            const void *pkt,
+                            pj_size_t size);
+
+    /**
+     * This function is called by the stream to send RTCP packet using the
+     * transport.
+     */
+    pj_status_t (*send_rtcp)(pjmedia_transport *tp,
+                             const void *pkt,
+                             pj_size_t size);
+
+};
+
+
+/**
+ * @see pjmedia_transport_op.
+ */
+typedef struct pjmedia_transport_op pjmedia_transport_op;
+
+
+/**
+ * This structure declares stream transport. A stream transport is called
+ * by the stream to transmit a packet, and will notify stream when
+ * incoming packet is arrived.
+ */
+struct pjmedia_transport
+{
+    char                  name[PJ_MAX_OBJ_NAME];
+
+    pjmedia_transport_op *op;
+};
+
+
+
+/**
  * Create a media stream based on the specified parameter. After the stream
  * has been created, application normally would want to get the media port 
@@ -115 +188 @@
  *                      buffer needs to preallocate some storage.
  * @param info          Stream information.
+ * @param tp            Stream transport instance used to transmit 
+ *                      and receive RTP/RTCP packets to/from the underlying 
+ *                      transport. 
  * @param user_data     Arbitrary user data (for future callback feature).
  * @param p_stream      Pointer to receive the media stream.
@@ -123 +199 @@
                                            pj_pool_t *pool,
                                            const pjmedia_stream_info *info,
+                                           pjmedia_transport *tp,
                                            void *user_data,
                                            pjmedia_stream **p_stream);
@@ -152 +229 @@
 
 /**
+ * Get the media transport object associated with this stream.
+ *
+ * @param st            The media stream.
+ *
+ * @return              The transport object being used by the stream.
+ */
+PJ_DECL(pjmedia_transport*) pjmedia_stream_get_transport(pjmedia_stream *st);
+
+
+/**
  * Start the media stream. This will start the appropriate channels
  * in the media stream, depending on the media direction that was set
@@ -225 +312 @@
 
 /**
- * Retrieve the incoming DTMF digits from the stream. Note that the digits
- * buffer will not be NULL terminated.
+ * Retrieve the incoming DTMF digits from the stream, and remove the digits
+ * from stream's DTMF buffer. Note that the digits buffer will not be NULL 
+ * terminated.
  *
  * @param stream        The media stream.

pjproject/trunk/pjmedia/src/pjmedia/session.c

@@ -65 +65 @@
                                            pj_pool_t *pool,
                                            pjmedia_endpt *endpt,
-                                           const pjmedia_sock_info *skinfo,
                                            const pjmedia_sdp_session *local,
                                            const pjmedia_sdp_session *remote,
@@ -343 +342 @@
     }
 
-    /* Copy skinfo */
-    if (skinfo)
-        si->sock_info = *skinfo;
-
     /* Leave SSRC to random. */
     si->ssrc = pj_rand();
@@ -365 +360 @@
                                unsigned max_streams,
                                pjmedia_session_info *si,
-                               const pjmedia_sock_info skinfo[],
                                const pjmedia_sdp_session *local,
                                const pjmedia_sdp_session *remote)
@@ -382 +376 @@
         status = pjmedia_stream_info_from_sdp( &si->stream_info[i], pool,
                                                endpt, 
-                                               (skinfo ? &skinfo[i] : NULL),
                                                local, remote, i);
         if (status != PJ_SUCCESS)
@@ -398 +391 @@
 PJ_DEF(pj_status_t) pjmedia_session_create( pjmedia_endpt *endpt, 
                                             const pjmedia_session_info *si,
+                                            pjmedia_transport *transports[],
                                             void *user_data,
                                             pjmedia_session **p_session )
@@ -433 +427 @@
         status = pjmedia_stream_create(endpt, session->pool,
                                        &session->stream_info[i],
+                                       (transports?transports[i]:NULL),
                                        session,
                                        &session->stream[i]);

pjproject/trunk/pjmedia/src/pjmedia/stream.c

@@ -86 +86 @@
     void                    *user_data;     /**< User data.                 */
 
+    pjmedia_transport       *transport;     /**< Stream transport.          */
+
     pjmedia_codec           *codec;         /**< Codec instance being used. */
     pjmedia_codec_param      codec_param;   /**< Codec param.               */
@@ -92 +94 @@
     pjmedia_jbuf            *jb;            /**< Jitter buffer.             */
 
-    pjmedia_sock_info        skinfo;        /**< Transport info.            */
-    pj_sockaddr_in           rem_rtp_addr;  /**< Remote RTP address.        */
-    pj_sockaddr_in           rem_rtcp_addr; /**< Remote RTCP address.       */
-
-
     pjmedia_rtcp_session     rtcp;          /**< RTCP for incoming RTP.     */
 
-    pj_ioqueue_key_t        *rtp_key;       /**< RTP ioqueue key.           */
-    pj_ioqueue_op_key_t      rtp_op_key;    /**< The pending read op key.   */
-    pj_sockaddr_in           rtp_src_addr;  /**< addr of src pkt from remote*/
-    unsigned                 rtp_src_cnt;   /**< if different, # of pkt rcv */
-    int                      rtp_addrlen;   /**< Address length.            */
-
-    pj_ioqueue_key_t        *rtcp_key;      /**< RTCP ioqueue key.          */
-    pj_ioqueue_op_key_t      rtcp_op_key;   /**< The pending read op key.   */
-    pj_size_t                rtcp_pkt_size; /**< Size of RTCP packet buf.   */
-    char                     rtcp_pkt[512]; /**< RTCP packet buffer.        */
     pj_uint32_t              rtcp_last_tx;  /**< RTCP tx time in timestamp  */
     pj_uint32_t              rtcp_interval; /**< Interval, in timestamp.    */
-    int                      rtcp_addrlen;  /**< Address length.            */
 
     /* RFC 2833 DTMF transmission queue: */
@@ -424 +410 @@
         
         pjmedia_rtcp_pkt *rtcp_pkt;
-        pj_ssize_t size;
         int len;
-        pj_status_t status;
 
         pjmedia_rtcp_build_rtcp(&stream->rtcp, &rtcp_pkt, &len);
-        size = len;
-        status = pj_sock_sendto(stream->skinfo.rtcp_sock, rtcp_pkt, &size, 0,
-                                &stream->rem_rtcp_addr, 
-                                sizeof(stream->rem_rtcp_addr));
-#if 0
-        if (status != PJ_SUCCESS) {
-            char errmsg[PJ_ERR_MSG_SIZE];
-            
-            pj_strerror(status, errmsg, sizeof(errmsg));
-            PJ_LOG(4,(port->info.name.ptr, "Error sending RTCP: %s [%d]",
-                                 errmsg, status));
-        }
-#endif
+
+        (*stream->transport->op->send_rtcp)(stream->transport, 
+                                            rtcp_pkt, len);
 
         stream->rtcp_last_tx = timestamp;
@@ -587 +561 @@
     /* Send. */
     sent = frame_out.size+sizeof(pjmedia_rtp_hdr);
-    status = pj_sock_sendto(stream->skinfo.rtp_sock, channel->out_pkt, 
-                            &sent, 0, &stream->rem_rtp_addr, 
-                            sizeof(stream->rem_rtp_addr));
-    if (status != PJ_SUCCESS)
-        return status;
+
+    (*stream->transport->op->send_rtp)(stream->transport,
+                                       channel->out_pkt, sent);
+
 
     /* Update stat */
@@ -683 +656 @@
 
 /*
- * This callback is called by ioqueue framework on receipt of packets
+ * This callback is called by stream transport on receipt of packets
  * in the RTP socket. 
  */
-static void on_rx_rtp( pj_ioqueue_key_t *key, 
-                       pj_ioqueue_op_key_t *op_key, 
+static void on_rx_rtp( pjmedia_stream *stream, 
+                       const void *pkt,
                        pj_ssize_t bytes_read)
 
 {
-    pjmedia_stream *stream = pj_ioqueue_get_user_data(key);
     pjmedia_channel *channel = stream->dec;
+    const pjmedia_rtp_hdr *hdr;
+    const void *payload;
+    unsigned payloadlen;
+    pjmedia_rtp_status seq_st;
     pj_status_t status;
 
-    
-    PJ_UNUSED_ARG(op_key);
-
-
-    /*
-     * Loop while we have packet.
+
+    /* Update RTP and RTCP session. */
+    status = pjmedia_rtp_decode_rtp(&channel->rtp, pkt, bytes_read,
+                                    &hdr, &payload, &payloadlen);
+    if (status != PJ_SUCCESS) {
+        LOGERR_((stream->port.info.name.ptr, "RTP decode error", status));
+        return;
+    }
+
+
+    /* Inform RTCP session */
+    pjmedia_rtcp_rx_rtp(&stream->rtcp, pj_ntohs(hdr->seq),
+                        pj_ntohl(hdr->ts), payloadlen);
+
+    /* Handle incoming DTMF. */
+    if (hdr->pt == stream->rx_event_pt) {
+        handle_incoming_dtmf(stream, payload, payloadlen);
+        return;
+    }
+
+
+    /* Update RTP session (also checks if RTP session can accept
+     * the incoming packet.
      */
-    do {
-        const pjmedia_rtp_hdr *hdr;
-        const void *payload;
-        unsigned payloadlen;
-        pjmedia_rtp_status seq_st;
-
-        /* Go straight to read next packet if bytes_read == 0.
+    pjmedia_rtp_session_update(&channel->rtp, hdr, &seq_st);
+    if (seq_st.status.value) {
+        TRC_  ((stream->port.info.name.ptr, 
+                "RTP status: badpt=%d, badssrc=%d, dup=%d, "
+                "outorder=%d, probation=%d, restart=%d", 
+                seq_st.status.flag.badpt,
+                seq_st.status.flag.badssrc,
+                seq_st.status.flag.dup,
+                seq_st.status.flag.outorder,
+                seq_st.status.flag.probation,
+                seq_st.status.flag.restart));
+
+        if (seq_st.status.flag.badpt) {
+            PJ_LOG(4,(stream->port.info.name.ptr,
+                      "Bad RTP pt %d (expecting %d)",
+                      hdr->pt, channel->rtp.out_pt));
+        }
+    }
+
+    /* Skip bad RTP packet */
+    if (seq_st.status.flag.bad)
+        return;
+
+
+    /* Put "good" packet to jitter buffer, or reset the jitter buffer
+     * when RTP session is restarted.
+     */
+    pj_mutex_lock( stream->jb_mutex );
+    if (seq_st.status.flag.restart) {
+        status = pjmedia_jbuf_reset(stream->jb);
+        PJ_LOG(4,(stream->port.info.name.ptr, "Jitter buffer reset"));
+
+    } else {
+        /*
+         * Packets may contain more than one frames, while the jitter
+         * buffer can only take one frame per "put" operation. So we need
+         * to ask the codec to "parse" the payload into multiple frames.
          */
-        if (bytes_read == 0)
-            goto read_next_packet;
-
-        if (bytes_read < 0)
-            goto read_next_packet;
-
-        /* Update RTP and RTCP session. */
-        status = pjmedia_rtp_decode_rtp(&channel->rtp, 
-                                        channel->in_pkt, bytes_read, 
-                                        &hdr, &payload, &payloadlen);
+        enum { MAX = 16 };
+        pj_timestamp ts;
+        unsigned i, count = MAX;
+        unsigned samples_per_frame;
+        pjmedia_frame frames[MAX];
+
+        /* Get the timestamp of the first sample */
+        ts.u64 = pj_ntohl(hdr->ts);
+
+        /* Parse the payload. */
+        status = (*stream->codec->op->parse)(stream->codec,
+                                             (void*)payload,
+                                             payloadlen,
+                                             &ts,
+                                             &count,
+                                             frames);
         if (status != PJ_SUCCESS) {
-            LOGERR_((stream->port.info.name.ptr, "RTP decode error", status));
-            goto read_next_packet;
+            LOGERR_((stream->port.info.name.ptr, 
+                     "Codec parse() error", 
+                     status));
+            count = 0;
         }
 
-
-        /* Inform RTCP session */
-        pjmedia_rtcp_rx_rtp(&stream->rtcp, pj_ntohs(hdr->seq),
-                            pj_ntohl(hdr->ts), payloadlen);
-
-        /* Handle incoming DTMF. */
-        if (hdr->pt == stream->rx_event_pt) {
-            handle_incoming_dtmf(stream, payload, payloadlen);
-            goto read_next_packet;
+        /* Put each frame to jitter buffer. */
+        samples_per_frame = stream->codec_param.info.frm_ptime * 
+                            stream->codec_param.info.clock_rate *
+                            stream->codec_param.info.channel_cnt /
+                            1000;
+                            
+        for (i=0; i<count; ++i) {
+            unsigned ext_seq;
+
+            ext_seq = (unsigned)(frames[i].timestamp.u64 /
+                                 samples_per_frame);
+            pjmedia_jbuf_put_frame(stream->jb, frames[i].buf, 
+                                   frames[i].size, ext_seq);
+
         }
-
-
-        /* Update RTP session (also checks if RTP session can accept
-         * the incoming packet.
-         */
-        pjmedia_rtp_session_update(&channel->rtp, hdr, &seq_st);
-        if (seq_st.status.value) {
-            TRC_  ((stream->port.info.name.ptr, 
-                    "RTP status: badpt=%d, badssrc=%d, dup=%d, "
-                    "outorder=%d, probation=%d, restart=%d", 
-                    seq_st.status.flag.badpt,
-                    seq_st.status.flag.badssrc,
-                    seq_st.status.flag.dup,
-                    seq_st.status.flag.outorder,
-                    seq_st.status.flag.probation,
-                    seq_st.status.flag.restart));
-
-            if (seq_st.status.flag.badpt) {
-                PJ_LOG(4,(stream->port.info.name.ptr,
-                          "Bad RTP pt %d (expecting %d)",
-                          hdr->pt, channel->rtp.out_pt));
-            }
-        }
-
-        /* Skip bad RTP packet */
-        if (seq_st.status.flag.bad)
-            goto read_next_packet;
-
-
-        /* See if source address of RTP packet is different than the 
-         * configured address.
-         */
-        if ((stream->rem_rtp_addr.sin_addr.s_addr != 
-             stream->rtp_src_addr.sin_addr.s_addr) ||
-            (stream->rem_rtp_addr.sin_port != stream->rtp_src_addr.sin_port))
-        {
-            stream->rtp_src_cnt++;
-
-            if (stream->rtp_src_cnt >= PJMEDIA_RTP_NAT_PROBATION_CNT) {
-            
-                stream->rem_rtp_addr = stream->rtp_src_addr;
-                stream->rtp_src_cnt = 0;
-
-                PJ_LOG(4,(stream->port.info.name.ptr,
-                          "Remote RTP address switched to %s:%d",
-                          pj_inet_ntoa(stream->rtp_src_addr.sin_addr),
-                          pj_ntohs(stream->rtp_src_addr.sin_port)));
-            }
-        }
-
-
-
-        /* Put "good" packet to jitter buffer, or reset the jitter buffer
-         * when RTP session is restarted.
-         */
-        pj_mutex_lock( stream->jb_mutex );
-        if (seq_st.status.flag.restart) {
-            status = pjmedia_jbuf_reset(stream->jb);
-            PJ_LOG(4,(stream->port.info.name.ptr, "Jitter buffer reset"));
-
-        } else {
-            /*
-             * Packets may contain more than one frames, while the jitter
-             * buffer can only take one frame per "put" operation. So we need
-             * to ask the codec to "parse" the payload into multiple frames.
-             */
-            enum { MAX = 16 };
-            pj_timestamp ts;
-            unsigned i, count = MAX;
-            unsigned samples_per_frame;
-            pjmedia_frame frames[MAX];
-
-            /* Get the timestamp of the first sample */
-            ts.u64 = pj_ntohl(hdr->ts);
-
-            /* Parse the payload. */
-            status = (*stream->codec->op->parse)(stream->codec,
-                                                 (void*)payload,
-                                                 payloadlen,
-                                                 &ts,
-                                                 &count,
-                                                 frames);
-            if (status != PJ_SUCCESS) {
-                LOGERR_((stream->port.info.name.ptr, 
-                         "Codec parse() error", 
-                         status));
-                count = 0;
-            }
-
-            /* Put each frame to jitter buffer. */
-            samples_per_frame = stream->codec_param.info.frm_ptime * 
-                                stream->codec_param.info.clock_rate *
-                                stream->codec_param.info.channel_cnt /
-                                1000;
-                                
-            for (i=0; i<count; ++i) {
-                unsigned ext_seq;
-
-                ext_seq = (unsigned)(frames[i].timestamp.u64 /
-                                     samples_per_frame);
-                pjmedia_jbuf_put_frame(stream->jb, frames[i].buf, 
-                                       frames[i].size, ext_seq);
-
-            }
-        }
-        pj_mutex_unlock( stream->jb_mutex );
-
-
-        /* Check if now is the time to transmit RTCP SR/RR report.
-         * We only do this when stream direction is "decoding only", 
-         * because otherwise check_tx_rtcp() will be handled by put_frame()
-         */
-        if (stream->dir == PJMEDIA_DIR_DECODING) {
-            check_tx_rtcp(stream, pj_ntohl(hdr->ts));
-        }
-
-        if (status != 0) {
-            LOGERR_((stream->port.info.name.ptr, "Jitter buffer put() error", 
-                    status));
-            goto read_next_packet;
-        }
-
-
-read_next_packet:
-        bytes_read = channel->in_pkt_size;
-        stream->rtp_addrlen = sizeof(stream->rtp_src_addr);
-        status = pj_ioqueue_recvfrom( stream->rtp_key,
-                                      &stream->rtp_op_key,
-                                      channel->in_pkt, 
-                                      &bytes_read, 0,
-                                      &stream->rtp_src_addr, 
-                                      &stream->rtp_addrlen);
-
-        if (status != PJ_SUCCESS) {
-            bytes_read = -status;
-        }
-
-    } while (status == PJ_SUCCESS ||
-             status == PJ_STATUS_FROM_OS(OSERR_ECONNRESET));
-
-    if (status != PJ_SUCCESS && status != PJ_EPENDING) {
-        char errmsg[PJ_ERR_MSG_SIZE];
-
-        pj_strerror(status, errmsg, sizeof(errmsg));
-        PJ_LOG(4,(stream->port.info.name.ptr, 
-                  "Error reading RTP packet: %s [status=%d]. "
-                  "RTP stream thread quitting!",
-                  errmsg, status));
-    }
-}
-
-
-/*
- * This callback is called by ioqueue framework on receipt of packets
+    }
+    pj_mutex_unlock( stream->jb_mutex );
+
+
+    /* Check if now is the time to transmit RTCP SR/RR report.
+     * We only do this when stream direction is "decoding only", 
+     * because otherwise check_tx_rtcp() will be handled by put_frame()
+     */
+    if (stream->dir == PJMEDIA_DIR_DECODING) {
+        check_tx_rtcp(stream, pj_ntohl(hdr->ts));
+    }
+
+    if (status != 0) {
+        LOGERR_((stream->port.info.name.ptr, "Jitter buffer put() error", 
+                status));
+        return;
+    }
+}
+
+
+/*
+ * This callback is called by stream transport on receipt of packets
  * in the RTCP socket. 
  */
-static void on_rx_rtcp( pj_ioqueue_key_t *key, 
-                        pj_ioqueue_op_key_t *op_key, 
+static void on_rx_rtcp( pjmedia_stream *stream,
+                        const void *pkt, 
                         pj_ssize_t bytes_read)
 {
-    pjmedia_stream *stream = pj_ioqueue_get_user_data(key);
-    pj_status_t status;
-
-    PJ_UNUSED_ARG(op_key);
-
-    do {
-        if (bytes_read > 0) {
-            pjmedia_rtcp_rx_rtcp(&stream->rtcp, stream->rtcp_pkt, 
-                                 bytes_read);
-        }
-
-        bytes_read = stream->rtcp_pkt_size;
-        stream->rtcp_addrlen = sizeof(stream->rem_rtcp_addr);
-        status = pj_ioqueue_recvfrom( stream->rtcp_key,
-                                      &stream->rtcp_op_key,
-                                      stream->rtcp_pkt,
-                                      &bytes_read, 0,
-                                      &stream->rem_rtcp_addr,
-                                      &stream->rtcp_addrlen);
-
-    } while (status == PJ_SUCCESS);
-
-    if (status != PJ_SUCCESS && status != PJ_EPENDING) {
-        char errmsg[PJ_ERR_MSG_SIZE];
-
-        pj_strerror(status, errmsg, sizeof(errmsg));
-        PJ_LOG(4,(stream->port.info.name.ptr, 
-                  "Error reading RTCP packet: %s [status=%d]",
-                  errmsg, status));
-    }
-
+    pjmedia_rtcp_rx_rtcp(&stream->rtcp, pkt, bytes_read);
 }
 
@@ -994 +868 @@
                                            pj_pool_t *pool,
                                            const pjmedia_stream_info *info,
+                                           pjmedia_transport *tp,
                                            void *user_data,
                                            pjmedia_stream **p_stream)
@@ -999 +874 @@
 {
     pjmedia_stream *stream;
-    pj_ioqueue_callback ioqueue_cb;
-    pj_uint16_t rtcp_port;
     unsigned jb_init, jb_max, jb_min_pre, jb_max_pre;
     pj_status_t status;
@@ -1037 +910 @@
     stream->dir = info->dir;
     stream->user_data = user_data;
-    stream->skinfo = info->sock_info;
-    stream->rem_rtp_addr = info->rem_addr;
-    rtcp_port = (pj_uint16_t) (pj_ntohs(info->rem_addr.sin_port)+1);
-    stream->rem_rtcp_addr = stream->rem_rtp_addr;
-    stream->rem_rtcp_addr.sin_port = pj_htons(rtcp_port);
     stream->rtcp_interval = (PJMEDIA_RTCP_INTERVAL + (pj_rand() % 8000)) * 
                             info->fmt.clock_rate / 1000;
@@ -1048 +916 @@
     stream->rx_event_pt = info->rx_event_pt ? info->rx_event_pt : -1;
     stream->last_dtmf = -1;
+
+    /* Attach transport */
+    status = (*tp->op->attach)(tp, stream, &info->rem_addr, 
+                               sizeof(info->rem_addr), &on_rx_rtp,
+                               &on_rx_rtcp);
+    if (status != PJ_SUCCESS)
+        goto err_cleanup;
+
+    stream->transport = tp;
 
 
@@ -1161 +1038 @@
         goto err_cleanup;
 
-    /*  Register RTP socket to ioqueue */
-    pj_memset(&ioqueue_cb, 0, sizeof(ioqueue_cb));
-    ioqueue_cb.on_read_complete = &on_rx_rtp;
-
-    status = pj_ioqueue_register_sock( pool, 
-                                       pjmedia_endpt_get_ioqueue(endpt), 
-                                       stream->skinfo.rtp_sock,
-                                       stream, &ioqueue_cb, &stream->rtp_key);
-    if (status != PJ_SUCCESS)
-        goto err_cleanup;
-
-    /* Init pending operation key. */
-    pj_ioqueue_op_key_init(&stream->rtp_op_key, sizeof(stream->rtp_op_key));
-
-    /* Bootstrap the first recvfrom() operation. */
-    on_rx_rtp( stream->rtp_key, &stream->rtp_op_key, 0);
-
-
-    /* Register RTCP socket to ioqueue. */
-    if (stream->skinfo.rtcp_sock != PJ_INVALID_SOCKET) {
-        pj_memset(&ioqueue_cb, 0, sizeof(ioqueue_cb));
-        ioqueue_cb.on_read_complete = &on_rx_rtcp;
-
-        status = pj_ioqueue_register_sock( pool, 
-                                           pjmedia_endpt_get_ioqueue(endpt),
-                                           stream->skinfo.rtcp_sock,
-                                           stream, &ioqueue_cb, 
-                                           &stream->rtcp_key);
-        if (status != PJ_SUCCESS)
-            goto err_cleanup;
-    }
-
-    /* Init pending operation key. */
-    pj_ioqueue_op_key_init(&stream->rtcp_op_key, sizeof(stream->rtcp_op_key));
-
-    stream->rtcp_pkt_size = sizeof(stream->rtcp_pkt);
-
-    /* Bootstrap the first recvfrom() operation. */
-    on_rx_rtcp( stream->rtcp_key, &stream->rtcp_op_key, 0);
+
 
     /* Success! */
@@ -1228 +1067 @@
 
 
-    /* Unregister from ioqueue. */
-    if (stream->rtp_key) {
-        pj_ioqueue_unregister(stream->rtp_key);
-        stream->rtp_key = NULL;
-    }
-    if (stream->rtcp_key) {
-        pj_ioqueue_unregister(stream->rtcp_key);
-        stream->rtcp_key = NULL;
+    /* Detach from transport */
+    if (stream->transport) {
+        (*stream->transport->op->detach)(stream->transport, stream);
+        stream->transport = NULL;
     }
 
@@ -1266 +1101 @@
     *p_port = &stream->port;
     return PJ_SUCCESS;
+}
+
+
+/*
+ * Get the transport object
+ */
+PJ_DEF(pjmedia_transport*) pjmedia_stream_get_transport(pjmedia_stream *st)
+{
+    return st->transport;
 }
 

pjproject/trunk/pjsip-apps/src/samples/simpleua.c

@@ -74 +74 @@
 static pjmedia_endpt        *g_med_endpt;   /* Media endpoint.          */
 static pjmedia_sock_info     g_med_skinfo;  /* Socket info for media    */
+static pjmedia_transport    *g_med_transport;/* Media stream transport  */
 
 /* Call variables: */
@@ -281 +282 @@
 
     
+    /* Create media transport */
+    status = pjmedia_transport_udp_attach(g_med_endpt, NULL, &g_med_skinfo,
+                                          &g_med_transport);
+    if (status != PJ_SUCCESS) {
+        app_perror(THIS_FILE, "Unable to create media transport", status);
+        return 1;
+    }
+
     /*
      * If URL is specified, then make call immediately.
@@ -612 +621 @@
      * We only support one stream per session for now.
      */
-    status = pjmedia_session_info_from_sdp(inv->dlg->pool, g_med_endpt, 1,
-                                           &sess_info, &g_med_skinfo,
+    status = pjmedia_session_info_from_sdp(inv->dlg->pool, g_med_endpt, 
+                                           1, &sess_info, 
                                            local_sdp, remote_sdp);
     if (status != PJ_SUCCESS) {
@@ -630 +639 @@
      */
     status = pjmedia_session_create( g_med_endpt, &sess_info,
-                                     NULL, &g_med_session );
+                                     &g_med_transport, NULL, &g_med_session );
     if (status != PJ_SUCCESS) {
         app_perror( THIS_FILE, "Unable to create media session", status);

pjproject/trunk/pjsip-apps/src/samples/siprtp.c

@@ -1277 +1277 @@
 
     status = pjmedia_stream_info_from_sdp(&audio->si, inv->pool, app.med_endpt,
-                                          NULL, local_sdp, remote_sdp, 0);
+                                          local_sdp, remote_sdp, 0);
     if (status != PJ_SUCCESS) {
         app_perror(THIS_FILE, "Error creating stream info from SDP", status);

pjproject/trunk/pjsip-apps/src/samples/streamutil.c

@@ -111 +111 @@
 {
     pjmedia_stream_info info;
+    pjmedia_transport *transport;
     pj_status_t status;
 
@@ -130 +131 @@
 
 
-    /* Create RTP socket */
-    status = pj_sock_socket(PJ_AF_INET, PJ_SOCK_DGRAM, 0, 
-                            &info.sock_info.rtp_sock);
-    PJ_ASSERT_RETURN(status == PJ_SUCCESS, status);
-
-
-    /* Bind RTP socket to local port */
-    info.sock_info.rtp_addr_name.sin_family = PJ_AF_INET;
-    info.sock_info.rtp_addr_name.sin_port = pj_htons(local_port);
-
-    status = pj_sock_bind(info.sock_info.rtp_sock, 
-                          &info.sock_info.rtp_addr_name,
-                          sizeof(pj_sockaddr_in));
-    if (status != PJ_SUCCESS) {
-        app_perror(THIS_FILE, "Unable to bind RTP socket", status);
-        pj_sock_close(info.sock_info.rtp_sock);
+    /* Create media transport */
+    status = pjmedia_transport_udp_create(med_endpt, NULL, local_port,
+                                          &transport);
+    if (status != PJ_SUCCESS)
         return status;
-    }
-
-
-    /* Create RTCP socket */
-    status = pj_sock_socket(PJ_AF_INET, PJ_SOCK_DGRAM, 0,
-                            &info.sock_info.rtcp_sock);
-    PJ_ASSERT_RETURN(status == PJ_SUCCESS, status);
-
-
-    /* Bind RTP socket to local port + 1 */
-    ++local_port;
-    info.sock_info.rtcp_addr_name.sin_family = PJ_AF_INET;
-    info.sock_info.rtcp_addr_name.sin_port = pj_htons(local_port);
-
-    status = pj_sock_bind(info.sock_info.rtcp_sock, 
-                          &info.sock_info.rtcp_addr_name,
-                          sizeof(pj_sockaddr_in));
-    if (status != PJ_SUCCESS) {
-        app_perror(THIS_FILE, "Unable to bind RTCP socket", status);
-        pj_sock_close(info.sock_info.rtp_sock);
-        pj_sock_close(info.sock_info.rtcp_sock);
-        return status;
-    }
 
 
@@ -176 +142 @@
      */
 
-    status = pjmedia_stream_create( med_endpt, pool, &info, NULL, p_stream);
+    status = pjmedia_stream_create( med_endpt, pool, &info, 
+                                    transport, NULL, p_stream);
 
     if (status != PJ_SUCCESS) {
         app_perror(THIS_FILE, "Error creating stream", status);
-        pj_sock_close(info.sock_info.rtp_sock);
-        pj_sock_close(info.sock_info.rtcp_sock);
+        pjmedia_transport_udp_close(transport);
         return status;
     }
@@ -212 +178 @@
     pjmedia_port *stream_port;
     char tmp[10];
-    pj_status_t status;
+    pj_status_t status; 
 
 
@@ -514 +480 @@
     /* Destroy stream */
     if (stream) {
+        pjmedia_transport *tp;
+
+        tp = pjmedia_stream_get_transport(stream);
         pjmedia_stream_destroy(stream);
+        pjmedia_transport_udp_close(tp);
     }
 

pjproject/trunk/pjsip/include/pjsua-lib/pjsua.h

@@ -103 +103 @@
                                          call was triggered by xfer.        */
     pjmedia_sock_info    skinfo;    /**< Preallocated media sockets.        */
+    pjmedia_transport   *med_tp;    /**< Media transport.                   */
     void                *app_data;  /**< Application data.                  */
     pj_timer_entry       refresh_tm;/**< Timer to send re-INVITE.           */

pjproject/trunk/pjsip/src/pjsua-lib/pjsua_call.c

@@ -98 +98 @@
 
 
-/* Close and reopen socket. */
-static pj_status_t reopen_sock( pj_sock_t *sock, pj_sockaddr_in *addr)
-{
-    pj_status_t status;
-
-    status = pj_sock_socket(PJ_AF_INET, PJ_SOCK_DGRAM, 0, sock);
-    if (status != PJ_SUCCESS) {
-        pjsua_perror(THIS_FILE, "Unable to create socket", status);
-        return status;
-    }
-
-    status = pj_sock_bind(*sock, addr, sizeof(pj_sockaddr_in));
-    if (status != PJ_SUCCESS) {
-        pjsua_perror(THIS_FILE, "Unable to re-bind RTP/RTCP socket", status);
-        return status;
-    }
-
-    return PJ_SUCCESS;
-}
-
 /*
  * Destroy the call's media
@@ -131 +111 @@
 
     if (call->session) {
-        pj_sockaddr_in rtp_addr, rtcp_addr;
-        int addrlen;
-
-        addrlen = sizeof(rtp_addr);
-        pj_sock_getsockname(call->skinfo.rtp_sock, &rtp_addr, &addrlen);
-
-        addrlen = sizeof(rtcp_addr);
-        pj_sock_getsockname(call->skinfo.rtcp_sock, &rtcp_addr, &addrlen);
-
         /* Destroy session (this will also close RTP/RTCP sockets). */
         pjmedia_session_destroy(call->session);
-
-        /* Close and reopen RTP socket.
-         * This is necessary to get the socket unregistered from ioqueue,
-         * when IOCompletionPort is used.
-         */
-        reopen_sock(&call->skinfo.rtp_sock, &rtp_addr);
-
-        /* Close and reopen RTCP socket too. */
-        reopen_sock(&call->skinfo.rtcp_sock, &rtcp_addr);
-
         call->session = NULL;
 
@@ -1012 +973 @@
      */    
     status = pjmedia_session_info_from_sdp( call->inv->dlg->pool, 
-                                            pjsua.med_endpt, 1,
-                                            &sess_info, &call->skinfo,
+                                            pjsua.med_endpt, 
+                                            1,&sess_info, 
                                             local_sdp, remote_sdp);
     if (status != PJ_SUCCESS) {
@@ -1036 +997 @@
     /* Create session based on session info. */
     status = pjmedia_session_create( pjsua.med_endpt, &sess_info,
+                                     &call->med_tp,
                                      call, &call->session );
     if (status != PJ_SUCCESS) {

pjproject/trunk/pjsip/src/pjsua-lib/pjsua_core.c

@@ -321 +321 @@
               pj_inet_ntoa(skinfo->rtp_addr_name.sin_addr), 
               pj_ntohs(skinfo->rtp_addr_name.sin_port)));
-    PJ_LOG(4,(THIS_FILE, "RTCP UDP socket reachable at %s:%d",
+    PJ_LOG(4,(THIS_FILE, "RTCP socket reachable at %s:%d",
               pj_inet_ntoa(skinfo->rtcp_addr_name.sin_addr), 
               pj_ntohs(skinfo->rtcp_addr_name.sin_port)));
@@ -780 +780 @@
     for (i=0; i<(int)pjsua.max_calls; ++i) {
         status = init_sockets(i==0, &pjsua.calls[i].skinfo);
+        if (status == PJ_SUCCESS)
+            status = pjmedia_transport_udp_attach(pjsua.med_endpt, NULL,
+                                                  &pjsua.calls[i].skinfo,
+                                                  &pjsua.calls[i].med_tp);
         if (status != PJ_SUCCESS) {
             pjsua_perror(THIS_FILE, "init_sockets() has returned error", 
@@ -787 +791 @@
                 pj_sock_close(pjsua.sip_sock);
             while (i >= 0) {
-                pj_sock_close(pjsua.calls[i].skinfo.rtp_sock);
-                pj_sock_close(pjsua.calls[i].skinfo.rtcp_sock);
+                pjmedia_transport_udp_close(pjsua.calls[i].med_tp);
             }
             return status;
@@ -1049 +1052 @@
 #endif  /* PJMEDIA_HAS_L16_CODEC */
 
+
+    /* Close transports */
+    for (i=0; i<pjsua.call_cnt; ++i) {
+        pjmedia_transport_udp_close(pjsua.calls[i].med_tp);
+    }
+
     /* Destroy media endpoint. */