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

import java.io.*;
import java.net.*;
import java.util.*;

import jp.ossc.nimbus.core.*;
import jp.ossc.nimbus.daemon.*;
import jp.ossc.nimbus.service.queue.Queue;
import jp.ossc.nimbus.service.queue.DefaultQueueService;
import jp.ossc.nimbus.util.SynchronizeMonitor;
import jp.ossc.nimbus.util.WaitSynchronizeMonitor;

/**
 * クラスタサービス。<p>
 * 分散環境下で、複数のサーバ上にいる特定のサービスが、１つだけ起動している状態にするクラスタ構成サービスである。<br>
 *
 * @author M.Takata
 */
public class ClusterService extends ServiceBase implements ClusterServiceMBean{
    
    private static final long serialVersionUID = 4503189967951662029L;
    
    protected static final int MESSAGE_ID_ADD_REQ = 1;
    protected static final int MESSAGE_ID_MEMBER_CHANGE_REQ = 2;
    protected static final int MESSAGE_ID_MEMBER_MERGE_REQ = 3;
    protected static final int MESSAGE_ID_MEMBER_MERGE_RES = 4;
    protected static final int MESSAGE_ID_MAIN_HELLO_REQ = 5;
    protected static final int MESSAGE_ID_MAIN_REQ = 6;
    protected static final int MESSAGE_ID_MAIN_RES = 7;
    protected static final int MESSAGE_ID_HELLO_REQ = 8;
    protected static final int MESSAGE_ID_HELLO_RES = 9;
    protected static final int MESSAGE_ID_BYE_REQ = 10;
    
    protected ServiceName targetServiceName;
    protected ServiceName[] clusterListenerServiceNames;
    protected List<ClusterListener> listeners;
    
    protected String multicastGroupAddress;
    protected int multicastPort = 1500;
    protected int timeToLive = -1;
    protected String localAddress;
    protected String[] unicastMemberAddresses;
    protected int unicastPort = 1500;
    
    protected int receiveBufferSize = 1024;
    protected long heartBeatInterval = 1000;
    protected long heartBeatResponseTimeout = 500;
    protected int heartBeatRetryCount = 1;
    protected long addMemberResponseTimeout = 500;
    protected boolean isClient;
    protected long lostTimeout = 500;
    
    protected transient GlobalUID uid;
    protected transient InetAddress group;
    protected transient DatagramSocket socket;
    protected transient Daemon clusterMessageReceiver;
    protected transient Daemon heartBeater;
    protected transient Daemon eventHandler;
    protected transient Queue<ClusterEvent> eventQueue;
    protected transient boolean isMain;
    protected transient boolean isMainDoubt;
    
    protected transient List<GlobalUID> members;
    protected transient Set<GlobalUID> clientMembers;
    protected transient List<InetAddress> unicastMembers;
    
    protected final SynchronizeMonitor addMonitor = new WaitSynchronizeMonitor();
    
    protected transient boolean isMainRequesting;
    protected transient Set<GlobalUID> mainReqMembers;
    
    protected final SynchronizeMonitor helloMonitor = new WaitSynchronizeMonitor();
    protected transient GlobalUID helloTarget;
    protected transient Serializable option;
    protected transient boolean isJoinOnStart = true;
    protected transient boolean isJoin;
    protected transient boolean isJoining;
    protected final String sequenceLock = "SEQUENCE";
    protected transient int currentSequence;
    protected transient int maxWindowCount;
    
    public void setTargetServiceName(ServiceName name){
        targetServiceName = name;
    }
    public ServiceName getTargetServiceName(){
        return targetServiceName;
    }
    
    public void setClusterListenerServiceNames(ServiceName[] names){
        clusterListenerServiceNames = names;
    }
    public ServiceName[] getClusterListenerServiceNames(){
        return clusterListenerServiceNames;
    }
    
    public void setMulticastGroupAddress(String ip){
        multicastGroupAddress = ip;
    }
    public String getMulticastGroupAddress(){
        return multicastGroupAddress;
    }
    
    public void setMulticastPort(int port){
        multicastPort = port;
    }
    public int getMulticastPort(){
        return multicastPort;
    }
    
    public void setUnicastMemberAddresses(String[] addresses){
        unicastMemberAddresses = addresses;
    }
    public String[] getUnicastMemberAddresses(){
        return unicastMemberAddresses;
    }
    
    public void setUnicastPort(int port){
        unicastPort = port;
    }
    public int getUnicastPort(){
        return unicastPort;
    }
    
    public void setReceiveBufferSize(int size){
        receiveBufferSize = size;
    }
    public int getReceiveBufferSize(){
        return receiveBufferSize;
    }
    
    public void setTimeToLive(int ttl){
        timeToLive = ttl;
    }
    public int getTimeToLive(){
        return timeToLive;
    }
    
    public void setLocalAddress(String ip){
        localAddress = ip;
    }
    public String getLocalAddress(){
        return localAddress;
    }
    
    public void setOption(Serializable opt){
        option = opt;
        if(uid != null){
            uid.setOption(opt);
        }
    }
    public Serializable getOption(){
        return option;
    }
    
    public void setHeartBeatInterval(long interval){
        heartBeatInterval = interval;
    }
    public long getHeartBeatInterval(){
        return heartBeatInterval;
    }
    
    public void setHeartBeatResponseTimeout(long timeout){
        heartBeatResponseTimeout = timeout;
    }
    public long getHeartBeatResponseTimeout(){
        return heartBeatResponseTimeout;
    }
    public void setHeartBeatRetryCount(int count){
        heartBeatRetryCount = count;
    }
    public int getHeartBeatRetryCount(){
        return heartBeatRetryCount;
    }
    
    public void setAddMemberResponseTimeout(long timeout){
        addMemberResponseTimeout = timeout;
    }
    public long getAddMemberResponseTimeout(){
        return addMemberResponseTimeout;
    }
    
    public void setLostTimeout(long timeout){
        lostTimeout = timeout;
    }
    public long getLostTimeout(){
        return lostTimeout;
    }
    
    public void setClient(boolean isClient){
        this.isClient = isClient;
    }
    public boolean isClient(){
        return isClient;
    }
    
    public void setJoinOnStart(boolean isJoin){
        isJoinOnStart = isJoin;
    }
    public boolean isJoinOnStart(){
        return isJoinOnStart;
    }
    
    public boolean isMain(){
        return isMain;
    }
    public List<? extends Object> getMembers(){
        synchronized(members){
            return new ArrayList<GlobalUID>(members);
        }
    }
    public Object getUID(){
        return uid;
    }
    
    public boolean isJoin(){
        return isJoin;
    }
    
    public void addClusterListener(ClusterListener listener){
        if(getState() == State.STARTED){
            if(isJoin()){
                try{
                    listener.memberInit(isClient ? null : uid, new ArrayList<GlobalUID>(members));
                }catch(Exception e){
                }
                try{
                    if(isMain){
                        listener.changeMain();
                    }else{
                        listener.changeSub();
                    }
                }catch(Exception e){
                }
            }
            synchronized(listeners){
                List<ClusterListener> tmp = new ArrayList<ClusterListener>(listeners);
                tmp.add(listener);
                listeners = tmp;
            }
        }else{
            listeners.add(listener);
        }
    }
    
    public void removeClusterListener(ClusterListener listener){
        if(getState() == State.STARTED){
            synchronized(listeners){
                List<ClusterListener> tmp = new ArrayList<ClusterListener>(listeners);
                tmp.remove(listener);
                listeners = tmp;
            }
        }else{
            listeners.add(listener);
        }
    }
    
    public int getMaxWindowCount(){
        return maxWindowCount;
    }
    
    public void createService() throws Exception{
        members = Collections.synchronizedList(new ArrayList<GlobalUID>());
        clientMembers = Collections.synchronizedSet(new HashSet<GlobalUID>());
        mainReqMembers = Collections.synchronizedSet(new HashSet<GlobalUID>());
        listeners = new ArrayList<ClusterListener>();
        unicastMembers = Collections.synchronizedList(new ArrayList<InetAddress>());
    }
    
    public void startService() throws Exception{
        
        if(clusterListenerServiceNames != null){
            for(int i = 0; i < clusterListenerServiceNames.length; i++){
                listeners.add(
                    (ClusterListener)ServiceManagerFactory.getServiceObject(
                        clusterListenerServiceNames[i]
                    )
                );
            }
        }
        
        if(targetServiceName != null){
            DefaultClusterListenerService listener = new DefaultClusterListenerService();
            listener.setTargetServiceName(targetServiceName);
            listener.setClusterService(this);
            listener.create();
            listener.start();
            listeners.add(listener);
        }
        
        uid = new GlobalUID(localAddress, option);
        
        if(multicastGroupAddress == null && (unicastMemberAddresses == null || unicastMemberAddresses.length == 0)){
            throw new IllegalArgumentException("MulticastGroupAddress and UnicastMemberAddresses is null.");
        }
        
        eventQueue = new DefaultQueueService<ClusterEvent>();
        ((Service)eventQueue).create();
        ((Service)eventQueue).start();
        eventQueue.accept();
        
        connect();
        
        eventHandler = new Daemon(new EventHandler());
        eventHandler.setName(
            "Nimbus Cluster EventHandler " + getServiceNameObject()
        );
        eventHandler.start();
        
        clusterMessageReceiver = new Daemon(new MessageReceiver());
        clusterMessageReceiver.setName(
            "Nimbus Cluster MessageReceiver " + getServiceNameObject()
        );
        clusterMessageReceiver.start();
        
        heartBeater = new Daemon(new HeartBeater());
        heartBeater.setName(
            "Nimbus Cluster HeartBeater " + getServiceNameObject()
        );
        heartBeater.suspend();
        heartBeater.start();
        
        if(isJoinOnStart){
            join();
        }
    }
    
    public void stopService() throws Exception{
        
        heartBeater.stop(100);
        heartBeater = null;
        
        clusterMessageReceiver.stop(100);
        clusterMessageReceiver = null;
        
        eventHandler.stop(100);
        eventHandler = null;
        
        eventQueue.release();
        
        leave();
        
        if(socket != null){
            if(group != null && group.isMulticastAddress()){
                try{
                    ((MulticastSocket)socket).leaveGroup(group);
                }catch(IOException e){
                }
            }
            socket.close();
        }
        isMain = false;
        isMainDoubt = false;
        group = null;
        members.clear();
        clientMembers.clear();
        mainReqMembers.clear();
    }
    
    public void destroyService() throws Exception{
        uid = null;
        members = null;
        clientMembers = null;
        mainReqMembers = null;
    }
    
    private synchronized void connect() throws IOException{
        if(socket != null){
            socket.close();
        }
        if(multicastGroupAddress != null){
            group = InetAddress.getByName(multicastGroupAddress);
            final InetSocketAddress address = new InetSocketAddress(uid.getAddress(), multicastPort);
            socket = group.isMulticastAddress() ? new MulticastSocket(address) : new DatagramSocket(address);
            if(group.isMulticastAddress() && timeToLive >= 0){
                ((MulticastSocket)socket).setTimeToLive(timeToLive);
            }
            if(group.isMulticastAddress()){
                ((MulticastSocket)socket).joinGroup(group);
            }
        }else{
            unicastMembers.clear();
            for(int i = 0; i < unicastMemberAddresses.length; i++){
                final InetAddress unicastMemberAddress = InetAddress.getByName(unicastMemberAddresses[i]);
                unicastMembers.add(unicastMemberAddress);
            }
            socket = new DatagramSocket(new InetSocketAddress(uid.getAddress(), unicastPort));
        }
    }
    
    public synchronized void join() throws Exception{
        if(isJoin){
            return;
        }
        isJoining = true;
        try{
            synchronized(addMonitor){
                addMonitor.initMonitor();
                    sendMessage(MESSAGE_ID_ADD_REQ);
                addMonitor.waitMonitor(addMemberResponseTimeout);
            }
            if(members.size() == 0 || members.get(0).equals(uid)){
                if(!isClient && !members.contains(uid)){
                    synchronized(members){
                        members.add(uid);
                    }
                }
                processMemberInit(members);
                if(!isClient){
                    try{
                        synchronized(members){
                            isMain = true;
                            isMainDoubt = false;
                            isMainRequesting = false;
                            if(group == null){
                                synchronized(clientMembers){
                                    clientMembers.clear();
                                }
                            }
                            getLogger().write(
                                MSG_ID_CHANGE_OPERATION_SYSTEM,
                                getServiceNameObject()
                            );
                        }
                        processChangeMain();
                    }catch(Exception e){
                        isMain = false;
                        processChangeSub();
                        sendMessage(MESSAGE_ID_BYE_REQ);
                        clusterMessageReceiver.stop(100);
                        if(socket != null){
                            if(group != null && group.isMulticastAddress()){
                                try{
                                    ((MulticastSocket)socket).leaveGroup(group);
                                }catch(IOException e2){
                                }
                            }
                            socket.close();
                        }
                        isJoin = false;
                        throw e;
                    }
                }
            }else{
                processMemberInit(members);
                if(!isClient){
                    getLogger().write(
                        MSG_ID_CHANGE_STANDBY_SYSTEM,
                        getServiceNameObject()
                    );
                }
            }
        }finally{
            isJoining = false;
        }
        isJoin = true;
        heartBeater.resume();
    }
    
    public synchronized void leave(){
        if(!isJoin){
            return;
        }
        isJoin = false;
        if(heartBeater != null){
            heartBeater.suspend();
        }
        try{
            sendMessage(MESSAGE_ID_BYE_REQ);
        }catch(Exception e){
        }
        synchronized(members){
            members.clear();
        }
        if(!isClient){
            isMain = false;
            processChangeSub();
        }
        getLogger().write(
            MSG_ID_CHANGE_STANDBY_SYSTEM,
            getServiceNameObject()
        );
    }
    
    public ClusterService createClient(){
        ClusterService client = new ClusterService();
        client.multicastGroupAddress = multicastGroupAddress;
        client.multicastPort = multicastPort;
        client.timeToLive = timeToLive;
        client.unicastMemberAddresses = unicastMemberAddresses;
        client.unicastPort = unicastPort;
        client.receiveBufferSize = receiveBufferSize;
        client.heartBeatInterval = heartBeatInterval;
        client.heartBeatResponseTimeout = heartBeatResponseTimeout;
        client.heartBeatRetryCount = heartBeatRetryCount;
        client.addMemberResponseTimeout = addMemberResponseTimeout;
        client.lostTimeout = lostTimeout;
        client.isClient = true;
        return client;
    }
    
    protected void sendMessage(int messageId) throws IOException{
        sendMessage(messageId, null);
    }
    
    protected void sendMessage(int messageId, GlobalUID toUID) throws IOException{
        sendMessage(messageId, uid, toUID);
    }
    
    protected void sendMessage(
        int messageId,
        GlobalUID fromUID,
        GlobalUID toUID
    ) throws IOException{
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        ObjectOutputStream oos = new ObjectOutputStream(baos);
        try{
            oos.writeInt(messageId);
            switch(messageId){
            case MESSAGE_ID_MAIN_HELLO_REQ:
            case MESSAGE_ID_HELLO_REQ:
            case MESSAGE_ID_HELLO_RES:
            case MESSAGE_ID_BYE_REQ:
            case MESSAGE_ID_MEMBER_CHANGE_REQ:
            case MESSAGE_ID_MEMBER_MERGE_REQ:
            case MESSAGE_ID_MEMBER_MERGE_RES:
            case MESSAGE_ID_MAIN_RES:
                fromUID = (GlobalUID)fromUID.clone();
                fromUID.setOption(null);
                break;
            case MESSAGE_ID_ADD_REQ:
                break;
            default:
            }
            oos.writeObject(fromUID);
            if(toUID != null){
                toUID = (GlobalUID)toUID.clone();
                toUID.setOption(null);
            }
            oos.writeObject(toUID);
            switch(messageId){
            case MESSAGE_ID_MAIN_HELLO_REQ:
            case MESSAGE_ID_HELLO_RES:
                break;
            case MESSAGE_ID_ADD_REQ:
            case MESSAGE_ID_HELLO_REQ:
            case MESSAGE_ID_BYE_REQ:
                oos.writeBoolean(isClient);
                break;
            case MESSAGE_ID_MEMBER_CHANGE_REQ:
            case MESSAGE_ID_MEMBER_MERGE_REQ:
            case MESSAGE_ID_MEMBER_MERGE_RES:
                oos.writeInt(members.size());
                for(int i = 0, imax = members.size(); i < imax; i++){
                    oos.writeObject(members.get(i));
                }
                break;
            case MESSAGE_ID_MAIN_RES:
                oos.writeBoolean(!isMain);
                break;
            default:
            }
            oos.close();
            final byte[] bytes = baos.toByteArray();
            if(bytes.length <= 0){
                return;
            }
            List<DatagramPacket> packets = new ArrayList<DatagramPacket>();
            Window window = new Window();
            synchronized(sequenceLock){
                window.sequence = currentSequence++;
            }
            window.data = bytes;
            List<byte[]> windows = window.divide(receiveBufferSize);
            maxWindowCount = Math.max(maxWindowCount, windows.size());
            if(group == null){
                if(toUID == null){
                    Set<InetAddress> toMembers = new LinkedHashSet<InetAddress>();
                    toMembers.addAll(unicastMembers);
                    for(int i = 0, imax = members.size(); i < imax; i++){
                        toMembers.add(((GlobalUID)members.get(i)).getAddress());
                    }
                    switch(messageId){
                    case MESSAGE_ID_MEMBER_CHANGE_REQ:
                    case MESSAGE_ID_MEMBER_MERGE_RES:
                        Object[] clients = clientMembers.toArray();
                        for(int i = 0; i < clients.length; i++){
                            toMembers.add(((GlobalUID)clients[i]).getAddress());
                        }
                    }
                    InetAddress[] toAddresses = (InetAddress[])toMembers.toArray(new InetAddress[toMembers.size()]);
                    for(int i = 0; i < windows.size(); i++){
                        byte[] windowData = windows.get(i);
                        for(int j = 0; j < toAddresses.length; j++){
                            packets.add(
                                new DatagramPacket(
                                    windowData,
                                    windowData.length,
                                    toAddresses[j],
                                    unicastPort
                                )
                            );
                        }
                    }
                }else{
                    for(int i = 0; i < windows.size(); i++){
                        byte[] windowData = windows.get(i);
                        packets.add(
                            new DatagramPacket(
                                windowData,
                                windowData.length,
                                toUID.getAddress(),
                                unicastPort
                            )
                        );
                    }
                }
            }else{
                for(int i = 0; i < windows.size(); i++){
                    byte[] windowData = windows.get(i);
                    packets.add(
                        new DatagramPacket(
                            windowData,
                            windowData.length,
                            group,
                            multicastPort
                        )
                    );
                }
            }
            for(int i = 0; i < packets.size(); i++){
                socket.send(packets.get(i));
            }
        }finally{
            if(oos != null){
                oos.close();
            }
        }
    }
    
    protected void handleMessage(InputStream is){
        ObjectInputStream ois = null;
        try{
            ois = new ObjectInputStream(is);
            final int messageId = ois.readInt();
            final GlobalUID fromUID = (GlobalUID)ois.readObject();
            final GlobalUID toUID = (GlobalUID)ois.readObject();
            if(uid.equals(fromUID)
                 || (toUID != null && !uid.equals(toUID))){
                return;
            }
            int memberSize = 0;
            List<GlobalUID> newMembers = null;
            boolean isMemberChange = false;
            List<GlobalUID> tmpOldMembers = null;
            List<GlobalUID> tmpNewMembers = null;
            switch(messageId){
            case MESSAGE_ID_ADD_REQ:
                if(!isMain){
                    break;
                }
                if(ois.readBoolean()){
                    if(group == null){
                        synchronized(clientMembers){
                            clientMembers.add(fromUID);
                        }
                    }
                    sendMessage(MESSAGE_ID_MEMBER_CHANGE_REQ, fromUID);
                }else{
                    synchronized(members){
                        if(!members.contains(fromUID)){
                            tmpOldMembers = new ArrayList<GlobalUID>(members);
                            members.add(fromUID);
                            tmpNewMembers = new ArrayList<GlobalUID>(members);
                            isMemberChange = true;
                        }
                    }
                    sendMessage(MESSAGE_ID_MEMBER_CHANGE_REQ);
                    if(isMemberChange){
                        eventQueue.push(new ClusterEvent(ClusterEvent.EVENT_MEMBER_CHANGE, tmpOldMembers, tmpNewMembers));
                    }
                }
                break;
            case MESSAGE_ID_MEMBER_CHANGE_REQ:
                if(isMain
                    || (members.size() != 0 && members.indexOf(fromUID) != 0)){
                    break;
                }
                memberSize = ois.readInt();
                newMembers = new ArrayList<GlobalUID>();
                if(memberSize > 0){
                    for(int i = 0; i < memberSize; i++){
                        newMembers.add((GlobalUID)ois.readObject());
                    }
                }
                boolean isNotMember = false;
                synchronized(members){
                    if(!members.equals(newMembers)){
                        if(isClient || newMembers.contains(uid)){
                            tmpOldMembers = new ArrayList<GlobalUID>(members);
                            tmpNewMembers = new ArrayList<GlobalUID>(newMembers);
                            members = Collections.synchronizedList(newMembers);
                            isMemberChange = true;
                        }else{
                            isNotMember = true;
                        }
                    }
                }
                if(addMonitor.isWait() && members.contains(uid)){
                    addMonitor.notifyAllMonitor();
                    isMemberChange = false;
                }
                if(isNotMember){
                    synchronized(addMonitor){
                        addMonitor.initMonitor();
                        sendMessage(MESSAGE_ID_ADD_REQ);
                        try{
                            addMonitor.waitMonitor(addMemberResponseTimeout);
                        }catch(InterruptedException e){}
                    }
                }else if(isMemberChange && !isJoining){
                    eventQueue.push(new ClusterEvent(ClusterEvent.EVENT_MEMBER_CHANGE, tmpOldMembers, tmpNewMembers));
                }
                break;
            case MESSAGE_ID_MEMBER_MERGE_REQ:
                if(!isMain){
                    break;
                }
                memberSize = ois.readInt();
                newMembers = new ArrayList<GlobalUID>();
                if(memberSize > 0){
                    for(int i = 0; i < memberSize; i++){
                        newMembers.add((GlobalUID)ois.readObject());
                    }
                }
                synchronized(members){
                    newMembers.removeAll(members);
                    if(newMembers.size() != 0){
                        tmpOldMembers = new ArrayList<GlobalUID>(members);
                        tmpNewMembers = new ArrayList<GlobalUID>(newMembers);
                        members.addAll(newMembers);
                        isMemberChange = true;
                    }
                }
                sendMessage(MESSAGE_ID_MEMBER_MERGE_RES);
                if(isMemberChange){
                    eventQueue.push(new ClusterEvent(ClusterEvent.EVENT_MEMBER_CHANGE, tmpOldMembers, tmpNewMembers));
                }
                break;
            case MESSAGE_ID_MEMBER_MERGE_RES:
                memberSize = ois.readInt();
                newMembers = new ArrayList<GlobalUID>();
                if(memberSize > 0){
                    for(int i = 0; i < memberSize; i++){
                        newMembers.add((GlobalUID)ois.readObject());
                    }
                }
                if(!isClient && isMain && newMembers.indexOf(uid) != 0){
                    eventQueue.push(new ClusterEvent(ClusterEvent.EVENT_CHANGE_SUB));
                    isMain = false;
                    isMainDoubt = false;
                    isMainRequesting = false;
                    getLogger().write(
                        MSG_ID_CHANGE_STANDBY_SYSTEM,
                        getServiceNameObject()
                    );
                }
                synchronized(members){
                    if(newMembers.contains(uid)
                        && !members.equals(newMembers)){
                        tmpOldMembers = new ArrayList<GlobalUID>(members);
                        tmpNewMembers = new ArrayList<GlobalUID>(newMembers);
                        members = Collections.synchronizedList(newMembers);
                        isMemberChange = true;
                    }
                }
                if(isMemberChange){
                    eventQueue.push(new ClusterEvent(ClusterEvent.EVENT_MEMBER_CHANGE, tmpOldMembers, tmpNewMembers));
                }
                break;
            case MESSAGE_ID_MAIN_HELLO_REQ:
                if(!isMain){
                    if(members.size() == 1 && !addMonitor.isWait()){
                        sendMessage(MESSAGE_ID_ADD_REQ, fromUID);
                    }
                }else if(!isMainDoubt && uid.compareTo(fromUID) > 0){
                    isMainDoubt = true;
                    sendMessage(MESSAGE_ID_MEMBER_MERGE_REQ, fromUID);
                }
                break;
            case MESSAGE_ID_MAIN_REQ:
                if(!isClient){
                    sendMessage(MESSAGE_ID_MAIN_RES, fromUID);
                }
                break;
            case MESSAGE_ID_MAIN_RES:
                if(isClient){
                    break;
                }
                if(isMainRequesting){
                    if(ois.readBoolean()){
                        synchronized(mainReqMembers){
                            mainReqMembers.remove(fromUID);
                            if(mainReqMembers.size() == 0){
                                try{
                                    isMain = true;
                                    isMainRequesting = false;
                                    if(group == null){
                                        synchronized(clientMembers){
                                            clientMembers.clear();
                                        }
                                    }
                                    getLogger().write(
                                        MSG_ID_CHANGE_OPERATION_SYSTEM,
                                        getServiceNameObject()
                                    );
                                    eventQueue.push(new ClusterEvent(ClusterEvent.EVENT_CHANGE_MAIN));
                                }catch(Exception e){
                                    getLogger().write(
                                        MSG_ID_FAILED_CHANGE_ACTIVE_SYSTEM,
                                        e,
                                        getServiceNameObject()
                                    );
                                    stop();
                                }
                            }
                        }
                    }else{
                        isMainRequesting = false;
                        mainReqMembers.clear();
                    }
                }
                break;
            case MESSAGE_ID_HELLO_REQ:
                if((isMain && ois.readBoolean()) || members.contains(fromUID)){
                    sendMessage(MESSAGE_ID_HELLO_RES, fromUID);
                }else if(isMain){
                    synchronized(members){
                        if(!members.contains(fromUID)){
                            tmpOldMembers = new ArrayList<GlobalUID>(members);
                            members.add(fromUID);
                            tmpNewMembers = new ArrayList<GlobalUID>(members);
                            isMemberChange = true;
                        }
                    }
                    sendMessage(MESSAGE_ID_MEMBER_CHANGE_REQ);
                    if(isMemberChange){
                        eventQueue.push(new ClusterEvent(ClusterEvent.EVENT_MEMBER_CHANGE, tmpOldMembers, tmpNewMembers));
                    }
                }
                break;
            case MESSAGE_ID_HELLO_RES:
                synchronized(helloMonitor){
                    if(helloTarget != null && helloTarget.equals(fromUID)){
                        helloMonitor.notifyMonitor();
                    }
                }
                break;
            case MESSAGE_ID_BYE_REQ:
                if(isMain){
                    if(ois.readBoolean()){
                        if(group == null){
                            synchronized(clientMembers){
                                clientMembers.remove(fromUID);
                            }
                        }
                    }else{
                        synchronized(members){
                            if(members.contains(fromUID)){
                                tmpOldMembers = new ArrayList<GlobalUID>(members);
                                members.remove(fromUID);
                                tmpNewMembers = new ArrayList<GlobalUID>(members);
                                isMemberChange = true;
                            }
                        }
                        if(isMemberChange){
                            sendMessage(MESSAGE_ID_MEMBER_CHANGE_REQ);
                            eventQueue.push(new ClusterEvent(ClusterEvent.EVENT_MEMBER_CHANGE, tmpOldMembers, tmpNewMembers));
                        }
                    }
                }else if(!ois.readBoolean()){
                    if(isMainRequesting){
                        synchronized(mainReqMembers){
                            mainReqMembers.remove(fromUID);
                        }
                    }
                    synchronized(members){
                        if(members.contains(fromUID)){
                            tmpOldMembers = new ArrayList<GlobalUID>(members);
                            members.remove(fromUID);
                            tmpNewMembers = new ArrayList<GlobalUID>(members);
                            eventQueue.push(new ClusterEvent(ClusterEvent.EVENT_MEMBER_CHANGE, tmpOldMembers, tmpNewMembers));
                        }
                        if(members.indexOf(uid) == 0){
                            if(!isClient && members.size() == 1){
                                isMain = true;
                                isMainRequesting = false;
                                if(group == null){
                                    synchronized(clientMembers){
                                        clientMembers.clear();
                                    }
                                }
                                getLogger().write(
                                    MSG_ID_CHANGE_OPERATION_SYSTEM,
                                    getServiceNameObject()
                                );
                                eventQueue.push(new ClusterEvent(ClusterEvent.EVENT_CHANGE_MAIN));
                            }else if(!isMainRequesting){
                                synchronized(mainReqMembers){
                                    mainReqMembers.clear();
                                    mainReqMembers.addAll(members);
                                    mainReqMembers.remove(uid);
                                    isMainRequesting = true;
                                }
                                sendMessage(MESSAGE_ID_MAIN_REQ);
                            }
                        }
                    }
                }
                break;
            default:
            }
        }catch(ClassNotFoundException e){
            getLogger().write(MSG_ID_MESSAGE_IO_ERROR, e);
        }catch(IOException e){
            getLogger().write(MSG_ID_MESSAGE_IO_ERROR, e);
        }finally{
            if(ois != null){
                try{
                    ois.close();
                }catch(IOException e2){}
            }
        }
    }
    
    protected void processMemberInit(List<GlobalUID> members){
        Object[] tmpListeners = listeners.toArray();
        for(int i = 0; i < tmpListeners.length; i++){
            ClusterListener listener = (ClusterListener)tmpListeners[i];
            try{
                listener.memberInit(isClient ? null : uid, new ArrayList<GlobalUID>(members));
            }catch(Exception e){
            }
        }
    }
    
    protected void processMemberChange(List<GlobalUID> oldMembers, List<GlobalUID> newMembers){
        if(oldMembers.equals(newMembers)){
            return;
        }
        Object[] tmpListeners = listeners.toArray();
        for(int i = 0; i < tmpListeners.length; i++){
            ClusterListener listener = (ClusterListener)tmpListeners[i];
            try{
                listener.memberChange(new ArrayList<GlobalUID>(oldMembers), new ArrayList<GlobalUID>(newMembers));
            }catch(Exception e){
            }
        }
    }
    
    protected void processChangeMain() throws Exception{
        Object[] tmpListeners = listeners.toArray();
        for(int i = 0; i < tmpListeners.length; i++){
            ClusterListener listener = (ClusterListener)tmpListeners[i];
            listener.changeMain();
        }
    }
    
    protected void processChangeSub(){
        Object[] tmpListeners = listeners.toArray();
        for(int i = 0; i < tmpListeners.length; i++){
            ClusterListener listener = (ClusterListener)tmpListeners[i];
            listener.changeSub();
        }
    }
    
    public static class GlobalUID extends jp.ossc.nimbus.net.GlobalUID{
        
        private static final long serialVersionUID = 2185113122895103559L;
        
        protected Serializable option;
        
        public GlobalUID(String localAddress, Serializable option) throws UnknownHostException{
            super(localAddress);
            this.option = option;
        }
        
        public Object getOption(){
            return option;
        }
        protected void setOption(Object opt){
            option = (Serializable)opt;
        }
    }
    
    protected class MessageReceiver implements DaemonRunnable<Window>{
        
        private Map<Window, Window> windowMap = new LinkedHashMap<Window, Window>();
        
        public boolean onStart(){return true;}
        public boolean onStop(){return true;}
        public boolean onSuspend(){return true;}
        public boolean onResume(){return true;}
        
        public Window provide(DaemonControl ctrl) throws Throwable{
            final DatagramPacket packet = new DatagramPacket(new byte[receiveBufferSize], receiveBufferSize);
            try{
                socket.receive(packet);
                if(windowMap.size() != 0){
                    final long currentTime = System.currentTimeMillis();
                    Iterator<Window> itr = windowMap.values().iterator();
                    while(itr.hasNext()){
                        Window window = itr.next();
                        if(currentTime - window.receiveTime > lostTimeout){
                            itr.remove();
                        }else{
                            break;
                        }
                    }
                }
                ByteArrayInputStream bais = new ByteArrayInputStream(
                    packet.getData(),
                    0,
                    packet.getLength()
                );
                DataInputStream dis = new DataInputStream(bais);
                Window window = new Window();
                window.read(dis);
                if(window.isComplete()){
                    return window;
                }else{
                    Window tmp = windowMap.get(window);
                    if(tmp == null){
                        windowMap.put(window, window);
                        return null;
                    }else if(tmp.addWindow(window)){
                        return windowMap.remove(tmp);
                    }else{
                        return null;
                    }
                }
            }catch(SocketException e){
                try{
                    connect();
                }catch(IOException e2){}
                return null;
            }catch(IOException e){
                getLogger().write(MSG_ID_MESSAGE_IO_ERROR, e);
                return null;
            }
        }
        
        public void consume(Window window, DaemonControl ctrl)
         throws Throwable{
            if(window == null || (!isJoin && !isJoining)){
                return;
            }
            handleMessage(new ByteArrayInputStream(window.getData()));
        }
        
        public void garbage(){}
    }
    
    protected class HeartBeater implements DaemonRunnable<Object>{
        
        protected long lastSendTime = -1;
        protected int heartBeatFailedCount;
        protected GlobalUID targetMember;
        
        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 processTime = lastSendTime >= 0 ? System.currentTimeMillis() - lastSendTime : 0l;
            if(heartBeatInterval > processTime){
                Thread.sleep(heartBeatInterval - processTime);
            }
            return null;
        }
        
        public void consume(Object received, DaemonControl ctrl)
         throws Throwable{
            if(!isJoin){
                return;
            }
            lastSendTime = System.currentTimeMillis();
            if(isMain){
                sendMessage(MESSAGE_ID_MAIN_HELLO_REQ);
            }
            GlobalUID member = null;
            synchronized(members){
                if(isClient){
                    if(members.size() > 0){
                        member = (GlobalUID)members.get(0);
                    }
                }else{
                    if(members.size() > 1){
                        int index = members.indexOf(uid);
                        if(index == -1){
                            return;
                        }
                        if(index == members.size() - 1){
                            index = 0;
                        }else{
                            index += 1;
                        }
                        member = (GlobalUID)members.get(index);
                        if(!member.equals(targetMember)){
                            heartBeatFailedCount = 0;
                        }
                    }
                }
            }
            if(isClient && member == null){
                synchronized(addMonitor){
                    addMonitor.initMonitor();
                    sendMessage(MESSAGE_ID_ADD_REQ);
                    try{
                        addMonitor.waitMonitor(addMemberResponseTimeout);
                    }catch(InterruptedException e){
                        return;
                    }
                }
            }else if(member != null && !member.equals(uid)){
                targetMember = member;
                try{
                    boolean isNotify = false;
                    synchronized(helloMonitor){
                        helloTarget = member;
                        helloMonitor.initMonitor();
                        sendMessage(MESSAGE_ID_HELLO_REQ, helloTarget);
                        try{
                            isNotify = helloMonitor.waitMonitor(heartBeatResponseTimeout);
                        }catch(InterruptedException e){
                            return;
                        }
                    }
                    if(isNotify){
                        heartBeatFailedCount = 0;
                    }else{
                        heartBeatFailedCount++;
                        if(heartBeatFailedCount - 1 >= heartBeatRetryCount){
                            if(isClient){
                                isNotify = false;
                                synchronized(addMonitor){
                                    addMonitor.initMonitor();
                                    sendMessage(MESSAGE_ID_ADD_REQ);
                                    try{
                                        isNotify = addMonitor.waitMonitor(addMemberResponseTimeout);
                                    }catch(InterruptedException e){
                                        return;
                                    }
                                }
                                if(!isNotify){
                                    boolean isMemberChange = false;
                                    List<GlobalUID> tmpOldMembers = null;
                                    List<GlobalUID> tmpNewMembers = null;
                                    synchronized(members){
                                        if(members.size() != 0){
                                            tmpOldMembers = new ArrayList<GlobalUID>(members);
                                            members.clear();
                                            tmpNewMembers = new ArrayList<GlobalUID>(members);
                                            isMemberChange = true;
                                        }
                                    }
                                    if(isMemberChange){
                                        eventQueue.push(new ClusterEvent(ClusterEvent.EVENT_MEMBER_CHANGE, tmpOldMembers, tmpNewMembers));
                                    }
                                }
                            }else{
                                if(isMainRequesting){
                                    synchronized(mainReqMembers){
                                        mainReqMembers.remove(member);
                                    }
                                }
                                boolean isMemberChange = false;
                                List<GlobalUID> tmpOldMembers = null;
                                List<GlobalUID> tmpNewMembers = null;
                                synchronized(members){
                                    if(members.contains(member)){
                                        tmpOldMembers = new ArrayList<GlobalUID>(members);
                                        members.remove(member);
                                        tmpNewMembers = new ArrayList<GlobalUID>(members);
                                        isMemberChange = true;
                                    }
                                }
                                if(isMain){
                                    sendMessage(MESSAGE_ID_MEMBER_CHANGE_REQ);
                                }else if(!isClient){
                                    sendMessage(MESSAGE_ID_BYE_REQ, member, null);
                                }
                                if(isMemberChange){
                                    eventQueue.push(new ClusterEvent(ClusterEvent.EVENT_MEMBER_CHANGE, tmpOldMembers, tmpNewMembers));
                                }
                            }
                        }
                    }
                }catch(IOException e){
                }
            }
        }
        
        public void garbage(){}
    }
    
    protected class EventHandler implements DaemonRunnable<ClusterEvent>{
        
        public boolean onStart(){return true;}
        public boolean onStop(){return true;}
        public boolean onSuspend(){return true;}
        public boolean onResume(){return true;}
        
        public ClusterEvent provide(DaemonControl ctrl) throws Throwable{
            return eventQueue.get(1000);
        }
        public void consume(ClusterEvent event, DaemonControl ctrl){
            if(event == null){
                return;
            }
            switch(event.event){
            case ClusterEvent.EVENT_CHANGE_MAIN:
                try{
                    processChangeMain();
                }catch(Exception e){
                    getLogger().write(
                        MSG_ID_FAILED_CHANGE_ACTIVE_SYSTEM,
                        e,
                        getServiceNameObject()
                    );
                }
                break;
            case ClusterEvent.EVENT_CHANGE_SUB:
                processChangeSub();
                break;
            case ClusterEvent.EVENT_MEMBER_INIT:
                processMemberInit(event.newMembers);
                break;
            case ClusterEvent.EVENT_MEMBER_CHANGE:
                processMemberChange(event.oldMembers, event.newMembers);
                break;
            }
        }
        
        public void garbage(){}
    }
    
    protected static class ClusterEvent{
        public static final int EVENT_CHANGE_MAIN   = 1;
        public static final int EVENT_CHANGE_SUB    = 2;
        public static final int EVENT_MEMBER_INIT   = 3;
        public static final int EVENT_MEMBER_CHANGE = 4;
        public final int event;
        public final List<GlobalUID> oldMembers;
        public final List<GlobalUID> newMembers;
        
        public ClusterEvent(int event){
            this.event = event;
            this.oldMembers = null;
            this.newMembers = null;
        }
        
        public ClusterEvent(int event, List<GlobalUID> members){
            this.event = event;
            this.oldMembers = null;
            this.newMembers = members;
        }
        
        public ClusterEvent(int event, List<GlobalUID> oldMembers, List<GlobalUID> newMembers){
            this.event = event;
            this.oldMembers = oldMembers;
            this.newMembers = newMembers;
        }
    }
    
    protected static class Window implements Comparable<Window>{
        
        private static final int HEADER_LENGTH = 4 + 2 + 2 + 4;
        public int sequence;
        
        public short windowCount;
        public short windowNo;
        public long receiveTime;
        public byte[] data;
        private List<Window> windows;
        
        public List<byte[]> divide(int windowSize) throws IOException{
            List<byte[]> result = new ArrayList<byte[]>();
            if(data == null || data.length <= (windowSize - HEADER_LENGTH)){
                ByteArrayOutputStream baos = new ByteArrayOutputStream();
                DataOutputStream dos = new DataOutputStream(baos);
                dos.writeInt(sequence);
                dos.writeShort(1);
                dos.writeShort(0);
                dos.writeInt(data == null ? 0 : data.length);
                if(data != null && data.length != 0){
                    dos.write(data);
                }
                dos.flush();
                result.add(baos.toByteArray());
            }else{
                int offset = 0;
                short no = 0;
                short tmpWindowCount = (short)Math.ceil((double)data.length / (double)(windowSize - HEADER_LENGTH));
                ByteArrayOutputStream baos = new ByteArrayOutputStream();
                DataOutputStream dos = new DataOutputStream(baos);
                do{
                    dos.writeInt(sequence);
                    dos.writeShort(tmpWindowCount);
                    dos.writeShort(no);
                    int dataLength = Math.min(windowSize - HEADER_LENGTH, data.length - offset);
                    dos.writeInt(dataLength);
                    if(dataLength != 0){
                        dos.write(data, offset, dataLength);
                    }
                    dos.flush();
                    result.add(baos.toByteArray());
                    baos.reset();
                    offset += dataLength;
                    no++;
                }while(data.length > offset);
            }
            return result;
        }
        
        public boolean addWindow(Window window){
            if(isComplete()){
                return true;
            }
            if(windows == null){
                windows = new ArrayList<Window>(windowCount);
            }
            if(windows.size() == 0){
                windows.add(this);
            }
            windows.add(window);
            if(windowCount <= windows.size()){
                Collections.sort(windows);
                return true;
            }else{
                return false;
            }
        }
        
        public byte[] getData() throws IOException{
            if(!isComplete()){
                return null;
            }
            if(windows == null){
                return data;
            }
            
            ByteArrayOutputStream baos = new ByteArrayOutputStream();
            for(int i = 0, imax = windows.size(); i < imax; i++){
                Window w = windows.get(i);
                baos.write(w.data);
            }
            return baos.toByteArray();
        }
        
        public boolean isComplete(){
            return windowCount == 1 || (windows != null && windowCount <= windows.size());
        }
        
        public void read(DataInput in) throws IOException{
            sequence = in.readInt();
            windowCount = in.readShort();
            windowNo = in.readShort();
            int length = in.readInt();
            data = new byte[length];
            in.readFully(data, 0, length);
            receiveTime = System.currentTimeMillis();
        }
        
        public boolean equals(Object o){
            if(o == this){
                return true;
            }
            if(o == null || !(o instanceof Window)){
                return false;
            }
            Window cmp = (Window)o;
            return sequence == cmp.sequence;
        }
        
        public int hashCode(){
            return sequence;
        }
        
        public int compareTo(Window o){
            if(windowNo == o.windowNo){
                return 0;
            }else{
                return windowNo > o.windowNo ? 1 : -1;
            }
        }
    }
}