Transmitting and Receiving RTP over Custom Transport Layer

Transmitting and Receiving RTP over Custom Transport Layer [RTPConnector] Problem The default implementation of RTPManager uses UDP/IP for transport. This sample shows how you can override that and use RTP over your own custom transport mechanism. Solution An implementation of RTPConnector, RTPSocketAdapter is built to support a custom transport mechanism. For simplicity, UDP sockets are used in this example to illustrate the concepts. The AVTransmit and AVReceive samples were modified to make use of RTPSocketAdapter for sending and receiving of the RTP data. Requirements Related Classes How to run this sample Source Code See Also Requirements Platform: JDK 1.1.6 or later JMF API: 2.1.1 or later. Implementation: AJ, WPP, SPP * Hardware: [optional] Capture card required for live capture. * AJ = All Java, WPP = Windows Performance Pack, SPP = Solaris Performance Pack

Related Classes

javax.media.Manager javax.media.Processor javax.media.protocol.DataSource javax.media.rtp.RTPManager javax.media.rtp.RTPConnector

How to run this sample

Download and compile the source files. Run AVTransmit3 and specifies the RTP session address to send to. Run AVReceive3 and specifies the RTP session addresses to receive from. For example, java AVTransmit3 224.1.2.3 42050 file:/temp/foo.mov java AVReceive3 224.1.2.3/42050 224.1.2.3/42052 The above example assumes that foo.mov contains 2 different tracks and so 2 sessions were created on the client side to receive both streams.

Source Code

RTPSocketAdapter.java AVTransmit3.java AVReceive3.java

See Also

JMF 2.0 Programmer's Guide

——————————————————————————————————————————————————————————————————

源代码链接已失效，网上找了个

/*
* @(#)RTPSocketAdapter.java 1.2 01/03/13
*
* Copyright (c) 1999-2001 Sun Microsystems, Inc. All Rights Reserved.
*
* Sun grants you ("Licensee") a non-exclusive, royalty free, license to use,
* modify and redistribute this software in source and binary code form,
* provided that i) this copyright notice and license appear on all copies of
* the software; and ii) Licensee does not utilize the software in a manner
* which is disparaging to Sun.
*
* This software is provided "AS IS," without a warranty of any kind. ALL
* EXPRESS OR IMPLIED CONDITIONS, REPRESENTATIONS AND WARRANTIES, INCLUDING ANY
* IMPLIED WARRANTY OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE OR
* NON-INFRINGEMENT, ARE HEREBY EXCLUDED. SUN AND ITS LICENSORS SHALL NOT BE
* LIABLE FOR ANY DAMAGES SUFFERED BY LICENSEE AS A RESULT OF USING, MODIFYING
* OR DISTRIBUTING THE SOFTWARE OR ITS DERIVATIVES. IN NO EVENT WILL SUN OR ITS
* LICENSORS BE LIABLE FOR ANY LOST REVENUE, PROFIT OR DATA, OR FOR DIRECT,
* INDIRECT, SPECIAL, CONSEQUENTIAL, INCIDENTAL OR PUNITIVE DAMAGES, HOWEVER
* CAUSED AND REGARDLESS OF THE THEORY OF LIABILITY, ARISING OUT OF THE USE OF
* OR INABILITY TO USE SOFTWARE, EVEN IF SUN HAS BEEN ADVISED OF THE
* POSSIBILITY OF SUCH DAMAGES.
*
* This software is not designed or intended for use in on-line control of
* aircraft, air traffic, aircraft navigation or aircraft communications; or in
* the design, construction, operation or maintenance of any nuclear
* facility. Licensee represents and warrants that it will not use or
* redistribute the Software for such purposes.
*/

package edu.depaul.mocha.collab.audio.component;
import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.MulticastSocket;
import java.net.SocketException;

import javax.media.protocol.ContentDescriptor;
import javax.media.protocol.PushSourceStream;
import javax.media.protocol.SourceTransferHandler;
import javax.media.rtp.OutputDataStream;
import javax.media.rtp.RTPConnector;

/**
* An implementation of RTPConnector based on UDP sockets.
*/
public class RTPSocketAdapter implements RTPConnector {

DatagramSocket dataSock;
DatagramSocket ctrlSock;

InetAddress addr;
int port;

SockInputStream dataInStrm = null, ctrlInStrm = null;
SockOutputStream dataOutStrm = null, ctrlOutStrm = null;

public RTPSocketAdapter(InetAddress addr, int port) throws IOException {
this(addr, port, 1);
}

public RTPSocketAdapter(InetAddress addr, int port, int ttl) throws IOException {

try {

if (addr.isMulticastAddress()) {
dataSock = new MulticastSocket(port);
ctrlSock = new MulticastSocket(port+1);
((MulticastSocket)dataSock).joinGroup(addr);
((MulticastSocket)dataSock).setTimeToLive(ttl);
((MulticastSocket)ctrlSock).joinGroup(addr);
((MulticastSocket)ctrlSock).setTimeToLive(ttl);
} else {
dataSock = new DatagramSocket(port, InetAddress.getLocalHost());
ctrlSock = new DatagramSocket(port+1, InetAddress.getLocalHost());
}

} catch (SocketException e) {
throw new IOException(e.getMessage());
}

this.addr = addr;
this.port = port;
}

/**
* Returns an input stream to receive the RTP data.
*/
public PushSourceStream getDataInputStream() throws IOException {
if (dataInStrm == null) {
dataInStrm = new SockInputStream(dataSock, addr, port);
dataInStrm.start();
}
return dataInStrm;
}

/**
* Returns an output stream to send the RTP data.
*/
public OutputDataStream getDataOutputStream() throws IOException {
if (dataOutStrm == null)
dataOutStrm = new SockOutputStream(dataSock, addr, port);
return dataOutStrm;
}

/**
* Returns an input stream to receive the RTCP data.
*/
public PushSourceStream getControlInputStream() throws IOException {
if (ctrlInStrm == null) {
ctrlInStrm = new SockInputStream(ctrlSock, addr, port+1);
ctrlInStrm.start();
}
return ctrlInStrm;
}

/**
* Returns an output stream to send the RTCP data.
*/
public OutputDataStream getControlOutputStream() throws IOException {
if (ctrlOutStrm == null)
ctrlOutStrm = new SockOutputStream(ctrlSock, addr, port+1);
return ctrlOutStrm;
}

/**
* Close all the RTP, RTCP streams.
*/
public void close() {
if (dataInStrm != null)
dataInStrm.kill();
if (ctrlInStrm != null)
ctrlInStrm.kill();
dataSock.close();
ctrlSock.close();
}

/**
* Set the receive buffer size of the RTP data channel.
* This is only a hint to the implementation. The actual implementation
* may not be able to do anything to this.
*/
public void setReceiveBufferSize( int size) throws IOException {
dataSock.setReceiveBufferSize(size);
}

/**
* Get the receive buffer size set on the RTP data channel.
* Return -1 if the receive buffer size is not applicable for
* the implementation.
*/
public int getReceiveBufferSize() {
try {
return dataSock.getReceiveBufferSize();
} catch (Exception e) {
return -1;
}
}

/**
* Set the send buffer size of the RTP data channel.
* This is only a hint to the implementation. The actual implementation
* may not be able to do anything to this.
*/
public void setSendBufferSize( int size) throws IOException {
dataSock.setSendBufferSize(size);
}

/**
* Get the send buffer size set on the RTP data channel.
* Return -1 if the send buffer size is not applicable for
* the implementation.
*/
public int getSendBufferSize() {
try {
return dataSock.getSendBufferSize();
} catch (Exception e) {
return -1;
}
}

/**
* Return the RTCP bandwidth fraction. This value is used to
* initialize the RTPManager. Check RTPManager for more detauls.
* Return -1 to use the default values.
*/
public double getRTCPBandwidthFraction() {
return -1;
}

/**
* Return the RTCP sender bandwidth fraction. This value is used to
* initialize the RTPManager. Check RTPManager for more detauls.
* Return -1 to use the default values.
*/
public double getRTCPSenderBandwidthFraction() {
return -1;
}

/**
* An inner class to implement an OutputDataStream based on UDP sockets.
*/
class SockOutputStream implements OutputDataStream {

DatagramSocket sock;
InetAddress addr;
int port;

public SockOutputStream(DatagramSocket sock, InetAddress addr, int port) {
this.sock = sock;
this.addr = addr;
this.port = port;
}

public int write(byte data[], int offset, int len) {
try {
sock.send(new DatagramPacket(data, offset, len, addr, port));
} catch (Exception e) {
return -1;
}
return len;
}
}

/**
* An inner class to implement an PushSourceStream based on UDP sockets.
*/
class SockInputStream extends Thread implements PushSourceStream {

DatagramSocket sock;
InetAddress addr;
int port;
boolean done = false;
boolean dataRead = false;

SourceTransferHandler sth = null;

public SockInputStream(DatagramSocket sock, InetAddress addr, int port) {
this.sock = sock;
this.addr = addr;
this.port = port;
}

public int read(byte buffer[], int offset, int length) {
DatagramPacket p = new DatagramPacket(buffer, offset, length, addr, port);
try {
sock.receive(p);
} catch (IOException e) {
return -1;
}
synchronized (this) {
dataRead = true;
notify();
}
return p.getLength();
}

public synchronized void start() {
super.start();
if (sth != null) {
dataRead = true;
notify();
}
}

public synchronized void kill() {
done = true;
notify();
}

public int getMinimumTransferSize() {
return 2 * 1024; // twice the MTU size, just to be safe.
}

public synchronized void setTransferHandler(SourceTransferHandler sth) {
this.sth = sth;
dataRead = true;
notify();
}

// Not applicable.
public ContentDescriptor getContentDescriptor() {
return null;
}

// Not applicable.
public long getContentLength() {
return LENGTH_UNKNOWN;
}

// Not applicable.
public boolean endOfStream() {
return false;
}

// Not applicable.
public Object[] getControls() {
return new Object[0];
}

// Not applicable.
public Object getControl(String type) {
return null;
}

/**
* Loop and notify the transfer handler of new data.
*/
public void run() {
while (!done) {

synchronized (this) {
while (!dataRead && !done) {
try {
wait();
} catch (InterruptedException e) { }
}
dataRead = false;
}

if (sth != null && !done) {
sth.transferData(this);
}
}
}
}
}

/*
* @(#)AVTransmit3.java 1.2 01/03/13
*
* Copyright (c) 1999-2001 Sun Microsystems, Inc. All Rights Reserved.
*
* Sun grants you ("Licensee") a non-exclusive, royalty free, license to use,
* modify and redistribute this software in source and binary code form,
* provided that i) this copyright notice and license appear on all copies of
* the software; and ii) Licensee does not utilize the software in a manner
* which is disparaging to Sun.
*
* This software is provided "AS IS," without a warranty of any kind. ALL
* EXPRESS OR IMPLIED CONDITIONS, REPRESENTATIONS AND WARRANTIES, INCLUDING ANY
* IMPLIED WARRANTY OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE OR
* NON-INFRINGEMENT, ARE HEREBY EXCLUDED. SUN AND ITS LICENSORS SHALL NOT BE
* LIABLE FOR ANY DAMAGES SUFFERED BY LICENSEE AS A RESULT OF USING, MODIFYING
* OR DISTRIBUTING THE SOFTWARE OR ITS DERIVATIVES. IN NO EVENT WILL SUN OR ITS
* LICENSORS BE LIABLE FOR ANY LOST REVENUE, PROFIT OR DATA, OR FOR DIRECT,
* INDIRECT, SPECIAL, CONSEQUENTIAL, INCIDENTAL OR PUNITIVE DAMAGES, HOWEVER
* CAUSED AND REGARDLESS OF THE THEORY OF LIABILITY, ARISING OUT OF THE USE OF
* OR INABILITY TO USE SOFTWARE, EVEN IF SUN HAS BEEN ADVISED OF THE
* POSSIBILITY OF SUCH DAMAGES.
*
* This software is not designed or intended for use in on-line control of
* aircraft, air traffic, aircraft navigation or aircraft communications; or in
* the design, construction, operation or maintenance of any nuclear
* facility. Licensee represents and warrants that it will not use or
* redistribute the Software for such purposes.
*/

package edu.depaul.mocha.collab.audio.component;
import java.awt.Dimension;
import java.io.IOException;
import java.net.InetAddress;
import javax.media.Codec;
import javax.media.Control;
import javax.media.Controller;
import javax.media.ControllerClosedEvent;
import javax.media.ControllerEvent;
import javax.media.ControllerListener;
import javax.media.Format;
import javax.media.MediaLocator;
import javax.media.NoDataSourceException;
import javax.media.NoProcessorException;
import javax.media.Owned;
import javax.media.Player;
import javax.media.Processor;
import javax.media.control.QualityControl;
import javax.media.control.TrackControl;
import javax.media.format.VideoFormat;
import javax.media.protocol.ContentDescriptor;
import javax.media.protocol.DataSource;
import javax.media.protocol.PushBufferDataSource;
import javax.media.protocol.PushBufferStream;
import javax.media.rtp.RTPManager;
import javax.media.rtp.SendStream;

public class AVTransmit3 {

// Input MediaLocator
// Can be a file or http or capture source
private MediaLocator locator;
private String ipAddress;
private int portBase;

private Processor processor = null;
private RTPManager rtpMgrs[];
private DataSource dataOutput = null;

public AVTransmit3(MediaLocator locator,
String ipAddress,
String pb,
Format format) {

this.locator = locator;
this.ipAddress = ipAddress;
Integer integer = Integer.valueOf(pb);
if (integer != null)
this.portBase = integer.intValue();
}

/**
* Starts the transmission. Returns null if transmission started ok.
* Otherwise it returns a string with the reason why the setup failed.
*/
public synchronized String start() {
String result;

// Create a processor for the specified media locator
// and program it to output JPEG/RTP
result = createProcessor();
if (result != null)
return result;

// Create an RTP session to transmit the output of the
// processor to the specified IP address and port no.
result = createTransmitter();
if (result != null) {
processor.close();
processor = null;
return result;
}

// Start the transmission
processor.start();

return null;
}

/**
* Stops the transmission if already started
*/
public void stop() {
synchronized (this) {
if (processor != null) {
processor.stop();
processor.close();
processor = null;
for (int i = 0; i < rtpMgrs.length; i++) {
rtpMgrs[i].removeTargets( "Session ended.");
rtpMgrs[i].dispose();
}
}
}
}

private String createProcessor() {
if (locator == null)
return "Locator is null";

DataSource ds = null;

/* THIS CODE FOR TESTING
Vector datasources = javax.media.Manager.getDataSourceList(locator.getProtocol());

for (Iterator iter = datasources.iterator(); iter.hasNext();)
{
System.out.println(iter.next());
}
*/

try {
ds = javax.media.Manager.createDataSource(locator);

}
catch (NoDataSourceException e)
{
/*We'll handle this exception better in iteration 2*/
return "Couldn't create DataSource";
}
catch (IOException e)
{
/*We'll handle this exception better in iteration 2*/
return "IOException Caught"; //e.printStackTrace();
}

// Try to create a processor to handle the input media locator
try {
processor = javax.media.Manager.createProcessor(ds);
} catch (NoProcessorException npe) {
return "Couldn't create processor";
} catch (IOException ioe) {
return "IOException creating processor";
}

// Wait for it to configure
boolean result = waitForState(processor, Processor.Configured);
if (result == false)
return "Couldn't configure processor";

// Get the tracks from the processor
TrackControl [] tracks = processor.getTrackControls();

// Do we have atleast one track?
if (tracks == null || tracks.length < 1)
return "Couldn't find tracks in processor";

// Set the output content descriptor to RAW_RTP
// This will limit the supported formats reported from
// Track.getSupportedFormats to only valid RTP formats.
ContentDescriptor cd = new ContentDescriptor(ContentDescriptor.RAW_RTP);
processor.setContentDescriptor(cd);

Format supported[];
Format chosen;
boolean atLeastOneTrack = false;

// Program the tracks.
for (int i = 0; i < tracks.length; i++) {
if (tracks[i].isEnabled()) {

supported = tracks[i].getSupportedFormats();

// We've set the output content to the RAW_RTP.
// So all the supported formats should work with RTP.
// We'll just pick the first one.

if (supported.length > 0) {
if (supported[0] instanceof VideoFormat) {
// For video formats, we should double check the
// sizes since not all formats work in all sizes.
chosen = checkForVideoSizes(tracks[i].getFormat(),
supported[0]);
} else
chosen = supported[0];
tracks[i].setFormat(chosen);
System.err.println("Track " + i + " is set to transmit as:");
System.err.println(" " + chosen);
atLeastOneTrack = true;
} else
tracks[i].setEnabled(false);
} else
tracks[i].setEnabled(false);
}

if (!atLeastOneTrack)
return "Couldn't set any of the tracks to a valid RTP format";

// Realize the processor. This will internally create a flow
// graph and attempt to create an output datasource for JPEG/RTP
// audio frames.
result = waitForState(processor, Controller.Realized);
if (result == false)
return "Couldn't realize processor";

// Set the JPEG quality to .5.
setJPEGQuality(processor, 0.5f);

// Get the output data source of the processor
dataOutput = processor.getDataOutput();

return null;
}

/**
* Use the RTPManager API to create sessions for each media
* track of the processor.
*/
private String createTransmitter() {

// Cheated. Should have checked the type.
PushBufferDataSource pbds = (PushBufferDataSource)dataOutput;
PushBufferStream pbss[] = pbds.getStreams();

rtpMgrs = new RTPManager[pbss.length];
SendStream sendStream;
int port;
for (int i = 0; i < pbss.length; i++) {
try {
rtpMgrs[i] = RTPManager.newInstance();

port = portBase + 2*i;

// Initialize the RTPManager with the RTPSocketAdapter
rtpMgrs[i].initialize(new RTPSocketAdapter(
InetAddress.getByName(ipAddress),
port));

System.err.println( "Created RTP session: " + ipAddress + " " + port);

sendStream = rtpMgrs[i].createSendStream(dataOutput, i);
sendStream.start();
} catch (Exception e) {
return e.getMessage();
}
}

return null;
}

/**
* For JPEG and H263, we know that they only work for particular
* sizes. So we'll perform extra checking here to make sure they
* are of the right sizes.
*/
Format checkForVideoSizes(Format original, Format supported) {

int width, height;
Dimension size = ((VideoFormat)original).getSize();
Format jpegFmt = new Format(VideoFormat.JPEG_RTP);
Format h263Fmt = new Format(VideoFormat.H263_RTP);

if (supported.matches(jpegFmt)) {
// For JPEG, make sure width and height are divisible by 8.
width = (size.width % 8 == 0 ? size.width :
(int)(size.width / 8) * 8);
height = (size.height % 8 == 0 ? size.height :
(int)(size.height / 8) * 8);
} else if (supported.matches(h263Fmt)) {
// For H.263, we only support some specific sizes.
if (size.width < 128) {
width = 128;
height = 96;
} else if (size.width < 176) {
width = 176;
height = 144;
} else {
width = 352;
height = 288;
}
} else {
// We don't know this particular format. We'll just
// leave it alone then.
return supported;
}

return (new VideoFormat(null,
new Dimension(width, height),
Format.NOT_SPECIFIED,
null,
Format.NOT_SPECIFIED)).intersects(supported);
}

/**
* Setting the encoding quality to the specified value on the JPEG encoder.
* 0.5 is a good default.
*/
void setJPEGQuality(Player p, float val) {

Control cs[] = p.getControls();
QualityControl qc = null;
VideoFormat jpegFmt = new VideoFormat(VideoFormat.JPEG);

// Loop through the controls to find the Quality control for
// the JPEG encoder.
for (int i = 0; i < cs.length; i++) {

if (cs[i] instanceof QualityControl &&
cs[i] instanceof Owned) {
Object owner = ((Owned)cs[i]).getOwner();

// Check to see if the owner is a Codec.
// Then check for the output format.
if (owner instanceof Codec) {
Format fmts[] = ((Codec)owner).getSupportedOutputFormats(null);
for (int j = 0; j < fmts.length; j++) {
if (fmts[j].matches(jpegFmt)) {
qc = (QualityControl)cs[i];
qc.setQuality(val);
System.err.println("- Setting quality to " +
val + " on " + qc);
break;
}
}
}
if (qc != null)
break;
}
}
}

/****************************************************************
* Convenience methods to handle processor's state changes.
****************************************************************/

private Integer stateLock = new Integer(0);
private boolean failed = false;

Integer getStateLock() {
return stateLock;
}

void setFailed() {
failed = true;
}

private synchronized boolean waitForState(Processor p, int state) {
p.addControllerListener(new StateListener());
failed = false;

// Call the required method on the processor
if (state == Processor.Configured) {
p.configure();
} else if (state == Processor.Realized) {
p.realize();
}

// Wait until we get an event that confirms the
// success of the method, or a failure event.
// See StateListener inner class
while (p.getState() < state && !failed) {
synchronized (getStateLock()) {
try {
getStateLock().wait();
} catch (InterruptedException ie) {
return false;
}
}
}

if (failed)
return false;
else
return true;
}

/****************************************************************
* Inner Classes
****************************************************************/

class StateListener implements ControllerListener {

public void controllerUpdate(ControllerEvent ce) {

// If there was an error during configure or
// realize, the processor will be closed
if (ce instanceof ControllerClosedEvent)
setFailed();

// All controller events, send a notification
// to the waiting thread in waitForState method.
if (ce instanceof ControllerEvent) {
synchronized (getStateLock()) {
getStateLock().notifyAll();
}
}
}
}

/****************************************************************
* Sample Usage for AVTransmit3 class
****************************************************************/

public static void main(String [] args) {
// We need three parameters to do the transmission
// For example,
// java AVTransmit3 file:/C:/media/test.mov 129.130.131.132 42050

if (args.length < 3) {
prUsage();
}

Format fmt = null;
int i = 0;

// Create a audio transmit object with the specified params.
AVTransmit3 at = new AVTransmit3(new MediaLocator(args[i]),
args[i+1], args[i+2], fmt);
// Start the transmission
String result = at.start();

// result will be non-null if there was an error. The return
// value is a String describing the possible error. Print it.
if (result != null) {
System.err.println("Error : " + result);
System.exit(0);
}

System.err.println("Start transmission for 60 seconds...");

// Transmit for 60 seconds and then close the processor
// This is a safeguard when using a capture data source
// so that the capture device will be properly released
// before quitting.
// The right thing to do would be to have a GUI with a
// "Stop" button that would call stop on AVTransmit3
try {
Thread.sleep(60000);
} catch (InterruptedException ie) {
}

// Stop the transmission
at.stop();

System.err.println("...transmission ended.");

System.exit(0);
}

static void prUsage() {
System.err.println("Usage: AVTransmit3 <sourceURL> <destIP> <destPortBase>");
System.err.println(" <sourceURL>: input URL or file name");
System.err.println(" <destIP>: multicast, broadcast or unicast IP address for the transmission");
System.err.println(" <destPortBase>: network port numbers for the transmission.");
System.err.println(" The first track will use the destPortBase.");
System.err.println(" The next track will use destPortBase + 2 and so on.\n");
System.exit(0);
}
}

/*
* @(#)AVReceive3.java 1.2 01/03/13
*
* Copyright (c) 1999-2001 Sun Microsystems, Inc. All Rights Reserved.
*
* Sun grants you ("Licensee") a non-exclusive, royalty free, license to use,
* modify and redistribute this software in source and binary code form,
* provided that i) this copyright notice and license appear on all copies of
* the software; and ii) Licensee does not utilize the software in a manner
* which is disparaging to Sun.
*
* This software is provided "AS IS," without a warranty of any kind. ALL
* EXPRESS OR IMPLIED CONDITIONS, REPRESENTATIONS AND WARRANTIES, INCLUDING ANY
* IMPLIED WARRANTY OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE OR
* NON-INFRINGEMENT, ARE HEREBY EXCLUDED. SUN AND ITS LICENSORS SHALL NOT BE
* LIABLE FOR ANY DAMAGES SUFFERED BY LICENSEE AS A RESULT OF USING, MODIFYING
* OR DISTRIBUTING THE SOFTWARE OR ITS DERIVATIVES. IN NO EVENT WILL SUN OR ITS
* LICENSORS BE LIABLE FOR ANY LOST REVENUE, PROFIT OR DATA, OR FOR DIRECT,
* INDIRECT, SPECIAL, CONSEQUENTIAL, INCIDENTAL OR PUNITIVE DAMAGES, HOWEVER
* CAUSED AND REGARDLESS OF THE THEORY OF LIABILITY, ARISING OUT OF THE USE OF
* OR INABILITY TO USE SOFTWARE, EVEN IF SUN HAS BEEN ADVISED OF THE
* POSSIBILITY OF SUCH DAMAGES.
*
* This software is not designed or intended for use in on-line control of
* aircraft, air traffic, aircraft navigation or aircraft communications; or in
* the design, construction, operation or maintenance of any nuclear
* facility. Licensee represents and warrants that it will not use or
* redistribute the Software for such purposes.
*/

package edu.depaul.mocha.collab.audio.component;
import java.awt.Component;
import java.awt.Dimension;
import java.net.InetAddress;
import java.util.Vector;

import javax.media.ControllerErrorEvent;
import javax.media.ControllerEvent;
import javax.media.ControllerListener;
import javax.media.Player;
import javax.media.RealizeCompleteEvent;
import javax.media.control.BufferControl;
import javax.media.protocol.DataSource;
import javax.media.rtp.Participant;
import javax.media.rtp.RTPControl;
import javax.media.rtp.RTPManager;
import javax.media.rtp.ReceiveStream;
import javax.media.rtp.ReceiveStreamListener;
import javax.media.rtp.SessionListener;
import javax.media.rtp.event.ByeEvent;
import javax.media.rtp.event.NewParticipantEvent;
import javax.media.rtp.event.NewReceiveStreamEvent;
import javax.media.rtp.event.ReceiveStreamEvent;
import javax.media.rtp.event.RemotePayloadChangeEvent;
import javax.media.rtp.event.SessionEvent;
import javax.media.rtp.event.StreamMappedEvent;
import javax.swing.BoxLayout;
import javax.swing.JFrame;
import javax.swing.JLabel;
import javax.swing.JPanel;

import edu.depaul.mocha.applications.swingim.IMUtilities;
import edu.depaul.mocha.applications.swingim.OpenVideoListener;
import edu.depaul.mocha.collab.audiovideo.window.IMChatAVWindow;
import edu.depaul.mocha.collab.data.DataAccessor;
import edu.depaul.mocha.collab.im.services.IMChatServices;
import edu.depaul.mocha.ui.resource.MochaResourceManager;
import edu.depaul.mocha.ui.resource.MochaResourceManagerFactory;
import edu.depaul.mocha.ui.resource.MochaSession;
import edu.depaul.mocha.ui.resource.MochaSessionManager;

/**
* AVReceive3 to receive RTP transmission using the RTPConnector.
* @version $Revision: 1.34 $
*/
public class AVReceive3 implements ReceiveStreamListener, SessionListener,
ControllerListener
{
String sessions[] = null;
RTPManager mgrs[] = null;
Vector<PlayerWindow> playerWindows = null;
PlayerWindow pw = null;

boolean dataReceived = false;
Object dataSync = new Object();
String partner = "";
String name = "";

/**
* Constructor for AVReceive3.
* @param sessions String[]
* @param Partner String
*/
public AVReceive3(String sessions[], String Partner) {
this.sessions = sessions;
this.partner = Partner;

// Get the name of the user
DataAccessor da = new DataAccessor();
name = String.format("%1s %2s", da.getJIDByJIDLoginID(IMChatServices.getJIDName(partner)).getFirstName(), da.getJIDByJIDLoginID(IMChatServices.getJIDName(partner)).getLastName());
}

/**
* Method initialize.
* @return boolean
*/
public boolean initialize() {

try {
mgrs = new RTPManager[sessions.length];
playerWindows = new Vector<PlayerWindow>();

SessionLabel session;

// Open the RTP sessions.
for (int i = 0; i < sessions.length; i++) {

// Parse the session addresses.
try {
session = new SessionLabel(sessions[i]);
} catch (IllegalArgumentException e) {
System.err.println("Failed to parse the session address given: " + sessions[i]);
return false;
}

System.err.println(" - Open RTP session for: addr: " + session.addr + " port: " + session.port + " ttl: " + session.ttl);

mgrs[i] = (RTPManager) RTPManager.newInstance();
mgrs[i].addSessionListener(this);
mgrs[i].addReceiveStreamListener(this);

// Initialize the RTPManager with the RTPSocketAdapter
mgrs[i].initialize(new RTPSocketAdapter(
InetAddress.getByName(session.addr),
session.port, session.ttl));

// You can try out some other buffer size to see
// if you can get better smoothness.
BufferControl bc = (BufferControl)mgrs[i].getControl("javax.media.control.BufferControl");
if (bc != null)
bc.setBufferLength(350);
}

} catch (Exception e){
System.err.println("Cannot create the RTP Session: " + e.getMessage());
return false;
}

// Wait for data to arrive before moving on.

long then = System.currentTimeMillis();
long waitingPeriod = 10000; // wait for a maximum of 10 secs.

try{
synchronized (dataSync) {
while (!dataReceived &&
System.currentTimeMillis() - then < waitingPeriod) {
if (!dataReceived)
System.err.println(" - Waiting for RTP data to arrive...");
dataSync.wait(500);
}
}
} catch (Exception e) {}

if (!dataReceived) {
System.err.println("No RTP data was received.");
close();
return false;
}

return true;
}

/**
* Method isDone.
* @return boolean
*/
public boolean isDone() {
return playerWindows.size() == 0;
}

/**
* Close the players and the session managers.
*/
public void close() {

for (int i = 0; i < playerWindows.size(); i++) {
try {
((PlayerWindow)playerWindows.elementAt(i)).close();
} catch (Exception e) {}
}

playerWindows.removeAllElements();

// close the RTP session.
for (int i = 0; i < mgrs.length; i++) {
if (mgrs[i] != null) {
mgrs[i].removeTargets( "Closing session from AVReceive3");
mgrs[i].dispose();
mgrs[i] = null;
}
}
}

/**
* Method find.
* @param p Player
* @return PlayerWindow
*/
PlayerWindow find(Player p) {
for (int i = 0; i < playerWindows.size(); i++) {
PlayerWindow pw = (PlayerWindow)playerWindows.elementAt(i);
if (pw.player == p)
return pw;
}
return null;
}

/**
* Method find.
* @param strm ReceiveStream
* @return PlayerWindow
*/
PlayerWindow find(ReceiveStream strm) {
for (int i = 0; i < playerWindows.size(); i++) {
PlayerWindow pw = (PlayerWindow)playerWindows.elementAt(i);
if (pw.stream == strm)
return pw;
}
return null;
}

/**
* SessionListener.
* @param evt SessionEvent
* @see javax.media.rtp.SessionListener#update(SessionEvent)
*/
public synchronized void update(SessionEvent evt) {
if (evt instanceof NewParticipantEvent) {
Participant p = ((NewParticipantEvent)evt).getParticipant();
System.err.println(" - A new participant had just joined: " + p.getCNAME());
}
}

/**
* ReceiveStreamListener
* @param evt ReceiveStreamEvent
* @see javax.media.rtp.ReceiveStreamListener#update(ReceiveStreamEvent)
*/
public synchronized void update( ReceiveStreamEvent evt) {

Participant participant = evt.getParticipant(); // could be null.
ReceiveStream stream = evt.getReceiveStream(); // could be null.

if (evt instanceof RemotePayloadChangeEvent) {

System.err.println(" - Received an RTP PayloadChangeEvent.");
System.err.println("Sorry, cannot handle payload change.");

/* Handle this later
System.exit(0);
*/
}

else if (evt instanceof NewReceiveStreamEvent) {

try {
stream = ((NewReceiveStreamEvent)evt).getReceiveStream();
DataSource ds = stream.getDataSource();

// Find out the formats.
RTPControl ctl = (RTPControl)ds.getControl("javax.media.rtp.RTPControl");
if (ctl != null){
System.err.println(" - Recevied new RTP stream: " + ctl.getFormat());
} else
System.err.println(" - Recevied new RTP stream");

if (participant == null)
System.err.println(" The sender of this stream had yet to be identified.");
else {
System.err.println(" The stream comes from: " + participant.getCNAME());
}

// create a player by passing datasource to the Media Manager
Player p = javax.media.Manager.createPlayer(ds);
if (p == null)
return;

p.addControllerListener(this);
p.realize();
PlayerWindow pw = new PlayerWindow(p, stream);
playerWindows.addElement(pw);

// Notify intialize() that a new stream had arrived.
synchronized (dataSync) {
dataReceived = true;
dataSync.notifyAll();
}

} catch (Exception e) {
System.err.println("NewReceiveStreamEvent exception " + e.getMessage());
return;
}

}

else if (evt instanceof StreamMappedEvent) {

if (stream != null && stream.getDataSource() != null) {
DataSource ds = stream.getDataSource();
// Find out the formats.
RTPControl ctl = (RTPControl)ds.getControl("javax.media.rtp.RTPControl");
System.err.println(" - The previously unidentified stream ");
if (ctl != null)
System.err.println(" " + ctl.getFormat());
System.err.println(" had now been identified as sent by: " + participant.getCNAME());
}
}

else if (evt instanceof ByeEvent) {

System.err.println(" - Got \"bye\" from: " + participant.getCNAME());
PlayerWindow pw = find(stream);
if (pw != null) {
pw.close();
playerWindows.removeElement(pw);
}
}

}

/**
* ControllerListener for the Players.
* @param ce ControllerEvent
* @see javax.media.ControllerListener#controllerUpdate(ControllerEvent)
*/
public synchronized void controllerUpdate(ControllerEvent ce) {

Player p = (Player)ce.getSourceController();

if (p == null)
return;

// Get this when the internal players are realized.
if (ce instanceof RealizeCompleteEvent) {
//PlayerWindow
pw = find(p);
if (pw == null) {
// Some strange happened.
System.err.println("Internal error!");
}

MochaSession session = MochaSessionManager.getSession();
MochaResourceManager resource = MochaResourceManagerFactory.createResourceManager();

if(session.getAttribute("video_chat") == null) {
// audio chat
(new OpenVideoListener(partner, true)).openWindow();
} else {
// video chat
(new OpenVideoListener(partner, false)).openWindow();
}

IMChatAVWindow avWindow;
avWindow = (IMChatAVWindow) session.getAttribute("avPanel_" + partner);

// format avWindow
avWindow.setResizable(true);

if (p.getVisualComponent()!= null) {

JPanel vPlayer = new PlayerPanel(p);
avWindow.addVideoPlayer(vPlayer);

}
else {

JPanel vPlayer = new PlayerPanel(p);
avWindow.addAudioPanel(vPlayer);
}

IMUtilities.advanceWindow();

session.setAttribute("avWindow_" + partner, avWindow);
session.setAttribute("video_chat", resource);

p.start();
}

if (ce instanceof ControllerErrorEvent) {
p.removeControllerListener(this);
PlayerWindow pw = find(p);
if (pw != null) {
pw.close();
playerWindows.removeElement(pw);
}
System.err.println("AVReceive3 internal error: " + ce);
}
}

public void closegui() {
pw.close();
}

/**
* A utility class to parse the session addresses.
* @author Paul Lee
* @version $Revision: 1.34 $
*/
class SessionLabel {

public String addr = null;
public int port;
public int ttl = 1;

/**
* Constructor for SessionLabel.
* @param session String
* @throws IllegalArgumentException
*/
SessionLabel(String session) throws IllegalArgumentException {

int off;
String portStr = null, ttlStr = null;

if (session != null && session.length() > 0) {
while (session.length() > 1 && session.charAt(0) == '/')
session = session.substring(1);

// Now see if there's a addr specified.
off = session.indexOf('/');
if (off == -1) {
if (!session.equals(""))
addr = session;
} else {
addr = session.substring(0, off);
session = session.substring(off + 1);
// Now see if there's a port specified
off = session.indexOf('/');
if (off == -1) {
if (!session.equals(""))
portStr = session;
} else {
portStr = session.substring(0, off);
session = session.substring(off + 1);
// Now see if there's a ttl specified
off = session.indexOf('/');
if (off == -1) {
if (!session.equals(""))
ttlStr = session;
} else {
ttlStr = session.substring(0, off);
}
}
}
}

if (addr == null)
throw new IllegalArgumentException();

if (portStr != null) {
try {
Integer integer = Integer.valueOf(portStr);
if (integer != null)
port = integer.intValue();
} catch (Throwable t) {
throw new IllegalArgumentException();
}
} else
throw new IllegalArgumentException();

if (ttlStr != null) {
try {
Integer integer = Integer.valueOf(ttlStr);
if (integer != null)
ttl = integer.intValue();
} catch (Throwable t) {
throw new IllegalArgumentException();
}
}
}
}

/**
* GUI classes for the Player.
* @author Paul Lee
* @version $Revision: 1.34 $
*/
class PlayerWindow extends JFrame {

/**
*
*/
private static final long serialVersionUID = -2372692127029870388L;
Player player;
ReceiveStream stream;

/**
* Constructor for PlayerWindow.
* @param p Player
* @param strm ReceiveStream
*/
PlayerWindow(Player p, ReceiveStream strm) {
player = p;
stream = strm;
}

public void initialize() {
add(new PlayerPanel(player));
}

public void close() {
player.close();
setVisible(false);
dispose();
}

public void addNotify() {
super.addNotify();
pack();
}
}

/**
* GUI classes for the Player.
* @author Paul Lee
* @author Matt Choi
* @version $Revision: 1.34 $
*/
class PlayerPanel extends JPanel {

/**
*
*/
private static final long serialVersionUID = -1227817100348697684L;
Component vc, cc;
JLabel vLabel, aLabel;

/**
* Constructor for PlayerPanel.
* @param p Player
*/
PlayerPanel(Player p) {
setLayout(new BoxLayout(this, BoxLayout.PAGE_AXIS));
if ((vc = p.getVisualComponent()) != null){
vLabel = new JLabel(" Video");
vc.setSize(100, 165);
add(vLabel);
add(vc);
} else {
vLabel = new JLabel(" Audio");
add(vLabel);
}
if ((cc = p.getControlPanelComponent()) != null) {
cc.setSize(100, 23);
add(cc);
}
}

/**
* Method getPreferredSize.
* @return Dimension
*/
public Dimension getPreferredSize() {
int w = 0, h = 0;
if (vc != null) {
w = 125;
h = 135;
} else {
w = 125;
h = 17;
}
if (cc != null) {
h += 17;
}

return new Dimension(w, h);
}
}

/**
* Method main.
* @param argv String[]
*/
public static void main(String argv[]) {
if (argv.length == 0)
prUsage();

AVReceive3 avReceive = new AVReceive3(argv, "");
if (!avReceive.initialize()) {
System.err.println("Failed to initialize the sessions.");

/* Handle better later
System.exit(-1);
*/
}

// Check to see if AVReceive3 is done.
try {
while (!avReceive.isDone())
Thread.sleep(1000);
} catch (Exception e) {}

System.err.println("Exiting AVReceive3");
}

static void prUsage() {
System.err.println("Usage: AVReceive3 <session> <session> ...");
System.err.println(" <session>: <address>/<port>/<ttl>");
System.exit(0);
}

}// end of AVReceive3