rtpreceive.java

rtp接收端

package	RTPReceive;
import java.awt.*;
import java.net.*;
import java.awt.event.*;
import java.util.Vector;
import javax.media.*;
import javax.media.rtp.*;
import javax.media.rtp.event.*;
import javax.media.protocol.DataSource;
import javax.media.control.BufferControl;

public class RTPReceive implements ReceiveStreamListener,SessionListener,ControlListener
{
	String session[] = null;
	
	RTPManager mgrs[] = null;
	
	Vector playerWindows = null;
	
	boolean dataReceived = false;
	
	Object dataSync = new Object();
	
	public RTPReceive(String session[])
	{
		this.sessions = sessions;	
	}
	
	
	protected boolean initialize()
	{
		try
		{
			playerWindows = new Vector();
			
			mgrs = new RTPManager[sessions.length];
			
			SessionLabel session;
			
			for(int i =0; i < sessions.length;i++)
			{
				try{session = new SessionLabel(sessions[i]);}
				catch(IllegalArgumentException e)
				{
					System.err.println("Failed to parse the session address given:" + session[i]);
					return false;	
				}
				
				
				System.err.println(" - open RTP session for: addr: " + session.addr + " port: " + session.port);
				
				mgrs[i] = (RTPManager)RTPManager.newInstance();
				
				mgrs[i].addSessionListenter(this);
				
				mgrs[i].addReceiveStreamListenter(this);
				
				InetAddress ipAddr = InetAddress.getByName(ipAddress);
				
				SessionAddress localAddr = null;
				
				SessionAddress destAddr = null;
				
				if(ipAddr.isMulticastAddress())
				{
					localAddr = new SessionAddress(ipAddr,session.port);
					destAddr = new SessionAddress(ipAddr,session.port);		
				}
				else
				{
					localAddr = new SessionAddress(InetAddress.getLocalHost(),session.port);
					destAddr = 	new SessionAddress(ipAddr,session.port);
				}
				
				mgrs[i].initialize(localAddr);
				
				BufferControl bc = (BufferControl)mgrs[i].getControl("javax.media.control.BufferControl");
				
				if(bc != null)
					bc.setBufferLength(350);
					
				mgrs[i].addTarget(destAddr);							
			}
		}
		catch(Exception e)
		{
			System.err.println("Cannot create the RTP Session:" + e.getMessage());
			return false;
		}
		
		
		long then = System.currentTimeMillis();
		
		long waitingPeriod = 60000;
		
		try
		{
			synchronized (dataSync)
			{
				while(!dataReceived&&System.currentTimeMillis() - then < waitingPeriod)	
				{
					if(!dataReceived)
						System.err.println(" - Waiting for RTP data to arrive ... ");
					
					dataSync.wait(1000);	
				}
			}
		}
		catch(Exception e){}
		
		if(!dataReceived)
		{
			System.err.println(" No RTP data was received. ");
			close();
			return false;	
		}
		
		return true;				
	}
	
	
	protected void close()
	{
		for(int i =0; i < playerWindows.seze();i++)
		{
			try
			{
				((playerWindows)playerWindows.elementAt(i)).close();		
			}
			catch(Exception e){}			
		}
		
		
		playerWindows.removeAllElements();
		
		for(int i =0; i < mgrs.length;i++)
		{
			if(mgrs[i] != null)
			{
				mgrs[i].removeTargets("Closing session from RTPReceive");
				mgrs[i].dispose();
				mgrs[i] = null;	
			}	
		}	
	}
	
	
	public boolean isDone()
	{
		return 	playerWindows.size() == 0;
	}
	
	PlayerWindow find(Player p)
	{
		for(int i =0; i < playerWindows.seze();i++)
		{
			PlayerWindow pw = (PlayerWindow)playerWindows.elementAt(i);
			if(pw.player == p) return pw;	
		}
		
		return null;	
	}
	
	
	PlayerWindow find(ReceiveStream strm)
	{
		for(int i =0; i < playerWindows.seze();i++)
		{
			PlayerWindow pw = (PlayerWindow)playerWindows.elementAt(i);
			if(pw.stream == strm) return pw;	
		}
		
		return null;	
	}
	
	
/*-------------------下面的方法实现了ReceiveStreamListener监听接口--------------*/
	public synchronized void update(ReceiveStreamEvent evt)
	{
		RTPManager mgr = (RTPManager)evt.getSource();
		
		Participant participant = evt.getParticipant();
		
		ReceiveStream stream = evt.getReceiveStream();
		
		if(evt instanceof NewReceiveStreamEvent)
		{
			try
			{
				stream = ((newReceiveStreamEvent)evt).getReceiveStream();
				
				DataSource ds = stream.getDataSource();
				
				RTPControl ctl = (RTPControl)ds.getControl("javax.media.rtp.RTPControl");
				
				if(ctl != null)
					System.err.println(" - Received new RTP stream: " + ctl.getFormat());
				else
					System.err.println(" - Received 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.getCNAEM());
					
				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);
				
				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();
				
				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.getCNAEM());	
			}
		}
		else if(evt instanceof ByeEvent)
		{
			System.err.println(" - Got \"bye\" from: " + participant.getCNAEM());
			
			PlayerWindow pw = find(stream);
			
			if(pw != null)
			{
				pw.close();
				
				playerWindows.removeElement(pw);	
			}	
		}			
	}
	
	
/*-------------------下面的方法实现了SessionListener监听接口--------------*/
	public synchronized void update(SessionEvent evt)
	{
		if(evt instanceof NewParitcipantEvent)
		{
			Paritcipant p = ((NewParitcipantEvent)evt).getParitcipant();
			
			System.err.println(" -A new paritcipant had just joined: " + p.getCNAEM());				
		}	
	}
	

/*-------------------下面的方法实现了ControllerListener监听接口--------------*/	
	public synchronized void controllerupdate(ControllerEvent ce)
	{
		Player p = (Player)ce.getSourceController();
		
		if(p == null) return;
		
		if(ce instanceof RealizeCompleteEvent)
		{
			PlayerWindow pw = find(p);
			
			if(pw == null)
			{
				System.err.println(" Internal error! ");
				
				return;	
			}
			
			pw.initialize();
			
			pw.setVisible(true);
			
			p.start();
		}
		
		if(ce instanceof  ControllerErrorEvent)
		{
			p.removeControllerListener(this);
			
			PlayWindow pw = find(p);
			
			if(pw != null)
			{
				pw.close();
				
				playerWindows.removeElement(pw);	
			}
			
			System.err.println(" RTPReceive internal error: " + ce);	
		}	
	}
	
	
/*-------------------下面为内部类：解析回话地址--------------*/

	class SessionLabel
	{
		public String addr = null;
		
		public int port;
		
		SessionLabel(String session) throws IllegalArgumentException
		{
			int off;
			
			String portStr = null;
			
			if(session != null && session.length() > 0)
			{
				while (session.length() > 1 && session.charAt(0) == '/')
					session = session.substring(1);
					
				off = session.indexOf('/');
				
				if(off == -1)
				{
					if (!session.equals(""))
						addr = session;	
				}
				else
				{
					addr = 	session.substring(0,off);
					
					session = session.substring(off+1);
					
					off = session.indexOf('/');
					
					if(off == -1)
					{
						if (!session.equals(""))
							portStr = session;	
					}
					
				}
					
			}
			
			if(addr == null)
				throw new IllegalArgumentException();
				
			if (portStr != null)
			{
				try
				{
					Integer iinteger = Integer.valueOf(portStr);
					
					if(iinteger != null) port = iinteger.intValue();
				}
				catch(Throwable t)
				{
					throw new IllegalArgumentException();
				}					
			}
			else
				throw new IllegalArgumentException();			
		}	
	}	
	
/*-------------------下面为内部类：播放窗口类--------------*/	
	class PlayerWindow extends Frame
	{
		Player player;
		
		ReceiveStream stream;
		
		PlayerWindow(Player p,ReceiveStream strm)
		{
			player = p;	stream = strm;
		}
		
		public void initialize()
		{
			add(new PlayPanel(player));	
		}
		
		public void addNotify()
		{
			super.addNotify();
			pack();	
		}
		
		public void close()
		{
			player.close();setVisible(false);dispose();	
		}	
	}
	
	
/*-------------------下面为内部类：播放器组件类--------------*/	
	class PlayPanel extends Panel
	{
		Componet vc,cc;
		
		PlayPanel(Player P)
		{
			SETLayout(new BorderLayout());
			
			if((vc = p.getVisualComponet())	!= null)
				add("Center",vc);
				
			if((cc = p.getControlPanelComponet()) != null)
				add("South",cc);				
		}
			
	}
	
	public static void main(String argv[])
	{
		if(argv.length == 0)
			prUsage();
			
		RTPReceive rtpReceive = new RTPReceive(argv);
		
		if(!rtpReceive.initialize())
		{
			System.err.println(" Fialed to initialize the session.");
			
			System.exit(-1);	
		}
		
		try
		{
			while (!rtpReceive.isDone())
				Thread.sleep(1000);			
		}
		catch(Exception e){}
		
		System.err.println("Exiting RTPReceive.");		
	}
	
	
	static void prUsage()
	{
		System.err.println("Usage: RTPReceive <session> <session> ... ");
		
		System.err.println("   <session>: <address>/<port>");
		
		System.exit(0);	
	}		
}