Index: pjproject/trunk/pjmedia/src/pjmedia/stream.c
===================================================================
--- pjproject/trunk/pjmedia/src/pjmedia/stream.c	(revision 450)
+++ pjproject/trunk/pjmedia/src/pjmedia/stream.c	(revision 452)
@@ -86,4 +86,6 @@
     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,24 +94,8 @@
     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,22 +410,10 @@
 	
 	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,9 +561,8 @@
     /* 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,247 +656,148 @@
 
 /*
- * 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,4 +868,5 @@
 					   pj_pool_t *pool,
 					   const pjmedia_stream_info *info,
+					   pjmedia_transport *tp,
 					   void *user_data,
 					   pjmedia_stream **p_stream)
@@ -999,6 +874,4 @@
 {
     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,9 +910,4 @@
     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,4 +916,13 @@
     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,43 +1038,5 @@
 	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,12 +1067,8 @@
 
 
-    /* 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,4 +1101,13 @@
     *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;
 }
 
