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

import java.util.Collections;
import java.util.Set;
import java.util.HashSet;
import java.util.List;
import java.util.ArrayList;
import java.util.Map;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Iterator;
import java.io.Serializable;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.io.ObjectInputStream;
import java.rmi.RemoteException;

import jp.ossc.nimbus.core.Service;
import jp.ossc.nimbus.core.ServiceManagerFactory;
import jp.ossc.nimbus.service.distribute.ClusterService;
import jp.ossc.nimbus.service.distribute.ClusterListener;

/**
 * {@link ClientConnection}をクラスタ化するClientConnectionインタフェース実装クラス。<p>
 * 
 * @author M.Takata
 */
public class ClusterClientConnectionImpl implements ClientConnection, ClusterListener, MessageListener, Serializable{
    
    private static final long serialVersionUID = 4277728721026624133L;
    
    private transient Object uid;
    private transient ClusterService cluster;
    private String connectErrorMessageId;
    private String reconnectMessageId;
    private String noConnectErrorMessageId;
    private long failoverBufferTime;
    private transient boolean isConnected;
    private transient boolean isConnecting;
    private transient List<ClusterService.GlobalUID> members;
    private transient Map<ClusterService.GlobalUID, ClusterConnectionFactoryService.ClusterOption> connectionMap;
    private transient Map<String,Set<String>> subjects;
    private transient MessageListener messageListener;
    private transient Object id;
    private transient String serviceManagerName;
    private boolean isDistribute;
    private boolean isMultiple;
    private boolean isReceiveOwnMessage;
    private boolean isFlexibleConnect;
    private transient Object currentUID;
    private transient boolean isStartReceive;
    private transient long fromTime;
    private transient Message latestMessage;
    
    public ClusterClientConnectionImpl(ClusterService cluster){
        setCluster(cluster);
    }
    
    public void setCluster(ClusterService cluster){
        this.cluster = cluster;
        uid = this.cluster == null ? null : this.cluster.getUID();
    }
    
    public void setConnectErrorMessageId(String id){
        connectErrorMessageId = id;
    }
    
    public void setReconnectMessageId(String id){
        reconnectMessageId = id;
    }
    
    public void setNoConnectErrorMessageId(String id){
        noConnectErrorMessageId = id;
    }
    
    public void setServiceManagerName(String name){
        serviceManagerName = name;
    }
    
    public void setDistribute(boolean isDistribute){
        this.isDistribute = isDistribute;
    }
    
    public void setMultiple(boolean isMultiple){
        this.isMultiple = isMultiple;
    }
    
    public void setFailoverBufferTime(long time){
        failoverBufferTime = time;
    }
    
    public void setReceiveOwnMessage(boolean isReceive){
        this.isReceiveOwnMessage = isReceive;
    }
    
    public void setFlexibleConnect(boolean isFlexible){
        isFlexibleConnect = isFlexible;
    }
    
    public synchronized void connect() throws ConnectException{
        connect((Object)null);
    }
    
    public synchronized void connect(Object id) throws ConnectException{
        if(isConnected){
            return;
        }
        isConnecting = true;
        try{
            if(cluster.getState() != Service.State.STARTED){
                try{
                    cluster.create();
                    cluster.setClient(true);
                    cluster.addClusterListener(this);
                    cluster.start();
                    this.id = id == null ? cluster.getUID() : id;
                    cluster.join();
                }catch(Exception e){
                    cluster.stop();
                    cluster.destroy();
                    throw new ConnectException(e);
                }
            }else{
                this.id = id == null ? cluster.getUID() : id;
                cluster.addClusterListener(this);
            }
            if(!isFlexibleConnect && (connectionMap == null || connectionMap.size() == 0)){
                throw new ConnectException("No cluster member.");
            }
            isConnected = true;
        }finally{
            isConnecting = false;
        }
    }
    
    public List<ClusterService.GlobalUID> getClusterMembers(){
        return members;
    }
    
    public ClientConnection getClusterClientConnection(ClusterService.GlobalUID member){
        if(uid != null && uid.equals(member)){
            return this;
        }
        ClusterConnectionFactoryService.ClusterOption clusterOption = connectionMap.get(member);
        return clusterOption == null ? null : clusterOption.clientConnection;
    }
    
    @SuppressWarnings("unchecked")
    private void updateConnectionList(){
        List<ClusterService.GlobalUID> memberList = (List<ClusterService.GlobalUID>)cluster.getMembers();
        List<ClusterService.GlobalUID> tmpMembers = new ArrayList<ClusterService.GlobalUID>();
        Map<ClusterService.GlobalUID, ClusterConnectionFactoryService.ClusterOption> tmpConnectionMap = new LinkedHashMap<ClusterService.GlobalUID, ClusterConnectionFactoryService.ClusterOption>();
        ClusterService.GlobalUID[] members = (ClusterService.GlobalUID[])memberList.toArray(new ClusterService.GlobalUID[memberList.size()]);
        for(int i = 0; i < members.length; i++){
            ClusterConnectionFactoryService.ClusterOption clusterOption = null;
            if(connectionMap != null && connectionMap.containsKey(members[i])){
                clusterOption = connectionMap.get(members[i]);
            }else{
                clusterOption = (ClusterConnectionFactoryService.ClusterOption)members[i].getOption();
            }
            if(clusterOption != null){
                if(messageListener != null
                    && (isReceiveOwnMessage
                        || (!isReceiveOwnMessage  && uid != null && !uid.equals(members[i])))){
                    clusterOption.clientConnection.setMessageListener(this);
                }
                tmpMembers.add(members[i]);
                if(uid != null && !uid.equals(members[i])){
                    tmpConnectionMap.put(members[i], clusterOption);
                }
            }
        }
        connectionMap = tmpConnectionMap;
        this.members = tmpMembers;
    }
    
    public synchronized void addSubject(String subject) throws MessageSendException{
        addSubject(subject, null);
    }
    
    public synchronized void addSubject(String subject, String[] keys) throws MessageSendException{
        if(!isConnected){
            throw new MessageSendException("Not connected.");
        }
        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]);
            }
        }
        if(connectionMap != null){
            if(isMultiple){
                Iterator<ClusterConnectionFactoryService.ClusterOption> connections = connectionMap.values().iterator();
                while(connections.hasNext()){
                    ClientConnection connection = connections.next().clientConnection;
                    connection.addSubject(subject, keys);
                }
            }else if(currentUID != null){
                ClientConnection connection = connectionMap.get(currentUID).clientConnection;
                connection.addSubject(subject, keys);
            }
        }
    }
    
    public synchronized void removeSubject(String subject) throws MessageSendException{
        removeSubject(subject, null);
    }
    
    public synchronized void removeSubject(String subject, String[] keys) throws MessageSendException{
        if(!isConnected){
            throw new MessageSendException("Not connected.");
        }
        if(connectionMap != null){
            if(isMultiple){
                Iterator<ClusterConnectionFactoryService.ClusterOption> connections = connectionMap.values().iterator();
                while(connections.hasNext()){
                    ClientConnection connection = connections.next().clientConnection;
                    connection.removeSubject(subject, keys);
                }
            }else if(currentUID != null){
                ClientConnection connection = connectionMap.get(currentUID).clientConnection;
                connection.removeSubject(subject, keys);
            }
        }
        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 synchronized void startReceive() throws MessageSendException{
        startReceive(-1);
    }
    
    public synchronized void startReceive(long from) throws MessageSendException{
        if(!isConnected){
            throw new MessageSendException("Not connected.");
        }
        if(connectionMap != null){
            if(isMultiple){
                Iterator<ClusterConnectionFactoryService.ClusterOption> connections = connectionMap.values().iterator();
                while(connections.hasNext()){
                    ClientConnection connection = connections.next().clientConnection;
                    connection.startReceive(from);
                }
            }else if(currentUID != null){
                ClientConnection connection = connectionMap.get(currentUID).clientConnection;
                connection.startReceive(from);
            }
        }
        isStartReceive = true;
        fromTime = from;
    }
    
    public boolean isStartReceive(){
        return isStartReceive;
    }
    
    public synchronized void stopReceive() throws MessageSendException{
        if(!isConnected){
            throw new MessageSendException("Not connected.");
        }
        if(connectionMap != null){
            if(isMultiple){
                Iterator<ClusterConnectionFactoryService.ClusterOption> connections = connectionMap.values().iterator();
                while(connections.hasNext()){
                    ClientConnection connection = connections.next().clientConnection;
                    connection.stopReceive();
                }
            }else if(currentUID != null){
                ClientConnection connection = connectionMap.get(currentUID).clientConnection;
                connection.stopReceive();
            }
        }
    }
    
    public Set<String> getSubjects(){
        final Set<String> result = new HashSet<String>();
        if(connectionMap != null){
            if(isMultiple){
                Iterator<ClusterConnectionFactoryService.ClusterOption> connections = connectionMap.values().iterator();
                while(connections.hasNext()){
                    ClientConnection connection = connections.next().clientConnection;
                    result.addAll(connection.getSubjects());
                    break;
                }
            }else if(currentUID != null){
                ClientConnection connection = connectionMap.get(currentUID).clientConnection;
                result.addAll(connection.getSubjects());
            }
        }
        return result;
    }
    
    public Set<String> getKeys(String subject){
        final Set<String> result = new HashSet<String>();
        if(connectionMap != null){
            if(isMultiple){
                Iterator<ClusterConnectionFactoryService.ClusterOption> connections = connectionMap.values().iterator();
                while(connections.hasNext()){
                    ClientConnection connection = connections.next().clientConnection;
                    result.addAll(connection.getKeys(subject));
                    break;
                }
            }else if(currentUID != null){
                ClientConnection connection = connectionMap.get(currentUID).clientConnection;
                result.addAll(connection.getKeys(subject));
            }
        }
        return result;
    }
    
    public synchronized void setMessageListener(MessageListener listener){
        messageListener = listener;
        if(connectionMap != null){
            if(isMultiple){
                Iterator<ClusterConnectionFactoryService.ClusterOption> connections = connectionMap.values().iterator();
                while(connections.hasNext()){
                    ClientConnection connection = connections.next().clientConnection;
                    connection.setMessageListener(this);
                }
            }else if(currentUID != null){
                ClientConnection connection = connectionMap.get(currentUID).clientConnection;
                connection.setMessageListener(this);
            }
        }
    }
    
    public synchronized boolean isConnected(){
        return isConnected;
    }
    
    public Object getId(){
        return id;
    }
    
    public synchronized void close(){
        if(!isConnected){
            return;
        }
        id = null;
        currentUID = null;
        cluster.removeClusterListener(this);
        cluster.stop();
        if(connectionMap != null){
            List<ClusterConnectionFactoryService.ClusterOption> connections = new ArrayList<ClusterConnectionFactoryService.ClusterOption>(connectionMap.values());
            for(int i = 0, imax = connections.size(); i < imax; i++){
                ClientConnection connection = connections.get(i).clientConnection;
                connection.close();
            }
            connectionMap = null;
        }
        isConnected = false;
    }
    
    public void onMessage(Message message){
        if(messageListener != null){
            latestMessage = message;
            messageListener.onMessage(message);
        }
    }
    
    private synchronized boolean connect(ClientConnection connection) throws MessageCommunicateException{
        connection.setServiceManagerName(serviceManagerName);
        if(!connection.isConnected()){
            connection.connect(id);
            return true;
        }
        return false;
    }
    
    private synchronized void addSubject(ClientConnection connection) throws MessageCommunicateException{
        if(subjects != null){
            Object[] subjectArray = subjects.keySet().toArray();
            for(int j = 0; j < subjectArray.length; j++){
                Object subject = subjectArray[j];
                Set<String> keySet = subjects.get(subject);
                if(keySet != null){
                    String[] keys = keySet.toArray(new String[keySet.size()]);
                    boolean containsNull = false;
                    List<String> keyList = new ArrayList<String>();
                    for(int k = 0; k < keys.length; k++){
                        if(keys[k] == null){
                            containsNull = true;
                        }else{
                            keyList.add(keys[k]);
                        }
                    }
                    if(containsNull){
                        connection.addSubject((String)subject);
                        keys = keyList.toArray(new String[keyList.size()]);
                    }
                    if(keys != null && keys.length != 0){
                        connection.addSubject((String)subject, keys);
                    }
                }
            }
        }
    }
    
    private synchronized void startReceive(ClientConnection connection) throws MessageCommunicateException{
        if(isStartReceive && !connection.isStartReceive()){
            if(isMultiple){
                connection.startReceive(-1l);
            }else{
                long time = fromTime;
                if(latestMessage != null){
                    time = latestMessage.getReceiveTime() - failoverBufferTime;
                }
                connection.startReceive(time);
            }
        }
    }
    
    public synchronized void memberInit(Object myId, List<? extends Object> members){
        updateConnectionList();
        Object member = null;
        if(!isConnected && !isConnecting){
            return;
        }
        if(isMultiple){
            Iterator<ClusterConnectionFactoryService.ClusterOption> connections = connectionMap.values().iterator();
            while(connections.hasNext()){
                ClientConnection connection = connections.next().clientConnection;
                try{
                    if(!connection.isConnected() && connect(connection) && isConnected && reconnectMessageId != null){
                        ServiceManagerFactory.getLogger().write(
                            reconnectMessageId,
                            (Object)null,
                            connection
                        );
                    }
                    addSubject(connection);
                    startReceive(connection);
                }catch(MessageCommunicateException e){
                    if(connectErrorMessageId != null){
                        ServiceManagerFactory.getLogger().write(
                            connectErrorMessageId,
                            e,
                            connection
                        );
                    }
                }
            }
        }else{
            if(isDistribute){
                Iterator<Map.Entry<ClusterService.GlobalUID, ClusterConnectionFactoryService.ClusterOption>> entries = connectionMap.entrySet().iterator();
                int cilentCount = 0;
                while(entries.hasNext()){
                    Map.Entry<ClusterService.GlobalUID, ClusterConnectionFactoryService.ClusterOption> entry = entries.next();
                    ClientConnectionFactory factory = entry.getValue().clusterClientConnectionFactory;
                    int count = 0;
                    try{
                        count = factory.getClientCount();
                    }catch(RemoteException e){
                        continue;
                    }
                    if(member == null || cilentCount > count){
                        cilentCount = count;
                        member = entry.getKey();
                    }
                }
            }else{
                if(connectionMap.size() != 0){
                    member = connectionMap.keySet().iterator().next();
                }
            }
            if(member != null){
                ClusterConnectionFactoryService.ClusterOption option = connectionMap.get(member);
                ClientConnection connection = option.clientConnection;
                try{
                    if(!connection.isConnected() && connect(connection) && isConnected && reconnectMessageId != null){
                        ServiceManagerFactory.getLogger().write(
                            reconnectMessageId,
                            (Object)null,
                            connection
                        );
                    }
                    id = connection.getId();
                    addSubject(connection);
                    startReceive(connection);
                    currentUID = member;
                }catch(MessageCommunicateException e){
                    if(connectErrorMessageId != null){
                        ServiceManagerFactory.getLogger().write(
                            connectErrorMessageId,
                            e,
                            connection
                        );
                    }
                }
            }
        }
    }
    
    public synchronized void memberChange(List<? extends Object> oldMembers, List<? extends Object> newMembers){
        if(isMultiple){
            Set<Object> removedMembers = new HashSet<Object>(oldMembers);
            removedMembers.removeAll(newMembers);
            Iterator<Object> rmMembers = removedMembers.iterator();
            while(rmMembers.hasNext()){
                ClusterService.GlobalUID rmMember = (ClusterService.GlobalUID)rmMembers.next();
                ClusterConnectionFactoryService.ClusterOption option = (ClusterConnectionFactoryService.ClusterOption)connectionMap.get(rmMember);
                if(option != null){
                    option.clientConnection.close();
                }
            }
            updateConnectionList();
            Iterator<ClusterConnectionFactoryService.ClusterOption> connections = connectionMap.values().iterator();
            while(connections.hasNext()){
                ClientConnection connection = connections.next().clientConnection;
                try{
                    if(!connection.isConnected() && connect(connection) && reconnectMessageId != null){
                        ServiceManagerFactory.getLogger().write(
                            reconnectMessageId,
                            (Object)null,
                            connection
                        );
                    }
                    addSubject(connection);
                    startReceive(connection);
                }catch(MessageCommunicateException e){
                    if(connectErrorMessageId != null){
                        ServiceManagerFactory.getLogger().write(
                            connectErrorMessageId,
                            e,
                            connection
                        );
                    }
                }
            }
        }else{
            ClientConnection currentConnection = null;
            if(currentUID != null && connectionMap.containsKey(currentUID)){
                currentConnection = ((ClusterConnectionFactoryService.ClusterOption)connectionMap.get(currentUID)).clientConnection;
            }
            updateConnectionList();
            if(connectionMap.size() == 0){
                if((isConnected || isConnecting) && noConnectErrorMessageId != null){
                    ServiceManagerFactory.getLogger().write(
                        noConnectErrorMessageId,
                        this
                    );
                }
                if(currentConnection != null){
                    currentConnection.close();
                }
                id = null;
                currentUID = null;
                return;
            }
            if(!isConnected && !isConnecting){
                return;
            }
            Object member = null;
            if(isDistribute){
                if(currentUID == null || !connectionMap.containsKey(currentUID)){
                    int cilentCount = 0;
                    Iterator<Map.Entry<ClusterService.GlobalUID, ClusterConnectionFactoryService.ClusterOption>> entries = connectionMap.entrySet().iterator();
                    while(entries.hasNext()){
                        Map.Entry<ClusterService.GlobalUID, ClusterConnectionFactoryService.ClusterOption> entry = entries.next();
                        ClientConnectionFactory factory = entry.getValue().clusterClientConnectionFactory;
                        int count = 0;
                        try{
                            count = factory.getClientCount();
                        }catch(RemoteException e){
                            continue;
                        }
                        if(member == null || cilentCount > count){
                            cilentCount = count;
                            member = entry.getKey();
                        }
                    }
                }else if(currentConnection != null && !currentConnection.isConnected()){
                    member = currentUID;
                }
            }else{
                Object firstMember = connectionMap.keySet().iterator().next();
                if(currentUID == null || !currentUID.equals(firstMember)){
                    member = firstMember;
                }else if(currentConnection != null && !currentConnection.isConnected()){
                    member = currentUID;
                }
            }
            if(member != null && !member.equals(currentUID)){
                ClusterConnectionFactoryService.ClusterOption option = connectionMap.get(member);
                ClientConnection connection = option.clientConnection;
                String currentConnectionStr = null;
                if(currentConnection != null){
                    currentConnectionStr = currentConnection.toString();
                    currentConnection.close();
                }
                try{
                    if(!connection.isConnected() && connect(connection) && reconnectMessageId != null){
                        ServiceManagerFactory.getLogger().write(
                            reconnectMessageId,
                            currentConnectionStr,
                            connection
                        );
                    }
                    id = connection.getId();
                    addSubject(connection);
                    startReceive(connection);
                    currentUID = member;
                }catch(MessageCommunicateException e){
                    if(connectErrorMessageId != null){
                        ServiceManagerFactory.getLogger().write(
                            connectErrorMessageId,
                            e,
                            connection
                        );
                    }
                }
            }
        }
    }
    
    public void changeMain() throws Exception{}
    
    public void changeSub(){}
    
    public String toString(){
        final StringBuilder buf = new StringBuilder();
        buf.append(super.toString());
        buf.append('{');
        buf.append("id=").append(id);
        buf.append(", connectionMap=").append(connectionMap);
        buf.append(", subjects=").append(subjects);
        buf.append('}');
        return buf.toString();
    }
    
    private void writeObject(ObjectOutputStream out) throws IOException{
        out.defaultWriteObject();
        out.writeObject(cluster != null ? cluster.createClient() : null);
    }
    
    private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException{
        in.defaultReadObject();
        cluster = (ClusterService)in.readObject();
    }
}