rtpsend.java

本视频会议服务器端代码:根据自己定义的协议:能构穿透nat 需要java jmf框架 * 我开放源代码是为了更好的让java这门程序艺术更好的发展

/***************************************************************************
 * 一个RTP传输的线程,把它做为一个线程是为了将来能够多用户会议的时候能够重复利用
 *                2008年1月7号 完成
 *          writed by Yanglong (netBeans5.5)
 ***************************************************************************/
/*
 *     程序员杨龙:本视频会议服务器端代码:根据自己定义的协议:能构穿透nat 需要java jmf框架
 *     我开放源代码是为了更好的让java这门程序艺术更好的发展:您可以研究代码,您可以自己使用
 *     但是不要使用在商业用途上 让我们更好的把开源事业在中国做下去
 *     2008年 3月  杨龙 完成服务器端的框架 
 *     信箱 :[e-mail]
 *     QQ:274797328
 *     www.0351soft.com   www.lx08.com  
 */
package ytp.javaserver.server;

import java.awt.Dimension;
import java.io.IOException;
import java.net.InetAddress;
import javax.media.Controller;
import javax.media.ControllerClosedEvent;
import javax.media.ControllerEvent;
import javax.media.ControllerListener;
import javax.media.DataSink;
import javax.media.Format;
import javax.media.Manager;
import javax.media.MediaLocator;
import javax.media.NoDataSourceException;
import javax.media.NoProcessorException;
import javax.media.Processor;
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.InvalidSessionAddressException;
import javax.media.rtp.RTPManager;
import javax.media.rtp.SendStream;
import javax.media.rtp.SendStreamListener;
import javax.media.rtp.SessionAddress;
import javax.media.rtp.SessionListener;
import javax.media.rtp.event.LocalPayloadChangeEvent;
import javax.media.rtp.event.NewSendStreamEvent;
import javax.media.rtp.event.SendStreamEvent;
import javax.media.rtp.event.SessionEvent;
import javax.media.rtp.event.StreamClosedEvent;
import javax.media.rtp.rtcp.SourceDescription;
import javax.swing.JOptionPane;
import java.awt.*;

/**
 *
 * @author Administrator
 */
public class RTPsend implements Runnable {
    MediaLocator medialocal = null;
    DataSink datasink = null;
    Processor prosor  = null;
    DataSource dataSource = null;
    public int portBase;
    InetAddress inetRometo =null;
    int romport;
    RTPManager rtpMgrs[];
 /*   
    public static void main(String[] args) {
        File file = new File("C:\\pt.mpg");
        MediaLocator media =null;
        try {
            media = new MediaLocator(file.toURL());
        } catch (MalformedURLException ex) {
            ex.printStackTrace();
        }
        InetAddress inetrot = null;
        try {
            inetrot = InetAddress.getByName("192.168.1.110");
        } catch (UnknownHostException ex) {
            ex.printStackTrace();
        }
        
        RTPsend send = new RTPsend(media,inetrot,6700);
        (new Thread(send)).start();
    }
   */ 
    /** Creates a new instance of RTPsend */
    public RTPsend(MediaLocator medialocal) {
        this.medialocal = medialocal;
    }
    public void run() {
        if(this.init() != true) {
            this.init();
        }
    }
    public RTPsend(MediaLocator medialocal,InetAddress inetRometo,int romport) {
        this.medialocal = medialocal;
        this.inetRometo = inetRometo;
        this.romport    = romport;
        this.portBase   = romport;
        System.out.println(this.medialocal.toString()+"|"+this.inetRometo.toString()+romport);
    }
    
    public boolean init() {
        try {
            
            // prosor = Manager.createProcessor(medialocal);
            this.dataSource = Manager.createDataSource(this.medialocal);
            this.prosor = Manager.createProcessor(this.dataSource);
            prosor.addControllerListener(new Conlist(prosor));
            //在这个地方出现了一些问题 可能要延缓执行吧 等待
            boolean bret = this.waitForState(prosor,prosor.Configured);
            
            if(bret == false) {
                System.out.println("函数内部发生了错误");
            }
            
            
            TrackControl[] conTrons = prosor.getTrackControls();
            if(conTrons.length < 0 ) {
                return false;
            }
            ContentDescriptor cd = new ContentDescriptor(ContentDescriptor.RAW_RTP);
            prosor.setContentDescriptor(cd);
            
            Format supported[];
            Format chosen;
            boolean atLeastOneTrack = false;
            //获取每一个轨道上的数据流 并且将它们分别发送
            for(int i =0;i< conTrons.length;i++) {
                Format  format = conTrons[i].getFormat();
                if(conTrons[i].isEnabled()) {
                    supported = conTrons[i].getSupportedFormats();
                    if(supported.length>0) {
                        //有多种支持的多媒体格式
                        //检查视频格式
                        if(supported[0] instanceof VideoFormat) {
                            //视频格式 检查尺寸是否合格
                            chosen = this.checkForVideoSizes(conTrons[i].getFormat(),supported[0]);
                        }else {
                            //其他的格式
                            chosen = supported[0];
                            conTrons[i].setFormat(chosen);
                            atLeastOneTrack = true;
                        }
                        
                        
                    }else {
                        //没有支持的格式
                        conTrons[i].setEnabled(false);
                    }
                }else {//这个格式不能识别 返回
                }
                //循环结束
            }
            
            bret = waitForState(prosor, Controller.Realized);
            if(!bret) {
                return  false;
            }
            //获取输出的数据源
            this.dataSource = prosor.getDataOutput();
            
            PushBufferDataSource pbds = (PushBufferDataSource)this.dataSource;
            PushBufferStream pbss[] = pbds.getStreams();
            
            rtpMgrs = new RTPManager[pbss.length];
            SendStream sendStream;
            int port;
            
            SourceDescription srcDesList[];
            
            for (int i = 0; i < pbss.length; i++) {
                try {
                    rtpMgrs[i] = RTPManager.newInstance();
                    
                    port = this.portBase + 2*i;
                    
                    //先加入本地的地址吧
                    SessionAddress local = new SessionAddress(InetAddress.getLocalHost(),port);
                    
                    rtpMgrs[i].initialize(local);
                    
                    
                    SessionAddress remotoAdd = new SessionAddress(this.inetRometo,port);
                    rtpMgrs[i].addTarget(remotoAdd);
                    
                    System.err.println( "Created RTP session: " + port+"romoto:"+remotoAdd.getDataHostAddress()+"prot:"+remotoAdd.getDataPort());
                    
                    sendStream = rtpMgrs[i].createSendStream(dataSource, i);
                    rtpMgrs[i].addSendStreamListener(new streamSend());
                    rtpMgrs[i].addSessionListener(new sessionlist());
                    sendStream.start();
                    //
                    System.out.println("传输开始");
                    
                } catch (InvalidSessionAddressException  e) {
                    System.out.println(e.toString());
                    System.out.println("发生错误!检查"+e.getMessage());
                    
                }catch (Exception  e) {
                    System.out.println(e.toString());
                    System.out.println("发生总体错误!检查"+e.getMessage());
                    
                }
            }
            
            
            
        } catch (NoDataSourceException ex) {
            System.out.println("错误发生了");
        } catch (IOException ex) {
            System.out.println("错误发生了");
        }catch(NoProcessorException e) {
            System.out.println(":"+e.toString());
        }
        
        System.out.println("结束了");
        return true;
    }
    //检测状态
    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;
    }
    
    /**
     * 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.
     */
    public 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);
    }
    
    
    /*******************************************************************************************
     ***   **/
    // 检测处理器状态
    class Conlist implements  ControllerListener {
        
        Processor process;
        
        public Conlist(Processor process) {
            
            
            this.process = process;
        }
        public synchronized 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();
                }
            }
        }
    }
    public void stop() {
	synchronized (this) {
	    if (prosor != null) {
		prosor.stop();
		prosor.close();
		prosor = null;
		for (int i = 0; i < rtpMgrs.length; i++) {
		    rtpMgrs[i].removeTargets( "Session ended.");
		    rtpMgrs[i].dispose();
                    System.out.println("会话管理器关闭");
		}
	    }
	}
    }
   /****************************************************************************************
    *                             监控数据发送时候的状态                                     *
    *****************************************************************************************/
    class streamSend implements SendStreamListener
    {
        public synchronized void update(SendStreamEvent event)
        {
           
            if(event instanceof NewSendStreamEvent)
            {
                System.out.println("数据产生了");
            }
            if(event instanceof StreamClosedEvent)
            {
                System.out.println("数据结束");
            }
            if(event instanceof LocalPayloadChangeEvent)
            {
                System.out.println("数据状态改变");
            }
        };
    }
    /******************************************************************************************
     *                          session 控制                                                  *
     ******************************************************************************************/
    class sessionlist implements SessionListener
    {
        public synchronized void update(SessionEvent event)
        {
            System.out.println("产生时间");
           
        };
    }
    
    
    
}