filesender.java

LUMAQQ源代码 JAVA 地球人都知道

    /**
     * 建立文件传输连接
     * @param sk
     * @param channel
     * @param address
     * @throws IOException
     */
    private void establishConnection(SelectionKey sk, DatagramChannel channel, InetSocketAddress address) throws IOException {
        // 根据是直接端口还是本地端口设置连接方式
        Boolean b = (Boolean)sk.attachment();
        major = b.booleanValue();
        // 连接
        channel.connect(address);
        // 把这个channel的引用交给dc
        dc = channel;
        // say hello
        sayHello(fcp, buffer);
        // 设置当前状态为传送中
        fileTransferStatus = FT_SAYING_HELLO;
        // 初始化滑窗和分片缓冲区
        initSlideWindow(1, 0, fragments);
        initFragmentBuffer(localFile, 1, maxFragmentSize, fragments);
        // 启动heart beat线程
        hbThread = new HeartBeatThread(this);
        hbThread.start();    
    }

    /* (non-Javadoc)
     * @see edu.tsinghua.lumaqq.qq.filetrans.FileWatcher#finish()
     */
    public void finish() {
        // 设置文件传输状态为false，关闭文件守望者
        shutdown();
        setFileTransferStatus(FT_NONE);
        fireFileFinishedEvent();
    }

    /* (non-Javadoc)
     * @see edu.tsinghua.lumaqq.qq.filetrans.FileWatcher#start()
     */
    public void start() {
	    try {
            // 创建Selector
            selector = Selector.open();
        } catch (IOException e) {
            log.error("无法创建Selector");
            return;
        }
        useUdp = true;
        // 启动端口
        startMajorPort();
        startMinorPort();
        // 启动selector
        new Thread(this).start();
    }  
    
    /**
     * 启动直接端口的监听
     */
    private void startMajorPort() {        
	    try {
            dcMajor = DatagramChannel.open();
            dcMajor.configureBlocking(false);
            try {
	            dcMajor.socket().bind(new InetSocketAddress(Util.getIpStringFromBytes(myInternetIp), 0));                
            } catch (SocketException e) {
                dcMajor.socket().bind(new InetSocketAddress(Util.getIpStringFromBytes(myLocalIp), 0));
            }
            myMajorPort = dcMajor.socket().getLocalPort();
            dcMajor.register(selector, SelectionKey.OP_READ, new Boolean(true));
        } catch (Exception e) {
            log.error(e.getMessage());
        }
    }
    
    /**
     * 启动本地端口的监听
     */
    private void startMinorPort() {
	    try {	    
            dcMinor = DatagramChannel.open();
            dcMinor.configureBlocking(false);
            DatagramSocket ds = dcMinor.socket();
            ds.bind(new InetSocketAddress(Util.getIpStringFromBytes(myLocalIp), 0));
            myMinorPort = ds.getLocalPort();
            dcMinor.register(selector, SelectionKey.OP_READ, new Boolean(false));
        } catch (Exception e) {
            log.error(e.getMessage());
        }
    }
    
    /**
     * 初始化分片缓冲区
     * @param file RandomAccessFile对象
     * @param size buffer大小
     * @param fz 分片大小
     * @param max 文件最大的分片序号
     */
    public void initFragmentBuffer(RandomAccessFile file, int size, int fz, int max) {
        if(fb == null)
            fb = new FragmentBuffer(file, size, fz, max);
    }
    
    /**
     * 发送sender的buffer内容 
     */
    public void send() {
        try {
            if(useUdp)
                dc.write(buffer);
        } catch (IOException e) {
            log.error(e.getMessage());
        }
    }
    
    /**
     * 发送buffer中的内容
     */
    public void send(ByteBuffer buffer) {
        try {
            if(useUdp)
                dc.write(buffer);
        } catch (IOException e) {
            log.error(e.getMessage());
        }
    }
    
    /**
     * 发送heart beat 
     * @param seq 序号
     */
    protected void sendHeartBeat(FileDataPacket fdp, ByteBuffer buffer, char seq) {
        fdp.setCommand(QQ.QQ_FILE_CMD_HEART_BEAT);
        fdp.setHeartBeatSequence(seq);
        fdp.fill(buffer);
        buffer.flip();
        send(buffer);
        log.debug("Heart Beat " + (int)seq + " 已发送");
    }
    
    /**
     * say hello
     */
    protected void sayHello(FileControlPacket fcp, ByteBuffer buffer) {
        fcp.setCommand(QQ.QQ_FILE_CMD_SENDER_SAY_HELLO);
        if(condition == QQ.QQ_SAME_LAN)
            fcp.setHelloByte(QQ.QQ_SAME_IN_TO_SAME_IN_HELLO);
        else if(condition == QQ.QQ_HE_IS_BEHIND_FIREWALL)
            fcp.setHelloByte(QQ.QQ_OUT_TO_IN_HELLO);
        else if(condition == QQ.QQ_I_AM_BEHIND_FIREWALL)
            fcp.setHelloByte(QQ.QQ_IN_TO_OUT_HELLO);
        else if(condition == QQ.QQ_NONE_BEHIND_FIREWALL)
            fcp.setHelloByte(QQ.QQ_OUT_TO_OUT_HELLO);
        fcp.fill(buffer);
        buffer.flip();
        send(buffer);
        buffer.rewind();
        send(buffer);
        log.debug("Hello 包已经发送");
    }
    
    /**
     * 发送窗口中还没有收到确认的分片
     */
    protected void sendFragment(FileDataPacket fdp, ByteBuffer buffer) {
        // 由于这是heart beat线程调用的，所以需要同步
        synchronized(window) {
	        if(window.isFinished()) {
	            fileTransferStatus = FT_SENDING_EOF;
	            sendEOF(fdp, buffer);
	        } else {
	            fdp.setCommand(QQ.QQ_FILE_CMD_FILE_OP);
	            fdp.setInfoType(QQ.QQ_FILE_DATA_INFO);
	            int low = window.getLow();
	            int high = window.getHigh();
	            int mask = window.getMask();
	            for(int i = low, j = 1; i <= high; i++) {
	                // 根据mask判断这个分片是否已经得到了确认，如果没有，则重发
	                if((mask & j) == 0) {
	                    fdp.setPacketIndex(packetSN++);
	                    fdp.setFragmentIndex(i);
	                    if(high == fragments)
	                        fdp.setFragmentLength(fileSize % maxFragmentSize);
	                    else
	                        fdp.setFragmentLength(maxFragmentSize);
	                    fdp.setFragmentData(fb.getFragment(i));	                        
	                    fdp.setFragmentOffset(i * maxFragmentSize);
	                    fdp.fill(buffer);
	                    buffer.flip();
	                    send(buffer);
	                }
	                j <<= 1;
	            }                
            }
        }
    }
    
    /**
     * 发送文件EOF信息
     */
    protected void sendEOF(FileDataPacket fdp, ByteBuffer buffer) {
        fdp.setCommand(QQ.QQ_FILE_CMD_FILE_OP);
        fdp.setInfoType(QQ.QQ_FILE_EOF);
        fdp.setPacketIndex(packetSN++);
        fdp.fill(buffer);
        buffer.flip();
        send(buffer);
    }
    
    /**
     * 发送文件基本信息
     */
    protected void sendBasic(FileDataPacket fdp, ByteBuffer buffer) {
        fdp.setCommand(QQ.QQ_FILE_CMD_FILE_OP);
        fdp.setInfoType(QQ.QQ_FILE_BASIC_INFO);
        fdp.setPacketIndex(packetSN++);
        fdp.fill(buffer);
        buffer.flip();
        send(buffer);
    }
    
    /**
     * 发送传输结束信息
     */
    protected void sendFinish(FileDataPacket fdp, ByteBuffer buffer) {
        fdp.setCommand(QQ.QQ_FILE_CMD_TRANSFER_FINISHED);
        fdp.fill(buffer);
        buffer.flip();
        for(int i = 0; i < 4; i++) {
            buffer.rewind();
            send(buffer);
        }
    }
    
    /**
     * @return Returns the suspend.
     */
    public boolean isSuspend() {
        return suspend;
    }
    
    /**
     * @param suspend The suspend to set.
     */
    public void setSuspend(boolean suspend) {
        this.suspend = suspend;
    }
}