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

import java.util.Arrays;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.ArrayList;
import java.util.Map;
import java.util.Set;
import java.util.Iterator;
import java.util.Collection;
import java.util.Collections;

import jp.ossc.nimbus.core.ServiceBase;
import jp.ossc.nimbus.core.ServiceName;
import jp.ossc.nimbus.core.ServiceManagerFactory;
import jp.ossc.nimbus.service.cache.CacheMap;
import jp.ossc.nimbus.service.distribute.ClusterService;
import jp.ossc.nimbus.service.distribute.ClusterListener;
import jp.ossc.nimbus.service.publish.RequestMessageListener;
import jp.ossc.nimbus.service.publish.Message;
import jp.ossc.nimbus.service.publish.MessageException;
import jp.ossc.nimbus.service.publish.MessageSendException;
import jp.ossc.nimbus.service.publish.RequestServerConnection;
import jp.ossc.nimbus.service.publish.ServerConnectionFactory;
import jp.ossc.nimbus.service.publish.MessageReceiver;
import jp.ossc.nimbus.util.SynchronizeMonitor;
import jp.ossc.nimbus.util.WaitSynchronizeMonitor;

/**
 * 分散共有コンテキスト。<p>
 *
 * @author M.Takata
 */
public class DistributedSharedContextService<K> extends ServiceBase implements DistributedSharedContext<K>, RequestMessageListener, ClusterListener, DistributedSharedContextServiceMBean<K>{
    
    private static final long serialVersionUID = -8599934979277211223L;
    
    private ServiceName requestConnectionFactoryServiceName;
    private ServiceName clientCacheMapServiceName;
    private ServiceName serverCacheMapServiceName;
    private ServiceName clusterServiceName;
    private ServiceName sharedContextKeyDistributorServiceName;
    private SharedContextKeyDistributor keyDistributor;
    private ServiceName sharedContextStoreServiceName;
    private SharedContextStore<K> sharedContextStore;
    
    private String subject = DEFAULT_SUBJECT;
    private String clientSubject;
    private boolean isClient;
    
    private long synchronizeTimeout = 10000l;
    private long rehashTimeout = 10000l;
    private long defaultTimeout = 1000l;
    
    private boolean isClearBeforeSave = true;
    
    private int distributedSize = 2;
    private int replicationSize = 2;
    
    private RequestServerConnection serverConnection;
    private MessageReceiver messageReceiver;
    private ClusterService cluster;
    private Message targetMessage;
    
    private SharedContextService<K>[] sharedContextArray;
    private DistributeInfo<K> distributeInfo;
    private boolean isRehashEnabled = true;
    
    private boolean isManagedDataNode;
    private List<SharedContextUpdateListener<K>> updateListeners;
    
    private ServiceName[] sharedContextUpdateListenerServiceNames;
    
    public void setSharedContextKeyDistributorServiceName(ServiceName name){
        sharedContextKeyDistributorServiceName = name;
    }
    public ServiceName getSharedContextKeyDistributorServiceName(){
        return sharedContextKeyDistributorServiceName;
    }
    
    public void setDistributedSize(int size) throws IllegalArgumentException{
        if(size <= 1){
            throw new IllegalArgumentException("DistributedSize must be 2 or more." + size);
        }
        distributedSize = size;
    }
    public int getDistributedSize(){
        return distributedSize;
    }
    
    public void setReplicationSize(int size) throws IllegalArgumentException{
        if(size <= 1){
            throw new IllegalArgumentException("ReplicationSize must be 2 or more." + size);
        }
        replicationSize = size;
    }
    public int getReplicationSize(){
        return replicationSize;
    }
    
    public void setRequestConnectionFactoryServiceName(ServiceName name){
        requestConnectionFactoryServiceName = name;
    }
    public ServiceName getRequestConnectionFactoryServiceName(){
        return requestConnectionFactoryServiceName;
    }
    
    public void setClusterServiceName(ServiceName name){
        clusterServiceName = name;
    }
    public ServiceName getClusterServiceName(){
        return clusterServiceName;
    }
    
    public void setSubject(String subject){
        this.subject = subject;
    }
    public String getSubject(){
        return subject;
    }
    
    public void setClient(boolean isClient) throws SharedContextSendException, SharedContextTimeoutException{
        if(this.isClient == isClient){
            return;
        }
        this.isClient = isClient;
        if(getState() == State.STARTED){
            try{
                messageReceiver.removeMessageListener(this);
                messageReceiver.addSubject(this, isClient ? clientSubject :  subject);
            }catch(MessageSendException e){
                throw new SharedContextSendException(e);
            }
            synchronize();
        }
    }
    public boolean isClient(){
        return isClient;
    }
    
    public void setRehashEnabled(boolean isEnabled) throws SharedContextSendException, SharedContextTimeoutException{
        if(getState() == State.STARTED){
            try{
                Message message = serverConnection.createMessage(subject, Integer.toString(DistributedSharedContextEvent.EVENT_REHASH_SWITCH));
                message.setSubject(clientSubject, null);
                Set<Object> receiveClients = serverConnection.getReceiveClientIds(message);
                if(receiveClients.size() != 0){
                    message.setObject(new DistributedSharedContextEvent(DistributedSharedContextEvent.EVENT_REHASH_SWITCH, isEnabled ? Boolean.TRUE : Boolean.FALSE));
                    Message[] responses = serverConnection.request(
                        message,
                        isClient ? clientSubject : subject,
                        null,
                        0,
                        defaultTimeout
                    );
                    if(responses == null || responses.length == 0){
                        throw new SharedContextTimeoutException();
                    }
                }
            }catch(MessageException e){
                throw new SharedContextSendException(e);
            }catch(MessageSendException e){
                throw new SharedContextSendException(e);
            }
        }
        isRehashEnabled = isEnabled;
    }
    public boolean isRehashEnabled(){
        return isRehashEnabled;
    }
    
    public void setClientCacheMapServiceName(ServiceName name){
        clientCacheMapServiceName = name;
    }
    public ServiceName getClientCacheMapServiceName(){
        return clientCacheMapServiceName;
    }
    
    public void setServerCacheMapServiceName(ServiceName name){
        serverCacheMapServiceName = name;
    }
    public ServiceName getServerCacheMapServiceName(){
        return serverCacheMapServiceName;
    }
    
    public void setSharedContextStoreServiceName(ServiceName name){
        sharedContextStoreServiceName = name;
    }
    public ServiceName getSharedContextStoreServiceName(){
        return sharedContextStoreServiceName;
    }
    
    public void setClearBeforeSave(boolean isClear){
        isClearBeforeSave = isClear;
    }
    public boolean isClearBeforeSave(){
        return isClearBeforeSave;
    }
    
    public void setSynchronizeTimeout(long timeout){
        synchronizeTimeout = timeout;
    }
    public long getSynchronizeTimeout(){
        return synchronizeTimeout;
    }
    
    public void setRehashTimeout(long timeout){
        rehashTimeout = timeout;
    }
    public long getRehashTimeout(){
        return rehashTimeout;
    }
    
    public void setDefaultTimeout(long timeout){
        defaultTimeout = timeout;
    }
    public long getDefaultTimeout(){
        return defaultTimeout;
    }
    
    public void setManagedDataNode(boolean isManage){
        isManagedDataNode = isManage;
    }
    public boolean isManagedDataNode(){
        return isManagedDataNode;
    }
    
    public void setSharedContextUpdateListenerServiceNames(ServiceName[] names){
        sharedContextUpdateListenerServiceNames = names;
    }
    public ServiceName[] getSharedContextUpdateListenerServiceNames(){
        return sharedContextUpdateListenerServiceNames;
    }
    
    /**
     * サービスの開始処理を行う。<p>
     *
     * @exception Exception サービスの開始処理に失敗した場合
     */
    @SuppressWarnings("unchecked")
    public void startService() throws Exception{
        if(sharedContextKeyDistributorServiceName == null){
            MD5HashSharedContextKeyDistributorService defaultKeyDistributor = new MD5HashSharedContextKeyDistributorService();
            defaultKeyDistributor.create();
            defaultKeyDistributor.start();
            keyDistributor = defaultKeyDistributor;
        }else{
            keyDistributor = ServiceManagerFactory.getServiceObject(sharedContextKeyDistributorServiceName);
        }
        if(sharedContextStoreServiceName != null){
            sharedContextStore = ServiceManagerFactory.getServiceObject(sharedContextStoreServiceName);
        }
        if(requestConnectionFactoryServiceName == null){
            throw new IllegalArgumentException("RequestConnectionFactoryServiceName must be specified.");
        }
        
        if(sharedContextUpdateListenerServiceNames != null){
            for(int i = 0; i < sharedContextUpdateListenerServiceNames.length; i++){
                addSharedContextUpdateListener(
                    (SharedContextUpdateListener<K>)ServiceManagerFactory.getServiceObject(sharedContextUpdateListenerServiceNames[i])
                );
            }
        }
        ServerConnectionFactory factory = ServiceManagerFactory.getServiceObject(requestConnectionFactoryServiceName);
        serverConnection = (RequestServerConnection)factory.getServerConnection();
        messageReceiver = ServiceManagerFactory.getServiceObject(requestConnectionFactoryServiceName);
        clientSubject = subject + CLIENT_SUBJECT_SUFFIX;
        targetMessage = serverConnection.createMessage(subject, null);
        messageReceiver.addSubject(this, isClient ? clientSubject :  subject);
        if(clusterServiceName == null){
            throw new IllegalArgumentException("ClusterServiceName must be specified.");
        }
        cluster = ServiceManagerFactory.getServiceObject(clusterServiceName);
        
        sharedContextArray = new SharedContextService[distributedSize];
        for(int i = 0; i < sharedContextArray.length; i++){
            sharedContextArray[i] = new SharedContextService<K>();
            if(isManagedDataNode){
                sharedContextArray[i].setServiceManagerName(getServiceManagerName());
                sharedContextArray[i].setServiceName(
                    getServiceName() + '$' + i
                );
            }
            sharedContextArray[i].create();
            sharedContextArray[i].setRequestConnectionFactoryServiceName(requestConnectionFactoryServiceName);
            sharedContextArray[i].setClusterServiceName(clusterServiceName);
            if(clientCacheMapServiceName != null){
                sharedContextArray[i].setClientCacheMap((CacheMap<K,Object>)ServiceManagerFactory.getServiceObject(clientCacheMapServiceName));
            }
            if(serverCacheMapServiceName != null){
                sharedContextArray[i].setServerCacheMap((CacheMap<K,Object>)ServiceManagerFactory.getServiceObject(serverCacheMapServiceName));
            }
            if(sharedContextStoreServiceName != null){
                sharedContextArray[i].setSharedContextStoreServiceName(sharedContextStoreServiceName);
            }
            sharedContextArray[i].setSubject(subject + "$" + i);
            sharedContextArray[i].setClient(true);
            sharedContextArray[i].setSynchronizeTimeout(synchronizeTimeout);
            sharedContextArray[i].setDefaultTimeout(defaultTimeout);
            sharedContextArray[i].setSynchronizeOnStart(false);
            sharedContextArray[i].setSaveOnlyMain(true);
            sharedContextArray[i].setClearBeforeSave(false);
            sharedContextArray[i].setLoadOnStart(false);
            if(updateListeners != null){
                for(int j = 0; j < updateListeners.size(); j++){
                    sharedContextArray[i].addSharedContextUpdateListener(updateListeners.get(j));
                }
            }
            sharedContextArray[i].start();
        }
        distributeInfo = new DistributeInfo<K>(getId(), distributedSize);
        cluster.addClusterListener(this);
        rehash();
    }
    
    /**
     * サービスの停止処理を行う。<p>
     *
     * @exception Exception サービスの停止処理に失敗した場合
     */
    public void stopService() throws Exception{
        for(int i = 0; i < sharedContextArray.length; i++){
            sharedContextArray[i].stop();
            sharedContextArray[i].destroy();
        }
        
        if(messageReceiver != null){
            messageReceiver.removeMessageListener(this);
        }
        if(cluster != null){
            cluster.removeClusterListener(this);
        }
    }
    
    public void synchronize() throws SharedContextSendException, SharedContextTimeoutException{
        synchronize(synchronizeTimeout <= 0 ? 0 : synchronizeTimeout * sharedContextArray.length);
    }
    
    public synchronized void synchronize(long timeout) throws SharedContextSendException, SharedContextTimeoutException{
        if(sharedContextArray == null || sharedContextArray.length == 0){
            return;
        }
        for(int i = 0; i < sharedContextArray.length; i++){
            long start = System.currentTimeMillis();
            final boolean isNoTimeout = timeout <= 0;
            timeout = isNoTimeout ? timeout : timeout - (System.currentTimeMillis() - start);
            if(!isNoTimeout && timeout < 0){
                throw new SharedContextTimeoutException();
            }
            sharedContextArray[i].synchronize(timeout);
        }
    }
    
    public void rehash() throws SharedContextSendException, SharedContextTimeoutException{
        rehash(rehashTimeout);
    }
    
    @SuppressWarnings("unchecked")
    public synchronized void rehash(long timeout) throws SharedContextSendException, SharedContextTimeoutException{
        if(!isRehashEnabled){
            return;
        }
        if(isMain()){
            try{
                Message message = serverConnection.createMessage(subject, Integer.toString(DistributedSharedContextEvent.EVENT_GET_DIST_INFO));
                Set<Object> receiveClients = serverConnection.getReceiveClientIds(message);
                if(receiveClients.size() == 0){
                    DistributeGrid grid = new DistributeGrid();
                    grid.addDistributeInfo(distributeInfo);
                    grid.rehash();
                    distributeInfo.apply(distributeInfo, sharedContextArray);
                }else{
                    message.setObject(new DistributedSharedContextEvent(DistributedSharedContextEvent.EVENT_GET_DIST_INFO, new Long(timeout)));
                    long start = System.currentTimeMillis();
                    Message[] responses = serverConnection.request(message, 0, timeout);
                    if(responses != null && responses.length >= receiveClients.size()){
                        final boolean isNoTimeout = timeout <= 0;
                        timeout = isNoTimeout ? timeout : timeout - (System.currentTimeMillis() - start);
                        if(!isNoTimeout && timeout < 0){
                            throw new SharedContextTimeoutException();
                        }
                        DistributeGrid grid = new DistributeGrid();
                        grid.addDistributeInfo(distributeInfo);
                        for(int i = 0; i < responses.length; i++){
                            grid.addDistributeInfo((DistributeInfo<K>)responses[i].getObject());
                        }
                        grid.rehash();
                        RehashResponseCallBack callback = new RehashResponseCallBack();
                        Map<Object, DistributeInfo<K>> increaseDistributeInfos = grid.getIncreaseDistributeInfos();
                        DistributeInfo<K> info = increaseDistributeInfos.remove(getId());
                        if(info != null){
                            info.apply(distributeInfo, sharedContextArray);
                        }
                        if(increaseDistributeInfos.size() != 0){
                            callback.setResponseCount(increaseDistributeInfos.size());
                            Iterator<DistributeInfo<K>> infos = increaseDistributeInfos.values().iterator();
                            while(infos.hasNext()){
                                info = infos.next();
                                Message rehashMessage = serverConnection.createMessage(subject, Integer.toString(DistributedSharedContextEvent.EVENT_REHASH));
                                rehashMessage.setObject(new DistributedSharedContextEvent(DistributedSharedContextEvent.EVENT_REHASH, info));
                                rehashMessage.addDestinationId(info.getId());
                                serverConnection.request(rehashMessage, 1, timeout, callback);
                            }
                            callback.waitResponse(timeout);
                        }
                        Map<Object, DistributeInfo<K>> decreaseDistributeInfos = grid.getDecreaseDistributeInfos();
                        info = decreaseDistributeInfos.remove(getId());
                        if(info != null){
                            info.apply(distributeInfo, sharedContextArray);
                        }
                        if(decreaseDistributeInfos.size() != 0){
                            callback.setResponseCount(decreaseDistributeInfos.size());
                            Iterator<DistributeInfo<K>> infos = decreaseDistributeInfos.values().iterator();
                            while(infos.hasNext()){
                                info = infos.next();
                                Message rehashMessage = serverConnection.createMessage(subject, Integer.toString(DistributedSharedContextEvent.EVENT_REHASH));
                                rehashMessage.setObject(new DistributedSharedContextEvent(DistributedSharedContextEvent.EVENT_REHASH, info));
                                rehashMessage.addDestinationId(info.getId());
                                serverConnection.request(rehashMessage, 1, timeout, callback);
                            }
                            callback.waitResponse(timeout);
                        }
                    }else{
                        throw new SharedContextTimeoutException();
                    }
                }
            }catch(MessageException e){
                throw new SharedContextSendException(e);
            }catch(MessageSendException e){
                throw new SharedContextSendException(e);
            }
        }else{
            try{
                Message message = serverConnection.createMessage(subject, Integer.toString(DistributedSharedContextEvent.EVENT_REHASH_REQUEST));
                Set<Object> receiveClients = serverConnection.getReceiveClientIds(message);
                if(receiveClients.size() != 0){
                    message.setObject(new DistributedSharedContextEvent(DistributedSharedContextEvent.EVENT_REHASH_REQUEST, new Long(timeout)));
                    Message[] responses = serverConnection.request(
                        message,
                        isClient ? clientSubject : subject,
                        null,
                        1,
                        timeout
                    );
                    if(responses != null && responses.length > 0){
                        Object ret = responses[0].getObject();
                        if(ret instanceof Throwable){
                            if(ret instanceof RuntimeException){
                                throw (RuntimeException)ret;
                            }else if(ret instanceof Error){
                                throw (Error)ret;
                            }else{
                                throw new SharedContextSendException((Throwable)ret);
                            }
                        }
                    }else{
                        throw new SharedContextTimeoutException();
                    }
                }
            }catch(MessageException e){
                throw new SharedContextSendException(e);
            }catch(MessageSendException e){
                throw new SharedContextSendException(e);
            }
        }
    }
    
    public synchronized void load(long timeout) throws Exception{
        if(isMain()){
            if(sharedContextStore != null){
                sharedContextStore.load(this);
            }else{
                throw new UnsupportedOperationException();
            }
        }else{
            try{
                Message message = serverConnection.createMessage(subject, null);
                Set<Object> receiveClients = serverConnection.getReceiveClientIds(message);
                if(receiveClients.size() != 0){
                    message.setObject(new DistributedSharedContextEvent(DistributedSharedContextEvent.EVENT_LOAD));
                    Message[] responses = serverConnection.request(
                        message,
                        isClient ? clientSubject : subject,
                        null,
                        1,
                        timeout
                    );
                    if(responses != null && responses.length > 0){
                        Object ret = responses[0].getObject();
                        if(ret instanceof Throwable){
                            if(ret instanceof RuntimeException){
                                throw (RuntimeException)ret;
                            }else if(ret instanceof Error){
                                throw (Error)ret;
                            }else{
                                throw new SharedContextSendException((Throwable)ret);
                            }
                        }
                    }else{
                        throw new SharedContextTimeoutException();
                    }
                }
            }catch(MessageException e){
                throw new SharedContextSendException(e);
            }catch(MessageSendException e){
                throw new SharedContextSendException(e);
            }
        }
    }
    
    public synchronized void save(long timeout) throws Exception{
        if(isMain()){
            if(sharedContextStore != null){
                if(isClearBeforeSave){
                    sharedContextStore.clear();
                }
                for(int i = 0; i < sharedContextArray.length; i++){
                    sharedContextArray[i].save(timeout);
                }
            }else{
                throw new UnsupportedOperationException();
            }
        }else{
            try{
                Message message = serverConnection.createMessage(subject, null);
                Set<Object> receiveClients = serverConnection.getReceiveClientIds(message);
                if(receiveClients.size() != 0){
                    message.setObject(new DistributedSharedContextEvent(DistributedSharedContextEvent.EVENT_SAVE, new Long(timeout)));
                    Message[] responses = serverConnection.request(
                        message,
                        isClient ? clientSubject : subject,
                        null,
                        1,
                        timeout
                    );
                    if(responses != null && responses.length > 0){
                        Object ret = responses[0].getObject();
                        if(ret instanceof Throwable){
                            if(ret instanceof RuntimeException){
                                throw (RuntimeException)ret;
                            }else if(ret instanceof Error){
                                throw (Error)ret;
                            }else{
                                throw new SharedContextSendException((Throwable)ret);
                            }
                        }
                    }else{
                        throw new SharedContextTimeoutException();
                    }
                }
            }catch(MessageException e){
                throw new SharedContextSendException(e);
            }catch(MessageSendException e){
                throw new SharedContextSendException(e);
            }
        }
    }
    
    public void lock(K key) throws SharedContextSendException, SharedContextTimeoutException{
        lock(key, defaultTimeout);
    }
    
    public void lock(K key, long timeout) throws SharedContextSendException, SharedContextTimeoutException{
        sharedContextArray[keyDistributor.selectDataNodeIndex(key, sharedContextArray.length)].lock(key, timeout);
    }
    
    public boolean unlock(K key) throws SharedContextSendException{
        return unlock(key, false);
    }
    
    public boolean unlock(K key, boolean force) throws SharedContextSendException{
        return sharedContextArray[keyDistributor.selectDataNodeIndex(key, sharedContextArray.length)].unlock(key, force);
    }
    
    public Object getLockOwner(K key){
        return sharedContextArray[keyDistributor.selectDataNodeIndex(key, sharedContextArray.length)].getLockOwner(key);
    }
    
    public Object put(K key, Object value) throws SharedContextSendException{
        return put(key, value, defaultTimeout);
    }
    
    public Object put(K key, Object value, long timeout) throws SharedContextSendException, SharedContextTimeoutException{
        return sharedContextArray[keyDistributor.selectDataNodeIndex(key, sharedContextArray.length)].put(key, value, timeout);
    }
    
    public void putAsynch(K key, Object value) throws SharedContextSendException{
        sharedContextArray[keyDistributor.selectDataNodeIndex(key, sharedContextArray.length)].putAsynch(key, value);
    }
    
    public void update(K key, SharedContextValueDifference diff) throws SharedContextSendException{
        update(key, diff, defaultTimeout);
    }
    
    public void update(K key, SharedContextValueDifference diff, long timeout) throws SharedContextSendException, SharedContextTimeoutException{
        sharedContextArray[keyDistributor.selectDataNodeIndex(key, sharedContextArray.length)].update(key, diff, timeout);
    }
    
    public void updateAsynch(K key, SharedContextValueDifference diff) throws SharedContextSendException{
        sharedContextArray[keyDistributor.selectDataNodeIndex(key, sharedContextArray.length)].updateAsynch(key, diff);
    }
    
    public Object remove(Object key) throws SharedContextSendException{
        return remove(key, defaultTimeout);
    }
    
    public Object remove(Object key, long timeout) throws SharedContextSendException, SharedContextTimeoutException{
        return sharedContextArray[keyDistributor.selectDataNodeIndex(key, sharedContextArray.length)].remove(key, timeout);
    }
    
    public void removeAsynch(Object key) throws SharedContextSendException{
        sharedContextArray[keyDistributor.selectDataNodeIndex(key, sharedContextArray.length)].removeAsynch(key);
    }
    
    public void putAll(Map<? extends K, ? extends Object> t){
        putAll(t, defaultTimeout);
    }
    
    public synchronized void putAll(Map<? extends K, ? extends Object> t, long timeout) throws SharedContextSendException, SharedContextTimeoutException{
        Iterator<?> entries = t.entrySet().iterator();
        while(entries.hasNext()){
            @SuppressWarnings("unchecked")
            Map.Entry<? extends K,? extends Object> entry = (Map.Entry<? extends K, ? extends Object>)entries.next();
            long start = System.currentTimeMillis();
            final boolean isNoTimeout = timeout <= 0;
            timeout = isNoTimeout ? timeout : timeout - (System.currentTimeMillis() - start);
            if(!isNoTimeout && timeout < 0){
                throw new SharedContextTimeoutException();
            }
            sharedContextArray[keyDistributor.selectDataNodeIndex(entry.getKey(), sharedContextArray.length)].put(entry.getKey(), entry.getValue(), timeout);
        }
    }
    
    public synchronized void putAllAsynch(Map<? extends K, ? extends Object> t) throws SharedContextSendException{
        Iterator<?> entries = t.entrySet().iterator();
        while(entries.hasNext()){
            @SuppressWarnings("unchecked")
            Map.Entry<? extends K,? extends Object> entry = (Map.Entry<? extends K, ? extends Object>)entries.next();
            sharedContextArray[keyDistributor.selectDataNodeIndex(entry.getKey(), sharedContextArray.length)].putAsynch(entry.getKey(), entry.getValue());
        }
    }
    
    public void clear() throws SharedContextSendException{
        clear(defaultTimeout <= 0 ? 0 : defaultTimeout * sharedContextArray.length);
    }
    
    public synchronized void clear(long timeout) throws SharedContextSendException, SharedContextTimeoutException{
        for(int i = 0; i < sharedContextArray.length; i++){
            long start = System.currentTimeMillis();
            final boolean isNoTimeout = timeout <= 0;
            timeout = isNoTimeout ? timeout : timeout - (System.currentTimeMillis() - start);
            if(!isNoTimeout && timeout < 0){
                throw new SharedContextTimeoutException();
            }
            sharedContextArray[i].clear(timeout);
        }
    }
    
    public synchronized void clearAsynch() throws SharedContextSendException{
        for(int i = 0; i < sharedContextArray.length; i++){
            sharedContextArray[i].clearAsynch();
        }
    }
    
    public Object get(Object key) throws SharedContextSendException{
        return get(key, defaultTimeout);
    }
    
    public Object get(Object key, long timeout) throws SharedContextSendException, SharedContextTimeoutException{
        return sharedContextArray[keyDistributor.selectDataNodeIndex(key, sharedContextArray.length)].get(key, timeout);
    }
    
    public Set<K> keySet() throws SharedContextSendException{
        return keySet(defaultTimeout <= 0 ? 0 : defaultTimeout * sharedContextArray.length);
    }
    
    public synchronized Set<K> keySet(long timeout) throws SharedContextSendException, SharedContextTimeoutException{
        Set<K> result = new HashSet<K>();
        for(int i = 0; i < sharedContextArray.length; i++){
            long start = System.currentTimeMillis();
            final boolean isNoTimeout = timeout <= 0;
            timeout = isNoTimeout ? timeout : timeout - (System.currentTimeMillis() - start);
            if(!isNoTimeout && timeout < 0){
                throw new SharedContextTimeoutException();
            }
            result.addAll(sharedContextArray[i].keySet(timeout));
        }
        return result;
    }
    
    public int size() throws SharedContextSendException{
        return size(defaultTimeout <= 0 ? 0 : defaultTimeout * sharedContextArray.length);
    }
    
    public synchronized int size(long timeout) throws SharedContextSendException, SharedContextTimeoutException{
        int result = 0;
        for(int i = 0; i < sharedContextArray.length; i++){
            long start = System.currentTimeMillis();
            final boolean isNoTimeout = timeout <= 0;
            timeout = isNoTimeout ? timeout : timeout - (System.currentTimeMillis() - start);
            if(!isNoTimeout && timeout < 0){
                throw new SharedContextTimeoutException();
            }
            result += sharedContextArray[i].size(timeout);
        }
        return result;
    }
    
    public boolean isEmpty() throws SharedContextSendException{
        return size() == 0 ? true : false;
    }
    
    public boolean containsKey(Object key) throws SharedContextSendException{
        return containsKey(key, defaultTimeout);
    }
    
    public boolean containsKey(Object key, long timeout) throws SharedContextSendException, SharedContextTimeoutException{
        return sharedContextArray[keyDistributor.selectDataNodeIndex(key, sharedContextArray.length)].containsKey(key, timeout);
    }
    
    public boolean containsValue(Object value) throws SharedContextSendException{
        return containsValue(value, defaultTimeout <= 0 ? 0 : defaultTimeout * sharedContextArray.length);
    }
    
    public boolean containsValue(Object value, long timeout) throws SharedContextSendException, SharedContextTimeoutException{
        for(int i = 0; i < sharedContextArray.length; i++){
            long start = System.currentTimeMillis();
            final boolean isNoTimeout = timeout <= 0;
            timeout = isNoTimeout ? timeout : timeout - (System.currentTimeMillis() - start);
            if(!isNoTimeout && timeout < 0){
                throw new SharedContextTimeoutException();
            }
            if(sharedContextArray[i].containsValue(value, timeout)){
                return true;
            }
        }
        return false;
    }
    
    public Collection<Object> values(){
        throw new UnsupportedOperationException();
    }
    
    public Set<Map.Entry<K, Object>> entrySet(){
        throw new UnsupportedOperationException();
    }
    
    public int getDataNodeIndex(Object key){
        return keyDistributor.selectDataNodeIndex(key, sharedContextArray.length);
    }
    
    public int size(int nodeIndex) throws SharedContextSendException, SharedContextTimeoutException{
        return sharedContextArray[nodeIndex].size();
    }
    
    public Set<K> keySet(int nodeIndex) throws SharedContextSendException, SharedContextTimeoutException{
        return sharedContextArray[nodeIndex].keySet();
    }
    
    public boolean isClient(int nodeIndex){
        return sharedContextArray[nodeIndex].isClient();
    }
    
    public boolean isMain(int nodeIndex){
        return sharedContextArray[nodeIndex].isMain();
    }
    
    public void addSharedContextUpdateListener(SharedContextUpdateListener<K> listener){
        if(updateListeners == null){
            updateListeners = Collections.synchronizedList(new ArrayList<SharedContextUpdateListener<K>>());
        }
        if(!updateListeners.contains(listener)){
            updateListeners.add(listener);
        }
        if(sharedContextArray != null){
            for(int i = 0; i < sharedContextArray.length; i++){
                sharedContextArray[i].addSharedContextUpdateListener(listener);
            }
        }
    }
    
    public void removeSharedContextUpdateListener(SharedContextUpdateListener<K> listener){
        if(updateListeners == null){
            return;
        }
        updateListeners.remove(listener);
        if(sharedContextArray != null){
            for(int i = 0; i < sharedContextArray.length; i++){
                sharedContextArray[i].removeSharedContextUpdateListener(listener);
            }
        }
    }
    
    @SuppressWarnings("unchecked")
    public String displayDistributeInfo() throws SharedContextSendException, SharedContextTimeoutException{
        DistributeGrid grid = new DistributeGrid();
        try{
            Message message = serverConnection.createMessage(subject, Integer.toString(DistributedSharedContextEvent.EVENT_GET_DIST_INFO));
            Set<Object> receiveClients = serverConnection.getReceiveClientIds(message);
            if(!isClient){
                grid.addDistributeInfo(distributeInfo);
            }
            if(receiveClients.size() != 0){
                long timeout = rehashTimeout;
                message.setObject(new DistributedSharedContextEvent(DistributedSharedContextEvent.EVENT_GET_DIST_INFO, new Long(timeout)));
                long start = System.currentTimeMillis();
                Message[] responses = serverConnection.request(
                    message,
                    isClient ? clientSubject : subject,
                    null,
                    0,
                    timeout
                );
                if(responses != null && responses.length >= receiveClients.size()){
                    final boolean isNoTimeout = timeout <= 0;
                    timeout = isNoTimeout ? timeout : timeout - (System.currentTimeMillis() - start);
                    if(!isNoTimeout && timeout < 0){
                        throw new SharedContextTimeoutException();
                    }
                    for(int i = 0; i < responses.length; i++){
                        grid.addDistributeInfo((DistributeInfo<K>)responses[i].getObject());
                    }
                }
            }
        }catch(MessageException e){
            throw new SharedContextSendException(e);
        }catch(MessageSendException e){
            throw new SharedContextSendException(e);
        }
        return grid.toString();
    }
    
    public void onMessage(Message message){
    }
    
    public Message onRequestMessage(Object sourceId, int sequence, Message message, String responseSubject, String responseKey){
        DistributedSharedContextEvent event = null;
        try{
            event = (DistributedSharedContextEvent)message.getObject();
        }catch(MessageException e){
            return null;
        }
        Message result = null;
        switch(event.type){
        case DistributedSharedContextEvent.EVENT_GET_DIST_INFO:
            result = onGetDistributeInfo(event, responseSubject, responseKey);
            break;
        case DistributedSharedContextEvent.EVENT_REHASH:
            result = onRehash(event, sourceId, sequence, responseSubject, responseKey);
            break;
        case DistributedSharedContextEvent.EVENT_REHASH_REQUEST:
            onRehashRequest(event, sourceId, sequence, responseSubject, responseKey);
            break;
        case DistributedSharedContextEvent.EVENT_SAVE:
            result = onSave(event, sourceId, sequence, responseSubject, responseKey);
            break;
        case DistributedSharedContextEvent.EVENT_LOAD:
            result = onLoad(event, responseSubject, responseKey);
            break;
        case DistributedSharedContextEvent.EVENT_REHASH_SWITCH:
            result = onRehashSwitch(event, responseSubject, responseKey);
            break;
        default:
        }
        return result;
    }
    
    protected Message createResponseMessage(String responseSubject, String responseKey, Object response){
        Message result = null;
        try{
            result = serverConnection.createMessage(responseSubject, responseKey);
            result.setObject(response);
        }catch(MessageException e){
            e.printStackTrace();
            // TODO
        }
        return result;
    }
    
    protected Message onGetDistributeInfo(DistributedSharedContextEvent event, String responseSubject, String responseKey){
        return createResponseMessage(responseSubject, responseKey, distributeInfo);
    }
    
    protected Message onRehashRequest(final DistributedSharedContextEvent event, final Object sourceId, final int sequence, final String responseSubject, final String responseKey){
        if(isMain()){
            Thread rehashThread = new Thread(){
                public void run(){
                    Message response = null;
                    try{
                        rehash(((Long)event.value).longValue());
                        response = createResponseMessage(responseSubject, responseKey, null);
                    }catch(Throwable th){
                        response = createResponseMessage(responseSubject, responseKey, th);
                    }
                    try{
                        serverConnection.response(sourceId, sequence, response);
                    }catch(MessageSendException e){
                        e.printStackTrace();
                        // TODO
                    }
                }
            };
            rehashThread.start();
        }
        return null;
    }
    
    protected Message onRehash(final DistributedSharedContextEvent event, final Object sourceId, final int sequence, final String responseSubject, final String responseKey){
        @SuppressWarnings("unchecked")
        final DistributeInfo<K> info = (DistributeInfo<K>)event.value;
        Thread rehashThread = new Thread(){
            public void run(){
                Message response = null;
                try{
                    info.apply(distributeInfo, sharedContextArray);
                        response = createResponseMessage(responseSubject, responseKey, null);
                }catch(Throwable th){
                    response = createResponseMessage(responseSubject, responseKey, th);
                }
                try{
                    serverConnection.response(sourceId, sequence, response);
                }catch(MessageSendException e){
                    e.printStackTrace();
                    // TODO
                }
            }
        };
        rehashThread.start();
        return null;
    }
    
    protected Message onSave(final DistributedSharedContextEvent event, final Object sourceId, final int sequence, final String responseSubject, final String responseKey){
        if(isMain()){
            Thread saveThread = new Thread(){
                public void run(){
                    Message response = null;
                    try{
                        long timeout = ((Long)event.value).longValue();
                        if(sharedContextStore != null){
                            if(isClearBeforeSave){
                                sharedContextStore.clear();
                            }
                            for(int i = 0; i < sharedContextArray.length; i++){
                                sharedContextArray[i].save(timeout);
                            }
                        }else{
                            throw new UnsupportedOperationException();
                        }
                        response = createResponseMessage(responseSubject, responseKey, null);
                    }catch(Throwable th){
                        response = createResponseMessage(responseSubject, responseKey, th);
                    }
                    try{
                        serverConnection.response(sourceId, sequence, response);
                    }catch(MessageSendException e){
                        // TODO
                        e.printStackTrace();
                    }
                }
            };
            saveThread.start();
        }
        return null;
    }
    
    protected synchronized Message onLoad(DistributedSharedContextEvent event, String responseSubject, String responseKey){
        if(isMain()){
            try{
                if(sharedContextStore != null){
                    sharedContextStore.load(this);
                }else{
                    throw new UnsupportedOperationException();
                }
                return createResponseMessage(responseSubject, responseKey, null);
            }catch(Throwable th){
                return createResponseMessage(responseSubject, responseKey, th);
            }
        }
        return null;
    }
    
    protected synchronized Message onRehashSwitch(DistributedSharedContextEvent event, String responseSubject, String responseKey){
        isRehashEnabled = ((Boolean)event.value).booleanValue();
        return createResponseMessage(responseSubject, responseKey, null);
    }
    
    public boolean isMain(){
        return isMain(cluster.getMembers());
    }
    
    private boolean isMain(List<? extends Object> members){
        if(isClient){
            return false;
        }else{
            Set<Object> targetMembers = serverConnection.getReceiveClientIds(targetMessage);
            Object myId = cluster.getUID();
            for(int i = 0, imax = members.size(); i < imax; i++){
                Object id = members.get(i);
                if(id.equals(myId)){
                    return true;
                }else if(targetMembers.contains(id)){
                    return false;
                }
            }
            return true;
        }
    }
    
    public Object getId(){
        return cluster == null ? null :  cluster.getUID();
    }
    
    public Object getMainId(){
        if(cluster == null){
            return null;
        }else{
            Object myId = cluster.getUID();
            List<? extends Object> members = cluster.getMembers();
            Set<Object> targetMembers = serverConnection.getReceiveClientIds(targetMessage);
            for(int i = 0, imax = members.size(); i < imax; i++){
                Object id = members.get(i);
                if(id.equals(myId)){
                    return myId;
                }else if(targetMembers.contains(id)){
                    return id;
                }
            }
            return myId;
        }
    }
    
    public void memberInit(Object myId, List<? extends Object>  members){
    }
    
    public void memberChange(List<? extends Object>  oldMembers, List<? extends Object>  newMembers){
        if(isMain(newMembers)){
            Set<? extends Object>  deadMembers = new HashSet<Object> (oldMembers);
            deadMembers.removeAll(newMembers);
            if(deadMembers.size() != 0){
                try{
                    rehash();
                }catch(Throwable th){
                    th.printStackTrace();
                    // TODO
                }
            }
        }
    }
    
    public void changeMain() throws Exception{
    }
    
    public void changeSub(){
    }
    
    public static class DistributedSharedContextEvent implements java.io.Externalizable{
        
        public static final byte EVENT_GET_DIST_INFO  = (byte)1;
        public static final byte EVENT_REHASH_REQUEST = (byte)2;
        public static final byte EVENT_REHASH         = (byte)3;
        public static final byte EVENT_SAVE           = (byte)4;
        public static final byte EVENT_LOAD           = (byte)5;
        public static final byte EVENT_REHASH_SWITCH  = (byte)6;
        
        public byte type;
        public Object value;
        
        public DistributedSharedContextEvent(){
        }
        
        public DistributedSharedContextEvent(byte type){
            this(type, null);
        }
        
        public DistributedSharedContextEvent(byte type, Object value){
            this.type = type;
            this.value = value;
        }
        
        public void writeExternal(java.io.ObjectOutput out) throws java.io.IOException{
            out.write(type);
            out.writeObject(value);
        }
        
        public void readExternal(java.io.ObjectInput in) throws java.io.IOException, ClassNotFoundException{
            type = (byte)in.read();
            value = in.readObject();
        }
        
        public String toString(){
            StringBuilder buf = new StringBuilder(super.toString());
            buf.append('{');
            buf.append("type=").append(type);
            buf.append(", value=").append(value);
            buf.append('}');
            return buf.toString();
        }
    }
    
    public static class DistributeInfo<K> implements java.io.Externalizable{
        
        private Object id;
        private boolean[] serverFlagArray;
        private int serverCount;
        
        public DistributeInfo(){
        }
        
        public DistributeInfo(Object id, int distributedSize){
            this.id = id;
            serverFlagArray = new boolean[distributedSize];
        }
        
        public Object getId(){
            return id;
        }
        
        public void setServer(int index){
            if(serverFlagArray[index]){
                return;
            }
            serverFlagArray[index] = true;
            serverCount++;
        }
        public void setClient(int index){
            if(!serverFlagArray[index]){
                return;
            }
            serverFlagArray[index] = false;
            serverCount--;
        }
        public boolean isServer(int index){
            return serverFlagArray[index];
        }
        public int size(){
            return serverFlagArray.length;
        }
        public int getServerCount(){
            return serverCount;
        }
        public int getMaxServerIndex(){
            if(getServerCount() == 0){
                return -1;
            }
            for(int i = serverFlagArray.length; --i >= 0;){
                if(serverFlagArray[i]){
                    return i;
                }
            }
            return -1;
        }
        
        public synchronized void apply(DistributeInfo<K> info, SharedContextService<K>[] contexts) throws SharedContextSendException, SharedContextTimeoutException{
            for(int i = 0; i < contexts.length; i++){
                final boolean isClient = !isServer(i);
                if(isClient != contexts[i].isClient()){
                    contexts[i].setClient(!isServer(i));
                    if(isClient){
                        info.setClient(i);
                    }else{
                        info.setServer(i);
                    }
                }
            }
        }
        
        public void writeExternal(java.io.ObjectOutput out) throws java.io.IOException{
            out.writeObject(id);
            out.writeInt(serverFlagArray.length);
            for(int i = 0; i < serverFlagArray.length; i++){
                out.write(serverFlagArray[i] ? 1 : 0);
            }
        }
        
        public void readExternal(java.io.ObjectInput in) throws java.io.IOException, ClassNotFoundException{
            id = in.readObject();
            serverFlagArray = new boolean[in.readInt()];
            for(int i = 0; i < serverFlagArray.length; i++){
                serverFlagArray[i] = in.read() == 1 ? true : false;
                if(serverFlagArray[i]){
                    serverCount++;
                }
            }
        }
    }
    
    private class DistributeGrid{
        private int[] serverCounts;
        private final Map<Object, DistributeInfo<K>> distributeInfos = new HashMap<Object, DistributeInfo<K>>();
        private final Map<Object, DistributeInfo<K>> increaseInfos = new HashMap<Object, DistributeInfo<K>>();
        private final Map<Object, DistributeInfo<K>> decreaseInfos = new HashMap<Object, DistributeInfo<K>>();
        
        public DistributeGrid(){
            serverCounts = new int[sharedContextArray.length];
        }
        
        public void addDistributeInfo(DistributeInfo<K> info){
            distributeInfos.put(info.getId(), info);
            for(int i = 0; i < serverCounts.length; i++){
                if(info.isServer(i)){
                    serverCounts[i]++;
                }
            }
        }
        
        @SuppressWarnings("unused")
        public int getServerCount(int index){
            return serverCounts[index];
        }
        
        @SuppressWarnings("unchecked")
        public void rehash(){
            Set<DistributeInfo<K>> infoSet = new HashSet<DistributeInfo<K>>(distributeInfos.values());
            DistributeInfo<K>[] infos = infoSet.toArray(new DistributeInfo[infoSet.size()]);
            for(int i = 0; i < serverCounts.length; i++){
                if(serverCounts[i] < replicationSize){
                    Arrays.sort(infos, DistributeInfoComparator.INSTANCE);
                    infos[0].setServer(i);
                    serverCounts[i]++;
                    increaseInfos.put(infos[0].getId(), infos[0]);
                    infoSet.remove(infos[0]);
                }
            }
            infos = infoSet.toArray(new DistributeInfo[infoSet.size()]);
            if(infos.length > 2){
                do{
                    Arrays.sort(infos, DistributeInfoComparator.INSTANCE);
                    DistributeInfo<K> maxInfo = infos[infos.length - 1];
                    DistributeInfo<K> minInfo = infos[0];
                    if(maxInfo.getServerCount() - minInfo.getServerCount() > 1){
                        final int index = maxInfo.getMaxServerIndex();
                        maxInfo.setClient(index);
                        decreaseInfos.put(maxInfo.getId(), maxInfo);
                        minInfo.setServer(index);
                        increaseInfos.put(minInfo.getId(), minInfo);
                    }else{
                        break;
                    }
                }while(true);
            }
        }
        
        public Map<Object, DistributeInfo<K>> getIncreaseDistributeInfos(){
            return increaseInfos;
        }
        
        public Map<Object, DistributeInfo<K>> getDecreaseDistributeInfos(){
            return decreaseInfos;
        }
        
        public String toString(){
            StringBuffer buf = new StringBuffer();
            @SuppressWarnings("unchecked")
            DistributeInfo<K>[] infos = distributeInfos.values().toArray(new DistributeInfo[distributeInfos.size()]);
            for(int i = 0, imax = infos.length; i < imax; i++){
                buf.append(infos[i].getId()).append(',');
                for(int j = 0, jmax = infos[i].size(); j < jmax; j++){
                    buf.append(infos[i].isServer(j) ? "S" : "C");
                    if(j != jmax - 1){
                        buf.append(',');
                    }
                }
                if(i != imax - 1){
                    buf.append('\n');
                }
            }
            return buf.toString();
        }
    }
    
    private static class DistributeInfoComparator<K> implements Comparator<DistributeInfo<K>>{
        
        @SuppressWarnings({"rawtypes", "unchecked"})
        public static final Comparator<DistributeInfo<?>> INSTANCE = new DistributeInfoComparator();
        
        public int compare(DistributeInfo<K> d1, DistributeInfo<K> d2){
            return d1.getServerCount() - d2.getServerCount();
        }
    }
    
    private class RehashResponseCallBack implements RequestServerConnection.ResponseCallBack{
        private SynchronizeMonitor monitor = new WaitSynchronizeMonitor();
        private boolean isTimeout;
        private Throwable throwable;
        private int responseCount;
        private int currentResponseCount;
        public RehashResponseCallBack(){
            monitor.initMonitor();
        }
        
        public void setResponseCount(int count){
            responseCount = count;
            currentResponseCount = 0;
            isTimeout = false;
            throwable = null;
        }
        
        public void onResponse(Message message, boolean isLast){
            currentResponseCount++;
            if(message == null){
                isTimeout = true;
                monitor.notifyMonitor();
            }else{
                Object response = null;
                try{
                    response = message.getObject();
                }catch(MessageException e){
                    throwable = e;
                    monitor.notifyMonitor();
                }
                if(response instanceof Throwable){
                    throwable = (Throwable)response;
                    monitor.notifyMonitor();
                }else if(currentResponseCount >= responseCount){
                    monitor.notifyMonitor();
                }
            }
        }
        
        public void waitResponse(long timeout) throws SharedContextSendException, SharedContextTimeoutException{
            try{
                if(!monitor.waitMonitor(timeout)){
                    throw new SharedContextTimeoutException();
                }
            }catch(InterruptedException e){
                throw new SharedContextTimeoutException();
            }
            if(isTimeout){
                throw new SharedContextTimeoutException();
            }
            if(throwable != null){
                if(throwable instanceof SharedContextSendException){
                    throw (SharedContextSendException)throwable;
                }else if(throwable instanceof SharedContextTimeoutException){
                    throw (SharedContextTimeoutException)throwable;
                }else if(throwable instanceof RuntimeException){
                    throw (RuntimeException)throwable;
                }else if(throwable instanceof Error){
                    throw (Error)throwable;
                }else{
                    throw new SharedContextSendException(throwable);
                }
            }
        }
    }
}