/*
 * 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.tcp;

import java.io.EOFException;
import java.io.IOException;
import java.io.ObjectInputStream;
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.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.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.Queue;
import jp.ossc.nimbus.service.queue.AsynchContext;
import jp.ossc.nimbus.service.queue.DefaultQueueService;
import jp.ossc.nimbus.service.queue.QueueHandler;
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.util.SynchronizeMonitor;
import jp.ossc.nimbus.util.WaitSynchronizeMonitor;

/**
 * TCPプロトコル用の{@link ServerConnection}インタフェース実装クラス。<p>
 *
 * @author M.Takata
 */
public class ServerConnectionImpl implements ServerConnection{
    
    private ServerSocket serverSocket;
    private ServerSocketChannel serverSocketChannel;
    private Selector selector;
    
    private Set<ClientImpl> clients = Collections.synchronizedSet(new LinkedHashSet<ClientImpl>());
    private Map<Object, ClientImpl> clientMap = Collections.synchronizedMap(new HashMap<Object, ClientImpl>());
    private int maxSendRetryCount;
    private Logger logger;
    private String sendErrorMessageId;
    private String sendErrorRetryOverMessageId;
    private Daemon clientAcceptor;
    private QueueHandlerContainerService<AsynchContext<SendRequest,?>> sendQueueHandlerContainer;
    private ClientDistributedQueueSelector queueSelector;
    private DistributedQueueHandlerContainerService<AsynchContext<SendRequest,?>> asynchSendQueueHandlerContainer;
    private long sendCount;
    private long sendProcessTime;
    private List<ServerConnectionListener> serverConnectionListeners;
    private Externalizer<Object> externalizer;
    private SocketFactory socketFactory;
    private List<MessageImpl> sendBufferList = Collections.synchronizedList(new ArrayList<MessageImpl>());
    private long sendBufferTime;
    
    public ServerConnectionImpl(
        ServerSocket serverSocket,
        Externalizer<Object> ext,
        int sendThreadSize,
        ServiceName sendQueueServiceName,
        int asynchSendThreadSize,
        ServiceName asynchSendQueueServiceName
    ) throws Exception{
        this.serverSocket = serverSocket;
        externalizer = ext;
        
        initSend(sendQueueServiceName, sendThreadSize);
        initAsynchSend(asynchSendQueueServiceName, asynchSendThreadSize);
        
        initClientAcceptor(serverSocket.getLocalSocketAddress());
        
    }
    
    public ServerConnectionImpl(
        ServerSocketChannel ssc,
        Externalizer<Object> ext,
        int sendThreadSize,
        ServiceName sendQueueServiceName,
        int asynchSendThreadSize,
        ServiceName asynchSendQueueServiceName,
        SocketFactory sf
    ) throws Exception{
        serverSocketChannel = ssc;
        socketFactory = sf;
        externalizer = ext;
        
        initSend(sendQueueServiceName, sendThreadSize);
        initAsynchSend(asynchSendQueueServiceName, asynchSendThreadSize);
        
        selector = Selector.open();
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT, null);
        initClientAcceptor(serverSocketChannel.socket().getLocalSocketAddress());
    }
    
    private void initClientAcceptor(SocketAddress localAddress){
        clientAcceptor = new Daemon(new ClientAcceptor());
        clientAcceptor.setName(
            "Nimbus Publish(TCP) ServerConnection ClientAcceptor " + localAddress
        );
        clientAcceptor.setDaemon(true);
        clientAcceptor.start();
    }
    
    private void initSend(ServiceName sendQueueServiceName, int sendThreadSize) throws Exception{
        if(sendThreadSize >= 2){
            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 clientQueueDistributedSize) throws Exception{
        if(clientQueueDistributedSize > 0){
            queueSelector = new ClientDistributedQueueSelector();
            queueSelector.create();
            queueSelector.setDistributedSize(clientQueueDistributedSize);
            if(queueFactoryServiceName != null){
                queueSelector.setQueueFactoryServiceName(queueFactoryServiceName);
            }
            queueSelector.start();
            
            asynchSendQueueHandlerContainer = new DistributedQueueHandlerContainerService<AsynchContext<SendRequest,?>>();
            asynchSendQueueHandlerContainer.create();
            asynchSendQueueHandlerContainer.setDistributedQueueSelector(queueSelector);
            asynchSendQueueHandlerContainer.setQueueHandler(new SendQueueHandler());
            asynchSendQueueHandlerContainer.start();
        }
    }
    
    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 setSendBufferTime(long time){
        sendBufferTime = time;
    }
    
    public Message createMessage(String subject, String key) throws MessageCreateException{
        final MessageImpl message = new MessageImpl();
        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;
    }
    
    public void send(Message message) throws MessageSendException{
        long startTime = System.currentTimeMillis();
        addSendBuffer((MessageImpl)message);
        if(clients.size() == 0){
            return;
        }
        try{
            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]);
                }
                int retryCount = -1;
                while(currentClients.size() != 0 && retryCount < maxSendRetryCount){
                    Iterator<ClientImpl> itr = currentClients.iterator();
                    while(itr.hasNext()){
                        ClientImpl client = itr.next();
                        try{
                            client.send(message);
                            itr.remove();
                        }catch(MessageSendException e){
                            if(logger != null){
                                if((retryCount + 1) >= maxSendRetryCount){
                                    if(sendErrorRetryOverMessageId != null){
                                        logger.write(
                                            sendErrorRetryOverMessageId,
                                            e,
                                            client,
                                            message
                                        );
                                    }
                                }else{
                                    if(sendErrorMessageId != null){
                                        logger.write(
                                            sendErrorMessageId,
                                            e,
                                            client,
                                            message
                                        );
                                    }
                                }
                            }
                        }
                    }
                    retryCount++;
                }
                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], message), responseQueue));
                }
                List<ClientImpl> errorClients = new ArrayList<ClientImpl>();
                for(int i = 0; i < clientArray.length; i++){
                    if(clientArray[i] == null){
                        continue;
                    }
                    AsynchContext<SendRequest,Object> 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);
                }
            }
        }finally{
            sendProcessTime += (System.currentTimeMillis() - startTime);
        }
    }
    
    public void sendAsynch(Message message){
        if(asynchSendQueueHandlerContainer == null){
            throw new UnsupportedOperationException();
        }
        addSendBuffer((MessageImpl)message);
        if(clients.size() == 0){
            return;
        }
        ClientImpl[] clientArray = (ClientImpl[])clients.toArray(new ClientImpl[clients.size()]);
        for(int i = 0; i < clientArray.length; i++){
            if(!clientArray[i].isStartReceive()
                || !clientArray[i].isTargetMessage(message)){
                continue;
            }
            asynchSendQueueHandlerContainer.push(new AsynchContext<SendRequest,Object>(new SendRequest(clientArray[i], message)));
        }
    }
    
    public long getSendCount(){
        return sendCount;
    }
    
    public void resetSendCount(){
        sendCount = 0;
        sendProcessTime = 0;
    }
    
    public long getAverageSendProcessTime(){
        return sendCount == 0 ? 0 : (sendProcessTime / sendCount);
    }
    
    public Set<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<String> getSubjects(Object id){
        ClientImpl client = clientMap.get(id);
        if(client == null){
            return null;
        }
        return client.getSubjects();
    }
    
    public Set<String> getKeys(Object id, String subject){
        ClientImpl client = clientMap.get(id);
        if(client == null){
            return null;
        }
        return client.getKeys(subject);
    }
    
    private void addSendBuffer(MessageImpl message){
        final long currentTime = System.currentTimeMillis();
        message.setSendTime(currentTime);
        synchronized(sendBufferList){
            sendBufferList.add(message);
            for(int i = 0, imax = sendBufferList.size(); i < imax; i++){
                MessageImpl msg = sendBufferList.get(0);
                if((currentTime - msg.getSendTime()) > sendBufferTime){
                    sendBufferList.remove(0);
                }else{
                    break;
                }
            }
            sendCount++;
        }
    }
    
    private List<MessageImpl> getSendMessages(long from){
        List<MessageImpl> result = new ArrayList<MessageImpl>();
        synchronized(sendBufferList){
            for(int i = sendBufferList.size(); --i >= 0; ){
                MessageImpl msg = sendBufferList.get(i);
                if(msg.getSendTime() >= from){
                    result.add(0, msg);
                }else{
                    break;
                }
            }
        }
        return result;
    }
    
    public int getSendBufferSize(){
        return sendBufferList.size();
    }
    
    public void close(){
        try{
            send(new MessageImpl(true));
        }catch(MessageSendException e){}
        
        if(clientAcceptor != null){
            clientAcceptor.stopNoWait();
            clientAcceptor = null;
        }
        if(sendQueueHandlerContainer != null){
            sendQueueHandlerContainer.stop();
            sendQueueHandlerContainer.destroy();
            sendQueueHandlerContainer = null;
        }
        if(asynchSendQueueHandlerContainer != null){
            asynchSendQueueHandlerContainer.stop();
            asynchSendQueueHandlerContainer.destroy();
            asynchSendQueueHandlerContainer = null;
        }
        if(queueSelector != null){
            queueSelector.stop();
            queueSelector.destroy();
            queueSelector = null;
        }
        
        if(serverSocket != null){
            try{
                serverSocket.close();
            }catch(IOException e){}
        }
        ClientImpl[] clientArray = (ClientImpl[])clients.toArray(new ClientImpl[clients.size()]);
        for(int i = 0; i < clientArray.length; i++){
            clientArray[i].close();
        }
    }
    
    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 StringBuilder buf = new StringBuilder();
        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> sendClientQueue;
        private List<ClientImpl> sendClients;
        
        public ClientAcceptor(){
            if(selector != null){
                
                DefaultQueueService<ClientImpl> sendQueue = new DefaultQueueService<ClientImpl>();
                try{
                    sendQueue.create();
                    sendQueue.start();
                }catch(Exception e){}
                sendClientQueue = sendQueue;
                sendClients = new ArrayList<ClientImpl>();
            }
        }
        
        public void addSendClient(ClientImpl client){
            sendClientQueue.push(client);
            selector.wakeup();
        }
        
        public boolean onStart(){
            if(sendClientQueue != null){
                sendClientQueue.accept();
            }
            return true;
        }
        public boolean onStop(){
            if(sendClientQueue != null){
                sendClientQueue.release();
            }
            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 = sendClientQueue.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;
                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);
                                channel.register(
                                    key.selector(),
                                    SelectionKey.OP_READ,
                                    client
                                );
                                clients.add(client);
                                clientMap.put(client.getId(), client);
                            }
                        }else if(key.isReadable()){
                            client = (ClientImpl)key.attachment();
                            client.receive(key);
                        }else if(key.isWritable()){
                            client = (ClientImpl)key.attachment();
                            sendClients.add(client);
                            client.setSendKey(key);
                        }else if(!key.isValid()){
                            key.cancel();
                        }
                    }catch(CancelledKeyException e){}
                }
                for(int i = 0, imax = sendClients.size(); i < imax; i++){
                    sendClients.remove(0).waitSend();
                }
            }else{
                if(!(paramObj instanceof Socket)){
                    return;
                }
                Socket socket = (Socket)paramObj;
                if(!socket.isBound() || socket.isClosed()){
                    return;
                }
                ClientImpl client = new ClientImpl(socket);
                clients.add(client);
                clientMap.put(client.getId(), client);
            }
        }
        public void garbage(){}
    }
    
    public class ClientImpl implements DaemonRunnable<Object>, Client{
        private SocketChannel socketChannel;
        private Socket socket;
        private Daemon requestDispatcher;
        private Map<String,Set<String>> subjects;
        private long sendCount;
        private long sendProcessTime;
        private boolean isEnabled = true;
        private Object id;
        private ByteBuffer lengthBuffer;
        private ByteBuffer dataBuffer;
        private int dataLength = 0;
        private long fromTime = -1;
        private boolean isStartReceive = false;
        private SynchronizeMonitor selectMonitor;
        private SynchronizeMonitor sendMonitor;
        private SelectionKey sendKey;
        
        public ClientImpl(SocketChannel sc){
            socketChannel = sc;
            socket = socketChannel.socket();
            subjects = Collections.synchronizedMap(new HashMap<String,Set<String>>());
            lengthBuffer = ByteBuffer.allocate(4);
            selectMonitor = new WaitSynchronizeMonitor();
            sendMonitor = new WaitSynchronizeMonitor();
        }
        
        public ClientImpl(Socket sock) throws IOException{
            socket = sock;
            subjects = Collections.synchronizedMap(new HashMap<String,Set<String>>());
            requestDispatcher = new Daemon(ClientImpl.this);
            requestDispatcher.setName(
                "Nimbus Publish(TCP) ServerConnection ClientRequestDisptcher " + socket.getRemoteSocketAddress()
            );
            requestDispatcher.setDaemon(true);
            requestDispatcher.start();
        }
        
        public SocketChannel getSocketChannel(){
            return socketChannel;
        }
        
        public Socket getSocket(){
            return socket;
        }
        
        public boolean isEnabled(){
            return isEnabled;
        }
        public void setEnabled(boolean isEnabled){
            this.isEnabled = isEnabled;
        }
        
        public boolean isStartReceive(){
            return isStartReceive;
        }
        
        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 || (socketChannel == null && socket == null)){
                return;
            }
            long startTime = System.currentTimeMillis();
            try{
                ByteArrayOutputStream baos = new ByteArrayOutputStream();
                ((MessageImpl)message).write(baos, externalizer);
                byte[] bytes = baos.toByteArray();
                if(socketChannel != null){
                    final ByteBuffer buf = ByteBuffer.allocate(bytes.length + 4);
                    buf.putInt(bytes.length);
                    buf.put(bytes);
                    buf.flip();
                    selectMonitor.initMonitor();
                    ((ClientAcceptor)clientAcceptor.getDaemonRunnable()).addSendClient(this);
                    selectMonitor.waitMonitor();
                    socketChannel.write(buf);
                    sendMonitor.notifyAllMonitor();
                }else{
                    DataOutputStream dos = new DataOutputStream(socket.getOutputStream());
                    dos.writeInt(bytes.length);
                    dos.write(bytes);
                    dos.flush();
                }
                sendCount++;
                sendProcessTime += (System.currentTimeMillis() - startTime);
            }catch(InterruptedException e){
                return;
            }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 setSendKey(SelectionKey key){
            sendKey = key;
            sendMonitor.initMonitor();
            selectMonitor.notifyAllMonitor();
        }
        
        public void waitSend() throws IOException{
            try{
                sendMonitor.waitMonitor();
            }catch(InterruptedException e){
                return;
            }finally{
                socketChannel.register(
                    sendKey.selector(),
                    SelectionKey.OP_READ,
                    this
                );
            }
        }
        
        public void receive(SelectionKey key){
            try{
                if(dataBuffer == null){
                    int 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);
                }
                
                int 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);
                boolean isClosed = false;
                if(externalizer == null){
                    ObjectInputStream ois = new ObjectInputStream(is);
                    isClosed = handleMessage((ClientMessage)ois.readObject());
                }else{
                    isClosed = handleMessage((ClientMessage)externalizer.readExternal(is));
                }
                if(isClosed){
                    key.cancel();
                }
            }catch(ClassNotFoundException e){
                e.printStackTrace();
            }catch(SocketTimeoutException e){
            }catch(IOException e){
                key.cancel();
                ClientImpl.this.close();
            }
        }
        
        public long getSendCount(){
            return sendCount;
        }
        
        public void resetSendCount(){
            sendCount = 0;
            sendProcessTime = 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(selectMonitor != null){
                selectMonitor.releaseMonitor();
                selectMonitor = null;
            }
            if(sendMonitor != null){
                sendMonitor.releaseMonitor();
                sendMonitor = null;
            }
            if(socketChannel != null){
                try{
                    socketChannel.finishConnect();
                    socketChannel.close();
                }catch(IOException e){
                }
                socketChannel = null;
            }
            if(socket != null){
                try{
                    socket.close();
                }catch(IOException e){
                }
                socket = null;
            }
            clients.remove(ClientImpl.this);
            clientMap.remove(id);
            if(serverConnectionListeners != null){
                for(ServerConnectionListener serverConnectionListener : serverConnectionListeners){
                    serverConnectionListener.onClose(ClientImpl.this);
                }
            }
            isStartReceive = false;
        }
        
        public String toString(){
            final StringBuilder buf = new StringBuilder();
            buf.append(super.toString());
            buf.append('{');
            buf.append("client=").append(socket == null ? null : socket.getRemoteSocketAddress());
            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);
        }
        
        private 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);
                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 ClientMessage.MESSAGE_START_RECEIVE:
                StartReceiveMessage startMessage = (StartReceiveMessage)message;
                fromTime = startMessage.getFrom();
                if(fromTime >= 0){
                    List<MessageImpl> messages = getSendMessages(fromTime);
                    for(int i = 0; i < messages.size(); i++){
                        Message msg = messages.get(i);
                        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;
                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 ClientImpl client;
        public Message message;
        public SendRequest(ClientImpl client, Message message){
            this.client = client;
            this.message = message;
        }
    }
    
    private class SendQueueHandler implements QueueHandler<AsynchContext<SendRequest,?>>{
        
        public void handleDequeuedObject(AsynchContext<SendRequest,?> asynchContext) throws Throwable{
            if(asynchContext == null){
                return;
            }
            SendRequest request = asynchContext.getInput();
            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,
                    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,
                    request.message
                );
            }
            asynchContext.setThrowable(th);
            asynchContext.response();
        }
    }
    
    private class ClientDistributedQueueSelector extends AbstractDistributedQueueSelectorService<AsynchContext<SendRequest,?>,ClientImpl>{
        
        private static final long serialVersionUID = 8050312124454494504L;
        
        protected ClientImpl getKey(AsynchContext<SendRequest,?> asynchContext){
            SendRequest request = asynchContext.getInput();
            return request.client;
        }
    }
}
