/*
 * This software is distributed under following license based on modified BSD
 * style license.
 * ----------------------------------------------------------------------
 * 
 * Copyright 2009 The Nimbus2 Project. All rights reserved.
 * 
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 * 
 * 1. Redistributions of source code must retain the above copyright notice,
 *    this list of conditions and the following disclaimer. 
 * 2. Redistributions in binary form must reproduce the above copyright notice,
 *    this list of conditions and the following disclaimer in the documentation
 *    and/or other materials provided with the distribution.
 * 
 * THIS SOFTWARE IS PROVIDED BY THE NIMBUS PROJECT ``AS IS'' AND ANY EXPRESS
 * OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN
 * NO EVENT SHALL THE NIMBUS PROJECT OR CONTRIBUTORS BE LIABLE FOR ANY
 * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
 * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
 * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
 * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
 * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 * 
 * The views and conclusions contained in the software and documentation are
 * those of the authors and should not be interpreted as representing official
 * policies, either expressed or implied, of the Nimbus2 Project.
 */
package jp.ossc.nimbus.service.publish.udp;

import java.io.EOFException;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.nio.ByteBuffer;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.channels.Selector;
import java.nio.channels.SelectionKey;
import java.nio.channels.ClosedSelectorException;
import java.nio.channels.CancelledKeyException;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketAddress;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.DatagramPacket;
import java.net.MulticastSocket;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import jp.ossc.nimbus.daemon.Daemon;
import jp.ossc.nimbus.daemon.DaemonControl;
import jp.ossc.nimbus.daemon.DaemonRunnable;
import jp.ossc.nimbus.core.Service;
import jp.ossc.nimbus.core.ServiceName;
import jp.ossc.nimbus.service.log.Logger;
import jp.ossc.nimbus.service.publish.Client;
import jp.ossc.nimbus.service.publish.Message;
import jp.ossc.nimbus.service.publish.MessageCreateException;
import jp.ossc.nimbus.service.publish.MessageSendException;
import jp.ossc.nimbus.service.publish.MessageException;
import jp.ossc.nimbus.service.publish.ServerConnection;
import jp.ossc.nimbus.service.publish.ServerConnectionListener;
import jp.ossc.nimbus.service.queue.AsynchContext;
import jp.ossc.nimbus.service.queue.Queue;
import jp.ossc.nimbus.service.queue.DefaultQueueService;
import jp.ossc.nimbus.service.queue.QueueHandler;
import jp.ossc.nimbus.service.queue.QueueHandlerContainer;
import jp.ossc.nimbus.service.queue.QueueHandlerContainerService;
import jp.ossc.nimbus.service.queue.AbstractDistributedQueueSelectorService;
import jp.ossc.nimbus.service.queue.DistributedQueueHandlerContainerService;
import jp.ossc.nimbus.service.io.Externalizer;
import jp.ossc.nimbus.net.SocketFactory;
import jp.ossc.nimbus.service.publish.tcp.ClientMessage;
import jp.ossc.nimbus.service.publish.tcp.AddMessage;
import jp.ossc.nimbus.service.publish.tcp.RemoveMessage;
import jp.ossc.nimbus.service.publish.tcp.StartReceiveMessage;

/**
 * UDPプロトコル用の{@link ServerConnection}インタフェース実装クラス。<p>
 *
 * @author M.Takata
 */
public class ServerConnectionImpl implements ServerConnection{
    
    private ServerSocket serverSocket;
    private ServerSocketChannel serverSocketChannel;
    private Selector selector;
    private DatagramSocket sendSocket;
    private InetAddress multicastAddress;
    private InetSocketAddress sendSocketAddress;
    private int destPort;
    
    private Set<ServerConnectionImpl.ClientImpl> clients = Collections.synchronizedSet(new LinkedHashSet<ServerConnectionImpl.ClientImpl>());
    private Map<Object, ServerConnectionImpl.ClientImpl> clientMap = Collections.synchronizedMap(new HashMap<Object, ServerConnectionImpl.ClientImpl>());
    private Set<ServerConnectionImpl.ClientImpl> newClients = Collections.synchronizedSet(new HashSet<ServerConnectionImpl.ClientImpl>());
    private int maxSendRetryCount;
    private Logger logger;
    private String sendErrorMessageId;
    private String sendErrorRetryOverMessageId;
    private String responseErrorMessageId;
    private String messageLostErrorMessageId;
    private Daemon clientAcceptor;
    private QueueHandlerContainerService<AsynchContext<SendRequest,?>> sendQueueHandlerContainer;
    private ClientDistributedQueueSelector queueSelector;
    private QueueHandlerContainer<AsynchContext<SendRequest,?>> asynchSendQueueHandlerContainer;
    private QueueHandlerContainerService<RequestHandleRequest> requestHandleQueueHandlerContainer;
    private long sendPacketCount;
    private long sendCount;
    private long sendProcessTime;
    private List<ServerConnectionListener> serverConnectionListeners;
    private Externalizer<Object> externalizer;
    private SocketFactory socketFactory;
    private int windowSize;
    private int currentSequence = 0;
    private int maxWindowCount;
    private List<MessageImpl> sendBufferList = Collections.synchronizedList(new ArrayList<MessageImpl>());
    private Map<Integer,List<Window>> sendBufferMap = Collections.synchronizedMap(new HashMap<Integer,List<Window>>());
    private long sendBufferTime;
    
    public ServerConnectionImpl(
        ServerSocket serverSocket,
        Externalizer<Object> ext,
        int sendThreadSize,
        ServiceName sendQueueServiceName,
        int asynchSendThreadSize,
        ServiceName asynchSendQueueServiceName,
        InetSocketAddress sendSocketAddress,
        InetAddress multicastAddress,
        int destPort
    ) throws Exception{
        this.serverSocket = serverSocket;
        this.sendSocketAddress = sendSocketAddress;
        this.multicastAddress = multicastAddress;
        this.destPort = destPort;
        externalizer = ext;
        
        initSendSocket();
        initSend(sendQueueServiceName, sendThreadSize, multicastAddress != null);
        initAsynchSend(asynchSendQueueServiceName, asynchSendThreadSize, multicastAddress != null);
        
        initClientAcceptor(serverSocket.getLocalSocketAddress());
    }
    
    public ServerConnectionImpl(
        ServerSocketChannel ssc,
        Externalizer<Object> ext,
        int sendThreadSize,
        ServiceName sendQueueServiceName,
        int asynchSendThreadSize,
        ServiceName asynchSendQueueServiceName,
        int requestHandleThreadSize,
        ServiceName requestHandleQueueServiceName,
        SocketFactory sf,
        InetSocketAddress sendSocketAddress,
        InetAddress multicastAddress,
        int destPort
    ) throws Exception{
        serverSocketChannel = ssc;
        socketFactory = sf;
        this.sendSocketAddress = sendSocketAddress;
        this.multicastAddress = multicastAddress;
        this.destPort = destPort;
        externalizer = ext;
        
        initSendSocket();
        initSend(sendQueueServiceName, sendThreadSize, multicastAddress != null);
        initAsynchSend(asynchSendQueueServiceName, asynchSendThreadSize, multicastAddress != null);
        initRequestHandle(requestHandleQueueServiceName, requestHandleThreadSize);
        
        selector = Selector.open();
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT, null);
        initClientAcceptor(serverSocketChannel.socket().getLocalSocketAddress());
    }
    
    private void initSendSocket() throws IOException{
        if(multicastAddress != null){
            sendSocket = multicastAddress.isMulticastAddress() ? new MulticastSocket(sendSocketAddress) : new DatagramSocket(sendSocketAddress);
        }else if(sendSocketAddress.getPort() != 0){
            sendSocket = new DatagramSocket(sendSocketAddress);
        }
    }
    
    private void initClientAcceptor(SocketAddress localAddress){
        clientAcceptor = new Daemon(new ClientAcceptor());
        clientAcceptor.setName(
            "Nimbus Publish(UDP) ServerConnection ClientAcceptor " + localAddress
        );
        clientAcceptor.setDaemon(true);
        clientAcceptor.start();
    }
    
    private void initSend(ServiceName sendQueueServiceName, int sendThreadSize, boolean isMulticast) throws Exception{
        if(!isMulticast && sendThreadSize >= 1){
            sendQueueHandlerContainer = new QueueHandlerContainerService<AsynchContext<SendRequest,?>>();
            sendQueueHandlerContainer.create();
            if(sendQueueServiceName == null){
                DefaultQueueService<AsynchContext<SendRequest,?>> sendQueue = new DefaultQueueService<AsynchContext<SendRequest,?>>();
                sendQueue.create();
                sendQueue.start();
                sendQueueHandlerContainer.setQueueService(sendQueue);
            }else{
                sendQueueHandlerContainer.setQueueServiceName(sendQueueServiceName);
            }
            sendQueueHandlerContainer.setQueueHandlerSize(sendThreadSize);
            sendQueueHandlerContainer.setQueueHandler(new SendQueueHandler());
            sendQueueHandlerContainer.start();
        }
    }
    
    private void initAsynchSend(ServiceName queueFactoryServiceName, int asynchSendThreadSize, boolean isMulticast) throws Exception{
        if(asynchSendThreadSize <= 0){
            return;
        }
        if(isMulticast){
            QueueHandlerContainerService<AsynchContext<SendRequest,?>> queueHandlerContainer = new QueueHandlerContainerService<AsynchContext<SendRequest,?>>();
            queueHandlerContainer.create();
            if(queueFactoryServiceName == null){
                DefaultQueueService<AsynchContext<SendRequest,?>> sendQueue = new DefaultQueueService<AsynchContext<SendRequest,?>>();
                sendQueue.create();
                sendQueue.start();
                queueHandlerContainer.setQueueService(sendQueue);
            }else{
                queueHandlerContainer.setQueueServiceName(queueFactoryServiceName);
            }
            queueHandlerContainer.setQueueHandlerSize(asynchSendThreadSize);
            queueHandlerContainer.setQueueHandler(new SendQueueHandler());
            queueHandlerContainer.start();
            asynchSendQueueHandlerContainer = queueHandlerContainer;
        }else{
            queueSelector = new ClientDistributedQueueSelector();
            queueSelector.create();
            queueSelector.setDistributedSize(asynchSendThreadSize);
            if(queueFactoryServiceName != null){
                queueSelector.setQueueFactoryServiceName(queueFactoryServiceName);
            }
            queueSelector.start();
            
            DistributedQueueHandlerContainerService<AsynchContext<SendRequest,?>> distributedQueueHandlerContainer = new DistributedQueueHandlerContainerService<AsynchContext<SendRequest,?>>();
            distributedQueueHandlerContainer.create();
            distributedQueueHandlerContainer.setDistributedQueueSelector(queueSelector);
            distributedQueueHandlerContainer.setQueueHandler(new SendQueueHandler());
            distributedQueueHandlerContainer.start();
            asynchSendQueueHandlerContainer = distributedQueueHandlerContainer;
        }
    }
    
    private void initRequestHandle(ServiceName requestHandleQueueServiceName, int requestHandleThreadSize) throws Exception{
        if(requestHandleThreadSize >= 1){
            requestHandleQueueHandlerContainer = new QueueHandlerContainerService<RequestHandleRequest>();
            requestHandleQueueHandlerContainer.create();
            if(requestHandleQueueServiceName == null){
                DefaultQueueService<RequestHandleRequest> requestHandleQueue = new DefaultQueueService<RequestHandleRequest>();
                requestHandleQueue.create();
                requestHandleQueue.start();
                requestHandleQueueHandlerContainer.setQueueService(requestHandleQueue);
            }else{
                requestHandleQueueHandlerContainer.setQueueServiceName(requestHandleQueueServiceName);
            }
            requestHandleQueueHandlerContainer.setQueueHandlerSize(requestHandleThreadSize);
            requestHandleQueueHandlerContainer.setQueueHandler(new RequestHandleQueueHandler());
            requestHandleQueueHandlerContainer.start();
        }
    }
    
    public void setTimeToLive(int ttl) throws IOException{
        if(multicastAddress != null && multicastAddress.isMulticastAddress() && ttl >= 0 && sendSocket != null){
            ((MulticastSocket)sendSocket).setTimeToLive(ttl);
        }
    }
    
    public void setWindowSize(int bytes){
        windowSize = bytes;
        if(sendSocket != null){
            try{
                int sendBufferSize = sendSocket.getSendBufferSize();
                if(sendBufferSize < windowSize){
                    sendSocket.setSendBufferSize(windowSize);
                }
            }catch(SocketException e){
            }
        }
    }
    
    public void setSendBufferTime(long time){
        sendBufferTime = time;
    }
    
    public void setMaxSendRetryCount(int count){
        maxSendRetryCount = count;
        if(sendQueueHandlerContainer != null){
            sendQueueHandlerContainer.setMaxRetryCount(maxSendRetryCount);
        }
        if(asynchSendQueueHandlerContainer != null){
            asynchSendQueueHandlerContainer.setMaxRetryCount(maxSendRetryCount);
        }
    }
    
    public void setLogger(Logger logger){
        this.logger = logger;
    }
    
    public void setSendErrorMessageId(String id){
        sendErrorMessageId = id;
    }
    
    public void setSendErrorRetryOverMessageId(String id){
        sendErrorRetryOverMessageId = id;
    }
    
    public void setResponseErrorMessageId(String id){
        responseErrorMessageId = id;
    }
    
    public void setMessageLostErrorMessageId(String id){
        messageLostErrorMessageId = id;
    }
    public String getMessageLostErrorMessageId(){
        return messageLostErrorMessageId;
    }
    
    public int getMaxWindowCount(){
        return maxWindowCount;
    }
    
    public double getAverageWindowCount(){
        return sendCount == 0 ? 0.0d : (double)sendPacketCount / (double)sendCount;
    }
    
    public long getAverageAsynchSendProcessTime(){
        return asynchSendQueueHandlerContainer == null ? 0 : asynchSendQueueHandlerContainer.getAverageHandleProcessTime();
    }
    
    public long getAverageRequestHandleProcessTime(){
        return requestHandleQueueHandlerContainer == null ? 0 : requestHandleQueueHandlerContainer.getAverageHandleProcessTime();
    }
    
    public Message createMessage(String subject, String key) throws MessageCreateException{
        final MessageImpl message = multicastAddress == null ? new MessageImpl() : new MulticastMessageImpl();
        message.setSubject(subject, key);
        return message;
    }
    
    public Message castMessage(Message message) throws MessageException{
        if(message instanceof MessageImpl){
            return message;
        }
        Message msg = createMessage(message.getSubject(), message.getKey());
        msg.setObject(message.getObject());
        return msg;
    }
    
    private void sendMessage(DatagramSocket sendSocket, InetAddress destAddress, MessageImpl message, int destPort, boolean isRetry) throws IOException{
        List<Window> windows = message.getWindows(windowSize, externalizer);
        maxWindowCount = Math.max(maxWindowCount, windows.size());
        List<DatagramPacket> packets = new ArrayList<DatagramPacket>();
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        DataOutputStream dos = new DataOutputStream(baos);
        for(int i = 0, imax = windows.size(); i < imax; i++){
            Window window = windows.get(i);
            baos.reset();
            window.write(dos);
            dos.flush();
            byte[] bytes = baos.toByteArray();
            DatagramPacket packet = new DatagramPacket(
                bytes,
                bytes.length,
                destAddress,
                destPort
            );
            packets.add(packet);
        }
        for(int i = 0, imax = packets.size(); i < imax; i++){
            DatagramPacket packet = packets.get(i);
            if(isRetry){
                for(int j = 0; j <= maxSendRetryCount; i++){
                    try{
                        synchronized(sendSocket){
                            sendSocket.send(packet);
                        }
                        break;
                    }catch(IOException e){
                        if(logger != null){
                            if(j >= maxSendRetryCount){
                                if(sendErrorRetryOverMessageId != null){
                                    logger.write(
                                        sendErrorRetryOverMessageId,
                                        e,
                                        destAddress + ":" + destPort,
                                        message
                                    );
                                }
                                throw e;
                            }else{
                                if(sendErrorMessageId != null){
                                    logger.write(
                                        sendErrorMessageId,
                                        e,
                                        destAddress + ":" + destPort,
                                        message
                                    );
                                }
                            }
                        }
                    }
                }
            }else{
                synchronized(sendSocket){
                    sendSocket.send(packet);
                }
            }
        }
    }
    
    public void send(Message message) throws MessageSendException{
        if(!(message instanceof MessageImpl)){
            throw new MessageSendException("Message is illegal class. " + (message == null ? null : message.getClass()));
        }
        long startTime = System.currentTimeMillis();
        try{
            Set<ClientImpl> firstClients = allocateSequence((MessageImpl)message);
            try{
                addSendBuffer((MessageImpl)message);
            }catch(IOException e){
                throw new MessageSendException("Send error : message=" + message, e);
            }
            if(clients.size() == 0){
                return;
            }
            if(multicastAddress == null){
                ClientImpl[] clientArray = clients.toArray(new ClientImpl[clients.size()]);
                if(sendQueueHandlerContainer == null){
                    List<ClientImpl> currentClients = new ArrayList<ClientImpl>();
                    for(int i = 0; i < clientArray.length; i++){
                        if(!clientArray[i].isStartReceive()
                            || !clientArray[i].isTargetMessage(message)){
                            continue;
                        }
                        currentClients.add(clientArray[i]);
                    }
                    Iterator<ClientImpl> itr = currentClients.iterator();
                    while(itr.hasNext()){
                        ClientImpl client = itr.next();
                        try{
                            client.send(message);
                            itr.remove();
                        }catch(MessageSendException e){
                        }
                    }
                    if(currentClients.size() != 0){
                        throw new MessageSendException("Send error : clients=" + currentClients + ", message=" + message);
                    }
                }else{
                    DefaultQueueService<AsynchContext<SendRequest,Object>> responseQueue = new DefaultQueueService<AsynchContext<SendRequest,Object>>();
                    try{
                        responseQueue.create();
                        responseQueue.start();
                    }catch(Exception e){
                        throw new MessageSendException(e);
                    }
                    responseQueue.accept();
                    for(int i = 0; i < clientArray.length; i++){
                        if(!clientArray[i].isStartReceive()
                            || !clientArray[i].isTargetMessage(message)){
                            clientArray[i] = null;
                            continue;
                        }
                        sendQueueHandlerContainer.push(new AsynchContext<SendRequest,Object>(new SendRequest(clientArray[i], (MessageImpl)message), responseQueue));
                    }
                    List<ClientImpl> errorClients = new ArrayList<ClientImpl>();
                    for(int i = 0; i < clientArray.length; i++){
                        if(clientArray[i] == null){
                            continue;
                        }
                        AsynchContext<SendRequest,?> asynchContext = responseQueue.get();
                        if(asynchContext.getThrowable() != null){
                            errorClients.add(asynchContext.getInput().client);
                        }
                    }
                    if(errorClients.size() != 0){
                        throw new MessageSendException("Send error : clients=" + errorClients + ", message=" + message);
                    }
                }
            }else{
                try{
                    if(firstClients != null){
                        Iterator<ClientImpl> firstClientItr = firstClients.iterator();
                        while(firstClientItr.hasNext()){
                            ClientImpl client = firstClientItr.next();
                            if(client.isStartReceive()){
                                client.send(message);
                            }
                        }
                    }
                    sendMessage(sendSocket, multicastAddress, (MessageImpl)message, destPort, true);
                }catch(IOException e){
                    throw new MessageSendException("Send error : dest=" + multicastAddress + ':' + destPort + ", message=" + message, e);
                }
            }
        }finally{
            sendProcessTime += (System.currentTimeMillis() - startTime);
        }
    }
    
    public void sendAsynch(Message message) throws MessageSendException{
        if(!(message instanceof MessageImpl)){
            throw new MessageSendException("Message is illegal class. " + (message == null ? null : message.getClass()));
        }
        if(asynchSendQueueHandlerContainer == null){
            throw new UnsupportedOperationException();
        }
        Set<ClientImpl> firstClients = allocateSequence((MessageImpl)message);
        try{
            addSendBuffer((MessageImpl)message);
        }catch(IOException e){
            throw new MessageSendException("Send error : message=" + message, e);
        }
        if(clients.size() == 0){
            return;
        }
        if(multicastAddress == null){
            ClientImpl[] clientArray = clients.toArray(new ClientImpl[clients.size()]);
            for(int i = 0; i < clientArray.length; i++){
                asynchSendQueueHandlerContainer.push(new AsynchContext<SendRequest,Object>(new SendRequest(clientArray[i], (MessageImpl)message)));
            }
        }else{
            asynchSendQueueHandlerContainer.push(new AsynchContext<SendRequest,Object>(new SendRequest(firstClients, (MessageImpl)message)));
        }
    }
    
    private synchronized Set<ClientImpl> allocateSequence(MessageImpl message){
        currentSequence++;
        message.setSequence(currentSequence);
        final long currentTime = System.currentTimeMillis();
        message.setSendTime(currentTime);
        Set<ClientImpl> result = null;
        if(newClients.size() != 0){
            final ClientImpl[] clientArray = newClients.toArray(new ClientImpl[newClients.size()]);
            for(int i = 0; i < clientArray.length; i++){
                if(clientArray[i].isStartReceive() && clientArray[i].isFirstMessage() && clientArray[i].isTargetMessage(message)){
                    if(clientArray[i].setFirstMessage(message)){
                        newClients.remove(clientArray[i]);
                        if(result == null){
                            result = new HashSet<ClientImpl>();
                        }
                        result.add(clientArray[i]);
                    }
                }
            }
        }
        return result;
    }
    
    private void addSendBuffer(MessageImpl message) throws IOException{
        final long currentTime = System.currentTimeMillis();
        List<Window> windows = message.getWindows(windowSize, externalizer);
        synchronized(sendBufferMap){
            sendBufferList.add(message);
            sendCount++;
            sendPacketCount += windows.size();
            sendBufferMap.put(new Integer(message.getSequence()), windows);
            for(int i = 0, imax = sendBufferList.size(); i < imax; i++){
                MessageImpl msg = sendBufferList.get(0);
                if((currentTime - msg.getSendTime()) > sendBufferTime){
                    sendBufferList.remove(0);
                    sendBufferMap.remove(new Integer(msg.getSequence()));
                }else{
                    break;
                }
            }
        }
    }
    
    private List<Window> getSendWindows(MessageId id){
        synchronized(sendBufferMap){
            return sendBufferMap.get(new Integer(id.sequence));
        }
    }
    
    private List<MessageImpl> getSendMessages(long from){
        List<MessageImpl> result = new ArrayList<MessageImpl>();
        synchronized(sendBufferMap){
            for(int i = sendBufferList.size(); --i >= 0; ){
                MessageImpl msg = sendBufferList.get(i);
                if(msg.getSendTime() >= from){
                    result.add(0, msg);
                }else{
                    break;
                }
            }
        }
        return result;
    }
    
    private List<MessageImpl> getSendMessages(MessageId from, MessageId to){
        List<MessageImpl> result = new ArrayList<MessageImpl>();
        synchronized(sendBufferMap){
            if(sendBufferList.size() == 0){
                return result;
            }
            int toIndex = to == null ? sendBufferList.size() - 1 : Collections.binarySearch(sendBufferList, to);
            if(toIndex < 0){
                result.add(sendBufferList.get(0));
                return result;
            }
            int fromIndex = Collections.binarySearch(sendBufferList, from);
            if(fromIndex < 0){
                fromIndex = -fromIndex - 1;
            }
            
            for(int i = fromIndex; i < toIndex; i++){
                MessageImpl msg = sendBufferList.get(i);
                result.add(msg);
            }
        }
        return result;
    }
    
    private Window getSendWindow(WindowId id){
        List<Window> windows = null;
        synchronized(sendBufferMap){
            windows = sendBufferMap.get(new Integer(id.sequence));
        }
        if(windows == null || windows.size() < id.windowNo){
            return null;
        }
        return windows.get((int)id.windowNo);
    }
    
    public int getMostOldSendBufferSequence(){
        synchronized(sendBufferMap){
            return sendBufferList.size() == 0 ? 0 : sendBufferList.get(0).sequence;
        }
    }
    
    public int getSendBufferSize(){
        return sendBufferList.size();
    }
    
    public long getSendCount(){
        return sendCount;
    }
    
    public long getSendPacketCount(){
        return sendPacketCount;
    }
    
    public void resetSendCount(){
        sendCount = 0;
        sendPacketCount = 0;
        sendProcessTime = 0;
    }
    
    public long getAverageSendProcessTime(){
        return sendCount == 0 ? 0 : (sendProcessTime / sendCount);
    }
    
    public Set<ServerConnectionImpl.ClientImpl> getClients(){
        return clients;
    }
    
    public int getClientCount(){
        return clients.size();
    }
    
    public Set<Object> getClientIds(){
        return new HashSet<Object>(clientMap.keySet());
    }
    
    public Set<Object> getReceiveClientIds(Message message){
        ClientImpl[] clientArray = clients.toArray(new ClientImpl[clients.size()]);
        Set<Object> result = new HashSet<Object>();
        for(int i = 0; i < clientArray.length; i++){
            if(clientArray[i].isTargetMessage(message)){
                result.add(clientArray[i].getId());
            }
        }
        return result;
    }
    
    public Set getSubjects(Object id){
        ClientImpl client = clientMap.get(id);
        if(client == null){
            return null;
        }
        return client.getSubjects();
    }
    
    public Set getKeys(Object id, String subject){
        ClientImpl client = clientMap.get(id);
        if(client == null){
            return null;
        }
        return client.getKeys(subject);
    }
    
    public synchronized void close(){
        ClientImpl[] clientArray = clients.toArray(new ClientImpl[clients.size()]);
        ServerCloseRequestMessage closeMsg = new ServerCloseRequestMessage();
        for(int i = 0; i < clientArray.length; i++){
            clientArray[i].response(closeMsg);
        }
        
        if(clientAcceptor != null){
            clientAcceptor.stop(1000);
            clientAcceptor = null;
        }
        if(sendQueueHandlerContainer != null){
            sendQueueHandlerContainer.stop();
            sendQueueHandlerContainer.destroy();
            sendQueueHandlerContainer = null;
        }
        if(asynchSendQueueHandlerContainer != null){
            ((Service)asynchSendQueueHandlerContainer).stop();
            ((Service)asynchSendQueueHandlerContainer).destroy();
            asynchSendQueueHandlerContainer = null;
        }
        if(queueSelector != null){
            queueSelector.stop();
            queueSelector.destroy();
            queueSelector = null;
        }
        
        for(int i = 0; i < clientArray.length; i++){
            clientArray[i].close();
        }
        
        if(sendSocket != null){
            sendSocket.close();
            sendSocket = null;
        }
        if(serverSocket != null){
            try{
                serverSocket.close();
            }catch(IOException e){}
            serverSocket = null;
        }
    }
    
    public void addServerConnectionListener(ServerConnectionListener listener){
        if(serverConnectionListeners == null){
            serverConnectionListeners = new ArrayList<ServerConnectionListener>();
        }
        if(!serverConnectionListeners.contains(listener)){
            serverConnectionListeners.add(listener);
        }
    }
    
    public void removeServerConnectionListener(ServerConnectionListener listener){
        if(serverConnectionListeners == null){
            return;
        }
        serverConnectionListeners.remove(listener);
        if(serverConnectionListeners.size() == 0){
            serverConnectionListeners = null;
        }
    }
    
    public String toString(){
        final StringBuffer buf = new StringBuffer();
        buf.append(super.toString());
        buf.append('{');
        buf.append("server=").append(serverSocket == null ? null : serverSocket.getLocalSocketAddress());
        buf.append('}');
        return buf.toString();
    }
    
    private class ClientAcceptor implements DaemonRunnable<Object>{
        private Queue<ClientImpl> responseClientQueue;
        private Queue<SelectionKey> readResponseQueue;
        private Queue<SelectionKey> writeResponseQueue;
        
        public ClientAcceptor(){
            if(selector != null){
                
                DefaultQueueService<ClientImpl> responseQueue = new DefaultQueueService<ClientImpl>();
                try{
                    responseQueue.create();
                    responseQueue.start();
                }catch(Exception e){}
                responseClientQueue = responseQueue;
                
                if(requestHandleQueueHandlerContainer != null){
                    DefaultQueueService<SelectionKey> queue = new DefaultQueueService<SelectionKey>();
                    try{
                        queue.create();
                        queue.start();
                    }catch(Exception e){}
                    readResponseQueue = queue;
                    
                    queue = new DefaultQueueService<SelectionKey>();
                    try{
                        queue.create();
                        queue.start();
                    }catch(Exception e){}
                    writeResponseQueue = queue;
                }
            }
        }
        
        public void addResponseClient(ClientImpl client){
            responseClientQueue.push(client);
            selector.wakeup();
        }
        
        public boolean onStart(){
            if(responseClientQueue != null){
                responseClientQueue.accept();
            }
            return true;
        }
        public boolean onStop(){return true;}
        public boolean onSuspend(){return true;}
        public boolean onResume(){return true;}
        public Object provide(DaemonControl ctrl) throws Throwable{
            if(selector != null){
                try{
                    return selector.select(1000) > 0 ? (Object)selector.selectedKeys() : (Object)this;
                }catch(ClosedSelectorException e){
                    return null;
                }catch(IOException e){
                    return this;
                }
            }else{
                try{
                    return serverSocket.accept();
                }catch(SocketTimeoutException e){
                    return this;
                }catch(SocketException e){
                    return null;
                }catch(IOException e){
                    return this;
                }
            }
        }
        @SuppressWarnings("unchecked")
        public void consume(Object paramObj, DaemonControl ctrl) throws Throwable{
            if(paramObj == null){
                close();
                return;
            }
            if(selector != null){
                ClientImpl client = null;
                boolean isAddKey = false;
                while((client = responseClientQueue.get(0)) != null){
                    client.getSocketChannel().register(
                        selector,
                        SelectionKey.OP_READ | SelectionKey.OP_WRITE,
                        client
                    );
                    isAddKey = true;
                }
                if(isAddKey){
                    if(selector.selectNow() > 0){
                        paramObj = selector.selectedKeys();
                    }
                }
                if(!(paramObj instanceof Set<?>)){
                    return;
                }
                Set<SelectionKey> keys = (Set<SelectionKey>)paramObj;
                int readResponseCount = 0;
                int writeResponseCount = 0;
                for(Iterator<SelectionKey> itr = keys.iterator(); itr.hasNext();){
                    SelectionKey key = itr.next();
                    itr.remove();
                    try{
                        if(key.isAcceptable()){
                            ServerSocketChannel ssc = (ServerSocketChannel)key.channel();
                            SocketChannel channel = ssc.accept();
                            if(channel != null){
                                channel.configureBlocking(false);
                                if(socketFactory != null){
                                    socketFactory.applySocketProperties(channel.socket());
                                }
                                client = new ClientImpl(channel, sendSocket == null ? new DatagramSocket(sendSocketAddress) : null);
                                client.setDestPort(destPort);
                                channel.register(
                                    key.selector(),
                                    SelectionKey.OP_READ,
                                    client
                                );
                                clients.add(client);
                                newClients.add(client);
                                clientMap.put(client.getId(), client);
                            }
                        }else if(key.isReadable()){
                            client = (ClientImpl)key.attachment();
                            if(requestHandleQueueHandlerContainer == null){
                                client.receive(key);
                            }else{
                                requestHandleQueueHandlerContainer.push(
                                    new RequestHandleRequest(client, key, readResponseQueue, RequestHandleRequest.REQUEST_TYPE_READ)
                                );
                                readResponseCount++;
                            }
                        }else if(key.isWritable()){
                            client = (ClientImpl)key.attachment();
                            if(requestHandleQueueHandlerContainer == null){
                                client.writeResponse(key);
                            }else{
                                requestHandleQueueHandlerContainer.push(
                                    new RequestHandleRequest(client, key, writeResponseQueue, RequestHandleRequest.REQUEST_TYPE_WRITE)
                                );
                                writeResponseCount++;
                            }
                        }else if(!key.isValid()){
                            key.cancel();
                        }
                    }catch(CancelledKeyException e){}
                }
                for(int i = 0; i < readResponseCount; i++){
                    readResponseQueue.get();
                }
                for(int i = 0; i < writeResponseCount; i++){
                    writeResponseQueue.get();
                }
            }else{
                if(!(paramObj instanceof Socket)){
                    return;
                }
                Socket socket = (Socket)paramObj;
                if(!socket.isBound() || socket.isClosed()){
                    return;
                }
                ClientImpl client = new ClientImpl(socket, sendSocket == null ? new DatagramSocket(sendSocketAddress) : null);
                clients.add(client);
                newClients.add(client);
                clientMap.put(client.getId(), client);
            }
        }
        public void garbage(){
            if(selector != null){
                try{
                    consume(selector.selectedKeys(), null);
                }catch(Throwable th){
                }
                responseClientQueue.release();
            }
        }
    }
    
    public class ClientImpl implements DaemonRunnable<Object>, Client{
        private DatagramSocket sendSocket;
        private SocketChannel socketChannel;
        private Socket socket;
        private Daemon requestDispatcher;
        private Map<String,Set<String>> subjects;
        private long sendCount;
        private long sendProcessTime;
        private long interpolateRequestCount;
        private boolean isEnabled = true;
        private Object id;
        private ByteBuffer lengthBuffer;
        private ByteBuffer dataBuffer;
        private int dataLength = 0;
        private InetAddress clientAddress;
        private int destPort;
        private long fromTime = -1;
        private boolean isStartReceive = false;
        private Message firstMessage;
        private MessageId latestFirstMessageId;
        private Queue<ByteBuffer> responseQueue;
        
        public ClientImpl(SocketChannel sc, DatagramSocket ss){
            socketChannel = sc;
            sendSocket = ss;
            socket = socketChannel.socket();
            clientAddress = ((InetSocketAddress)socket.getRemoteSocketAddress()).getAddress();
            subjects = Collections.synchronizedMap(new HashMap<String,Set<String>>());
            lengthBuffer = ByteBuffer.allocate(4);
            DefaultQueueService<ByteBuffer> queue = new DefaultQueueService<ByteBuffer>();
            try{
                queue.create();
                queue.start();
            }catch(Exception e){}
            queue.accept();
            responseQueue = queue;
        }
        
        public ClientImpl(Socket sock, DatagramSocket ss) throws IOException{
            socket = sock;
            sendSocket = ss;
            clientAddress = ((InetSocketAddress)socket.getRemoteSocketAddress()).getAddress();
            subjects = Collections.synchronizedMap(new HashMap<String,Set<String>>());
            requestDispatcher = new Daemon(ClientImpl.this);
            requestDispatcher.setName(
                "Nimbus Publish(UDP) ServerConnection ClientRequestDisptcher " + socket.getRemoteSocketAddress()
            );
            requestDispatcher.setDaemon(true);
            requestDispatcher.start();
        }
        
        public boolean isStartReceive(){
            return isStartReceive;
        }
        
        public boolean isFirstMessage(){
            return firstMessage == null;
        }
        public synchronized boolean setFirstMessage(Message msg){
            if(firstMessage == null){
                firstMessage = msg;
                return true;
            }else{
                return false;
            }
        }
        
        public SocketChannel getSocketChannel(){
            return socketChannel;
        }
        
        public Socket getSocket(){
            return socket;
        }
        
        public void setDestPort(int port){
            destPort = port;
        }
        public int getDestPort(){
            return destPort;
        }
        
        public boolean isEnabled(){
            return isEnabled;
        }
        public void setEnabled(boolean isEnabled){
            this.isEnabled = isEnabled;
        }
        
        public boolean isTargetMessage(Message message){
            if(!message.containsDestinationId(getId())){
                return false;
            }
            if(message.getSubject() != null){
                Set<String> sbjs = message.getSubjects();
                for(String subject : sbjs){
                    Set<String> keySet = subjects.get(subject);
                    String key = message.getKey(subject);
                    if(keySet == null){
                        continue;
                    }else if(keySet.contains(null) || keySet.contains(key)){
                        return true;
                    }
                }
            }
            return false;
        }
        
        public synchronized void send(Message message) throws MessageSendException{
            if(!isEnabled || (sendSocket == null && ServerConnectionImpl.this.sendSocket == null)){
                return;
            }
            if(latestFirstMessageId == null){
                if(((MessageImpl)message).isFirst()){
                    firstMessage = null;
                    latestFirstMessageId = (MessageId)message;
                }else if(firstMessage != null && firstMessage.equals(message)){
                    firstMessage = null;
                    message = (Message)((MessageImpl)message).clone();
                    ((MessageImpl)message).setFirst(true);
                    latestFirstMessageId = (MessageId)message;
                }
            }else{
                firstMessage = null;
            }
            long startTime = System.currentTimeMillis();
            try{
                if(multicastAddress != null){
                    MulticastMessageImpl msg = (MulticastMessageImpl)message;
                    msg = (MulticastMessageImpl)msg.clone();
                    msg.addToId(id);
                    message = msg;
                }
                sendMessage(sendSocket == null ? ServerConnectionImpl.this.sendSocket : sendSocket, multicastAddress == null ? clientAddress : multicastAddress, (MessageImpl)message, destPort, false);
                sendCount++;
                sendProcessTime += (System.currentTimeMillis() - startTime);
            }catch(SocketTimeoutException e){
                throw new MessageSendException(e);
            }catch(SocketException e){
                ClientImpl.this.close();
                throw new MessageSendException(e);
            }catch(IOException e){
                ClientImpl.this.close();
                throw new MessageSendException(e);
            }
        }
        
        public void receive(SelectionKey key){
            try{
                int readLength = 0;
                if(dataBuffer == null){
                    readLength = socketChannel.read(lengthBuffer);
                    if(readLength == 0){
                        return;
                    }else if(readLength == -1){
                        throw new EOFException("EOF in reading length.");
                    }
                    if(lengthBuffer.position() < 4){
                        return;
                    }
                    lengthBuffer.rewind();
                    dataLength = lengthBuffer.getInt();
                    lengthBuffer.clear();
                    if(dataLength <= 0){
                        throw new IOException("DataLength is illegal." + dataLength);
                    }
                    dataBuffer = ByteBuffer.allocate(dataLength);
                }
                
                readLength = socketChannel.read(dataBuffer);
                if(readLength == 0){
                    return;
                }else if(readLength == -1){
                    throw new EOFException("EOF in reading data.");
                }
                if(dataBuffer.position() < dataLength){
                    return;
                }
                dataBuffer.rewind();
                final byte[] dataBytes = new byte[dataLength];
                dataBuffer.get(dataBytes);
                dataBuffer = null;
                lengthBuffer.clear();
                ByteArrayInputStream is = new ByteArrayInputStream(dataBytes);
                ClientMessage clientMessage = null;
                if(externalizer == null){
                    ObjectInputStream ois = new ObjectInputStream(is);
                    clientMessage = (ClientMessage)ois.readObject();
                }else{
                    clientMessage = (ClientMessage)externalizer.readExternal(is);
                }
                if(handleMessage(clientMessage)){
                    key.cancel();
                }
            }catch(ClassNotFoundException e){
                e.printStackTrace();
            }catch(SocketTimeoutException e){
            }catch(IOException e){
                key.cancel();
                ClientImpl.this.close();
            }
        }
        
        public void writeResponse(SelectionKey key) throws IOException{
            ByteBuffer buf = null;
            try{
                while((buf = responseQueue.get(0)) != null){
                    socketChannel.write(buf);
                }
                socketChannel.register(
                    key.selector(),
                    SelectionKey.OP_READ,
                    this
                );
            }catch(IOException e){
                key.cancel();
                ClientImpl.this.close();
            }
        }
        
        public void response(ServerMessage message){
            try{
                ByteArrayOutputStream baos = new ByteArrayOutputStream();
                if(externalizer == null){
                    ObjectOutputStream oos = new ObjectOutputStream(baos);
                    oos.writeObject(message);
                    oos.flush();
                    oos.close();
                }else{
                    externalizer.writeExternal(message, baos);
                }
                byte[] bytes = baos.toByteArray();
                if(socketChannel != null){
                    final ByteBuffer buf = ByteBuffer.allocate(bytes.length + 4);
                    buf.putInt(bytes.length);
                    buf.put(bytes);
                    buf.flip();
                    responseQueue.push(buf);
                    ((ClientAcceptor)clientAcceptor.getDaemonRunnable()).addResponseClient(this);
                    selector.wakeup();
                }else{
                    if(socket != null){
                        synchronized(socket){
                            if(socket != null){
                                DataOutputStream dos = new DataOutputStream(socket.getOutputStream());
                                dos.writeInt(bytes.length);
                                dos.write(bytes);
                                dos.flush();
                            }
                        }
                    }
                }
            }catch(IOException e){
                if(responseErrorMessageId != null){
                    logger.write(
                        responseErrorMessageId,
                        e,
                        this,
                        message
                    );
                }
            }
        }
        
        public long getSendCount(){
            return sendCount;
        }
        
        public void resetSendCount(){
            sendCount = 0;
            sendProcessTime = 0;
        }
        
        public long getInterpolateRequestCount(){
            return interpolateRequestCount;
        }
        
        public void resetInterpolateRequestCount(){
            interpolateRequestCount = 0;
        }
        
        public long getAverageSendProcessTime(){
            return sendCount == 0 ? 0 : (sendProcessTime / sendCount);
        }
        
        public synchronized void close(){
            Object id = getId();
            if(requestDispatcher != null){
                requestDispatcher.stopNoWait();
                requestDispatcher = null;
            }
            if(responseQueue != null){
                responseQueue.release();
                responseQueue = null;
            }
            if(socketChannel != null){
                synchronized(socketChannel){
                    try{
                        socketChannel.finishConnect();
                        socketChannel.close();
                    }catch(IOException e){
                    }
                    socketChannel = null;
                }
            }
            if(socket != null){
                synchronized(socket){
                    try{
                        socket.close();
                    }catch(IOException e){
                    }
                    socket = null;
                }
            }
            if(sendSocket != null){
                sendSocket.close();
                sendSocket = null;
            }
            clients.remove(ClientImpl.this);
            clientMap.remove(id);
            newClients.remove(ClientImpl.this);
            if(serverConnectionListeners != null){
                for(ServerConnectionListener serverConnectionListener : serverConnectionListeners){
                    serverConnectionListener.onClose(ClientImpl.this);
                }
            }
        }
        
        public String toString(){
            final StringBuilder buf = new StringBuilder();
            buf.append(super.toString());
            buf.append('{');
            buf.append("client=").append(clientAddress).append(':').append(destPort);
            buf.append(", subject=").append(subjects);
            buf.append(", isEnabled=").append(isEnabled);
            buf.append('}');
            return buf.toString();
        }
        
        public boolean onStart(){return true;}
        public boolean onStop(){return true;}
        public boolean onSuspend(){return true;}
        public boolean onResume(){return true;}
        public Object provide(DaemonControl ctrl) throws Throwable{
            try{
                DataInputStream dis = new DataInputStream(socket.getInputStream());
                int length = dis.readInt();
                byte[] bytes = new byte[length];
                dis.readFully(bytes);
                ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
                if(externalizer == null){
                    ObjectInputStream ois = new ObjectInputStream(bais);
                    return ois.readObject();
                }else{
                    return externalizer.readExternal(bais);
                }
            }catch(ClassNotFoundException e){
                return ClientImpl.this;
            }catch(SocketTimeoutException e){
                return ClientImpl.this;
            }catch(SocketException e){
                return null;
            }catch(EOFException e){
                return null;
            }catch(IOException e){
                return ClientImpl.this;
            }
        }
        
        public void consume(Object paramObj, DaemonControl ctrl) throws Throwable{
            if(paramObj == null){
                ClientImpl.this.close();
                return;
            }
            if(!(paramObj instanceof ClientMessage)){
                return;
            }
            ClientMessage message = (ClientMessage)paramObj;
            handleMessage(message);
        }
        
        public boolean handleMessage(ClientMessage message){
            Set<String> keySet = null;
            String[] keys = null;
            boolean result = false;
            switch(message.getMessageType()){
            case ClientMessage.MESSAGE_ID:
                IdMessage idMessage = (IdMessage)message;
                clientMap.remove(getId());
                this.id = idMessage.getId();
                clientMap.put(getId(), this);
                this.destPort = idMessage.getReceivePort();
                if(serverConnectionListeners != null){
                    for(ServerConnectionListener serverConnectionListener : serverConnectionListeners){
                        serverConnectionListener.onConnect(ClientImpl.this);
                    }
                }
                break;
            case ClientMessage.MESSAGE_ADD:
                List<String> addKeysList = Collections.synchronizedList(new ArrayList<String>());
                AddMessage addMessage = (AddMessage)message;
                keySet = subjects.get(addMessage.getSubject());
                if(keySet == null){
                    keySet = Collections.synchronizedSet(new HashSet<String>());
                    subjects.put(addMessage.getSubject(), keySet);
                }
                keys = addMessage.getKeys();
                if(keys == null){
                    if(keySet.add(null)){
                        addKeysList.add(null);
                    }
                }else{
                    for(int i = 0; i < keys.length; i++){
                        if(keySet.add(keys[i])){
                            addKeysList.add(keys[i]);
                        }
                    }
                }
                if(serverConnectionListeners != null && !addKeysList.isEmpty()){
                    String[] addkeys = addKeysList.toArray(new String[0]);
                    for(ServerConnectionListener serverConnectionListener : serverConnectionListeners){
                        serverConnectionListener.onAddSubject(ClientImpl.this, addMessage.getSubject(), addkeys);
                    }
                }
                break;
            case ClientMessage.MESSAGE_REMOVE:
                List<String> removeKeysList = Collections.synchronizedList(new ArrayList<String>());
                RemoveMessage removeMessage = (RemoveMessage)message;
                keySet = subjects.get(removeMessage.getSubject());
                if(keySet == null){
                    break;
                }
                keys = removeMessage.getKeys();
                if(keys == null){
                    if(keySet.remove(null)){
                        removeKeysList.add(null);
                    }
                    if(keySet.size() == 0){
                        subjects.remove(removeMessage.getSubject());
                    }
                }else{
                    for(int i = 0; i < keys.length; i++){
                        if(keySet.remove(keys[i])){
                            removeKeysList.add(keys[i]);
                        }
                    }
                    if(keySet.size() == 0){
                        subjects.remove(removeMessage.getSubject());
                    }
                }
                if(serverConnectionListeners != null && !removeKeysList.isEmpty()){
                    String[] removeKeys = removeKeysList.toArray(new String[0]);
                    for(ServerConnectionListener serverConnectionListener : serverConnectionListeners){
                        serverConnectionListener.onRemoveSubject(ClientImpl.this, removeMessage.getSubject(), removeKeys);
                    }
                }
                break;
            case jp.ossc.nimbus.service.publish.udp.ClientMessage.MESSAGE_INTERPOLATE_REQ:
                interpolateRequestCount++;
                InterpolateRequestMessage interpolateReqMessage = (InterpolateRequestMessage)message;
                InterpolateResponseMessage interpolateResMessage = new InterpolateResponseMessage();
                interpolateResMessage.setRequestId(interpolateReqMessage.getRequestId());
                MessageId latestMessageId = interpolateReqMessage.getLatestMessageId();
                if(latestMessageId != null){
                    List<MessageImpl> messages = getSendMessages(latestMessageId.next(), null);
                    for(int i = 0; i < messages.size(); i++){
                        MessageImpl msg = messages.get(i);
                        try{
                            interpolateResMessage.addWindows(msg.getWindows(windowSize, externalizer));
                        }catch(IOException e){}
                    }
                }else{
                    MessageId currentFirstMessageId = interpolateReqMessage.getCurrentFirstMessageId();
                    List<MessageId> lostIds = new ArrayList<MessageId>();
                    if(currentFirstMessageId != null && latestFirstMessageId != null){
                        List<MessageImpl> messages = getSendMessages(latestFirstMessageId, currentFirstMessageId);
                        if(messages.size() == 0){
                            lostIds.add(latestFirstMessageId);
                            lostIds = latestFirstMessageId.createMissingIds(currentFirstMessageId, lostIds);
                        }else{
                            for(int i = 0; i < messages.size(); i++){
                                MessageImpl msg = messages.get(i);
                                if(i == 0){
                                    if(!msg.equals(latestFirstMessageId)){
                                        lostIds.add(latestFirstMessageId);
                                        lostIds = latestFirstMessageId.createMissingIds(msg, lostIds);
                                    }
                                    msg = (MessageImpl)msg.clone();
                                    msg.setFirst(true);
                                    latestFirstMessageId = msg;
                                }
                                try{
                                    interpolateResMessage.addWindows(msg.getWindows(windowSize, externalizer));
                                }catch(IOException e){}
                            }
                        }
                    }
                    MessageId[] messageIds = interpolateReqMessage.getMessageIds();
                    if(messageIds != null){
                        for(int i = 0; i < messageIds.length; i++){
                            List<Window> windows = getSendWindows(messageIds[i]);
                            if(windows != null){
                                interpolateResMessage.addWindows(messageIds[i], windows);
                            }else{
                                lostIds.add(messageIds[i]);
                            }
                        }
                    }
                    WindowId[] windowIds = interpolateReqMessage.getWindowIds();
                    if(windowIds != null){
                        for(int i = 0; i < windowIds.length; i++){
                            Window window = getSendWindow(windowIds[i]);
                            if(window != null){
                                interpolateResMessage.addWindow(windowIds[i], window);
                            }else{
                                lostIds.add(windowIds[i]);
                            }
                        }
                    }
                    if(lostIds.size() != 0 && logger != null && messageLostErrorMessageId != null){
                        logger.write(
                            messageLostErrorMessageId,
                            ClientImpl.this,
                            lostIds,
                            new Integer(getMostOldSendBufferSequence())
                        );
                    }
                }
                response(interpolateResMessage);
                break;
            case ClientMessage.MESSAGE_START_RECEIVE:
                StartReceiveMessage startMessage = (StartReceiveMessage)message;
                fromTime = startMessage.getFrom();
                if(fromTime >= 0){
                    List<MessageImpl> messages = getSendMessages(fromTime);
                    boolean isFirstMessage = true;
                    for(int i = 0; i < messages.size(); i++){
                        MessageImpl msg = messages.get(i);
                        if(!isTargetMessage(msg)){
                            continue;
                        }
                        if(isFirstMessage){
                            msg = (MessageImpl)msg.clone();
                            msg.setFirst(true);
                            isFirstMessage = false;
                        }
                        try{
                            send(msg);
                        }catch(MessageSendException e){
                            if(logger != null && sendErrorRetryOverMessageId != null){
                                logger.write(
                                    sendErrorRetryOverMessageId,
                                    e,
                                    this,
                                    msg
                                );
                            }
                        }
                    }
                }
                isStartReceive = true;
                if(serverConnectionListeners != null){
                    for(ServerConnectionListener serverConnectionListener : serverConnectionListeners){
                        serverConnectionListener.onStartReceive(ClientImpl.this, fromTime);
                    }
                }
                break;
            case ClientMessage.MESSAGE_STOP_RECEIVE:
                isStartReceive = false;
                firstMessage = null;
                if(serverConnectionListeners != null){
                    for(ServerConnectionListener serverConnectionListener : serverConnectionListeners){
                        serverConnectionListener.onStopReceive(ClientImpl.this);
                    }
                }
                break;
            case ClientMessage.MESSAGE_BYE:
                ClientImpl.this.close();
                result = true;
                break;
            default:
            }
            return result;
        }
        
        public void garbage(){}
        
        public Set<String> getSubjects(){
            if(subjects == null){
                return null;
            }
            return subjects.keySet();
        }
        
        public Set<String> getKeys(String subject){
            if(subjects == null){
                return null;
            }
            return subjects.get(subject);
        }
        
        public Object getId(){
            return id == null ? (socket == null ? null : socket.getRemoteSocketAddress()) : id;
        }
    }
    
    private class SendRequest{
        public Set<ClientImpl> firstClients;
        public ClientImpl client;
        public MessageImpl message;
        public SendRequest(ClientImpl client, MessageImpl message){
            this.client = client;
            this.message = message;
        }
        public SendRequest(Set<ClientImpl> firstClients, MessageImpl message){
            this.firstClients = firstClients;
            this.message = message;
        }
    }
    
    private class SendQueueHandler implements QueueHandler<AsynchContext<SendRequest,?>>{
        
        public void handleDequeuedObject(AsynchContext<SendRequest,?> asynchContext) throws Throwable{
            if(asynchContext == null){
                return;
            }
            
            SendRequest request = (SendRequest)asynchContext.getInput();
            if(request.client == null){
                if(request.firstClients != null){
                    Iterator<ClientImpl> firstClientItr = request.firstClients.iterator();
                    while(firstClientItr.hasNext()){
                        ClientImpl client = firstClientItr.next();
                        if(client.isStartReceive()){
                            client.send(request.message);
                        }
                    }
                }
                sendMessage(sendSocket, multicastAddress, request.message, request.client.getDestPort(), false);
            }else{
                if(request.client.isStartReceive()){
                    request.client.send(request.message);
                }
            }
            asynchContext.response();
        }
        
        public boolean handleError(AsynchContext<SendRequest,?> asynchContext, Throwable th) throws Throwable{
            if(logger != null && sendErrorMessageId != null){
                SendRequest request = asynchContext.getInput();
                logger.write(
                    sendErrorMessageId,
                    th,
                    request.client == null ? (multicastAddress + ":" + request.client.getDestPort()) : request.client.toString(),
                    request.message
                );
            }
            return true;
        }
        
        public void handleRetryOver(AsynchContext<SendRequest,?> asynchContext, Throwable th) throws Throwable{
            if(logger != null && sendErrorRetryOverMessageId != null){
                SendRequest request = asynchContext.getInput();
                logger.write(
                    sendErrorRetryOverMessageId,
                    th,
                    request.client == null ? (multicastAddress + ":" + request.client.getDestPort()) : request.client.toString(),
                    request.message
                );
            }
            asynchContext.setThrowable(th);
            asynchContext.response();
        }
    }
    
    private static class RequestHandleRequest{
        
        public static final int REQUEST_TYPE_READ  = 1;
        public static final int REQUEST_TYPE_WRITE = 2;
        
        public ClientImpl client;
        public SelectionKey key;
        public Queue<SelectionKey> responseQueue;
        public int requestType;
        public RequestHandleRequest(ClientImpl client, SelectionKey key, Queue<SelectionKey> responseQueue, int type){
            this.client = client;
            this.key = key;
            this.responseQueue = responseQueue;
            requestType = type;
        }
    }
    
    private class RequestHandleQueueHandler implements QueueHandler<RequestHandleRequest>{
        
        public void handleDequeuedObject(RequestHandleRequest request) throws Throwable{
            if(request == null){
                return;
            }
            try{
                switch(request.requestType){
                case RequestHandleRequest.REQUEST_TYPE_READ:
                    request.client.receive(request.key);
                    break;
                case RequestHandleRequest.REQUEST_TYPE_WRITE:
                    request.client.writeResponse(request.key);
                    break;
                }
            }finally{
                request.responseQueue.push(request.key);
            }
        }
        
        public boolean handleError(RequestHandleRequest request, Throwable th) throws Throwable{
            return true;
        }
        
        public void handleRetryOver(RequestHandleRequest request, Throwable th) throws Throwable{
            throw th;
        }
    }
    
    private class ClientDistributedQueueSelector extends AbstractDistributedQueueSelectorService<AsynchContext<SendRequest,?>, ClientImpl>{
        
        private static final long serialVersionUID = 8661094319264622631L;
        
        protected ClientImpl getKey(AsynchContext<SendRequest,?> asynchContext){
            SendRequest request = (SendRequest)asynchContext.getInput();
            return request.client;
        }
    }
}
