porter.java

用java实现的

/*
* LumaQQ - Java QQ Client
*
* Copyright (C) 2004 notXX
*
* This program is free software; you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation; either version 2 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program; if not, write to the Free Software
* Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
*/
package edu.tsinghua.lumaqq.qq;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.DatagramChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
 * 发送接收线程
 * 
 * @author notxx
 */
final class Porter extends Thread {
	/** Logger */
	private static final Log log = LogFactory.getLog(Porter.class);

	/** 线程是否结束的标志 */
	protected boolean shutdown = false;
	/** 端口选择器 */
	protected Selector selector;
	/** 网络事件处理器 */
	protected INIOHandler handler;	
	
	/**
	 * 构造一个Porter.
	 */
	protected Porter() {
		try {
			selector = Selector.open();
		} catch (IOException e) {
			log.debug(e);
			throw new RuntimeException(e);
		}
		setName("Porter");
	}
	
	/**
	 * 注册一个SocketChannel，注册后的channel选择接收CONNECT事件
	 * @param channel
	 * @throws ClosedChannelException
	 */
	public void register(SocketChannel channel) throws ClosedChannelException {
	    channel.register(selector, SelectionKey.OP_CONNECT);
	}
	
	/**
	 * 注册一个DatagramChannel，注册后的channel选择接收READ事件
	 * @param channel
	 * @throws ClosedChannelException
	 */
	public void register(DatagramChannel channel) throws ClosedChannelException {
	    channel.register(selector, SelectionKey.OP_READ);
	}
	
	/**
	 * 设置网络事件的handler
	 * @param handler
	 */
	public void setNIOHandler(INIOHandler handler) {
	    this.handler = handler;
	}
	
	/**
	 * 不断运转维护所有注册的IPort对象.
	 * 通过调用它们的几个函数分别做到清空发送队列/填充接收队列/维护队列的功能.
	 * @see IPort#send(ByteBuffer)
	 * @see IPort#receive(ByteBuffer)
	 * @see IPort#maintain()
	 */
	public void run() {
		log.debug("发送接收线程已经启动");
	    while(!shutdown) {
			try {
			    int n = selector.select(3000);
			    // 如果要shutdown，关闭selector退出
			    if (shutdown) {
			        selector.close();
					break;			        
			    }
			    // 如果select返回大于0，处理事件
			    if(n > 0) {
			        for (Iterator i = selector.selectedKeys().iterator(); i.hasNext();) {
						// 得到下一个Key
						SelectionKey sk = (SelectionKey)i.next();
						i.remove();

						// 处理
			            if(sk.isConnectable())
			                handler.processConnect(sk);
			            else if (sk.isReadable())
			                handler.processRead(sk);
			        }
			    }
			    // 处理发送
			    handler.processWrite();
			} catch (IOException e) {
	            log.error(e.getMessage());
	            handler.processError(e);
	        } catch (PacketParseException e) {
	            log.error("包解析错误: " + e.getMessage());
	        }
		}
		log.debug("发送接收线程已经退出");
	}
	
    /**
     * 关闭porter
     */
    public void shutdown() {
	    shutdown = true;
	    if(selector != null)
	        selector.wakeup();
    }
    
    /**
     * 唤醒selector
     */
    public void wakeup() {
        selector.wakeup();
    }
}