JMF 利用RTPManager进行视频音频流传输聊天

 转：http://blog.csdn.net/maydie1989/article/details/6820482

http://download.csdn.net/detail/maydie1989/3636156

JMF(Java Media Framework)想必大家都不陌生  它为音频和视频等媒体内容的捕获、回放、传输和编码转换等提供了一个统一的架构 视频传输大概思路如下

其中有几个常用到的类  - >   ● 数据源（DataSource） ● 媒体定位器（MediaLocator） ● 播放器（Player） 

                                                 ● 处理器（Processor）  ● 管理器（Manager） ● RTP管理器（RTPManager）

一般发送端的情况是首先获得视频音频设备实例(CaptureDeviceInfo)   -> 然后得到 DataSource -> Processor -> 得到processor的输出DataSource,通过RTPManager以stream的形式发出

而接收端则是通过RTPManager得到ReceiveStream -> 还原后通过Player进行播放

发送端RTPTransmiter

public class RTPTransmiter
{
    private Processor processor;
    private Integer lock = new Integer(0);
    private DataSource dataOutput;
    private boolean failed = false;
    private RTPManager[] rtpMgrs;
    private String ipAddress;
    private int portBase;
    private int destPortBase; //接收端 基础端口
    public RTPTransmiter(String ipAddress,int portBase,int destPortBase) throws Exception
    {
    this.ipAddress = ipAddress;
    this.portBase = portBase;
    this.destPortBase = destPortBase;
    createProcessor() ;
    createTransmitter() ;
    processor.start();
    }

    private void createProcessor() throws Exception
    {
    Vector<?> devices = CaptureDeviceManager.getDeviceList(null);
    CaptureDeviceInfo diVideo = null;
    CaptureDeviceInfo diAudio = null;

        if (devices.size() > 2)
        {
            diVideo = (CaptureDeviceInfo) devices.elementAt(2);//2是视频捕捉器
             diAudio = (CaptureDeviceInfo) devices.elementAt(0);
              DataSource[] dataSources = new DataSource[2];
            dataSources[1] = Manager.createDataSource(diVideo.getLocator());
            dataSources[0] = Manager.createDataSource(diAudio.getLocator());
                    processor = Manager.createProcessor(Manager.createMergingDataSource(dataSources));//合并video和audio的 datasource
    }
    if(!waitForState(processor, Processor.Configured)) //configure processor
        throw new Exception("cant configured processor");
    ContentDescriptor cd = new ContentDescriptor(ContentDescriptor.RAW_RTP);
    processor.setContentDescriptor(cd);
    if(!waitForState(processor, Controller.Realized)) //realize processor
        throw new Exception("cant realized processor");
    setJPEGQuality(processor, 0.5f);//设置图像质量 百分比
    dataOutput = processor.getDataOutput();
    }

    private void createTransmitter()
    {
    PushBufferDataSource pbds = (PushBufferDataSource)dataOutput;
    PushBufferStream pbss[] = pbds.getStreams();
    System.out.println("pbss.length is: "+pbss.length);

    rtpMgrs = new RTPManager[pbss.length];//每一个通道由一个RTPManager负责
    SessionAddress localAddr, destAddr;
    InetAddress ipAddr;
    SendStream sendStream;
    int port;
    for (int i = 0; i < pbss.length; i++)
    {
         try
              {
        rtpMgrs[i] = RTPManager.newInstance();
        port = portBase + 2*i;
        int clientPort = destPortBase +2*i;

        ipAddr = InetAddress.getByName(ipAddress);

        localAddr = new SessionAddress( InetAddress.getLocalHost(),port); //本机测 所以IP都是InetAddress.getLocalHost()

        destAddr = new SessionAddress( InetAddress.getLocalHost(), clientPort);

        rtpMgrs[i].initialize( localAddr);

        rtpMgrs[i].addTarget( destAddr);

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

        sendStream = rtpMgrs[i].createSendStream(dataOutput, i);
        sendStream.start(); //开始传输
         }
              catch (Exception  e)
         {
        e.printStackTrace();
         }
        }

    }


    void setJPEGQuality(Player p, float val)
    {
        Control cs[] = p.getControls();
        QualityControl qc = null;
        VideoFormat jpegFmt = new VideoFormat(VideoFormat.JPEG);

        for (int i = 0; i < cs.length; i++)
        {
            if (cs[i] instanceof QualityControl && cs[i] instanceof Owned)
            {
                Object owner = ((Owned)cs[i]).getOwner();
                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;
            }
        }
    }

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

        if (state == Processor.Configured)
        {
            p.configure();
        }
        else if (state == Processor.Realized)
        {
            p.realize();
        }

        while (p.getState() < state && !failed)
        {
            synchronized (lock)
            {
                try
                {
                    lock.wait();
                }
                catch (InterruptedException ie)
                {
                    return false;
                }
            }
        }

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

    class StateListener implements ControllerListener
    {
        public void controllerUpdate(ControllerEvent ce)
        {
            if (ce instanceof ControllerClosedEvent)
                failed = true;

            if (ce instanceof ControllerEvent) {
                synchronized (lock) {
                    lock.notifyAll();
                }
            }
        }
   }
}

public class RTPTransmiter
{
    private Processor processor;
    private Integer lock = new Integer(0);
    private DataSource dataOutput;
    private boolean failed = false;
    private RTPManager[] rtpMgrs;
    private String ipAddress;
    private int portBase;
    private int destPortBase; //接收端 基础端口 
    public RTPTransmiter(String ipAddress,int portBase,int destPortBase) throws Exception
    {
	this.ipAddress = ipAddress;
	this.portBase = portBase;
	this.destPortBase = destPortBase;
	createProcessor() ;
	createTransmitter() ;
	processor.start();
    }
	
    private void createProcessor() throws Exception 
    {		
	Vector<?> devices = CaptureDeviceManager.getDeviceList(null);
	CaptureDeviceInfo diVideo = null;
	CaptureDeviceInfo diAudio = null;
         	
       	if (devices.size() > 2) 
       	{
	        diVideo = (CaptureDeviceInfo) devices.elementAt(2);//2是视频捕捉器
 	         diAudio = (CaptureDeviceInfo) devices.elementAt(0);
	          DataSource[] dataSources = new DataSource[2];
	        dataSources[1] = Manager.createDataSource(diVideo.getLocator());
	        dataSources[0] = Manager.createDataSource(diAudio.getLocator());
         	        processor = Manager.createProcessor(Manager.createMergingDataSource(dataSources));//合并video和audio的 datasource	        	
	}
 	if(!waitForState(processor, Processor.Configured)) //configure processor
 		throw new Exception("cant configured processor");
 	ContentDescriptor cd = new ContentDescriptor(ContentDescriptor.RAW_RTP); 
	processor.setContentDescriptor(cd);
 	if(!waitForState(processor, Controller.Realized)) //realize processor
 		throw new Exception("cant realized processor");
 	setJPEGQuality(processor, 0.5f);//设置图像质量 百分比 
 	dataOutput = processor.getDataOutput();
    }
    
    private void createTransmitter() 
    {
	PushBufferDataSource pbds = (PushBufferDataSource)dataOutput;
	PushBufferStream pbss[] = pbds.getStreams();
	System.out.println("pbss.length is: "+pbss.length);
		
	rtpMgrs = new RTPManager[pbss.length];//每一个通道由一个RTPManager负责
	SessionAddress localAddr, destAddr;
	InetAddress ipAddr;
	SendStream sendStream;
	int port;
	for (int i = 0; i < pbss.length; i++) 
	{
	     try 
              {
		rtpMgrs[i] = RTPManager.newInstance();	    
		port = portBase + 2*i;
		int clientPort = destPortBase +2*i;
				 
		ipAddr = InetAddress.getByName(ipAddress);
		
		localAddr = new SessionAddress( InetAddress.getLocalHost(),port); //本机测 所以IP都是InetAddress.getLocalHost()
				
		destAddr = new SessionAddress( InetAddress.getLocalHost(), clientPort);
		
		rtpMgrs[i].initialize( localAddr);
				
		rtpMgrs[i].addTarget( destAddr);
				
		System.err.println( "Created RTP session: " + ipAddress + " " + port);
				 
		sendStream = rtpMgrs[i].createSendStream(dataOutput, i);		
		sendStream.start(); //开始传输
	     } 
              catch (Exception  e) 
	     {
		e.printStackTrace();
	     }
        }
 
    }

    
    void setJPEGQuality(Player p, float val) 
    {
		Control cs[] = p.getControls();
		QualityControl qc = null;
		VideoFormat jpegFmt = new VideoFormat(VideoFormat.JPEG);
 
		for (int i = 0; i < cs.length; i++) 
		{
		    if (cs[i] instanceof QualityControl && cs[i] instanceof Owned) 
		    {
				Object owner = ((Owned)cs[i]).getOwner();
 				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;
		    }
		}
    }
    
    private synchronized boolean waitForState(Processor p, int state) 
    {
		p.addControllerListener(new StateListener());
		failed = false;
	
 		if (state == Processor.Configured) 
		{
		    p.configure();
		} 
		else if (state == Processor.Realized) 
		{
		    p.realize();
		}
 
		while (p.getState() < state && !failed) 
		{
		    synchronized (lock) 
		    {
				try 
				{
				    lock.wait();
				} 
				catch (InterruptedException ie) 
				{
				    return false;
				}
		    }
		}
	
		if (failed)
		    return false;
		else
		    return true;
    }
    
    class StateListener implements ControllerListener 
    {
    	public void controllerUpdate(ControllerEvent ce) 
    	{
     	    if (ce instanceof ControllerClosedEvent)
    			failed = true;

     	    if (ce instanceof ControllerEvent) {
	    		synchronized (lock) {
	    		    lock.notifyAll();
	    		}
    	    }
    	}
   }
}

RTPReciver 核心部分 需要为每个rtpManager添加 ReceiveStreamListener和SessionListener ,在ReceiveStreamListener的NewReceiveStreamEvent事件中添加ControllerListener

mgrs[i] = (RTPManager) RTPManager.newInstance();
//添加 SessionListener
mgrs[i].addSessionListener(this);
//添加 ReceiveStreamListener
mgrs[i].addReceiveStreamListener(this);

//得到服务器端IP地址
ipAddr = InetAddress.getByName("sender ip");
localAddr= new SessionAddress( InetAddress.getLocalHost(), sendPort);
destAddr = new SessionAddress( ipAddr, sendPort);
mgrs[i].initialize( localAddr);
BufferControl bc = (BufferControl)mgrs[i].getControl("javax.media.control.BufferControl");
if (bc != null)
    bc.setBufferLength(350);
mgrs[i].addTarget(destAddr);

mgrs[i] = (RTPManager) RTPManager.newInstance();
//添加 SessionListener
mgrs[i].addSessionListener(this);
//添加 ReceiveStreamListener
mgrs[i].addReceiveStreamListener(this);
				
//得到服务器端IP地址
ipAddr = InetAddress.getByName("sender ip");
localAddr= new SessionAddress( InetAddress.getLocalHost(), sendPort);
destAddr = new SessionAddress( ipAddr, sendPort);				
mgrs[i].initialize( localAddr);
BufferControl bc = (BufferControl)mgrs[i].getControl("javax.media.control.BufferControl");
if (bc != null)
    bc.setBufferLength(350);
mgrs[i].addTarget(destAddr);

关于ReceiveStreamListener

@Override
public void update(ReceiveStreamEvent evt)
{
    //从大到小进行定位，RTPManager->Participant->ReceiveStream
    RTPManager mgr = (RTPManager)evt.getSource();
    Participant participant = evt.getParticipant();
    ReceiveStream stream = evt.getReceiveStream();
    if (evt instanceof RemotePayloadChangeEvent){.....};//远程发送者改变负载
    else if(evt instanceof NewReceiveStreamEvent){.....<SPAN style="COLOR: #ff0000">需要添加ControllerListener</SPAN>};//接收到新的stream
    else if (evt instanceof ByeEvent){.....};//结束
}

	@Override
	public void update(ReceiveStreamEvent evt)
	{
    	//从大到小进行定位，RTPManager->Participant->ReceiveStream
		RTPManager mgr = (RTPManager)evt.getSource();  
		Participant participant = evt.getParticipant();
		ReceiveStream stream = evt.getReceiveStream(); 		
		if (evt instanceof RemotePayloadChangeEvent){.....};//远程发送者改变负载 
		else if(evt instanceof NewReceiveStreamEvent){.....需要添加ControllerListener};//接收到新的stream 	
		else if (evt instanceof ByeEvent){.....};//结束	
	}

关于SessionListener

@Override
public void update(SessionEvent evt)
{
             //有新的用户加入
    if (evt instanceof NewParticipantEvent)
    {
        Participant p = ((NewParticipantEvent)evt).getParticipant();
    }
}

	@Override
	public void update(SessionEvent evt)
	{
    	         //有新的用户加入
		if (evt instanceof NewParticipantEvent) 
		{
		    Participant p = ((NewParticipantEvent)evt).getParticipant();		    
		}
	}

关于ControllerListener

@Override
public void controllerUpdate(ControllerEvent ce)
{
    if (ce instanceof RealizeCompleteEvent){......};//player完成了realize
    if (ce instanceof ControllerErrorEvent){......};//出现异常
}

	@Override
	public void controllerUpdate(ControllerEvent ce)
	{
		if (ce instanceof RealizeCompleteEvent){......};//player完成了realize	
		if (ce instanceof ControllerErrorEvent){......};//出现异常		
	}

jmf的rtp应该是用udp做的...貌似
ps:话说 百度的题好难啊 不如腾讯的友好 哈哈哈