Revision: 85
Author: [email protected]
Date: Mon Feb 18 00:06:43 2013
Log: red5sip: added jitter buffer to RTP->RTMP channel
http://code.google.com/p/red5phone/source/detail?r=85
Modified:
/branches/red5sip/red5sip.iml
/branches/red5sip/src/java/org/red5/sip/app/RTPStreamReceiver.java
=======================================
--- /branches/red5sip/red5sip.iml Wed Feb 6 09:31:44 2013
+++ /branches/red5sip/red5sip.iml Mon Feb 18 00:06:43 2013
@@ -513,6 +513,7 @@
</orderEntry>
<orderEntry type="library" name="lib" level="project" />
<orderEntry type="library" name="Unnamed" level="project" />
+ <orderEntry type="library" name="libresample4j" level="project" />
</component>
</module>
=======================================
--- /branches/red5sip/src/java/org/red5/sip/app/RTPStreamReceiver.java Sun
Feb 17 20:38:26 2013
+++ /branches/red5sip/src/java/org/red5/sip/app/RTPStreamReceiver.java Mon
Feb 18 00:06:43 2013
@@ -1,5 +1,6 @@
package org.red5.sip.app;
+import java.io.IOException;
import java.net.DatagramSocket;
import local.net.RtpPacket;
@@ -105,23 +106,71 @@
return;
}
- byte[] codedBuffer = new byte[
sipCodec.getIncomingEncodedFrameSize() ];
+// byte[] codedBuffer = new byte[
sipCodec.getIncomingEncodedFrameSize() ];
byte[] internalBuffer = new
byte[sipCodec.getIncomingEncodedFrameSize() + RTP_HEADER_SIZE ];
RtpPacket rtpPacket = new RtpPacket( internalBuffer, 0 );
running = true;
- try {
+ final int BUFFER_LENGTH = 100;
+ final BytesBuffer buffer = new BytesBuffer(
sipCodec.getIncomingEncodedFrameSize(), BUFFER_LENGTH );
- rtp_socket.getDatagramSocket().setSoTimeout( SO_TIMEOUT );
+ Thread sendThread = new Thread(new Runnable() {
+ public void run() {
+ boolean ready = false;
+ int avail = 0;
+ while(running) {
+ byte[] codedBuffer = new byte[
sipCodec.getIncomingEncodedFrameSize() ];
+ synchronized (buffer) {
+ avail = buffer.available();
+ if(!ready) {
+ if(avail > 30) {
+ ready = true;
+ }
+ } else {
+ if(avail == 0 ) {
+ ready = false;
+ }
+ }
+ if(ready) {
+ buffer.take(codedBuffer, 0);
+ }
+ }
+ if(ready) {
+ timeStamp += sipCodec.getIncomingPacketization();
+ try {
+ rtmpUser.pushAudio(codedBuffer, timeStamp,
130);
+ try {
- float[] decodingBuffer = new float[
sipCodec.getIncomingDecodedFrameSize() ];
+ long pause =
sipCodec.getOutgoingPacketization();
+ if(avail > BUFFER_LENGTH / 2) {
+ pause -= 5;
+ }
+ if(avail > BUFFER_LENGTH / 5) {
+ pause -= 1;
+ }
+ log.trace("Sleep pause: " + pause);
+ //System.out.println("pushAudio, ts: " +
timeStamp + ", pause: " + pause);
+ Thread.sleep( pause, 800000 );
+
//Thread.sleep(sipCodec.getIncomingPacketization() - 1);
+ } catch (InterruptedException e) {
+
+ }
+ } catch (IOException e) {
+ log.error("rtmpUser.pushAudio", e);
+ }
+ }
+ Thread.yield();
+ }
+ }
+ });
+
+ sendThread.start();
+
+ try {
- println( "run",
- "internalBuffer.length = " + internalBuffer.length
- + ", codedBuffer.length = " + codedBuffer.length
- + ", decodingBuffer.length = " + decodingBuffer.length
+ "." );
+ rtp_socket.getDatagramSocket().setSoTimeout( SO_TIMEOUT );
while ( running ) {
@@ -137,10 +186,13 @@
if(payloadType < 20)
{
- System.arraycopy(packetBuffer, offset, codedBuffer, 0,
sipCodec.getIncomingEncodedFrameSize());
- //timeStamp = (System.currentTimeMillis() -
start);
- timeStamp +=
sipCodec.getIncomingPacketization();
-
rtmpUser.pushAudio(codedBuffer, timeStamp, 130);
+ //System.arraycopy(packetBuffer, offset, codedBuffer, 0,
sipCodec.getIncomingEncodedFrameSize());
+ synchronized (buffer) {
+ System.out.println(String.format("RTP:
ssrc[%d], cscr count %d", rtpPacket.getSscr(), rtpPacket.getCscrCount()));
+ buffer.push(packetBuffer, offset,
sipCodec.getIncomingEncodedFrameSize());
+ }
+// timeStamp +=
sipCodec.getIncomingPacketization();
+//
rtmpUser.pushAudio(codedBuffer, timeStamp, 130);
}
}
}
