selectandread.java

一套MMORPG手机网络游戏的服务端

/*
 * Created on 2005-4-20
 *
 * 系统的输入接口，新的连接请求在GameServer处进行处理后，会发送到这里，在这里注册到监听读的Selector
 * 上，随后读出数据，根据请求中的游戏代号从GameServer处取得对应的GameController，把后面的数据封入
 * GameEvent传递给GameController处理。
 * 
 * 这里要处理使用HTTP请求上行数据的玩家，其channel是要经常改变的，应该根据请求的ID来判断玩家是否在线
 */
package com.swing.server.server;

import java.nio.*;
import java.nio.channels.*;
import java.io.*;
import java.util.*;
import org.apache.log4j.Logger;

import com.swing.server.common.Attachment;
import com.swing.server.common.GameEvent;
import com.swing.server.common.Lifecycle;
import com.swing.server.common.LifecycleException;
import com.swing.server.common.NIOUtils;
import com.swing.server.common.Player;
import com.swing.server.server.controller.GameController;

/**
 * @author vampire_a
 * 
 * TODO To change the template for this generated type comment go to Window -
 * Preferences - Java - Code Style - Code Templates
 */
public class SelectAndRead implements Runnable, Lifecycle {
    /** log4j logger */
    private static Logger log = Logger.getLogger("SelectAndRead");

    private boolean ifRun;

    /** pending connections */
    private LinkedList newClients;

    //    /** timeout for the selector's select() call */
    //    private static final long SELECT_TIMEOUT = 250;

    /** the selector, multiplexes access to client channels */
    private Selector selector;

    /** reference to the GameServer */
    private GameServer gameServer;

    /**
     * Constructor.
     */
    public SelectAndRead(GameServer gameServer) {
        this.gameServer = gameServer;
        newClients = new LinkedList();
        ifRun = false;
    }

    /**
     * adds to the list of pending clients
     */
    public void addNewClient(SocketChannel clientChannel) {
        synchronized (newClients) {
            newClients.addLast(clientChannel);
        }
        // force selector to return
        // so our new client can get in the loop right away
        selector.wakeup();
    }

    /**
     * loop forever, first doing our select() then check for new connections
     */
    public void run() {
        try {
            selector = Selector.open();

            while (ifRun) {
                select();
                checkNewConnections();

                // sleep just a bit
                try {
                    Thread.sleep(30);
                } catch (InterruptedException e) {
                }
            }
        } catch (IOException e) {
            log.fatal("exception while opening Selector", e);
        }
    }

    /**
     * 检查是否有新的连接接入，如果有就注册到selector上
     */
    private void checkNewConnections() {
        synchronized (newClients) {
            while (newClients.size() > 0) {
                try {
                    SocketChannel clientChannel = (SocketChannel) newClients
                            .removeFirst();
                    clientChannel.configureBlocking(false);
                    clientChannel.register(selector, SelectionKey.OP_READ,
                            new Attachment());
                } catch (ClosedChannelException cce) {
                    log.error("channel closed", cce);
                } catch (IOException ioe) {
                    log.error("ioexception on clientChannel", ioe);
                }
            }
        }
    }

    /**
     * do our select, read from the channels and hand off events to
     * GameControllers 查看当前selector上所有channel的状态，如果有数据进入，就读取并封装到包内
     */
    private void select() {
        try {
            // this is a blocking select call but will
            // be interrupted when new clients come in
            selector.select();
            Set readyKeys = selector.selectedKeys();

            Iterator i = readyKeys.iterator();
            while (i.hasNext()) {
                SelectionKey key = (SelectionKey) i.next();
                i.remove();
                SocketChannel channel = (SocketChannel) key.channel();
                Attachment attachment = (Attachment) key.attachment();

                try {
                    // read from the channel
                    long nbytes = channel.read(attachment.readBuff);
                    // check for end-of-stream condition
                    if (nbytes == -1) {
                        log.info("disconnect: "
                                + channel.socket().getInetAddress()
                                + ", end-of-stream");
                        channel.close();
                    }

                    // check for a complete event
                    try {
                        if (attachment.readBuff.position() >= Attachment.HEADER_SIZE) {
                            // 接收完成，重置position和limit准备读取
                            attachment.readBuff.flip();

                            // 判断是否可以已经收到一条完整的请求
                            while (attachment.eventReady()) {
                                // 取得事件
                                GameEvent event = getEvent(attachment);
                                // 转发事件
                                delegateEvent(event, channel);
                                // 准备下次接收
                                attachment.reset();
                            }
                            // 把没处理的数据前移
                            attachment.readBuff.compact();
                            int z = 0;
                        }
                    } catch (IllegalArgumentException iae) {
                        log.error("illegal argument exception", iae);
                    }
                } catch (IOException ioe) {
                    log.warn("IOException during read(), closing channel:"
                            + channel.socket().getInetAddress());
                    channel.close();
                }
            }
        } catch (IOException ioe2) {
            log.warn("IOException during select(): " + ioe2.getMessage());
        } catch (Exception e) {
            log.error("exception during select()", e);
        }
    }

    /**
     * 用attachment.payload里的数据初始化event对象
     */
    private GameEvent getEvent(Attachment attachment) {
        ///////////////////////
        GameEvent event = null;
        ByteBuffer bb = ByteBuffer.wrap(attachment.payload, 0,
                attachment.payloadSize);

        // get the controller and tell it to instantiate an event for us
        GameController gc = gameServer
                .getGameControllerByCode(attachment.gameCode);

        if (gc == null) {
            log.error("No GameController for gameCode: " + attachment.gameCode);
            log.warn("Use DefaultGameController for gameCode: "
                    + attachment.gameCode);
            event = gameServer.defaultController.createGameEvent("DEFAULT");
            event.setKind(attachment.type);
            return event;
        }

        event = gc.createGameEvent(attachment.gameCode);
        event.setKind(attachment.type);
        // read the event from the payload
        event.read(bb);
        return event;
    }

    /**
     * pass off an event to the appropriate GameController based on the GameName
     * of the event 将event对象发送给适合的GameController进行处理，基于gameCode查找GameController
     */
    private void delegateEvent(GameEvent event, SocketChannel channel) {
        if (event != null && event.getGameCode() == null) {
            log.error("GameServer.handleEvent() : gameName is null");
            return;
        }

        GameController gc = gameServer.getGameControllerByCode(event
                .getGameCode());

        // 没有人处理的事件由defaultGameController处理
        if (gc == null) {
            ByteBuffer bb = ByteBuffer.allocate(50);
            NIOUtils.prepBuffer(event, bb);
            NIOUtils.channelWrite(channel, bb);
            return;
        }

        Player p = gc.getPlayerById(event.getPlayerId());
        if (p != null) {
            if (p.getChannel() != channel) {
                log.warn("player is on a new channel, must be reconnect.");
                p.setChannel(channel);
            }
        } else {
            // first time we see a playerId, create the Player object
            // and populate the channel, and also add to our lists
            p = gc.createPlayer(event.getKind());
            p.setPlayerId(event.getPlayerId());
            p.setChannel(channel);
            gc.addPlayer(p);
            log
                    .debug("delegate event, new player created and channel set, player:"
                            + p.getPlayerId() + ", channel: " + channel);
        }
        gc.handleEvent(event);
    }

    /*
     * (non-Javadoc)
     * 
     * @see com.swing.server.common.Lifecycle#start()
     */
    public void start() throws LifecycleException {
        ifRun = true;
        Thread t = new Thread(this);
        t.start();
    }

    /*
     * (non-Javadoc)
     * 
     * @see com.swing.server.common.Lifecycle#stop()
     */
    public void stop() throws LifecycleException {
        // TODO Auto-generated method stub
        log.warn("" + newClients.size() + " clients haven't been added!");
        newClients.clear();
        newClients = null;
        selector.wakeup();
        ifRun = false;

    }

}// SelectAndRead