Ticket #438: added delaybuf in splitcomb. Please see the checkin comment in ticket #438 for the details

git-svn-id: http://svn.pjsip.org/repos/pjproject/trunk@1840 74dad513-b988-da41-8d7b-12977e46ad98

2 files changed, 304 insertions, 188 deletions

diff --git a/pjmedia/src/pjmedia/delaybuf.c b/pjmedia/src/pjmedia/delaybuf.c
index 5a039a7c..70e27a5d 100644
--- a/pjmedia/src/pjmedia/delaybuf.c
+++ b/pjmedia/src/pjmedia/delaybuf.c
@@ -241,7 +241,7 @@ static void shrink_buffer(pjmedia_delay_buf *b, unsigned erase_cnt)
 
 	b->buf_cnt -= erase_cnt;
 
-	PJ_LOG(5,(b->obj_name,"Successfully shrinking %d samples, "
+	PJ_LOG(5,(b->obj_name,"Overflow, %d samples reduced, "
 		  "buf_cnt=%d", erase_cnt, b->buf_cnt));
     }
 
@@ -532,7 +532,7 @@ PJ_DEF(pj_status_t) pjmedia_delay_buf_reset(pjmedia_delay_buf *b)
 
     pj_lock_release(b->lock);
 
-    PJ_LOG(5,(b->obj_name,"Delay buffer resetted"));
+    PJ_LOG(5,(b->obj_name,"Delay buffer is reset"));
 
     return PJ_SUCCESS;
 }
diff --git a/pjmedia/src/pjmedia/splitcomb.c b/pjmedia/src/pjmedia/splitcomb.c
index 3beef551..84286c9d 100644
--- a/pjmedia/src/pjmedia/splitcomb.c
+++ b/pjmedia/src/pjmedia/splitcomb.c
@@ -17,6 +17,7 @@
  * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA 
  */
 #include <pjmedia/splitcomb.h>
+#include <pjmedia/delaybuf.h>
 #include <pjmedia/errno.h>
 #include <pj/assert.h>
 #include <pj/log.h>
@@ -28,12 +29,40 @@
 #define THIS_FILE	    "splitcomb.c"
 #define TMP_SAMP_TYPE	    pj_int16_t
 
-/* When delay buffer is used, we only need 1 frame buffering */
-#if defined(PJMEDIA_SOUND_USE_DELAYBUF) && PJMEDIA_SOUND_USE_DELAYBUF!=0
-#   define MAX_BUF_CNT	    1
-#else
-#   define MAX_BUF_CNT	    PJMEDIA_SOUND_BUFFER_COUNT
-#endif
+/* Maximum number of channels. */
+#define MAX_CHANNELS	    16
+
+/* Maximum number of buffers to be accommodated by delaybuf */
+#define MAX_BUF_CNT	    PJMEDIA_SOUND_BUFFER_COUNT
+
+/* Maximum number of burst before we pause the media flow */
+#define MAX_BURST	    (buf_cnt + 6)
+
+/* Maximum number of NULL frames received before we pause the
+ * media flow.
+ */
+#define MAX_NULL_FRAMES	    (rport->max_burst)
+
+
+/* Operations */
+#define OP_PUT		    (1)
+#define OP_GET		    (-1)
+
+/* Media flow directions */
+enum sc_dir
+{
+    /* This is the direction from the splitcomb to the downstream
+     * port(s), or when put_frame() is called to the splitcomb.
+     */
+    DIR_DOWNSTREAM,
+
+    /* This is the direction from the downstream port to the splitcomb,
+     * or when get_frame() is called to the splitcomb.
+     */
+    DIR_UPSTREAM
+};
+
+
 
 #if 0
 #   define TRACE_UP_(x)	PJ_LOG(5,x)
@@ -43,13 +72,6 @@
 #   define TRACE_DN_(x)
 #endif
 
-#if 1
-#   define LOG_UP_(x)	PJ_LOG(5,x)
-#   define LOG_DN_(x)	PJ_LOG(5,x)
-#else
-#   define LOG_UP_(x)
-#   define LOG_DN_(x)
-#endif
 
 /*
  * This structure describes the splitter/combiner.
@@ -64,7 +86,7 @@ struct splitcomb
     struct {
 	pjmedia_port *port;
 	pj_bool_t     reversed;
-    } port_desc[64];
+    } port_desc[MAX_CHANNELS];
 
     /* Temporary buffers needed to extract mono frame from
      * multichannel frame. We could use stack for this, but this
@@ -84,27 +106,53 @@ struct reverse_port
     struct splitcomb*parent;
     unsigned	     ch_num;
 
+    /* Maximum burst before media flow is suspended */
+    int		     max_burst;
+
+    /* Maximum NULL frames received before media flow is suspended. */
+    unsigned	     max_null_frames;
+
     /* A reverse port need a temporary buffer to store frame
      * (because of the different phase, see splitcomb.h for details). 
      * Since we can not expect get_frame() and put_frame() to be
-     * called evenly one after another, we use circular buffers to
-     * accomodate the "jitter".
+     * called evenly one after another, we use delay buffers to
+     * accomodate the burst.
+     *
+     * We maintain state for each direction, hence the array. The
+     * array is indexed by direction (sc_dir).
      */
-    unsigned	     buf_cnt;
+    struct {
 
-    /* Downstream is the direction when get_frame() is called to the
-     * splitter/combiner port.
-     */
-    unsigned	     dn_read_pos, dn_write_pos, 
-		     dn_overflow_pos, dn_underflow_pos;
-    pj_int16_t	    *dnstream_buf[MAX_BUF_CNT];
+	/* The delay buffer where frames will be stored */
+	pjmedia_delay_buf   *dbuf;
+
+	/* Flag to indicate that audio flow on this direction
+	 * is currently being suspended (perhaps because nothing
+	 * is processing the frame on the other end).
+	 */
+	pj_bool_t	paused;
+
+	/* Operation level. When the level exceeds a maximum value,
+	 * the media flow on this direction will be paused.
+	 */
+	int		level;
+
+	/* Timestamp. */
+	pj_timestamp	ts;
 
-    /* Upstream is the direction when put_frame() is called to the
-     * splitter/combiner port.
+	/* Number of NULL frames transmitted to this port so far.
+	 * NULL frame indicate that nothing is transmitted, and 
+	 * once we get too many of this, we should pause the media
+	 * flow to reduce processing.
+	 */
+	unsigned	null_cnt;
+
+    } buf[2];
+
+    /* Must have temporary put buffer for the delay buf,
+     * unfortunately.
      */
-    unsigned	     up_read_pos, up_write_pos, 
-		     up_overflow_pos, up_underflow_pos;
-    pj_int16_t	    *upstream_buf[MAX_BUF_CNT];
+    pj_int16_t	      *tmp_up_buf;
 };
 
 
@@ -228,11 +276,12 @@ PJ_DEF(pj_status_t) pjmedia_splitcomb_create_rev_channel( pj_pool_t *pool,
 				      unsigned options,
 				      pjmedia_port **p_chport)
 {
-    const pj_str_t name = pj_str("splitcomb-ch");
+    const pj_str_t name = pj_str("scomb-rev");
     struct splitcomb *sc = (struct splitcomb*) splitcomb;
     struct reverse_port *rport;
-    unsigned i;
+    unsigned buf_cnt;
     pjmedia_port *port;
+    pj_status_t status;
 
     /* Sanity check */
     PJ_ASSERT_RETURN(pool && splitcomb, PJ_EINVAL);
@@ -265,27 +314,37 @@ PJ_DEF(pj_status_t) pjmedia_splitcomb_create_rev_channel( pj_pool_t *pool,
     port->on_destroy = &rport_on_destroy;
 
 
-    rport->buf_cnt = options & 0xFF;
-    if (rport->buf_cnt == 0)
-	rport->buf_cnt = MAX_BUF_CNT;
+    buf_cnt = options & 0xFF;
+    if (buf_cnt == 0)
+	buf_cnt = MAX_BUF_CNT;
 
-    /* Create put buffers */
-    for (i=0; i<rport->buf_cnt; ++i) {
-	rport->dnstream_buf[i]=(pj_int16_t*)
-                               pj_pool_zalloc(pool, port->info.bytes_per_frame);
-	PJ_ASSERT_RETURN(rport->dnstream_buf[i], PJ_ENOMEM);
+    rport->max_burst = MAX_BURST;
+    rport->max_null_frames = MAX_NULL_FRAMES;
+
+    /* Create downstream/put buffers */
+    status = pjmedia_delay_buf_create(pool, "scomb-down",
+				      port->info.clock_rate,
+				      port->info.samples_per_frame,
+				      buf_cnt, -1, 0,
+				      &rport->buf[DIR_DOWNSTREAM].dbuf);
+    if (status != PJ_SUCCESS) {
+	return status;
     }
-    rport->dn_write_pos = rport->buf_cnt/2;
-
-    /* Create get buffers */
-    for (i=0; i<rport->buf_cnt; ++i) {
-	rport->upstream_buf[i] = (pj_int16_t*)
-				 pj_pool_zalloc(pool, 
-						port->info.bytes_per_frame);
-	PJ_ASSERT_RETURN(rport->upstream_buf[i], PJ_ENOMEM);
+
+    /* Create upstream/get buffers */
+    status = pjmedia_delay_buf_create(pool, "scomb-up",
+				      port->info.clock_rate,
+				      port->info.samples_per_frame,
+				      buf_cnt, -1, 0,
+				      &rport->buf[DIR_UPSTREAM].dbuf);
+    if (status != PJ_SUCCESS) {
+	pjmedia_delay_buf_destroy(rport->buf[DIR_DOWNSTREAM].dbuf);
+	return status;
     }
-    rport->up_write_pos = rport->buf_cnt/2;
 
+    /* And temporary upstream/get buffer */
+    rport->tmp_up_buf = (pj_int16_t*)
+	                pj_pool_alloc(pool, port->info.bytes_per_frame);
 
     /* Save port in the splitcomb */
     sc->port_desc[ch_num].port = &rport->base;
@@ -294,7 +353,7 @@ PJ_DEF(pj_status_t) pjmedia_splitcomb_create_rev_channel( pj_pool_t *pool,
 
     /* Done */
     *p_chport = port;
-    return PJ_SUCCESS;
+    return status;
 }
 
 
@@ -335,6 +394,55 @@ static void store_mono_frame( const pj_int16_t *in,
     }
 }
 
+/* Update operation on the specified direction  */
+static void op_update(struct reverse_port *rport, int dir, int op)
+{
+    char *dir_name[2] = {"downstream", "upstream"};
+
+    rport->buf[dir].level += op;
+
+    if (op == OP_PUT) {
+	rport->buf[dir].ts.u64 += rport->base.info.samples_per_frame;
+    }
+
+    if (rport->buf[dir].paused) {
+	if (rport->buf[dir].level < -rport->max_burst) {
+	    /* Prevent the level from overflowing and resets back to zero */
+	    rport->buf[dir].level = -rport->max_burst;
+	} else if (rport->buf[dir].level > rport->max_burst) {
+	    /* Prevent the level from overflowing and resets back to zero */
+	    rport->buf[dir].level = rport->max_burst;
+	} else {
+	    /* Level has fallen below max level, we can resume
+	     * media flow.
+	     */
+	    PJ_LOG(5,(rport->base.info.name.ptr, 
+		      "Resuming media flow on %s direction (level=%d)", 
+		      dir_name[dir], rport->buf[dir].level));
+	    rport->buf[dir].level = 0;
+	    rport->buf[dir].paused = PJ_FALSE;
+
+	    //This will cause disruption in audio, and it seems to be
+	    //working fine without this anyway, so we disable it for now.
+	    //pjmedia_delay_buf_learn(rport->buf[dir].dbuf);
+
+	}
+    } else {
+	if (rport->buf[dir].level >= rport->max_burst ||
+	    rport->buf[dir].level <= -rport->max_burst) 
+	{
+	    /* Level has reached maximum level, the other side of
+	     * rport is not sending/retrieving frames. Pause the
+	     * rport on this direction.
+	     */
+	    PJ_LOG(5,(rport->base.info.name.ptr, 
+		      "Pausing media flow on %s direction (level=%d)", 
+		      dir_name[dir], rport->buf[dir].level));
+	    rport->buf[dir].paused = PJ_TRUE;
+	}
+    }
+}
+
 
 /*
  * "Write" a multichannel frame. This would split the multichannel frame
@@ -353,7 +461,50 @@ static pj_status_t put_frame(pjmedia_port *this_port,
 
 	    if (!port) continue;
 
-	    pjmedia_port_put_frame(port, frame);
+	    if (!sc->port_desc[ch].reversed) {
+		pjmedia_port_put_frame(port, frame);
+	    } else {
+		struct reverse_port *rport = (struct reverse_port*)port;
+
+		/* Write zero port to delaybuf so that it doesn't underflow. 
+		 * If we don't do this, get_frame() on this direction will
+		 * cause delaybuf to generate missing frame and the last
+		 * frame transmitted to delaybuf will be replayed multiple
+		 * times, which doesn't sound good.
+		 */
+
+		/* Update the number of NULL frames received. Once we have too
+		 * many of this, we'll stop calling op_update() to let the
+		 * media be suspended.
+		 */
+
+		if (++rport->buf[DIR_DOWNSTREAM].null_cnt > 
+			rport->max_null_frames) 
+		{
+		    /* Prevent the counter from overflowing and resetting
+		     * back to zero
+		     */
+		    rport->buf[DIR_DOWNSTREAM].null_cnt = 
+			rport->max_null_frames + 1;
+		    continue;
+		}
+
+		/* Update rport state. */
+		op_update(rport, DIR_DOWNSTREAM, OP_PUT);
+
+		/* Discard frame if rport is paused on this direction */
+		if (rport->buf[DIR_DOWNSTREAM].paused)
+		    continue;
+
+		/* Generate zero frame. */
+		pjmedia_zero_samples(rport->tmp_up_buf, 
+				     this_port->info.samples_per_frame);
+
+		/* Put frame to delay buffer */
+		pjmedia_delay_buf_put(rport->buf[DIR_DOWNSTREAM].dbuf,
+				      rport->tmp_up_buf);
+
+	    }
 	}
 	return PJ_SUCCESS;
     }
@@ -373,17 +524,17 @@ static pj_status_t put_frame(pjmedia_port *this_port,
 	if (!port)
 	    continue;
 
+	/* Extract the mono frame to temporary buffer */
+	extract_mono_frame((const pj_int16_t*)frame->buf, sc->put_buf, ch, 
+			   this_port->info.channel_count, 
+			   frame->size * 8 / 
+			     this_port->info.bits_per_sample /
+			     this_port->info.channel_count);
+
 	if (!sc->port_desc[ch].reversed) {
 	    /* Write to normal port */
 	    pjmedia_frame mono_frame;
 
-	    /* Extract the mono frame */
-	    extract_mono_frame((const pj_int16_t*)frame->buf, sc->put_buf, ch, 
-			       this_port->info.channel_count, 
-			       frame->size * 8 / 
-				 this_port->info.bits_per_sample /
-				 this_port->info.channel_count);
-
 	    mono_frame.buf = sc->put_buf;
 	    mono_frame.size = frame->size / this_port->info.channel_count;
 	    mono_frame.type = frame->type;
@@ -395,36 +546,17 @@ static pj_status_t put_frame(pjmedia_port *this_port,
 	} else {
 	    /* Write to reversed phase port */
 	    struct reverse_port *rport = (struct reverse_port*)port;
-	    
-	    if (rport->dn_write_pos == rport->dn_read_pos) {
-
-		/* Only report overflow if the frame is constantly read
-		 * by the 'consumer' of the reverse port.
-		 * It is possible that nobody reads the buffer, so causing
-		 * overflow to happen rapidly, and writing log message this
-		 * way does not seem to be wise.
-		 */
-		if (rport->dn_read_pos != rport->dn_overflow_pos) {
-		    rport->dn_overflow_pos = rport->dn_read_pos;
-		    LOG_DN_((THIS_FILE, "Overflow in downstream direction"));
-		}
 
-		/* Adjust write position */
-		rport->dn_write_pos = 
-		    (rport->dn_write_pos + rport->buf_cnt/2) % 
-		    rport->buf_cnt;
-	    }
+	    /* Reset NULL frame counter */
+	    rport->buf[DIR_DOWNSTREAM].null_cnt = 0;
 
-	    /* Extract mono-frame and put it in downstream buffer */
-	    extract_mono_frame((const pj_int16_t*)frame->buf, 
-			       rport->dnstream_buf[rport->dn_write_pos],
-			       ch, this_port->info.channel_count, 
-			       frame->size * 8 / 
-				 this_port->info.bits_per_sample /
-				 this_port->info.channel_count);
+	    /* Update rport state. */
+	    op_update(rport, DIR_DOWNSTREAM, OP_PUT);
 
-	    rport->dn_write_pos = (rport->dn_write_pos + 1) %
-			           rport->buf_cnt;
+	    if (!rport->buf[DIR_DOWNSTREAM].paused) {
+		pjmedia_delay_buf_put(rport->buf[DIR_DOWNSTREAM].dbuf, 
+				      sc->put_buf);
+	    }
 	}
     }
 
@@ -444,21 +576,18 @@ static pj_status_t get_frame(pjmedia_port *this_port,
     unsigned ch;
     pj_bool_t has_frame = PJ_FALSE;
 
-    /* Clear output frame */
-    pjmedia_zero_samples((pj_int16_t*)frame->buf, 
-			 this_port->info.samples_per_frame);
-
     /* Read frame from each port */
     for (ch=0; ch < this_port->info.channel_count; ++ch) {
 	pjmedia_port *port = sc->port_desc[ch].port;
 	pjmedia_frame mono_frame;
 	pj_status_t status;
 
-	if (!port)
-	    continue;
+	if (!port) {
+	    pjmedia_zero_samples(sc->get_buf, 
+				 this_port->info.samples_per_frame /
+				    this_port->info.channel_count);
 
-	/* Read from the port */
-	if (sc->port_desc[ch].reversed == PJ_FALSE) {
+	} else if (sc->port_desc[ch].reversed == PJ_FALSE) {
 	    /* Read from normal port */
 	    mono_frame.buf = sc->get_buf;
 	    mono_frame.size = port->info.bytes_per_frame;
@@ -468,54 +597,37 @@ static pj_status_t get_frame(pjmedia_port *this_port,
 	    if (status != PJ_SUCCESS || 
 		mono_frame.type != PJMEDIA_FRAME_TYPE_AUDIO)
 	    {
-		continue;
+		pjmedia_zero_samples(sc->get_buf, 
+				     port->info.samples_per_frame);
 	    }
 
-	    /* Combine the mono frame into multichannel frame */
-	    store_mono_frame((const pj_int16_t*)mono_frame.buf, 
-			     (pj_int16_t*)frame->buf, ch,
-			     this_port->info.channel_count,
-			     mono_frame.size * 8 /
-				this_port->info.bits_per_sample);
-
 	    frame->timestamp.u64 = mono_frame.timestamp.u64;
 
 	} else {
 	    /* Read from temporary buffer for reverse port */
 	    struct reverse_port *rport = (struct reverse_port*)port;
 
-	    /* Check for underflows */
-	    if (rport->up_read_pos == rport->up_write_pos) {
+	    /* Update rport state. */
+	    op_update(rport, DIR_UPSTREAM, OP_GET);
 
-		/* Only report underflow if the buffer is constantly filled
-		 * up at the other side.
-		 * It is possible that nobody writes the buffer, so causing
-		 * underflow to happen rapidly, and writing log message this
-		 * way does not seem to be wise.
-		 */
-		if (rport->up_write_pos != rport->up_underflow_pos) {
-		    rport->up_underflow_pos = rport->up_write_pos;
-		    LOG_UP_((THIS_FILE, "Underflow in upstream direction"));
-		}
+	    if (!rport->buf[DIR_UPSTREAM].paused) {
+		pjmedia_delay_buf_get(rport->buf[DIR_UPSTREAM].dbuf, 
+				      sc->get_buf);
 
-		/* Adjust read position */
-		rport->up_read_pos = 
-		    (rport->up_write_pos - rport->buf_cnt/2) %
-		    rport->buf_cnt;
+	    } else {
+		pjmedia_zero_samples(sc->get_buf, 
+				      rport->base.info.samples_per_frame);
 	    }
 
-	    TRACE_UP_((THIS_FILE, "Upstream read at buffer pos %d", 
-		       rport->up_read_pos));
+	    frame->timestamp.u64 = rport->buf[DIR_UPSTREAM].ts.u64;
+	}
 
-	    /* Combine the mono frame into multichannel frame */
-	    store_mono_frame((const pj_int16_t*)rport->upstream_buf[rport->up_read_pos], 
-			     (pj_int16_t*)frame->buf, ch,
-			     this_port->info.channel_count,
-			     port->info.samples_per_frame);
+	/* Combine the mono frame into multichannel frame */
+	store_mono_frame(sc->get_buf, 
+			 (pj_int16_t*)frame->buf, ch,
+			 this_port->info.channel_count,
+			 this_port->info.samples_per_frame);
 
-	    rport->up_read_pos = (rport->up_read_pos + 1) %
-				   rport->buf_cnt;
-	}
 
 
 	has_frame = PJ_TRUE;
@@ -548,55 +660,72 @@ static pj_status_t rport_put_frame(pjmedia_port *this_port,
 				   const pjmedia_frame *frame)
 {
     struct reverse_port *rport = (struct reverse_port*) this_port;
-    unsigned count;
 
     pj_assert(frame->size <= rport->base.info.bytes_per_frame);
 
-    /* Check for overflows */
-    if (rport->up_write_pos == rport->up_read_pos) {
+    /* Handle NULL frame */
+    if (frame->type != PJMEDIA_FRAME_TYPE_AUDIO) {
+	TRACE_UP_((THIS_FILE, "Upstream write %d null samples at buf pos %d",
+		   this_port->info.samples_per_frame, rport->up_write_pos));
 
-	/* Only report overflow if the frame is constantly read
-	 * at the other end of the buffer (the multichannel side).
-	 * It is possible that nobody reads the buffer, so causing
-	 * overflow to happen rapidly, and writing log message this
-	 * way does not seem to be wise.
+	/* Write zero port to delaybuf so that it doesn't underflow. 
+	 * If we don't do this, get_frame() on this direction will
+	 * cause delaybuf to generate missing frame and the last
+	 * frame transmitted to delaybuf will be replayed multiple
+	 * times, which doesn't sound good.
 	 */
-	if (rport->up_read_pos != rport->up_overflow_pos) {
-	    rport->up_overflow_pos = rport->up_read_pos;
-	    LOG_UP_((THIS_FILE, "Overflow in upstream direction"));
+
+	/* Update the number of NULL frames received. Once we have too
+	 * many of this, we'll stop calling op_update() to let the
+	 * media be suspended.
+	 */
+	if (++rport->buf[DIR_UPSTREAM].null_cnt > rport->max_null_frames) {
+	    /* Prevent the counter from overflowing and resetting back 
+	     * to zero
+	     */
+	    rport->buf[DIR_UPSTREAM].null_cnt = rport->max_null_frames + 1;
+	    return PJ_SUCCESS;
 	}
 
-	/* Adjust the write position */
-	rport->up_write_pos = (rport->up_read_pos + rport->buf_cnt/2) %
-			       rport->buf_cnt;
-    }
+	/* Update rport state. */
+	op_update(rport, DIR_UPSTREAM, OP_PUT);
 
-    /* Handle NULL frame */
-    if (frame->type != PJMEDIA_FRAME_TYPE_AUDIO) {
-	TRACE_UP_((THIS_FILE, "Upstream write %d null samples at buf pos %d",
-		   this_port->info.samples_per_frame, rport->up_write_pos));
-	pjmedia_zero_samples(rport->upstream_buf[rport->up_write_pos],
+	/* Discard frame if rport is paused on this direction */
+	if (rport->buf[DIR_UPSTREAM].paused)
+	    return PJ_SUCCESS;
+
+	/* Generate zero frame. */
+	pjmedia_zero_samples(rport->tmp_up_buf, 
 			     this_port->info.samples_per_frame);
-	rport->up_write_pos = (rport->up_write_pos+1) % rport->buf_cnt;
-	return PJ_SUCCESS;
+
+	/* Put frame to delay buffer */
+	return pjmedia_delay_buf_put(rport->buf[DIR_UPSTREAM].dbuf, 
+				     rport->tmp_up_buf);
     }
 
     /* Not sure how to handle partial frame, so better reject for now */
     PJ_ASSERT_RETURN(frame->size == this_port->info.bytes_per_frame,
 		     PJ_EINVAL);
 
-    /* Copy normal frame to curcular buffer */
-    count = frame->size * 8 / this_port->info.bits_per_sample;
+    /* Reset NULL frame counter */
+    rport->buf[DIR_UPSTREAM].null_cnt = 0;
 
-    TRACE_UP_((THIS_FILE, "Upstream write %d samples at buf pos %d",
-	       count, rport->up_write_pos));
+    /* Update rport state. */
+    op_update(rport, DIR_UPSTREAM, OP_PUT);
 
+    /* Discard frame if rport is paused on this direction */
+    if (rport->buf[DIR_UPSTREAM].paused)
+	return PJ_SUCCESS;
 
-    pjmedia_copy_samples(rport->upstream_buf[rport->up_write_pos],
-			 (const pj_int16_t*) frame->buf, count);
-    rport->up_write_pos = (rport->up_write_pos+1) % rport->buf_cnt;
+    /* Unfortunately must copy to temporary buffer since delay buf
+     * modifies the frame content.
+     */
+    pjmedia_copy_samples(rport->tmp_up_buf, (const pj_int16_t*)frame->buf,
+		         this_port->info.samples_per_frame);
 
-    return PJ_SUCCESS;
+    /* Put frame to delay buffer */
+    return pjmedia_delay_buf_put(rport->buf[DIR_UPSTREAM].dbuf, 
+				 rport->tmp_up_buf);
 }
 
 
@@ -607,47 +736,34 @@ static pj_status_t rport_get_frame(pjmedia_port *this_port,
 				   pjmedia_frame *frame)
 {
     struct reverse_port *rport = (struct reverse_port*) this_port;
-    unsigned count;
-
-    count = rport->base.info.samples_per_frame;
-
-    frame->size = this_port->info.bytes_per_frame;
-    frame->type = PJMEDIA_FRAME_TYPE_AUDIO;
 
-    /* Check for underflows */
-    if (rport->dn_read_pos == rport->dn_write_pos) {
+    /* Update state */
+    op_update(rport, DIR_DOWNSTREAM, OP_GET);
 
-	/* Only report underflow if the buffer is constantly filled
-	 * up at the other side.
-	 * It is possible that nobody writes the buffer, so causing
-	 * underflow to happen rapidly, and writing log message this
-	 * way does not seem to be wise.
-	 */
-	if (rport->dn_write_pos != rport->dn_underflow_pos) {
-	    rport->dn_underflow_pos = rport->dn_write_pos;
-	    LOG_DN_((THIS_FILE, "Underflow in downstream direction"));
-	}
-
-	/* Adjust read position */
-	rport->dn_read_pos = 
-	    (rport->dn_write_pos - rport->buf_cnt/2) % rport->buf_cnt;
-	
+    /* Return no frame if media flow on this direction is being
+     * paused.
+     */
+    if (rport->buf[DIR_DOWNSTREAM].paused) {
+	frame->type = PJMEDIA_FRAME_TYPE_NONE;
+	return PJ_SUCCESS;
     }
 
-    /* Get the samples from the circular buffer */
-    pjmedia_copy_samples((pj_int16_t*)frame->buf, 
-			 rport->dnstream_buf[rport->dn_read_pos],
-			 count);
-    rport->dn_read_pos = (rport->dn_read_pos+1) % rport->buf_cnt;
+    /* Get frame from delay buffer */
+    frame->size = this_port->info.bytes_per_frame;
+    frame->type = PJMEDIA_FRAME_TYPE_AUDIO;
+    frame->timestamp.u64 = rport->buf[DIR_DOWNSTREAM].ts.u64;
 
-    return PJ_SUCCESS;
+    return pjmedia_delay_buf_get(rport->buf[DIR_DOWNSTREAM].dbuf, 
+				 (short*)frame->buf);
 }
 
 
 static pj_status_t rport_on_destroy(pjmedia_port *this_port)
 {
-    /* Nothing to do */
-    PJ_UNUSED_ARG(this_port);
+    struct reverse_port *rport = (struct reverse_port*) this_port;
+
+    pjmedia_delay_buf_destroy(rport->buf[DIR_DOWNSTREAM].dbuf);
+    pjmedia_delay_buf_destroy(rport->buf[DIR_UPSTREAM].dbuf);
 
     return PJ_SUCCESS;
 }