/*
 * 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.net.Socket;
import java.net.InetAddress;
import java.net.SocketAddress;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.channels.ServerSocketChannel;
import java.util.Set;
import java.util.HashSet;
import java.util.Map;
import java.util.HashMap;
import java.util.List;
import java.util.ArrayList;
import javax.net.ServerSocketFactory;
import javax.net.SocketFactory;

import jp.ossc.nimbus.core.ServiceBase;
import jp.ossc.nimbus.core.ServiceName;
import jp.ossc.nimbus.core.ServiceManagerFactory;
import jp.ossc.nimbus.service.repository.Repository;
import jp.ossc.nimbus.service.publish.ClientConnectionFactory;
import jp.ossc.nimbus.service.publish.ServerConnectionFactory;
import jp.ossc.nimbus.service.publish.ClientConnection;
import jp.ossc.nimbus.service.publish.ServerConnection;
import jp.ossc.nimbus.service.publish.ServerConnectionListener;
import jp.ossc.nimbus.service.publish.RemoteClientConnectionFactory;
import jp.ossc.nimbus.service.publish.ConnectionCreateException;
import jp.ossc.nimbus.service.io.Externalizer;

/**
 * TCPプロトコル用の{@link ClientConnectionFactory}及び{@link ServerConnectionFactory}インタフェース実装サービス。<p>
 * 
 * @author M.Takata
 */
public class ConnectionFactoryService extends ServiceBase implements ServerConnectionFactory, ClientConnectionFactory, ConnectionFactoryServiceMBean{
    
    private static final long serialVersionUID = 4621521654243947901L;
    
    private String clientAddressProertyName;
    private String clientPortProertyName;
    private int clientReconnectCount;
    private long clientReconnectInterval;
    private long clientReconnectBufferTime;
    private String serverAddress;
    private int serverPort;
    private int serverBacklog;
    private boolean isNIO;
    private ServiceName nioSocketFactoryServiceName;
    private ServiceName serverSocketFactoryServiceName;
    private ServiceName socketFactoryServiceName;
    private ServiceName jndiRepositoryServiceName;
    private String jndiName = DEFAULT_JNDI_NAME;
    private int rmiPort;
    private ServiceName[] serverConnectionListenerServiceNames;
    
    private int sendThreadSize = 1;
    private ServiceName sendQueueServiceName;
    private int asynchSendThreadSize;
    private ServiceName asynchSendQueueFactoryServiceName;
    private ServiceName externalizerServiceName;
    private int maxSendRetryCount;
    private long sendBufferTime = 5000;
    private String serverSendErrorMessageId = MSG_ID_SEND_ERROR;
    private String serverSendErrorRetryOverMessageId = MSG_ID_SEND_ERROR_RETRY_OVER;
    private String clientServerCloseMessageId = MSG_ID_SERVER_CLOSE;
    private String clientReceiveWarnMessageId = MSG_ID_RECEIVE_WARN;
    private String clientReceiveErrorMessageId = MSG_ID_RECEIVE_ERROR;
    
    private ServerSocketFactory serverSocketFactory;
    private ServerConnectionImpl serverConnection;
    private SocketFactory socketFactory;
    private jp.ossc.nimbus.net.SocketFactory nioSocketFactory;
    private Repository jndiRepository;
    private Externalizer<Object> externalizer;
    private List<ServerConnectionListener> serverConnectionListeners;
    
    public void setClientAddressProertyName(String name){
        clientAddressProertyName = name;
    }
    public String getClientAddressProertyName(){
        return clientAddressProertyName;
    }
    
    public void setClientPortProertyName(String name){
        clientPortProertyName = name;
    }
    public String getClientPortProertyName(){
        return clientPortProertyName;
    }
    
    public void setClientReconnectCount(int count){
        clientReconnectCount = count;
    }
    public int getClientReconnectCount(){
        return clientReconnectCount;
    }
    
    public void setClientReconnectInterval(long interval){
        clientReconnectInterval = interval;
    }
    public long getClientReconnectInterval(){
        return clientReconnectInterval;
    }
    
    public void setClientReconnectBufferTime(long interval){
        clientReconnectBufferTime = interval;
    }
    public long getClientReconnectBufferTime(){
        return clientReconnectBufferTime;
    }
    
    public void setServerAddress(String address){
        serverAddress = address;
    }
    public String getServerAddress(){
        return serverAddress;
    }
    
    public void setServerPort(int port){
        serverPort = port;
    }
    public int getServerPort(){
        return serverPort;
    }
    
    public boolean isNIO(){
        return isNIO;
    }
    public void setNIO(boolean isNIO){
        this.isNIO = isNIO;
    }
    
    public void setServerBacklog(int backlog){
        serverBacklog = backlog;
    }
    public int getServerBacklog(){
        return serverBacklog;
    }
    
    public void setServerSocketFactoryServiceName(ServiceName name){
        serverSocketFactoryServiceName = name;
    }
    public ServiceName getServerSocketFactoryServiceName(){
        return serverSocketFactoryServiceName;
    }
    
    public void setSocketFactoryServiceName(ServiceName name){
        socketFactoryServiceName = name;
    }
    public ServiceName getSocketFactoryServiceName(){
        return socketFactoryServiceName;
    }
    
    public void setNIOSocketFactoryServiceName(ServiceName name){
        nioSocketFactoryServiceName = name;
    }
    public ServiceName getNIOSocketFactoryServiceName(){
        return nioSocketFactoryServiceName;
    }
    
    public void setJndiName(String name){
        jndiName = name;
    }
    public String getJndiName(){
        return jndiName;
    }
    
    public void setJndiRepositoryServiceName(ServiceName name){
        jndiRepositoryServiceName = name;
    }
    public ServiceName getJndiRepositoryServiceName(){
        return jndiRepositoryServiceName;
    }
    
    public void setRMIPort(int port){
        rmiPort = port;
    }
    public int getRMIPort(){
        return rmiPort;
    }
    
    public void setSendQueueServiceName(ServiceName name){
        sendQueueServiceName = name;
    }
    public ServiceName getSendQueueServiceName(){
        return sendQueueServiceName;
    }
    
    public void setSendThreadSize(int threadSize){
        sendThreadSize = threadSize;
    }
    public int getSendThreadSize(){
        return sendThreadSize;
    }
    
    public void setAsynchSendQueueFactoryServiceName(ServiceName name){
        asynchSendQueueFactoryServiceName = name;
    }
    public ServiceName getAsynchSendQueueFactoryServiceName(){
        return asynchSendQueueFactoryServiceName;
    }
    
    public void setAsynchSendThreadSize(int threadSize){
        asynchSendThreadSize = threadSize;
    }
    public int getAsynchSendThreadSize(){
        return asynchSendThreadSize;
    }
    
    public void setExternalizerServiceName(ServiceName name){
        externalizerServiceName = name;
    }
    public ServiceName getExternalizerServiceName(){
        return externalizerServiceName;
    }
    
    public void setServerConnectionListenerServiceNames(ServiceName[] names){
        serverConnectionListenerServiceNames = names;
    }
    public ServiceName[] getServerConnectionListenerServiceNames(){
        return serverConnectionListenerServiceNames;
    }
    
    public void setMaxSendRetryCount(int count){
        maxSendRetryCount = count;
    }
    public int getMaxSendRetryCount(){
        return maxSendRetryCount;
    }
    
    public void setSendBufferTime(long time){
        sendBufferTime = time;
    }
    
    public long getSendBufferTime(){
        return sendBufferTime;
    }
    
    public void setServerSendErrorMessageId(String id){
        serverSendErrorMessageId = id;
    }
    public String getServerSendErrorMessageId(){
        return serverSendErrorMessageId;
    }
    
    public void setServerSendErrorRetryOverMessageId(String id){
        serverSendErrorRetryOverMessageId = id;
    }
    public String getServerSendErrorRetryOverMessageId(){
        return serverSendErrorRetryOverMessageId;
    }
    
    public void setClientServerCloseMessageId(String id){
        clientServerCloseMessageId = id;
    }
    public String getClientServerCloseMessageId(){
        return clientServerCloseMessageId;
    }
    
    public void setClientReceiveWarnMessageId(String id){
        clientReceiveWarnMessageId = id;
    }
    public String getClientReceiveWarnMessageId(){
        return clientReceiveWarnMessageId;
    }
    
    public void setClientReceiveErrorMessageId(String id){
        clientReceiveErrorMessageId = id;
    }
    public String getClientReceiveErrorMessageId(){
        return clientReceiveErrorMessageId;
    }
    
    public void setServerSocketFactory(ServerSocketFactory factory){
        serverSocketFactory = factory;
    }
    public ServerSocketFactory getServerSocketFactory(){
        return serverSocketFactory;
    }
    
    public void setSocketFactory(SocketFactory factory){
        socketFactory = factory;
    }
    public SocketFactory getSocketFactory(){
        return socketFactory;
    }
    
    public void setNIOSocketFactory(jp.ossc.nimbus.net.SocketFactory factory){
        nioSocketFactory = factory;
    }
    public jp.ossc.nimbus.net.SocketFactory getNIOSocketFactory(){
        return nioSocketFactory;
    }
    
    public void setExternalizer(Externalizer<Object> ext){
        externalizer = ext;
    }
    public Externalizer<Object> getExternalizer(){
        return externalizer;
    }
    
    public void addServerConnectionListener(ServerConnectionListener listener){
        if(serverConnectionListeners == null){
            serverConnectionListeners = new ArrayList<ServerConnectionListener>();
        }
        serverConnectionListeners.add(listener);
    }
    
    public void removeServerConnectionListener(ServerConnectionListener listener){
        if(serverConnectionListeners == null){
            return;
        }
        serverConnectionListeners.remove(listener);
    }
    
    public void clearServerConnectionListeners(){
        if(serverConnectionListeners == null){
            return;
        }
        serverConnectionListeners.clear();
    }
    
    public ServerConnectionListener[] getServerConnectionListeners(){
        return serverConnectionListeners == null ? null : (ServerConnectionListener[])serverConnectionListeners.toArray(new ServerConnectionListener[serverConnectionListeners.size()]);
    }
    
    public long getSendCount(){
        return serverConnection == null ? 0 : serverConnection.getSendCount();
    }
    
    public void resetSendCount(){
        if(serverConnection == null){
            return;
        }
        serverConnection.resetSendCount();
    }
    
    public long getAverageSendProcessTime(){
        return serverConnection == null ? 0 : serverConnection.getAverageSendProcessTime();
    }
    
    public Set<SocketAddress> getClients(){
        if(serverConnection == null){
            return new HashSet<SocketAddress>();
        }
        Set<ServerConnectionImpl.ClientImpl> clients = serverConnection.getClients();
        ServerConnectionImpl.ClientImpl[] clientArray = clients.toArray(new ServerConnectionImpl.ClientImpl[clients.size()]);
        Set<SocketAddress> result = new HashSet<SocketAddress>();
        for(int i = 0; i < clientArray.length; i++){
            Socket socket = clientArray[i].getSocket();
            if(socket == null){
                continue;
            }
            SocketAddress address = socket.getRemoteSocketAddress();
            if(address == null){
                continue;
            }
            result.add(address);
        }
        return result;
    }
    
    public int getClientSize(){
        return serverConnection.getClients().size();
    }
    
    public Set<SocketAddress> getEnabledClients(){
        if(serverConnection == null){
            return new HashSet<SocketAddress>();
        }
        Set<ServerConnectionImpl.ClientImpl> clients = serverConnection.getClients();
        ServerConnectionImpl.ClientImpl[] clientArray = clients.toArray(new ServerConnectionImpl.ClientImpl[clients.size()]);
        Set<SocketAddress> result = new HashSet<SocketAddress>();
        for(int i = 0; i < clientArray.length; i++){
            Socket socket = clientArray[i].getSocket();
            if(socket == null || !clientArray[i].isEnabled()){
                continue;
            }
            SocketAddress address = socket.getRemoteSocketAddress();
            if(address == null){
                continue;
            }
            result.add(address);
        }
        return result;
    }
    
    public Set<SocketAddress> getDisabledClients(){
        if(serverConnection == null){
            return new HashSet<SocketAddress>();
        }
        Set<ServerConnectionImpl.ClientImpl> clients = serverConnection.getClients();
        ServerConnectionImpl.ClientImpl[] clientArray = clients.toArray(new ServerConnectionImpl.ClientImpl[clients.size()]);
        Set<SocketAddress> result = new HashSet<SocketAddress>();
        for(int i = 0; i < clientArray.length; i++){
            Socket socket = clientArray[i].getSocket();
            if(socket == null || clientArray[i].isEnabled()){
                continue;
            }
            SocketAddress address = socket.getRemoteSocketAddress();
            if(address == null){
                continue;
            }
            result.add(address);
        }
        return result;
    }
    
    public void enabledClient(String address, int port){
        setEnabledClient(address, port, true);
    }
    
    public void disabledClient(String address, int port){
        setEnabledClient(address, port, false);
    }
    
    public Set<String> getSubjects(String address, int port){
        if(serverConnection == null){
            return new HashSet<String>();
        }
        Set<ServerConnectionImpl.ClientImpl> clients = serverConnection.getClients();
        ServerConnectionImpl.ClientImpl[] clientArray = clients.toArray(new ServerConnectionImpl.ClientImpl[clients.size()]);
        for(int i = 0; i < clientArray.length; i++){
            Socket socket = clientArray[i].getSocket();
            if(socket == null){
                continue;
            }
            InetSocketAddress remoteAddress = (InetSocketAddress)socket.getRemoteSocketAddress();
            if(remoteAddress == null){
                continue;
            }
            if((remoteAddress.getAddress().getHostAddress().equals(address)
                    || remoteAddress.getAddress().getHostName().equalsIgnoreCase(address))
                && port == remoteAddress.getPort()
            ){
                return clientArray[i].getSubjects();
            }
        }
        return new HashSet<String>();
    }
    
    public Set<String> getKeys(String address, int port, String subject){
        if(serverConnection == null){
            return new HashSet<String>();
        }
        Set<ServerConnectionImpl.ClientImpl> clients = serverConnection.getClients();
        ServerConnectionImpl.ClientImpl[] clientArray = clients.toArray(new ServerConnectionImpl.ClientImpl[clients.size()]);
        for(int i = 0; i < clientArray.length; i++){
            Socket socket = clientArray[i].getSocket();
            if(socket == null){
                continue;
            }
            InetSocketAddress remoteAddress = (InetSocketAddress)socket.getRemoteSocketAddress();
            if(remoteAddress == null){
                continue;
            }
            if((remoteAddress.getAddress().getHostAddress().equals(address)
                    || remoteAddress.getAddress().getHostName().equalsIgnoreCase(address))
                && port == remoteAddress.getPort()
            ){
                return clientArray[i].getKeys(subject);
            }
        }
        return new HashSet<String>();
    }
    
    private void setEnabledClient(String address, int port, boolean isEnabled){
        if(serverConnection == null){
            return;
        }
        Set<ServerConnectionImpl.ClientImpl> clients = serverConnection.getClients();
        ServerConnectionImpl.ClientImpl[] clientArray = clients.toArray(new ServerConnectionImpl.ClientImpl[clients.size()]);
        for(int i = 0; i < clientArray.length; i++){
            Socket socket = clientArray[i].getSocket();
            if(socket == null || clientArray[i].isEnabled() == isEnabled){
                continue;
            }
            InetSocketAddress remoteAddress = (InetSocketAddress)socket.getRemoteSocketAddress();
            if(remoteAddress == null){
                continue;
            }
            if((remoteAddress.getAddress().getHostAddress().equals(address)
                    || remoteAddress.getAddress().getHostName().equalsIgnoreCase(address))
                && (port <= 0 || port == remoteAddress.getPort())
            ){
                clientArray[i].setEnabled(isEnabled);
            }
        }
    }
    
    public Map<SocketAddress,Long> getSendCountsByClient(){
        if(serverConnection == null){
            return new HashMap<SocketAddress,Long>();
        }
        Set<ServerConnectionImpl.ClientImpl> clients = serverConnection.getClients();
        ServerConnectionImpl.ClientImpl[] clientArray = clients.toArray(new ServerConnectionImpl.ClientImpl[clients.size()]);
        Map<SocketAddress,Long> result = new HashMap<SocketAddress,Long>();
        for(int i = 0; i < clientArray.length; i++){
            Socket socket = clientArray[i].getSocket();
            if(socket == null){
                continue;
            }
            SocketAddress address = socket.getRemoteSocketAddress();
            if(address == null){
                continue;
            }
            result.put(address, new Long(clientArray[i].getSendCount()));
        }
        return result;
    }
    
    public Map<SocketAddress,Long> getAverageSendProcessTimesByClient(){
        if(serverConnection == null){
            return new HashMap<SocketAddress,Long>();
        }
        Set<ServerConnectionImpl.ClientImpl> clients = serverConnection.getClients();
        ServerConnectionImpl.ClientImpl[] clientArray = clients.toArray(new ServerConnectionImpl.ClientImpl[clients.size()]);
        Map<SocketAddress,Long> result = new HashMap<SocketAddress,Long>();
        for(int i = 0; i < clientArray.length; i++){
            Socket socket = clientArray[i].getSocket();
            if(socket == null){
                continue;
            }
            SocketAddress address = socket.getRemoteSocketAddress();
            if(address == null){
                continue;
            }
            result.put(address, new Long(clientArray[i].getAverageSendProcessTime()));
        }
        return result;
    }
    
    public void resetSendCountsByClient(){
        if(serverConnection == null){
            return;
        }
        Set<ServerConnectionImpl.ClientImpl> clients = serverConnection.getClients();
        ServerConnectionImpl.ClientImpl[] clientArray = clients.toArray(new ServerConnectionImpl.ClientImpl[clients.size()]);
        for(int i = 0; i < clientArray.length; i++){
            clientArray[i].resetSendCount();
        }
    }
    
    public void startService() throws Exception{
        if(clientReconnectCount > 0 && serverPort == 0){
            throw new IllegalArgumentException("When clientReconnectCount is more than 0, serverPort must not be 0.");
        }
        if(serverAddress == null){
            serverAddress = InetAddress.getLocalHost().getHostAddress();
        }
        if(externalizerServiceName != null){
            externalizer = ServiceManagerFactory
                .getServiceObject(externalizerServiceName);
        }
        if(isNIO){
            if(nioSocketFactoryServiceName != null){
                nioSocketFactory = ServiceManagerFactory
                    .getServiceObject(nioSocketFactoryServiceName);
            }
            ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.socket().setReuseAddress(true);
            serverSocketChannel.socket().bind(new InetSocketAddress(serverAddress, serverPort));
            if(serverPort == 0){
                serverPort = serverSocketChannel.socket().getLocalPort();
            }
            serverSocketChannel.configureBlocking(false);
            serverConnection = new ServerConnectionImpl(
                serverSocketChannel,
                externalizer,
                sendThreadSize,
                sendQueueServiceName,
                asynchSendThreadSize,
                asynchSendQueueFactoryServiceName,
                nioSocketFactory
            );
        }else{
        
            ServerSocket serverSocket = null;
            if(serverSocketFactory == null){
                if(serverSocketFactoryServiceName == null){
                    serverSocket = new ServerSocket(serverPort, serverBacklog, InetAddress.getByName(serverAddress)); 
                }else{
                    serverSocketFactory = ServiceManagerFactory
                        .getServiceObject(serverSocketFactoryServiceName);
                }
            }
            if(serverSocket == null){
                serverSocket = serverSocketFactory.createServerSocket(
                    serverPort,
                    serverBacklog,
                    InetAddress.getByName(serverAddress)
                );
            }
            if(serverPort == 0){
                serverPort = serverSocket.getLocalPort();
            }
            serverConnection = new ServerConnectionImpl(
                serverSocket,
                externalizer,
                sendThreadSize,
                sendQueueServiceName,
                asynchSendThreadSize,
                asynchSendQueueFactoryServiceName
            );
        }
        serverConnection.setLogger(getLogger());
        serverConnection.setMaxSendRetryCount(maxSendRetryCount);
        serverConnection.setSendBufferTime(sendBufferTime);
        serverConnection.setSendErrorMessageId(serverSendErrorMessageId);
        serverConnection.setSendErrorRetryOverMessageId(serverSendErrorRetryOverMessageId);
        if(serverConnectionListenerServiceNames != null){
            for(int i = 0; i < serverConnectionListenerServiceNames.length; i++){
                serverConnection.addServerConnectionListener(
                    (ServerConnectionListener)ServiceManagerFactory
                        .getServiceObject(serverConnectionListenerServiceNames[i])
                );
            }
        }
        if(serverConnectionListeners != null){
            for(int i = 0, imax = serverConnectionListeners.size(); i < imax; i++){
                serverConnection.addServerConnectionListener(
                    serverConnectionListeners.get(i)
                );
            }
        }
        
        if(socketFactoryServiceName != null){
            socketFactory = ServiceManagerFactory
                .getServiceObject(socketFactoryServiceName);
        }
        if(jndiRepositoryServiceName != null){
            jndiRepository = ServiceManagerFactory
                .getServiceObject(jndiRepositoryServiceName);
            
            RemoteClientConnectionFactory remoteClientConnectionFactory = new RemoteClientConnectionFactory(
                this,
                rmiPort
            );
            if(!jndiRepository.register(jndiName, remoteClientConnectionFactory)){
                throw new Exception("Could not register in jndiRepository.");
            }
        }
    }
    
    public void stopService() throws Exception{
        if(jndiRepository != null){
            jndiRepository.unregister(jndiName);
        }
        if(serverConnection != null){
            serverConnection.close();
            serverConnection = null;
        }
    }
    
    public ServerConnection getServerConnection() throws ConnectionCreateException{
        return serverConnection;
    }
    
    public ClientConnection getClientConnection() throws ConnectionCreateException{
        ClientConnectionImpl connection = new ClientConnectionImpl(serverAddress, serverPort, socketFactory, externalizer, getServiceNameObject());
        if(clientAddressProertyName != null){
            connection.setBindAddressProertyName(clientAddressProertyName);
        }
        if(clientPortProertyName != null){
            connection.setBindPortProertyName(clientPortProertyName);
        }
        connection.setServerCloseMessageId(clientServerCloseMessageId);
        connection.setReceiveWarnMessageId(clientReceiveWarnMessageId);
        connection.setReceiveErrorMessageId(clientReceiveErrorMessageId);
        connection.setReconnectCount(clientReconnectCount);
        connection.setReconnectInterval(clientReconnectInterval);
        connection.setReconnectBufferTime(clientReconnectBufferTime);
        return connection;
    }
    
    public int getClientCount(){
        return serverConnection == null ? 0 : serverConnection.getClientCount();
    }
}