/*
 * 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.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.DataOutputStream;
import java.io.DataInputStream;
import java.io.ByteArrayOutputStream;
import java.io.ByteArrayInputStream;
import java.io.EOFException;
import java.io.IOException;
import java.io.Serializable;
import java.net.Socket;
import java.net.DatagramSocket;
import java.net.DatagramPacket;
import java.net.MulticastSocket;
import java.net.InetAddress;
import java.net.SocketAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.util.Set;
import java.util.HashSet;
import java.util.SortedMap;
import java.util.Map;
import java.util.HashMap;
import java.util.TreeMap;
import java.util.Collections;
import java.util.List;
import java.util.ArrayList;
import java.util.Date;
import javax.net.SocketFactory;

import jp.ossc.nimbus.core.ServiceManagerFactory;
import jp.ossc.nimbus.core.ServiceManager;
import jp.ossc.nimbus.core.ServiceName;
import jp.ossc.nimbus.core.ServiceBase;
import jp.ossc.nimbus.core.ServiceBaseMBean;
import jp.ossc.nimbus.daemon.Daemon;
import jp.ossc.nimbus.daemon.DaemonRunnable;
import jp.ossc.nimbus.daemon.DaemonControl;
import jp.ossc.nimbus.service.publish.ClientConnection;
import jp.ossc.nimbus.service.publish.MessageListener;
import jp.ossc.nimbus.service.publish.ConnectException;
import jp.ossc.nimbus.service.publish.MessageSendException;
import jp.ossc.nimbus.service.io.Externalizer;
import jp.ossc.nimbus.service.queue.DefaultQueueService;
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;
import jp.ossc.nimbus.service.publish.tcp.StopReceiveMessage;
import jp.ossc.nimbus.service.publish.tcp.ByeMessage;
import jp.ossc.nimbus.util.SynchronizeMonitor;
import jp.ossc.nimbus.util.WaitSynchronizeMonitor;

/**
 * UDPプロトコル用の{@link ClientConnection}インタフェース実装クラス。<p>
 *
 * @author M.Takata
 */
public class ClientConnectionImpl implements ClientConnection, Serializable{
    
    private static final long serialVersionUID = 1542561082447814032L;
    
    public static final String BIND_ADDRESS_PROPERTY = "jp.ossc.nimbus.service.publish.udp.bindAdress";
    public static final String BIND_PORT_PROPERTY = "jp.ossc.nimbus.service.publish.udp.bindPort";
    
    private String address;
    private int port;
    private SocketFactory socketFactory;
    private String receiveAddress;
    private int receivePort;
    private Externalizer<Object> externalizer;
    
    private String bindAddressProertyName = BIND_ADDRESS_PROPERTY;
    private String bindPortProertyName = BIND_PORT_PROPERTY;
    private String serverCloseMessageId;
    private String receiveWarnMessageId;
    private String receiveErrorMessageId;
    private String messageLostErrorMessageId;
    private int reconnectCount;
    private long reconnectInterval;
    private long reconnectBufferTime;
    private int windowSize;
    private long missingWindowTimeout;
    private long newMessagePollingInterval;
    private ServiceName serverServiceName;
    private long responseTimeout = -1;
    
    private transient Socket socket;
    private transient InetAddress receiveGroup;
    private transient DatagramSocket receiveSocket;
    
    private transient Map<String,Set<String>> subjects;
    private transient MessageListener messageListener;
    private transient Daemon packetReceiveDaemon;
    private transient Daemon replyReceiveDaemon;
    private transient Daemon messageReceiveDaemon;
    private transient Daemon missingWindowCheckDaemon;
    private transient DefaultQueueService<DatagramPacket> receivePacketQueue;
    private transient boolean isClosing;
    private transient boolean isConnected;
    private transient boolean isReconnecting;
    private transient Object id;
    private transient String serviceManagerName;
    private transient ServiceName serviceName;
    private transient long receiveCount;
    private transient long receivePacketCount;
    private transient long onMessageProcessTime;
    private transient long noContinuousMessageCount;
    private transient long wasteWindowCount;
    private transient long missingWindowRequestCount;
    private transient long missingWindowRequestTimeoutCount;
    private transient long missingWindowResponseTime;
    private transient long newMessagePollingCount;
    private transient long newMessagePollingTimeoutCount;
    private transient long newMessagePollingResponseTime;
    private transient int requestId;
    private transient boolean isStartReceive;
    
    public ClientConnectionImpl(
        String address,
        int port,
        SocketFactory factory,
        String receiveAddress,
        int receivePort,
        Externalizer<Object> ext,
        ServiceName serverServiceName
    ){
        this.address = address;
        this.port = port;
        socketFactory = factory;
        this.receiveAddress = receiveAddress;
        this.receivePort = receivePort;
        externalizer = ext;
        this.serverServiceName = serverServiceName;
    }
    
    public void setBindAddressProertyName(String name){
        bindAddressProertyName = name;
    }
    public String getBindAddressProertyName(){
        return bindAddressProertyName;
    }
    
    public void setBindPortProertyName(String name){
        bindPortProertyName = name;
    }
    public String getBindPortProertyName(){
        return bindPortProertyName;
    }
    
    public void setServerCloseMessageId(String id){
        serverCloseMessageId = id;
    }
    public String getServerCloseMessageId(){
        return serverCloseMessageId;
    }
    
    public void setReceiveWarnMessageId(String id){
        receiveWarnMessageId = id;
    }
    public String getReceiveWarnMessageId(){
        return receiveWarnMessageId;
    }
    
    public void setReceiveErrorMessageId(String id){
        receiveErrorMessageId = id;
    }
    public String getReceiveErrorMessageId(){
        return receiveErrorMessageId;
    }
    
    public void setMessageLostErrorMessageId(String id){
        messageLostErrorMessageId = id;
    }
    public String getMessageLostErrorMessageId(){
        return messageLostErrorMessageId;
    }
    
    public void setReconnectCount(int count){
        reconnectCount = count;
    }
    public int getReconnectCount(){
        return reconnectCount;
    }
    
    public void setReconnectInterval(long interval){
        reconnectInterval = interval;
    }
    public long getReconnectInterval(){
        return reconnectInterval;
    }
    
    public void setReconnectBufferTime(long time){
        reconnectBufferTime = time;
    }
    public long getReconnectBufferTime(){
        return reconnectBufferTime;
    }
    
    public void setWindowSize(int bytes){
        windowSize = bytes;
    }
    public int getWindowSize(){
        return windowSize;
    }
    
    public void setMissingWindowTimeout(long interval){
        missingWindowTimeout = interval;
    }
    public long getMissingWindowTimeout(){
        return missingWindowTimeout;
    }
    
    public void setNewMessagePollingInterval(long interval){
        newMessagePollingInterval = interval;
    }
    public long getNewMessagePollingInterval(){
        return newMessagePollingInterval;
    }
    
    public void setResponseTimeout(long timeout){
        responseTimeout = timeout;
    }
    public long getResponseTimeout(){
        return responseTimeout;
    }
    
    private String getProperty(String name){
        String prop = System.getProperty(name);
        if(prop == null){
            prop = ServiceManagerFactory.getProperty(name);
        }
        return prop;
    }
    
    private InetAddress getBindAddress() throws UnknownHostException{
        String bindAddress = getProperty(bindAddressProertyName);
        InetAddress address = null;
        if(bindAddress == null){
            address = InetAddress.getLocalHost();
        }else{
            address = InetAddress.getByName(bindAddress);
        }
        return address;
    }
    
    private int getBindPort() throws NumberFormatException{
        String bindPort = getProperty(bindPortProertyName);
        int port = 0;
        if(bindPort != null){
            port = Integer.parseInt(bindPort);
        }
        return port;
    }
    
    public void setServiceManagerName(String name){
        serviceManagerName = name;
    }
    
    public void connect() throws ConnectException{
        connect(null);
    }
    
    public void connect(Object id) throws ConnectException{
        connect(id, false);
    }
    
    private synchronized void connect(Object id, boolean isReconnect) throws ConnectException{
        if(socket != null){
            return;
        }
        isConnected = false;
        try{
            try{
                if(socketFactory == null){
                    socket = new Socket(
                        address,
                        port,
                        getBindAddress(),
                        getBindPort()
                    );
                }else{
                    socket = socketFactory.createSocket(
                        address,
                        port,
                        getBindAddress(),
                        getBindPort()
                    );
                }
                if(responseTimeout > 0){
                    socket.setSoTimeout((int)responseTimeout);
                }
                if(!isReconnect){
                    if(receiveAddress != null){
                        receiveGroup = InetAddress.getByName(receiveAddress);
                        final InetSocketAddress address = new InetSocketAddress(getBindAddress(), receivePort);
                        receiveSocket = receiveGroup.isMulticastAddress() ? new MulticastSocket(address) : new DatagramSocket(address);
                        if(receiveGroup.isMulticastAddress()){
                            ((MulticastSocket)receiveSocket).joinGroup(receiveGroup);
                        }
                    }else{
                        receiveSocket = new DatagramSocket(new InetSocketAddress(getBindAddress(), receivePort));
                        if(receivePort == 0){
                            receivePort = receiveSocket.getLocalPort();
                        }
                    }
                    if(receiveSocket != null){
                        try{
                            int receiveBufferSize = receiveSocket.getReceiveBufferSize();
                            if(receiveBufferSize < windowSize){
                                receiveSocket.setReceiveBufferSize(windowSize);
                            }
                        }catch(SocketException e){
                        }
                    }
                    
                }
            }catch(UnknownHostException e){
                throw new ConnectException(e);
            }catch(NumberFormatException e){
                throw new ConnectException(e);
            }catch(IOException e){
                throw new ConnectException(e);
            }
            
            if(receivePacketQueue == null){
                receivePacketQueue = new DefaultQueueService<DatagramPacket>();
                try{
                    receivePacketQueue.create();
                    receivePacketQueue.start();
                }catch(Exception e){
                    throw new ConnectException(e);
                }
            }
            if(packetReceiveDaemon == null){
                packetReceiveDaemon = new Daemon(new PacketReceiver());
                packetReceiveDaemon.setDaemon(true);
                packetReceiveDaemon.setName("Nimbus Publish(UDP) ClientConnection SocketReader " + receiveSocket.getLocalSocketAddress());
                packetReceiveDaemon.start();
            }
            
            if(replyReceiveDaemon == null){
                replyReceiveDaemon = new Daemon(new ReplyReceiver());
                replyReceiveDaemon.setDaemon(true);
                replyReceiveDaemon.setName("Nimbus Publish(UDP) ClientConnection ReplyReceiver " + receiveSocket.getLocalSocketAddress());
                replyReceiveDaemon.start();
            }
            
            if(messageReceiveDaemon == null){
                messageReceiveDaemon = new Daemon(new MessageReceiver());
                messageReceiveDaemon.setDaemon(true);
                messageReceiveDaemon.setName("Nimbus Publish(UDP) ClientConnection MessageReceiver " + receiveSocket.getLocalSocketAddress());
                messageReceiveDaemon.start();
            }
            
            if(missingWindowCheckDaemon == null){
                missingWindowCheckDaemon = new Daemon(new MissingWindowChecker((MessageReceiver)messageReceiveDaemon.getDaemonRunnable()));
                missingWindowCheckDaemon.setDaemon(true);
                missingWindowCheckDaemon.setName("Nimbus Publish(UDP) ClientConnection MissingWindowChecker " + receiveSocket.getLocalSocketAddress());
                missingWindowCheckDaemon.start();
            }
            this.id = id == null ? socket.getLocalSocketAddress() : id;
            try{
                IdMessage message = new IdMessage(this.id);
                message.setReceivePort(receivePort);
                send(message);
            }catch(IOException e){
                throw new ConnectException(e);
            }
            if(serverServiceName != null){
                ServiceManager manager = ServiceManagerFactory.findManager(serviceManagerName == null ? serverServiceName.getServiceManagerName() : serviceManagerName);
                if(manager != null){
                    final ClientConnectionService ccs = new ClientConnectionService();
                    try{
                        String name = serverServiceName.getServiceName() + '$' + receiveSocket.getLocalSocketAddress();
                        name = name.replaceAll(":", "\\$");
                        if(!manager.isRegisteredService(name) && manager.registerService(name, ccs)){
                            serviceName = ccs.getServiceNameObject();
                            manager.createService(ccs.getServiceName());
                            manager.startService(ccs.getServiceName());
                        }
                    }catch(Exception e){
                        throw new ConnectException(e);
                    }
                }
            }
        }catch(ConnectException e){
            if(socket != null){
                try{
                    socket.close();
                }catch(IOException e2){}
                socket = null;
            }
            if(!isReconnect && receiveSocket != null){
                receiveSocket.close();
                receiveSocket = null;
            }
            throw e;
        }
        isConnected = true;
    }
    
    public void addSubject(String subject) throws MessageSendException{
        addSubject(subject, null);
    }
    
    public void addSubject(String subject, String[] keys) throws MessageSendException{
        if(socket == null){
            throw new MessageSendException("Not connected.");
        }
        if(subject == null){
            return;
        }
        try{
            send(new AddMessage(subject, keys));
        }catch(SocketTimeoutException e){
            throw new MessageSendException(e);
        }catch(SocketException e){
            throw new MessageSendException(e);
        }catch(IOException e){
            throw new MessageSendException(e);
        }
        if(subjects == null){
            subjects = Collections.synchronizedMap(new HashMap<String,Set<String>>());
        }
        Set<String> keySet = subjects.get(subject);
        if(keySet == null){
            keySet = Collections.synchronizedSet(new HashSet<String>());
            subjects.put(subject, keySet);
        }
        if(keys == null){
            keySet.add(null);
        }else{
            for(int i = 0; i < keys.length; i++){
                keySet.add(keys[i]);
            }
        }
    }
    
    public void removeSubject(String subject) throws MessageSendException{
        removeSubject(subject, null);
    }
    
    public void removeSubject(String subject, String[] keys) throws MessageSendException{
        if(socket == null){
            throw new MessageSendException("Not connected.");
        }
        if(subject == null){
            return;
        }
        try{
            send(new RemoveMessage(subject, keys));
        }catch(SocketTimeoutException e){
            throw new MessageSendException(e);
        }catch(SocketException e){
            throw new MessageSendException(e);
        }catch(IOException e){
            throw new MessageSendException(e);
        }
        if(subjects != null){
            Set<String> keySet = subjects.get(subject);
            if(keySet != null){
                if(keys == null){
                    keySet.remove(null);
                }else{
                    for(int i = 0; i < keys.length; i++){
                        keySet.remove(keys[i]);
                    }
                }
                if(keySet.size() == 0){
                    subjects.remove(subject);
                }
            }
        }
    }
    
    public void startReceive() throws MessageSendException{
        startReceive(-1);
    }
    
    public void startReceive(long from) throws MessageSendException{
        startReceive(from, false);
    }
    
    private void startReceive(long from, boolean isRestart) throws MessageSendException{
        if(socket == null){
            throw new MessageSendException("Not connected.");
        }
        if(!isRestart && isStartReceive){
            return;
        }
        try{
            isStartReceive = true;
            send(new StartReceiveMessage(from));
        }catch(SocketTimeoutException e){
            isStartReceive = false;
            throw new MessageSendException(e);
        }catch(SocketException e){
            isStartReceive = false;
            throw new MessageSendException(e);
        }catch(IOException e){
            isStartReceive = false;
            throw new MessageSendException(e);
        }
    }
    
    public boolean isStartReceive(){
        return isStartReceive;
    }
    
    public void stopReceive() throws MessageSendException{
        if(socket == null){
            throw new MessageSendException("Not connected.");
        }
        if(!isStartReceive){
            return;
        }
        try{
            send(new StopReceiveMessage());
            isStartReceive = false;
        }catch(SocketTimeoutException e){
            throw new MessageSendException(e);
        }catch(SocketException e){
            throw new MessageSendException(e);
        }catch(IOException e){
            throw new MessageSendException(e);
        }
    }
    
    public Set<String> getSubjects(){
        return subjects == null ? new HashSet<String>() : subjects.keySet();
    }
    
    public Set<String> getKeys(String subject){
        if(subjects == null){
            return new HashSet<String>();
        }
        Set<String> keySet = subjects.get(subject);
        return keySet == null ? new HashSet<String>() : keySet;
    }
    
    private void send(ClientMessage message) throws IOException{
        try{
            send(message, false);
        }catch(ClassNotFoundException e){
            // 起こらない
        }
    }
    
    private ServerMessage send(ClientMessage message, boolean reply) throws IOException, ClassNotFoundException{
        if(reply){
            synchronized(this){
                ((RequestReplyMessage)message).setRequestId(requestId++);
            }
        }
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        if(externalizer == null){
            ObjectOutputStream oos = new ObjectOutputStream(baos);
            oos.writeObject(message);
            oos.flush();
        }else{
            externalizer.writeExternal(message, baos);
        }
        byte[] bytes = baos.toByteArray();
        ReplyReceiver replyReceiver = null;
        if(reply){
            replyReceiver = (ReplyReceiver)replyReceiveDaemon.getDaemonRunnable();
            replyReceiver.initReply();
        }
        synchronized(this){
            DataOutputStream dos = new DataOutputStream(socket.getOutputStream());
            dos.writeInt(bytes.length);
            dos.write(bytes);
            dos.flush();
        }
        if(reply){
            return replyReceiver.getReply((RequestReplyMessage)message, responseTimeout);
        }else{
            return null;
        }
    }
    
    public void setMessageListener(MessageListener listener){
        messageListener = listener;
    }
    
    private void reconnect() throws ConnectException, MessageSendException{
        boolean isNowReconnecting = isReconnecting;
        synchronized(this){
            if(isNowReconnecting){
                return;
            }
            isReconnecting = true;
            try{
                if(socket != null){
                    try{
                        socket.close();
                    }catch(IOException e){}
                    socket = null;
                }
                ((MessageReceiver)messageReceiveDaemon.getDaemonRunnable()).reset();
                int tryCount = 0;
                boolean isSuccess = false;
                while(!isSuccess){
                    tryCount++;
                    try{
                        connect(id, true);
                        if(subjects != null){
                            Object[] subjectArray = subjects.keySet().toArray();
                            for(int i = 0; i < subjectArray.length; i++){
                                Object subject = subjectArray[i];
                                Set<String> keySet = subjects.get(subject);
                                if(keySet != null){
                                    String[] keys = (String[])keySet.toArray(new String[keySet.size()]);
                                    boolean containsNull = false;
                                    List<String> keyList = new ArrayList<String>();
                                    for(int j = 0; j < keys.length; j++){
                                        if(keys[j] == null){
                                            containsNull = true;
                                        }else{
                                            keyList.add(keys[j]);
                                        }
                                    }
                                    if(containsNull){
                                        addSubject((String)subject);
                                        keys = (String[])keyList.toArray(new String[keyList.size()]);
                                    }
                                    if(keys != null && keys.length != 0){
                                        addSubject((String)subject, keys);
                                    }
                                }
                            }
                        }
                        if(isStartReceive){
                            long time = -1;
                            MessageId latestMessageId = ((MessageReceiver)messageReceiveDaemon.getDaemonRunnable()).getLatestMessageId();
                            if(latestMessageId != null){
                                time = ((MessageReceiver)messageReceiveDaemon.getDaemonRunnable()).getLatestMessageReceiveTime() - reconnectBufferTime;
                            }
                            startReceive(time, true);
                        }
                        isSuccess = true;
                    }catch(ConnectException e){
                        if(tryCount >= reconnectCount){
                            throw e;
                        }else{
                            if(receiveWarnMessageId != null){
                                ServiceManagerFactory.getLogger().write(
                                    receiveWarnMessageId,
                                    e,
                                    this
                                );
                            }
                        }
                    }catch(MessageSendException e){
                        if(tryCount >= reconnectCount){
                            throw e;
                        }else{
                            if(receiveWarnMessageId != null){
                                ServiceManagerFactory.getLogger().write(
                                    receiveWarnMessageId,
                                    e,
                                    this
                                );
                            }
                        }
                    }
                    if(!isSuccess && reconnectInterval > 0){
                        try{
                            Thread.sleep(reconnectInterval);
                        }catch(InterruptedException e){
                            throw new ConnectException(e);
                        }
                    }
                }
            }finally{
                isReconnecting = false;
            }
        }
    }
    
    public boolean isConnected(){
        return isConnected;
    }
    
    public Object getId(){
        return id;
    }
    
    public synchronized void close(){
        isClosing = true;
        isConnected = false;
        if(serviceName != null){
            ServiceManagerFactory.unregisterService(
                serviceName.getServiceManagerName(),
                serviceName.getServiceName()
            );
            serviceName = null;
        }
        if(missingWindowCheckDaemon != null){
            missingWindowCheckDaemon.stopNoWait();
            missingWindowCheckDaemon = null;
        }
        if(messageReceiveDaemon != null){
            messageReceiveDaemon.stopNoWait();
            messageReceiveDaemon = null;
        }
        if(replyReceiveDaemon != null){
            replyReceiveDaemon.stopNoWait();
            replyReceiveDaemon = null;
        }
        if(socket != null){
            try{
                send(new ByeMessage());
            }catch(IOException e){
            }
            try{
                socket.close();
            }catch(IOException e){}
            socket = null;
        }
        if(packetReceiveDaemon != null){
            packetReceiveDaemon.stopNoWait();
            packetReceiveDaemon = null;
            receivePacketQueue.stop();
            receivePacketQueue.destroy();
            receivePacketQueue = null;
        }
        if(receiveSocket != null){
            receiveGroup = null;
            receiveSocket.close();
            receiveSocket = null;
        }
        isClosing = false;
    }
    
    public String toString(){
        final StringBuilder buf = new StringBuilder();
        buf.append(super.toString());
        buf.append('{');
        buf.append("id=").append(id);
        buf.append(", receiveAddress=").append(receiveAddress);
        buf.append(", receivePort=").append(receivePort);
        buf.append(", localAddress=").append(socket == null ? null : socket.getLocalSocketAddress());
        buf.append(", remoteAddress=").append(socket == null ? null : socket.getRemoteSocketAddress());
        buf.append(", subject=").append(subjects);
        buf.append('}');
        return buf.toString();
    }
    
    private class ReplyReceiver implements DaemonRunnable<Object>{
        public SynchronizeMonitor replyMonitor = new WaitSynchronizeMonitor();
        public List<Object> replyMessages = Collections.synchronizedList(new ArrayList<Object>());
        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();
                ServerMessage message = null;
                if(length > 0){
                    final byte[] dataBytes = new byte[length];
                    dis.readFully(dataBytes, 0, length);
                    ByteArrayInputStream is = new ByteArrayInputStream(dataBytes);
                    if(externalizer == null){
                        ObjectInputStream ois = new ObjectInputStream(is);
                        message = (ServerMessage)ois.readObject();
                    }else{
                        message = (ServerMessage)externalizer.readExternal(is);
                    }
                }else{
                    return null;
                }
                if(message != null && message.getMessageType() == ServerMessage.MESSAGE_SERVER_CLOSE_REQ){
                    if(serverCloseMessageId != null){
                        ServiceManagerFactory.getLogger().write(
                            serverCloseMessageId,
                            ClientConnectionImpl.this
                        );
                    }
                    close();
                    return null;
                }
                return message;
            }catch(SocketTimeoutException e){
                return null;
            }catch(SocketException e){
                replyMessages.add(e);
                replyMonitor.notifyAllMonitor();
                if(isClosing || !isConnected){
                    return null;
                }
                if(reconnectCount > 0){
                    try{
                        reconnect();
                        return null;
                    }catch(ConnectException e2){
                    }catch(MessageSendException e2){
                    }
                }
                if(receiveErrorMessageId != null){
                    ServiceManagerFactory.getLogger().write(
                        receiveErrorMessageId,
                        e,
                        ClientConnectionImpl.this
                    );
                }
                close();
                return null;
            }catch(EOFException e){
                replyMessages.add(e);
                replyMonitor.notifyAllMonitor();
                if(isClosing || !isConnected){
                    return null;
                }
                if(reconnectCount > 0){
                    try{
                        reconnect();
                        return null;
                    }catch(ConnectException e2){
                    }catch(MessageSendException e2){
                    }
                }
                if(receiveErrorMessageId != null){
                    ServiceManagerFactory.getLogger().write(
                        receiveErrorMessageId,
                        e,
                        ClientConnectionImpl.this
                    );
                }
                close();
                return null;
            }catch(ClassNotFoundException e){
                return e;
            }catch(IOException e){
                return e;
            }
        }
        public void consume(Object paramObj, DaemonControl ctrl) throws Throwable{
            if(paramObj == null){
                return;
            }
            replyMessages.add(paramObj);
            replyMonitor.notifyAllMonitor();
        }
        public void garbage(){}
        
        public void initReply(){
            replyMonitor.initMonitor();
        }
        
        public ServerMessage getReply(RequestReplyMessage request, long timeout) throws IOException, ClassNotFoundException{
            long start = System.currentTimeMillis();
            Object response = null;
            long wait = timeout;
            do{
                if(replyMessages.size() != 0){
                    response = replyMessages.remove(0);
                }else{
                    try{
                        if(wait == 0){
                            replyMonitor.waitMonitor();
                        }else{
                            replyMonitor.waitMonitor(wait);
                        }
                    }catch(InterruptedException e){
                        break;
                    }
                    if(replyMessages.size() != 0){
                        response = replyMessages.remove(0);
                    }
                }
                if(response != null && response instanceof RequestReplyMessage){
                    if(request.compareRequestId((RequestReplyMessage)response) > 0){
                        replyMonitor.initMonitor();
                        wait -= (System.currentTimeMillis() - start);
                    }else{
                        wait = 0;
                    }
                }else{
                    wait = 0;
                }
            }while(wait > 0);
            if(response == null){
                throw new SocketTimeoutException("wait_time=" + (System.currentTimeMillis() - start) + "[ms]");
            }
            if(response instanceof ServerMessage){
                return (ServerMessage)response;
            }else if(response instanceof IOException){
                throw (IOException)response;
            }else if(response instanceof ClassNotFoundException){
                throw (ClassNotFoundException)response;
            }
            return null;
        }
    }
    
    private class PacketReceiver implements DaemonRunnable<DatagramPacket>{
        
        public boolean onStart(){return true;}
        public boolean onStop(){return true;}
        public boolean onSuspend(){return true;}
        public boolean onResume(){return true;}
        public DatagramPacket provide(DaemonControl ctrl) throws Throwable{
            try{
                byte[] buf = new byte[windowSize];
                DatagramPacket packet = new DatagramPacket(buf, buf.length);
                receiveSocket.receive(packet);
                return packet;
            }catch(SocketException e){
                if(isClosing || !isConnected){
                    return null;
                }
                if(receiveErrorMessageId != null){
                    ServiceManagerFactory.getLogger().write(
                        receiveErrorMessageId,
                        e,
                        ClientConnectionImpl.this
                    );
                }
                close();
                return null;
            }catch(EOFException e){
                if(isClosing || !isConnected){
                    return null;
                }
                if(receiveErrorMessageId != null){
                    ServiceManagerFactory.getLogger().write(
                        receiveErrorMessageId,
                        e,
                        ClientConnectionImpl.this
                    );
                }
                close();
                return null;
            }catch(IOException e){
                if(isClosing || !isConnected){
                    return null;
                }
                if(receiveWarnMessageId != null){
                    ServiceManagerFactory.getLogger().write(
                        receiveWarnMessageId,
                        e,
                        ClientConnectionImpl.this
                    );
                }
                return null;
            }
        }
        public void consume(DatagramPacket packet, DaemonControl ctrl) throws Throwable{
            if(packet == null || receivePacketQueue == null || messageListener == null || !isStartReceive){
                return;
            }
            receivePacketCount++;
            receivePacketQueue.push(packet);
        }
        public void garbage(){}
    }
    
    private class MessageReceiver implements DaemonRunnable<DatagramPacket>{
        
        public MessageId latestMessageId;
        public long latestMessageReceiveTime;
        public SortedMap<MessageId,Window> missingWindowMap = Collections.synchronizedSortedMap(new TreeMap<MessageId,Window>());
        
        public boolean onStart(){return true;}
        public boolean onStop(){return true;}
        public boolean onSuspend(){return true;}
        public boolean onResume(){return true;}
        public DatagramPacket provide(DaemonControl ctrl) throws Throwable{
            return receivePacketQueue.get(1000);
        }
        public void consume(DatagramPacket packet, DaemonControl ctrl) throws Throwable{
            if(messageListener == null){
                return;
            }
            Window window = null;
            try{
                if(packet != null){
                    ByteArrayInputStream bais = new ByteArrayInputStream(packet.getData());
                    DataInputStream dis = new DataInputStream(bais);
                    window = new Window();
                    window.read(dis);
                }
                receiveWindow(window);
            }catch(IOException e){
                if(!isClosing && isConnected && receiveErrorMessageId != null){
                    ServiceManagerFactory.getLogger().write(
                        receiveErrorMessageId,
                        e,
                        ClientConnectionImpl.this
                    );
                }
                return;
            }catch(ClassNotFoundException e){
                if(!isClosing && isConnected && receiveErrorMessageId != null){
                    ServiceManagerFactory.getLogger().write(
                        receiveErrorMessageId,
                        e,
                        ClientConnectionImpl.this
                    );
                }
                return;
            }
        }
        
        public synchronized void receiveWindow(Window window) throws IOException, ClassNotFoundException{
            MessageImpl message = null;
            while((message = retrieveMessage(window)) != null){
                window = null;
                handleMessage(message);
            }
        }
        
        private void handleMessage(MessageImpl message){
            if(message == null || message.isLost()){
                return;
            }
            receiveCount++;
            long sTime = System.currentTimeMillis();
            messageListener.onMessage(message);
            onMessageProcessTime += (System.currentTimeMillis() - sTime);
        }
        
        private MessageImpl retrieveMessage(Window window) throws IOException, ClassNotFoundException{
            
            MessageId id = window == null ? null : window.toMessageId();
            MessageImpl message = null;
            if(window == null){
                if(missingWindowMap.size() == 0){
                    return null;
                }
                id = (MessageId)missingWindowMap.firstKey();
                window = (Window)missingWindowMap.get(id);
                if(window.isComplete() || window.isLost()){
                    message = window.getMessage(externalizer);
                }else{
                    return null;
                }
            }else if(window.isComplete()){
                message = window.getMessage(externalizer);
            }else{
                Window w = (Window)missingWindowMap.get(id);
                if(w == null){
                    if(latestMessageId == null || latestMessageId.compareTo(id) < 0){
                        synchronized(missingWindowMap){
                            missingWindowMap.put(id, window);
                        }
                    }else{
                        wasteWindowCount++;
                    }
                    return null;
                }else{
                    if(w.addWindow(window)){
                        return retrieveMessage(null);
                    }else{
                        return null;
                    }
                }
            }
            
            if(message == null){
                return null;
            }
            
            if(receiveAddress != null){
                MulticastMessageImpl multicastMessage = (MulticastMessageImpl)message;
                if(!multicastMessage.containsId(ClientConnectionImpl.this.id)){
                    synchronized(missingWindowMap){
                        missingWindowMap.remove(id);
                    }
                    return retrieveMessage(null);
                }
            }
            
            if(latestMessageId == null){
                if(!message.isFirst()){
                    if(!missingWindowMap.containsKey(id)){
                        synchronized(missingWindowMap){
                            missingWindowMap.put(id, window);
                        }
                    }
                    return null;
                }
            }else{
                if(latestMessageId.compareTo(id) >= 0){
                    wasteWindowCount++;
                    synchronized(missingWindowMap){
                        missingWindowMap.remove(id);
                    }
                    return null;
                }else if(!latestMessageId.isNext(message)){
                    noContinuousMessageCount++;
                    if(window != null && !missingWindowMap.containsKey(id)){
                        synchronized(missingWindowMap){
                            missingWindowMap.put(id, window);
                        }
                        return retrieveMessage(null);
                    }else{
                        return null;
                    }
                }
            }
            synchronized(missingWindowMap){
                missingWindowMap.remove(id);
            }
            latestMessageReceiveTime = message.getReceiveTime();
            latestMessageId = id;
            return message;
        }
        
        private synchronized void reset(){
            latestMessageId = null;
            missingWindowMap.clear();
        }
        
        public void garbage(){}
        
        public int getMissingWindowSize(){
            return missingWindowMap.size();
        }
        
        public Window getMissingWindow(MessageId id){
            return (Window)missingWindowMap.get(id);
        }
        
        public MessageId getLatestMessageId(){
            return latestMessageId;
        }
        public long getLatestMessageReceiveTime(){
            return latestMessageReceiveTime;
        }
        
        public List<Window> getMissingWindows(){
            if(missingWindowMap.size() == 0){
                return new ArrayList<Window>();
            }
            synchronized(missingWindowMap){
                return new ArrayList<Window>(missingWindowMap.values());
            }
        }
    }
    
    private class MissingWindowChecker implements DaemonRunnable<Object>{
        private long lastCheckTime;
        private long lastPollingTime;
        private MessageReceiver messageReceiver;
        
        public MissingWindowChecker(MessageReceiver receiver){
            messageReceiver = receiver;
        }
        
        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{
            long waitTime = missingWindowTimeout;
            if(lastCheckTime != 0){
                long checkInterval = System.currentTimeMillis() - lastCheckTime;
                waitTime -= checkInterval;
            }
            if(waitTime > 0){
                Thread.sleep(waitTime);
            }
            return null;
        }
        public void consume(Object paramObj, DaemonControl ctrl) throws Throwable{
            try{
                lastCheckTime = System.currentTimeMillis();
                List<Window> missingWindows = null;
                if(messageReceiver.getMissingWindowSize() == 0){
                    MessageId latestMessageId = messageReceiver.getLatestMessageId();
                    if(latestMessageId == null){
                        return;
                    }
                    if(missingWindowTimeout > lastCheckTime - messageReceiver.getLatestMessageReceiveTime()
                        || newMessagePollingInterval > lastCheckTime - lastPollingTime){
                        return;
                    }
                    lastPollingTime = lastCheckTime;
                    InterpolateRequestMessage request = new InterpolateRequestMessage();
                    request.setLatestMessageId(latestMessageId);
                    InterpolateResponseMessage response = null;
                    final long start = System.currentTimeMillis();
                    try{
                        newMessagePollingCount++;
                        response = (InterpolateResponseMessage)send(request, true);
                    }catch(SocketTimeoutException e){
                        newMessagePollingTimeoutCount++;
                    }
                    newMessagePollingResponseTime += (System.currentTimeMillis() - start);
                    List<Window> ws = response == null ? null : response.getWindows();
                    if(ws != null){
                        missingWindows = new ArrayList<Window>();
                        missingWindows.addAll(ws);
                    }
                }else{
                    MessageId lastMessageId = messageReceiver.getLatestMessageId();
                    final List<Window> windows = messageReceiver.getMissingWindows();
                    MessageId currentFirstMessageId = null;
                    if(lastMessageId == null && windows.size() != 0){
                        currentFirstMessageId = windows.get(0).toMessageId();
                    }
                    List<MessageId> missingMessageIds = null;
                    List<WindowId> missingWindowIds = null;
                    for(int i = 0; i < windows.size(); i++){
                        Window window = windows.get(i);
                        final long missingTime = lastCheckTime - window.getReceiveTime();
                        if(missingTime < missingWindowTimeout){
                            break;
                        }
                        if(lastMessageId != null){
                            missingMessageIds = lastMessageId.createMissingIds(window, missingMessageIds);
                        }
                        lastMessageId = window.toMessageId();
                        if(window.isComplete() || window.isLost()){
                            continue;
                        }
                        missingWindowIds = window.getMissingWindowIds(missingWindowIds);
                    }
                    if(currentFirstMessageId != null || missingMessageIds != null || missingWindowIds != null){
                        InterpolateRequestMessage request = new InterpolateRequestMessage();
                        if(currentFirstMessageId != null){
                            request.setCurrentFirstMessageId(currentFirstMessageId);
                        }
                        if(missingMessageIds != null){
                            request.setMessageIds(missingMessageIds.toArray(new MessageId[missingMessageIds.size()]));
                        }
                        if(missingWindowIds != null){
                            request.setWindowIds(missingWindowIds.toArray(new WindowId[missingWindowIds.size()]));
                        }
                        InterpolateResponseMessage response = null;
                        final long start = System.currentTimeMillis();
                        try{
                            missingWindowRequestCount++;
                            response = (InterpolateResponseMessage)send(request, true);
                        }catch(SocketTimeoutException e){
                            missingWindowRequestTimeoutCount++;
                        }
                        missingWindowResponseTime += (System.currentTimeMillis() - start);
                        if(response != null){
                            missingWindows = new ArrayList<Window>();
                            List<MessageId> lostIds = new ArrayList<MessageId>();
                            if(currentFirstMessageId != null){
                                List<Window> ws = response.getWindows();
                                if(ws != null){
                                    missingWindows.addAll(ws);
                                }
                            }
                            if(missingMessageIds != null){
                                for(int i = 0, imax = missingMessageIds.size(); i < imax; i++){
                                    MessageId id = missingMessageIds.get(i);
                                    List<Window> ws = response.getWindows(id);
                                    if(ws == null){
                                        lostIds.add(id);
                                        Window w = new Window();
                                        w.sequence = id.sequence;
                                        w.setWindowCount((short)1);
                                        w.setLost(true);
                                        missingWindows.add(w);
                                    }else{
                                        missingWindows.addAll(ws);
                                    }
                                }
                            }
                            if(missingWindowIds != null){
                                for(int i = 0, imax = missingWindowIds.size(); i < imax; i++){
                                    WindowId id = missingWindowIds.get(i);
                                    Window window = response.getWindow(id);
                                    if(window == null){
                                        lostIds.add(id);
                                        Window exists = messageReceiver.getMissingWindow(id.toMessageId());
                                        Window w = new Window();
                                        w.sequence = id.sequence;
                                        w.windowNo = id.windowNo;
                                        w.setWindowCount(exists == null ? 1 : exists.getWindowCount());
                                        w.setLost(true);
                                        missingWindows.add(w);
                                    }else{
                                        missingWindows.add(window);
                                    }
                                }
                            }
                            if(lostIds.size() != 0 && messageLostErrorMessageId != null){
                                ServiceManagerFactory.getLogger().write(
                                    messageLostErrorMessageId,
                                    ClientConnectionImpl.this,
                                    lostIds
                                );
                            }
                        }
                    }
                    if(missingWindows != null && missingWindows.size() != 0){
                        Collections.sort(missingWindows);
                        for(int i = 0, imax = missingWindows.size(); i < imax; i++){
                            Window window = (Window)missingWindows.get(i);
                            messageReceiver.receiveWindow(window);
                        }
                        receivePacketQueue.push(null);
                    }
                }
            }catch(IOException e){
                if(isClosing || !isConnected){
                    return;
                }
                if(receiveWarnMessageId != null){
                    ServiceManagerFactory.getLogger().write(
                        receiveWarnMessageId,
                        e,
                        ClientConnectionImpl.this
                    );
                }
            }catch(ClassNotFoundException e){
                if(isClosing || !isConnected){
                    return;
                }
                if(receiveWarnMessageId != null){
                    ServiceManagerFactory.getLogger().write(
                        receiveWarnMessageId,
                        e,
                        ClientConnectionImpl.this
                    );
                }
            }
        }
        
        public void garbage(){}
    }
    
    /**
     * UDPプロトコル用の{@link ClientConnection}の管理サービス。<p>
     *
     * @author M.Takata
     */
    public class ClientConnectionService extends ServiceBase implements ClientConnectionServiceMBean{
        
        private static final long serialVersionUID = 5243807973535652312L;
        
        public Set<String> getSubjects(){
            return ClientConnectionImpl.this.getSubjects();
        }
        
        public Set<String> getKeys(String subject){
            return ClientConnectionImpl.this.getKeys(subject);
        }
        
        public long getReceiveCount(){
            return ClientConnectionImpl.this.receiveCount;
        }
        
        public long getReceivePacketCount(){
            return ClientConnectionImpl.this.receivePacketCount;
        }
        
        public void resetCount(){
            ClientConnectionImpl.this.receiveCount = 0;
            ClientConnectionImpl.this.receivePacketCount = 0;
            ClientConnectionImpl.this.onMessageProcessTime = 0;
            ClientConnectionImpl.this.noContinuousMessageCount = 0;
            ClientConnectionImpl.this.wasteWindowCount = 0;
            ClientConnectionImpl.this.missingWindowRequestCount = 0;
            ClientConnectionImpl.this.missingWindowRequestTimeoutCount = 0;
            ClientConnectionImpl.this.missingWindowResponseTime = 0;
            ClientConnectionImpl.this.newMessagePollingCount = 0;
            ClientConnectionImpl.this.newMessagePollingTimeoutCount = 0;
            ClientConnectionImpl.this.newMessagePollingResponseTime = 0;
        }
        
        public long getAverageOnMessageProcessTime(){
            return ClientConnectionImpl.this.receiveCount == 0 ? 0 : (ClientConnectionImpl.this.onMessageProcessTime / ClientConnectionImpl.this.receiveCount);
        }
        
        public long getMissingWindowRequestCount(){
            return ClientConnectionImpl.this.missingWindowRequestCount;
        }
        
        public long getMissingWindowRequestTimeoutCount(){
            return ClientConnectionImpl.this.missingWindowRequestTimeoutCount;
        }
        
        public long getAverageMissingWindowResponseTime(){
            return ClientConnectionImpl.this.missingWindowRequestCount == 0 ? 0 : (ClientConnectionImpl.this.missingWindowResponseTime / ClientConnectionImpl.this.missingWindowRequestCount);
        }
        
        public long getNewMessagePollingCount(){
            return ClientConnectionImpl.this.newMessagePollingCount;
        }
        
        public long getNewMessagePollingTimeoutCount(){
            return ClientConnectionImpl.this.newMessagePollingTimeoutCount;
        }
        
        public long getAverageNewMessagePollingResponseTime(){
            return ClientConnectionImpl.this.newMessagePollingCount == 0 ? 0 : (ClientConnectionImpl.this.newMessagePollingResponseTime / ClientConnectionImpl.this.newMessagePollingCount);
        }
        
        public SocketAddress getLocalSocketAddress(){
            return ClientConnectionImpl.this.socket.getLocalSocketAddress();
        }
        
        public SocketAddress getRemoteSocketAddress(){
            return ClientConnectionImpl.this.socket.getRemoteSocketAddress();
        }
        
        public SocketAddress getReceiveSocketAddress(){
            return ClientConnectionImpl.this.receiveSocket.getLocalSocketAddress();
        }
        
        public long getReceivePacketQueueCount(){
            return receivePacketQueue == null ? 0 : receivePacketQueue.getCount();
        }
        
        public long getReceivePacketQueueCountDelta(){
            return receivePacketQueue == null ? 0 : receivePacketQueue.getCountDelta();
        }
        
        public long getReceivePacketQueueLastPushedTimeMillis(){
            return receivePacketQueue == null ? 0 : receivePacketQueue.getLastPushedTimeMillis();
        }
        
        public Date getReceivePacketQueueLastPushedTime(){
            return receivePacketQueue == null ? null : receivePacketQueue.getLastPushedTime();
        }
        
        public long getReceivePacketQueueDepth(){
            return receivePacketQueue == null ? 0 : receivePacketQueue.getDepth();
        }
        
        public long getReceivePacketQueueDepthDelta(){
            return receivePacketQueue == null ? 0 : receivePacketQueue.getDepthDelta();
        }
        
        public long getReceivePacketQueueMaxDepth(){
            return receivePacketQueue == null ? 0 : receivePacketQueue.getMaxDepth();
        }
        
        public long getNoContinuousMessageCount(){
            return noContinuousMessageCount;
        }
        
        public long getWasteWindowCount(){
            return wasteWindowCount;
        }
        
        public MessageId getLatestMessageId(){
            return messageReceiveDaemon == null ? null : ((MessageReceiver)messageReceiveDaemon.getDaemonRunnable()).getLatestMessageId();
        }
        
        public Date getLatestMessageReceiveTime(){
            return messageReceiveDaemon == null ? null : new Date(((MessageReceiver)messageReceiveDaemon.getDaemonRunnable()).getLatestMessageReceiveTime());
        }
        
        public int getMissingWindowSize(){
            return messageReceiveDaemon == null ? 0 : ((MessageReceiver)messageReceiveDaemon.getDaemonRunnable()).getMissingWindowSize();
        }
        
        public void connect() throws ConnectException{
            ClientConnectionImpl.this.connect();
        }
        
        public void connect(Object id) throws ConnectException{
            ClientConnectionImpl.this.connect(id);
        }
        
        public void startReceive() throws MessageSendException{
            ClientConnectionImpl.this.startReceive();
        }
        
        public void startReceive(long from) throws MessageSendException{
            ClientConnectionImpl.this.startReceive(from);
        }
        
        public void stopReceive() throws MessageSendException{
            ClientConnectionImpl.this.stopReceive();
        }
        
        public boolean isStartReceive(){
            return ClientConnectionImpl.this.isStartReceive();
        }
        
        public void addSubject(String subject) throws MessageSendException{
            ClientConnectionImpl.this.addSubject(subject);
        }
        
        public void addSubject(String subject, String[] keys) throws MessageSendException{
            ClientConnectionImpl.this.addSubject(subject, keys);
        }
        
        public void removeSubject(String subject) throws MessageSendException{
            ClientConnectionImpl.this.removeSubject(subject);
        }
        
        public void removeSubject(String subject, String[] keys) throws MessageSendException{
            ClientConnectionImpl.this.removeSubject(subject, keys);
        }
        
        public void reconnect() throws ConnectException, MessageSendException{
            ClientConnectionImpl.this.reconnect();
            if(packetReceiveDaemon != null && packetReceiveDaemon.isSusupend()){
                packetReceiveDaemon.resume();
            }
        }
        
        public boolean isConnected(){
            return ClientConnectionImpl.this.isConnected();
        }
        
        public void close(){
            ClientConnectionImpl.this.close();
        }
    }
    
    /**
     * UDPプロトコル用の{@link ClientConnection}の管理サービスのMBeanインタフェース。<p>
     *
     * @author M.Takata
     */
    public interface ClientConnectionServiceMBean extends ServiceBaseMBean{
        
        /**
         * サーバに要求を送信する送信ソケットのローカルソケットアドレスを取得する。<p>
         *
         * @return ローカルソケットアドレス
         */
        public SocketAddress getLocalSocketAddress();
        
        /**
         * サーバに要求を送信する送信ソケットのリモートソケットアドレスを取得する。<p>
         *
         * @return リモートソケットアドレス
         */
        public SocketAddress getRemoteSocketAddress();
        
        /**
         * 受信側のローカルソケットアドレスを取得する。<p>
         *
         * @return ローカルソケットアドレス
         */
        public SocketAddress getReceiveSocketAddress();
        
        /**
         * 登録されているサブジェクトを取得する。<p>
         *
         * @return 登録されているサブジェクトの集合
         */
        public Set<String> getSubjects();
        
        /**
         * 指定したサブジェクトに登録されているキーを取得する。<p>
         *
         * @return 登録されているキーの集合
         */
        public Set<String> getKeys(String subject);
        
        /**
         * 受信件数を取得する。<p>
         *
         * @return 受信件数
         */
        public long getReceiveCount();
        
        /**
         * 受信パケット件数を取得する。<p>
         *
         * @return 受信パケット件数
         */
        public long getReceivePacketCount();
        
        /**
         * 平均メッセージ処理時間を取得する。<p>
         *
         * @return 平均メッセージ処理時間[ms]
         */
        public long getAverageOnMessageProcessTime();
        
        /**
         * 補間要求の送信件数を取得する。<p>
         *
         * @return 補間要求の送信件数
         */
        public long getMissingWindowRequestCount();
        
        /**
         * 補間要求のタイムアウト件数を取得する。<p>
         *
         * @return 補間要求のタイムアウト件数
         */
        public long getMissingWindowRequestTimeoutCount();
        
        /**
         * 補間要求の平均応答時間[ms]を取得する。<p>
         *
         * @return 補間要求の平均応答時間
         */
        public long getAverageMissingWindowResponseTime();
        
        /**
         * 新メッセージポーリングの送信件数を取得する。<p>
         *
         * @return 新メッセージポーリングの送信件数
         */
        public long getNewMessagePollingCount();
        
        /**
         * 新メッセージポーリングのタイムアウト件数を取得する。<p>
         *
         * @return 新メッセージポーリングのタイムアウト件数
         */
        public long getNewMessagePollingTimeoutCount();
        
        /**
         * 新メッセージポーリングの平均応答時間[ms]を取得する。<p>
         *
         * @return 新メッセージポーリングの平均応答時間
         */
        public long getAverageNewMessagePollingResponseTime();
        
        /**
         * パケット到達時にメッセージの順序性が保たれていなかった回数を取得する。<p>
         *
         * @return メッセージの順序性が保たれていなかった回数
         */
        public long getNoContinuousMessageCount();
        
        /**
         * ロスト時の補完処理などで、無駄になったウィンドウの件数を取得する。<p>
         *
         * @return 無駄になったウィンドウの件数
         */
        public long getWasteWindowCount();
        
        /**
         * カウントをリセットする。<p>
         */
        public void resetCount();
        
        /**
         * パケット受信キューの件数を取得する。<p>
         * 
         * @return パケット受信キューの件数
         */
        public long getReceivePacketQueueCount();
        
        /**
         * パケット受信キューの件数の前回差を取得する。<p>
         * 
         * @return パケット受信キューの件数の前回差
         */
        public long getReceivePacketQueueCountDelta();
        
        /**
         * パケット受信キューの最終受信時間を取得する。<p>
         * 
         * @return パケット受信キューの最終受信時間
         */
        public long getReceivePacketQueueLastPushedTimeMillis();
        
        /**
         * パケット受信キューの最終受信時間を取得する。<p>
         * 
         * @return パケット受信キューの最終受信時間
         */
        public Date getReceivePacketQueueLastPushedTime();
        
        /**
         * パケット受信キューの滞留件数を取得する。<p>
         * 
         * @return パケット受信キューの滞留件数
         */
        public long getReceivePacketQueueDepth();
        
        /**
         * パケット受信キューの滞留件数の前回差を取得する。<p>
         * 
         * @return パケット受信キューの滞留件数の前回差
         */
        public long getReceivePacketQueueDepthDelta();
        
        /**
         * パケット受信キューの最大滞留件数を取得する。<p>
         * 
         * @return パケット受信キューの最大滞留件数
         */
        public long getReceivePacketQueueMaxDepth();
        
        /**
         * 最終受信メッセージIDを取得する。<p>
         *
         * @return 最終受信メッセージID
         */
        public MessageId getLatestMessageId();
        
        /**
         * 最終受信メッセージの受信時刻を取得する。<p>
         *
         * @return 最終受信メッセージの受信時刻
         */
        public Date getLatestMessageReceiveTime();
        
        /**
         * 受信順序を調整中のメッセージ件数を取得する。<p>
         *
         * @return 受信順序を調整中のメッセージ件数
         */
        public int getMissingWindowSize();
        
        /**
         * サーバと接続する。<p>
         *
         * @exception ConnectException サーバとの接続に失敗した場合
         */
        public void connect() throws ConnectException;
        
        /**
         * サーバと接続する。<p>
         *
         * @param id クライアントを識別するID
         * @exception ConnectException サーバとの接続に失敗した場合
         */
        public void connect(Object id) throws ConnectException;
        
        /**
         * 配信開始をサーバに要求する。<br>
         *
         * @exception MessageSendException サーバへの要求に失敗した場合
         */
        public void startReceive() throws MessageSendException;
        
        /**
         * 指定した過去の時間のデータから配信開始をサーバに要求する。<br>
         *
         * @param from 開始時間
         * @exception MessageSendException サーバへの要求に失敗した場合
         */
        public void startReceive(long from) throws MessageSendException;
        
        /**
         * 配信停止をサーバに要求する。<br>
         *
         * @exception MessageSendException サーバへの要求に失敗した場合
         */
        public void stopReceive() throws MessageSendException;
        
        /**
         * 配信開始しているかどうかを判定する。<br>
         *
         * @return 配信開始している場合true
         */
        public boolean isStartReceive();
        
        /**
         * 配信して欲しいサブジェクトをサーバに要求する。<br>
         *
         * @param subject サブジェクト
         * @exception MessageSendException サーバへの要求に失敗した場合
         */
        public void addSubject(String subject) throws MessageSendException;
        
        /**
         * 配信して欲しいサブジェクトとキーをサーバに要求する。<br>
         *
         * @param subject サブジェクト
         * @param keys キー
         * @exception MessageSendException サーバへの要求に失敗した場合
         */
        public void addSubject(String subject, String[] keys) throws MessageSendException;
        
        /**
         * 配信を解除して欲しいサブジェクトをサーバに要求する。<br>
         *
         * @param subject サブジェクト
         * @exception MessageSendException サーバへの要求に失敗した場合
         */
        public void removeSubject(String subject) throws MessageSendException;
        
        /**
         * 配信を解除して欲しいサブジェクトとキーをサーバに要求する。<br>
         *
         * @param subject サブジェクト
         * @param keys キー
         * @exception MessageSendException サーバへの要求に失敗した場合
         */
        public void removeSubject(String subject, String[] keys) throws MessageSendException;
        
        /**
         * サーバと再接続する。<p>
         *
         * @exception ConnectException サーバとの接続に失敗した場合
         * @exception MessageSendException サーバへの要求に失敗した場合
         */
        public void reconnect() throws ConnectException, MessageSendException;
        
        /**
         * 接続しているかどうかを判定する。<p>
         *
         * @return 接続している場合true
         */
        public boolean isConnected();
        
        /**
         * サーバと切断する。<p>
         */
        public void close();
    }
}