datareceive.java

简易聊天工具

import java.io.*; 
import java.awt.*; 
import java.net.*; 
import java.awt.event.*; 
import java.util.Vector; 
import javax.swing.*;
import javax.media.*; 
import javax.media.rtp.*; 
import javax.media.rtp.event.*; 
import javax.media.rtp.rtcp.*; 
import javax.media.protocol.*; 
import javax.media.protocol.DataSource; 
import javax.media.format.AudioFormat; 
import javax.media.format.VideoFormat; 
import javax.media.Format; 
import javax.media.format.FormatChangeEvent; 
import javax.media.control.BufferControl; 
// 接收RTP传输数据的类
public class DataReceive implements ReceiveStreamListener, SessionListener, ControllerListener 
{ 
	String sessions[] = null;                  // RTP会话字符串数组
	RTPManager mgrs[] = null; 					// RTP管理器数组
	Vector playerPanels = null; 				// 管理播放器面板的向量
	PlayerWindow pw = null;
	boolean dataReceived = false; 				// 是否接收到数据的标志
	Object dataSync = new Object();  			// 同步对象
	public DataReceive(String sessions[])        // 构造函数
	{ 
		this.sessions = sessions; 
	} 
	protected String initialize(int xg, int yg) 				// 初始化RTP会话，准备接收数据
	{ 
		try 
		{ 
			pw = new PlayerWindow(xg, yg);
			InetAddress ipAddr; 
			SessionAddress localAddr = new SessionAddress(); 
			SessionAddress destAddr; 
			mgrs = new RTPManager[sessions.length]; // 为每一个RTP会话建立一个管理器
			playerPanels = new Vector(); 
			SessionLabel session; 
			for (int i = 0; i < sessions.length; i++) // 对每一个RTP会话
			{ 
				try 
				{ 
					session = new SessionLabel(sessions[i]); // 解析RTP会话地址
				} catch (IllegalArgumentException e) 
				{ 
					return "Failed to parse the session"; 
				}
		  
				System.err.println("  - Open RTP session for: addr: " + session.addr + " port: " + session.port);

				mgrs[i] = (RTPManager) RTPManager.newInstance(); 	// 为每一个RTP会话产生一个RTP管理器
				mgrs[i].addSessionListener(this); 					// 添加会话监听
				mgrs[i].addReceiveStreamListener(this);	 			// 添加接收到数据的监听
				ipAddr = InetAddress.getByName(session.addr); 
		
				if( ipAddr.isMulticastAddress()) 					// 对于组播，本地和目的地的IP地址相同，均采用组播地址
				{ 
					localAddr= new SessionAddress( ipAddr, session.port); 
					destAddr = new SessionAddress( ipAddr, session.port); 
				} 
				else 
				{ 
					localAddr= new SessionAddress( InetAddress.getLocalHost(), session.port);	// 用本机IP地址和端口号构造源会话地址 
					destAddr = new SessionAddress( ipAddr, session.port); 						// 用目的机（发送端）的IP地址和端口号构造目的会话地址
				}
			
				mgrs[i].initialize( localAddr); 				// 将本机会话地址给RTP管理器
				
				BufferControl bc = (BufferControl)mgrs[i].getControl("javax.media.control.BufferControl"); 
				
				if (bc != null) bc.setBufferLength(350); 		// 设置缓冲区大小
				
				mgrs[i].addTarget(destAddr); 					 // 加入目的会话地址
				
			} 
		} catch (Exception e)
		{ 
			return "Cannot create the RTP Session" + "<<" + e.getMessage() + ">>"; 
		} 
		// 等待数据到达
		long then = System.currentTimeMillis(); 
		long waitingPeriod = 60000; 
		try
		{ 
			synchronized (dataSync) 
			{ 
				while (!dataReceived && System.currentTimeMillis() - then < waitingPeriod)  // 等待上面所设定的时间
				{ 
					dataSync.wait(1000); 				
				} 
			} 
		} catch (Exception e) {} 
		if (!dataReceived) 
		{ 													// 在设定的时间内没有等到数据
			close(); 
			return "No RTP data was received"; 		
		} 
		pw.setVisible(true);
		return "success"; 
	} 
	 // 判断数据是否接收完成
	public boolean isDone() 	
	{ 
		return playerPanels.size() == 0; 
	} 
		// 关闭播放器和会话管理器
	protected void close() 
	{ 
		for (int i = 0; i < playerPanels.size(); i++) 
		{ 
			try 
			{ 
				((PlayerPanel)playerPanels.elementAt(i)).close(); 
			} catch (Exception e) {} 
		} 
		playerPanels.removeAllElements(); 
		pw.dispose();
		// 关闭RTP会话
		for (int i = 0; i < mgrs.length; i++) 
		{ 
			if (mgrs[i] != null) 
			{ 
				mgrs[i].removeTargets( "Closing session from AVReceive2"); 
				mgrs[i].dispose(); 
				mgrs[i] = null; 
			} 
		} 
	} 
	// 通过播放器查找播放面板
	PlayerPanel find(Player p) 
	{ 
		for (int i = 0; i < playerPanels.size(); i++) 
		{ 
			PlayerPanel pp = (PlayerPanel)playerPanels.elementAt(i); 
			if (pp.player == p) return pp; 
		} 
		return null; 
	} 
		// 通过接收数据流查找播放面板
	PlayerPanel find(ReceiveStream strm) 
	{ 
		for (int i = 0; i < playerPanels.size(); i++) 
		{ 
			PlayerPanel pp = (PlayerPanel)playerPanels.elementAt(i); 
			if (pp.stream == strm) return pp; 
		} 
		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();  
					Player p = javax.media.Manager.createPlayer(ds); 	// 通过数据源构造一个媒体播放器
					if (p == null) return; 
					p.addControllerListener(this); 
					p.realize(); 
					PlayerPanel pp = new PlayerPanel(p, stream); 
					playerPanels.addElement(pp); 
				// 通知initialize()函数中的等待过程：已经接收到了一个新数据流
					synchronized (dataSync) 
					{ 
						dataReceived = true; 
						dataSync.notifyAll(); 
					} 
				} catch (Exception e) 
				{ 
					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.getCNAME()); 
				} 
			} else if (evt instanceof ByeEvent) 
			{ 
				System.err.println(" - Got \"bye\" from: " + participant.getCNAME()); 
				PlayerPanel pp = find(stream); 
				if (pp != null) 
				{ 
					pp.close(); 
					playerPanels.removeElement(pp); 
				} 
			} 
		} 
 /***************************************************************
  * 下面这个函数实现了 SessionListener 接口
  ****************************************************************/

  		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());
    		}
  		}
  /***************************************************************
  * 下面这个函数实现了 ControllerListener 接口
  ****************************************************************/

		public synchronized void controllerUpdate(ControllerEvent ce) 
		{ 
			Player p = (Player)ce.getSourceController(); 		// 得到事件源
			if (p == null) return; 
			if (ce instanceof RealizeCompleteEvent) 			// 播放器实现完成
			{ 
				PlayerPanel pp = find(p); 
				if (pp == null) 
				{ 
					System.err.println("Internal error!"); 
				} 
				pp.initialize(); 
				pw.add(pp); 
				p.start(); 
			} 
			if (ce instanceof ControllerErrorEvent) 
			{ 
				p.removeControllerListener(this); 
				PlayerPanel pp = find(p); 
				if (pp != null) 
				{ 
					pp.close(); 
					playerPanels.removeElement(pp); 
				} 
			} 
		} 
		// 解析会话地址的类
  	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;              					// 字符串值作为IP地址
            		}
        		}
        		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)            								// 如果没有给出IP地址，则报错
        	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();
        	}
    	}
  	}
  		// 播放窗口类
		class PlayerWindow extends JFrame 
		{ 
			int xGrid;						//横向网格数
			int yGrid;						//竖向网格数
			JPanel panel = null;			//添加播放器组件面板的面板
			public PlayerWindow(int xg, int yg)
			{
				xGrid = xg;
				yGrid = yg;
				setTitle("网络会议");
				Container contentPane = getContentPane();
				panel = new JPanel();
				panel.setLayout(new GridLayout(xGrid, yGrid));
				contentPane.add(panel);	
			}
			public void addPanel(PlayerPanel pp)	//添加播放器组件面板的方法
			{
					panel.add(pp);
					this.add(panel);	
			}
		} 
		// 播放器面板类
		class PlayerPanel extends Panel 
		{ 
			Player player; 			// 播放器对象
			ReceiveStream stream;  	// 接收数据流对象
			Component vc, cc; 
			// 构造函数
			PlayerPanel(Player p, ReceiveStream strm) 
			{ 
				player = p; 
				stream = strm; 
			}
			// 初始化 
			public void initialize() 
			{ 
				setLayout(new BorderLayout()); 
				if ((vc = player.getVisualComponent()) != null) 
				{
					add("Center", vc); 
				}
				if ((cc = player.getControlPanelComponent()) != null) 
				{
					add("South", cc); 
				}
			}
			 // 关闭播放器
			public void close() 
			{ 
				player.close(); 
				setVisible(false); 
				this.close();
			} 
		} 
}
	// end of DataReceive