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

import jp.ossc.nimbus.core.*;
import jp.ossc.nimbus.service.publish.*;
import jp.ossc.nimbus.service.cache.CacheMap;
import jp.ossc.nimbus.service.cache.CacheRemoveListener;
import jp.ossc.nimbus.service.cache.CachedReference;
import jp.ossc.nimbus.service.cache.IllegalCachedReferenceException;
import jp.ossc.nimbus.service.cache.KeyCachedReference;
import jp.ossc.nimbus.service.distribute.ClusterListener;
import jp.ossc.nimbus.service.distribute.ClusterService;
import jp.ossc.nimbus.util.SynchronizeMonitor;
import jp.ossc.nimbus.util.WaitSynchronizeMonitor;

/**
 * 共有コンテキスト。<p>
 * サーバ間でコンテキスト情報を共有する。<br>
 * 以下に、サービス定義例を示す。<br>
 * <pre>
 * &lt;?xml version="1.0" encoding="Shift_JIS"?&gt;
 * 
 * &lt;nimbus&gt;
 *     
 *     &lt;manager name="Sample"&gt;
 *         
 *         &lt;service name="Context"
 *                  code="jp.ossc.nimbus.service.context.SharedContextService"&gt;
 *             &lt;attribute name="RequestConnectionFactoryServiceName"&gt;#RequestConnectionFactory&lt;/attribute&gt;
 *             &lt;attribute name="ClusterServiceName"&gt;#Cluster&lt;/attribute&gt;
 *             &lt;depends&gt;RequestConnectionFactory&lt;/depends&gt;
 *             &lt;depends&gt;Cluster&lt;/depends&gt;
 *         &lt;/service&gt;
 *         
 *         &lt;service name="RequestConnectionFactory"
 *                  code="jp.ossc.nimbus.service.publish.RequestConnectionFactoryService"&gt;
 *             &lt;attribute name="ServerConnectionFactoryServiceName"&gt;#ServerConnectionFactory&lt;/attribute&gt;
 *             &lt;attribute name="MessageReceiverServiceName"&gt;#MessageReceiver&lt;/attribute&gt;
 *             &lt;depends&gt;
 *                 &lt;service name="MessageReceiver"
 *                          code="jp.ossc.nimbus.service.publish.MessageReceiverService"&gt;
 *                     &lt;attribute name="ClientConnectionFactoryServiceName"&gt;#ClientConnectionFactory&lt;/attribute&gt;
 *                     &lt;attribute name="StartReceiveOnStart"&gt;true&lt;/attribute&gt;
 *                     &lt;depends&gt;ClientConnectionFactory&lt;/depends&gt;
 *                 &lt;/service&gt;
 *             &lt;/depends&gt;
 *         &lt;/service&gt;
 *         
 *         &lt;service name="ClientConnectionFactory"
 *                  code="jp.ossc.nimbus.service.publish.ClusterConnectionFactoryService"&gt;
 *             &lt;attribute name="JndiRepositoryServiceName"&gt;#JNDIRepository&lt;/attribute&gt;
 *             &lt;attribute name="ClusterServiceName"&gt;#Cluster&lt;/attribute&gt;
 *             &lt;attribute name="ClientConnectionFactoryServiceName"&gt;#ServerConnectionFactory&lt;/attribute&gt;
 *             &lt;attribute name="Multiple"&gt;true&lt;/attribute&gt;
 *             &lt;attribute name="FlexibleConnect"&gt;true&lt;/attribute&gt;
 *             &lt;depends&gt;
 *                 &lt;service name="JNDIRepository"
 *                          code="jp.ossc.nimbus.service.repository.JNDIRepositoryService" /&gt;
 *             &lt;/depends&gt;
 *             &lt;depends&gt;
 *                 &lt;service name="Cluster"
 *                          code="jp.ossc.nimbus.service.keepalive.ClusterService"&gt;
 *                     &lt;attribute name="LocalAddress"&gt;0.0.0.0&lt;/attribute&gt;
 *                     &lt;attribute name="MulticastGroupAddress"&gt;224.1.1.1&lt;/attribute&gt;
 *                     &lt;attribute name="HeartBeatRetryCount"&gt;2&lt;/attribute&gt;
 *                     &lt;attribute name="JoinOnStart"&gt;false&lt;/attribute&gt;
 *                 &lt;/service&gt;
 *             &lt;/depends&gt;
 *         &lt;/service&gt;
 *         
 *         &lt;service name="ServerConnectionFactory"
 *                  code="jp.ossc.nimbus.service.publish.tcp.ConnectionFactoryService" /&gt;
 *         
 *     &lt;/manager&gt;
 *     
 * &lt;/nimbus&gt;
 * </pre>
 *
 * @author M.Takata
 */
public class SharedContextService<K> extends DefaultContextService<K, Object>
 implements SharedContext<K>, RequestMessageListener, CacheRemoveListener<Object>, ClusterListener, SharedContextServiceMBean<K, Object>, java.io.Serializable{
    
    private static final long serialVersionUID = -7616415086512838961L;
    
    private ServiceName requestConnectionFactoryServiceName;
    private RequestServerConnection serverConnection;
    private MessageReceiver messageReceiver;
    
    private ServiceName clusterServiceName;
    private ClusterService cluster;
    private ServiceName clientCacheMapServiceName;
    private ServiceName serverCacheMapServiceName;
    private CacheMap<K, Object> clientCacheMap;
    private CacheMap<K, Object> serverCacheMap;
    private CacheMap<K, Object> cacheMap;
    private ServiceName sharedContextStoreServiceName;
    private SharedContextStore<K> sharedContextStore;
    private boolean isClient;
    private ServiceName[] sharedContextUpdateListenerServiceNames;
    
    private long synchronizeTimeout = 5000l;
    
    private long defaultTimeout = 1000l;
    
    private String subject = DEFAULT_SUBJECT;
    private String clientSubject;
    
    private boolean isSynchronizeOnStart = true;
    private boolean isSaveOnlyMain;
    private boolean isLoadOnStart;
    private boolean isClearBeforeSave = true;
    
    private Map<Object, Lock> keyLockMap;
    private Map<Object, Set<K>> idLocksMap;
    private Message targetMessage;
    private List<SharedContextUpdateListener<K>> updateListeners;
    
    @Override
    public void setRequestConnectionFactoryServiceName(ServiceName name){
        requestConnectionFactoryServiceName = name;
    }
    @Override
    public ServiceName getRequestConnectionFactoryServiceName(){
        return requestConnectionFactoryServiceName;
    }
    
    @Override
    public void setClusterServiceName(ServiceName name){
        clusterServiceName = name;
    }
    @Override
    public ServiceName getClusterServiceName(){
        return clusterServiceName;
    }
    
    @Override
    public void setClientCacheMapServiceName(ServiceName name){
        clientCacheMapServiceName = name;
    }
    @Override
    public ServiceName getClientCacheMapServiceName(){
        return clientCacheMapServiceName;
    }
    
    @Override
    public void setServerCacheMapServiceName(ServiceName name){
        serverCacheMapServiceName = name;
    }
    @Override
    public ServiceName getServerCacheMapServiceName(){
        return serverCacheMapServiceName;
    }
    
    @Override
    public void setSharedContextStoreServiceName(ServiceName name){
        sharedContextStoreServiceName = name;
    }
    @Override
    public ServiceName getSharedContextStoreServiceName(){
        return sharedContextStoreServiceName;
    }
    
    @Override
    public void setSubject(String subject){
        this.subject = subject;
    }
    @Override
    public String getSubject(){
        return subject;
    }
    
    @Override
    public synchronized 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);
            }
            if(cacheMap != null){
                Object[] keys = null;
                synchronized(context){
                    keys = (Object[])super.keySet().toArray();
                }
                for(int i = 0; i < keys.length; i++){
                    cacheMap.remove(keys[i]);
                }
            }
            if(isClient){
                if(clientCacheMapServiceName != null){
                    cacheMap = ServiceManagerFactory.getServiceObject(clientCacheMapServiceName);
                }else if(clientCacheMap != null){
                    cacheMap = clientCacheMap;
                }
            }else{
                if(serverCacheMapServiceName != null){
                    cacheMap = ServiceManagerFactory.getServiceObject(serverCacheMapServiceName);
                }else if(serverCacheMap != null){
                    cacheMap = serverCacheMap;
                }
            }
            synchronize();
        }
    }
    
    @Override
    public boolean isClient(){
        return isClient;
    }
    
    @Override
    public void setSynchronizeOnStart(boolean isSynch){
        isSynchronizeOnStart = isSynch;
    }
    @Override
    public boolean isSynchronizeOnStart(){
        return isSynchronizeOnStart;
    }
    
    @Override
    public void setSaveOnlyMain(boolean isSave){
        isSaveOnlyMain = isSave;
    }
    @Override
    public boolean isSaveOnlyMain(){
        return isSaveOnlyMain;
    }
    
    @Override
    public void setLoadOnStart(boolean isLoad){
        isLoadOnStart = isLoad;
    }
    @Override
    public boolean isLoadOnStart(){
        return isLoadOnStart;
    }
    
    @Override
    public void setClearBeforeSave(boolean isClear){
        isClearBeforeSave = isClear;
    }
    @Override
    public boolean isClearBeforeSave(){
        return isClearBeforeSave;
    }
    
    @Override
    public void setSynchronizeTimeout(long timeout){
        synchronizeTimeout = timeout;
    }
    @Override
    public long getSynchronizeTimeout(){
        return synchronizeTimeout;
    }
    
    @Override
    public void setDefaultTimeout(long timeout){
        defaultTimeout = timeout;
    }
    @Override
    public long getDefaultTimeout(){
        return defaultTimeout;
    }
    
    public void setClientCacheMap(CacheMap<K, Object> map){
        clientCacheMap = map;
    }
    public void setServerCacheMap(CacheMap<K, Object> map){
        serverCacheMap = map;
    }
    
    public void setSharedContextUpdateListenerServiceNames(ServiceName[] names){
        sharedContextUpdateListenerServiceNames = names;
    }
    public ServiceName[] getSharedContextUpdateListenerServiceNames(){
        return sharedContextUpdateListenerServiceNames;
    }
    
    /**
     * サービスの生成処理を行う。<p>
     *
     * @exception Exception サービスの生成処理に失敗した場合
     */
    @Override
    public void createService() throws Exception{
        super.createService();
        keyLockMap = Collections.synchronizedMap(new HashMap<Object, Lock>());
        idLocksMap = Collections.synchronizedMap(new HashMap<Object, Set<K>>());
    }
    
    /**
     * サービスの開始処理を行う。<p>
     *
     * @exception Exception サービスの開始処理に失敗した場合
     */
    @SuppressWarnings("unchecked")
    public void startService() throws Exception{
        super.startService();
        if(requestConnectionFactoryServiceName == null){
            throw new IllegalArgumentException("RequestConnectionFactoryServiceName must be specified.");
        }
        if(isClient){
            if(clientCacheMapServiceName != null){
                cacheMap = ServiceManagerFactory.getServiceObject(clientCacheMapServiceName);
            }else if(clientCacheMap != null){
                cacheMap = clientCacheMap;
            }
        }else{
            if(serverCacheMapServiceName != null){
                cacheMap = ServiceManagerFactory.getServiceObject(serverCacheMapServiceName);
            }else if(serverCacheMap != null){
                cacheMap = serverCacheMap;
            }
        }
        if(sharedContextStoreServiceName != null){
            sharedContextStore = ServiceManagerFactory.getServiceObject(sharedContextStoreServiceName);
        }else if(isLoadOnStart){
            throw new IllegalArgumentException("SharedContextStoreServiceName 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();
        targetMessage = serverConnection.createMessage(subject, null);
        messageReceiver = ServiceManagerFactory.getServiceObject(requestConnectionFactoryServiceName);
        clientSubject = subject + CLIENT_SUBJECT_SUFFIX;
        messageReceiver.addSubject(this, isClient ? clientSubject :  subject);
        if(clusterServiceName == null){
            throw new IllegalArgumentException("ClusterServiceName must be specified.");
        }
        cluster = ServiceManagerFactory.getServiceObject(clusterServiceName);
        cluster.addClusterListener(this);
        if(isMain() && isLoadOnStart){
            sharedContextStore.load(this);
        }
        if(isSynchronizeOnStart && !isMain()){
            synchronize();
        }
    }
    
    /**
     * サービスの停止処理を行う。<p>
     *
     * @exception Exception サービスの停止処理に失敗した場合
     */
    @Override
    public void stopService() throws Exception{
        if(messageReceiver != null){
            messageReceiver.removeMessageListener(this);
        }
        if(cluster != null){
            cluster.removeClusterListener(this);
        }
        super.stopService();
    }
    
    /**
     * サービスの破棄処理を行う。<p>
     * インスタンス変数を破棄する。<br>
     *
     * @exception Exception サービスの破棄処理に失敗した場合
     */
    @Override
    public void destroyService() throws Exception{
        keyLockMap = null;
        idLocksMap = null;
        super.destroyService();
    }
    
    @Override
    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 SharedContextEvent(SharedContextEvent.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);
            }
        }
    }
    
    @Override
    public synchronized void save(long timeout) throws Exception{
        if(!isMain()){
            try{
                Message message = serverConnection.createMessage(subject, null);
                Set<Object> receiveClients = serverConnection.getReceiveClientIds(message);
                if(receiveClients.size() != 0){
                    message.setObject(new SharedContextEvent(SharedContextEvent.EVENT_SAVE));
                    Message[] responses = serverConnection.request(
                        message,
                        isClient ? clientSubject : subject,
                        null,
                        isSaveOnlyMain ? 1 : 0,
                        timeout
                    );
                    if(responses != null
                        && (isSaveOnlyMain ? responses.length > 0 : responses.length >= receiveClients.size())
                    ){
                        for(int i = 0; i < responses.length; i++){
                            Object ret = responses[i].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);
            }
        }
        if(!isClient && (isMain() || !isSaveOnlyMain)){
            if(sharedContextStore != null){
                if(isClearBeforeSave){
                    sharedContextStore.clear();
                }
                sharedContextStore.save(this);
            }else{
                throw new UnsupportedOperationException();
            }
        }
    }
    
    @Override
    public void synchronize() throws SharedContextSendException, SharedContextTimeoutException{
        synchronize(synchronizeTimeout);
    }
    
    @Override
    public synchronized void synchronize(long timeout) throws SharedContextSendException, SharedContextTimeoutException{
        if(isClient){
            if(cacheMap != null){
                Object[] keys = null;
                synchronized(context){
                    keys = (Object[])super.keySet().toArray();
                }
                for(int i = 0; i < keys.length; i++){
                    cacheMap.remove(keys[i]);
                }
            }
            super.clear();
        }else if(isMain()){
            try{
                Message message = serverConnection.createMessage(subject, null);
                message.setSubject(clientSubject, null);
                Set<Object> receiveClients = serverConnection.getReceiveClientIds(message);
                if(receiveClients.size() != 0){
                    message.setObject(new SharedContextEvent(SharedContextEvent.EVENT_SYNCH, null, new Long(timeout)));
                    Message[] responses = serverConnection.request(message, 0, timeout);
                    if(responses != null && responses.length >= receiveClients.size()){
                        for(int i = 0; i < responses.length; i++){
                            if(responses[i].getObject() == null || !((Boolean)responses[i].getObject()).booleanValue()){
                                throw new SharedContextSendException("It faild to synchronize.");
                            }
                        }
                    }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, null);
                message.setObject(new SharedContextEvent(SharedContextEvent.EVENT_GET_ALL));
                Message[] responses = serverConnection.request(message, 1, timeout);
                if(responses != null && responses.length > 0){
                    @SuppressWarnings("unchecked")
                    Map<K, Object> result = (Map<K, Object>)responses[0].getObject();
                    super.clear();
                    if(result != null){
                        Iterator<Map.Entry<K, Object>> entries = result.entrySet().iterator();
                        while(entries.hasNext()){
                            Map.Entry<K, Object> entry = entries.next();
                            super.put(entry.getKey(), wrapCachedReference(entry.getKey(), entry.getValue()));
                        }
                    }
                }else{
                    throw new SharedContextTimeoutException();
                }
            }catch(MessageException e){
                throw new SharedContextSendException(e);
            }catch(MessageSendException e){
                throw new SharedContextSendException(e);
            }
        }
    }
    
    protected Object wrapCachedReference(K key, Object value){
        if(cacheMap == null){
            return value;
        }else{
            cacheMap.put(key, value);
            CachedReference<Object> ref = cacheMap.getCachedReference(key);
            ref.addCacheRemoveListener(this);
            return ref;
        }
    }
    
    protected Object unwrapCachedReference(Object value, boolean notify, boolean remove){
        if(value == null){
            return null;
        }
        if(cacheMap == null){
            return value;
        }else{
            @SuppressWarnings("unchecked")
            CachedReference<Object> ref = (CachedReference<Object>)value;
            if(remove){
                ref.remove(this);
            }
            return ref.get(this, notify);
        }
    }
    
    @Override
    public void lock(K key) throws SharedContextSendException, SharedContextTimeoutException{
        lock(key, defaultTimeout);
    }
    
    @Override
    public void lock(K key, long timeout) throws SharedContextSendException, SharedContextTimeoutException{
        final Object id = cluster.getUID();
        Lock lock = null;
        synchronized(keyLockMap){
            lock = (Lock)keyLockMap.get(key);
            if(lock == null){
                lock = new Lock(key);
                keyLockMap.put(key, lock);
            }
        }
        Object lockedOwner = lock.getOwner();
        if(id.equals(lockedOwner)){
            return;
        }
        if(isMain()){
            final long start = System.currentTimeMillis();
            if(lock.acquire(id, timeout)){
                final boolean isNoTimeout = timeout <= 0;
                timeout = isNoTimeout ? timeout : timeout - (System.currentTimeMillis() - start);
                if(!isNoTimeout && timeout < 0){
                    lock.release(id, false);
                    throw new SharedContextTimeoutException();
                }else{
                    try{
                        Message message = serverConnection.createMessage(subject, key == null ? null : key.toString());
                        message.setSubject(clientSubject, key == null ? null : key.toString());
                        Set<Object> receiveClients = serverConnection.getReceiveClientIds(message);
                        if(receiveClients.size() != 0){
                            message.setObject(new SharedContextEvent(SharedContextEvent.EVENT_GOT_LOCK, key, new Object[]{id, new Long(timeout)}));
                            Message[] responses = serverConnection.request(message, 0, timeout);
                            if(responses != null && responses.length >= receiveClients.size()){
                                for(int i = 0; i < responses.length; i++){
                                    Object ret = responses[i].getObject();
                                    if(ret instanceof Throwable){
                                        unlock(key);
                                        if(ret instanceof RuntimeException){
                                            throw (RuntimeException)ret;
                                        }else if(ret instanceof Error){
                                            throw (Error)ret;
                                        }else{
                                            throw new SharedContextSendException((Throwable)ret);
                                        }
                                    }else if(ret == null || !((Boolean)ret).booleanValue()){
                                        unlock(key);
                                        throw new SharedContextTimeoutException();
                                    }
                                }
                            }else{
                                unlock(key);
                                throw new SharedContextTimeoutException();
                            }
                        }
                    }catch(MessageException e){
                        unlock(key);
                        throw new SharedContextSendException(e);
                    }catch(MessageSendException e){
                        unlock(key);
                        throw new SharedContextSendException(e);
                    }catch(RuntimeException e){
                        unlock(key);
                        throw e;
                    }catch(Error e){
                        unlock(key);
                        throw e;
                    }catch(Throwable th){
                        unlock(key);
                        throw new SharedContextSendException(th);
                    }
                }
            }else{
                throw new SharedContextTimeoutException();
            }
        }else{
            try{
                Message message = serverConnection.createMessage(subject, key == null ? null : key.toString());
                message.setObject(new SharedContextEvent(SharedContextEvent.EVENT_GET_LOCK, key, new Long(timeout)));
                Message[] responses = serverConnection.request(
                    message,
                    isClient ? clientSubject : subject,
                    key == null ? null : key.toString(),
                    1,
                    timeout
                );
                if(responses != null && responses.length > 0){
                    Object ret = responses[0].getObject();
                    if(ret instanceof Throwable){
                        if(ret instanceof RuntimeException){
                            unlock(key);
                            throw (RuntimeException)ret;
                        }else if(ret instanceof Error){
                            unlock(key);
                            throw (Error)ret;
                        }else{
                            unlock(key);
                            throw new SharedContextSendException((Throwable)ret);
                        }
                    }else if(ret == null || !((Boolean)ret).booleanValue()){
                        unlock(key);
                        throw new SharedContextTimeoutException();
                    }
                }else{
                    unlock(key);
                    throw new SharedContextTimeoutException();
                }
            }catch(MessageException e){
                unlock(key);
                throw new SharedContextSendException(e);
            }catch(MessageSendException e){
                unlock(key);
                throw new SharedContextSendException(e);
            }
            if(!lock.acquire(id, timeout)){
                unlock(key);
                throw new SharedContextTimeoutException();
            }
        }
    }
    
    @Override
    public boolean unlock(K key) throws SharedContextSendException{
        return unlock(key, false);
    }
    
    @Override
    public boolean unlock(K key, boolean force) throws SharedContextSendException{
        Lock lock = null;
        synchronized(keyLockMap){
            lock = (Lock)keyLockMap.get(key);
        }
        Object id = cluster.getUID();
        if(force && lock != null && lock.getOwner() != null){
            id = lock.getOwner();
        }
        try{
            Message message = serverConnection.createMessage(subject, key == null ? null : key.toString());
            message.setSubject(clientSubject, key == null ? null : key.toString());
            message.setObject(new SharedContextEvent(SharedContextEvent.EVENT_RELEASE_LOCK, key, id));
            serverConnection.sendAsynch(message);
        }catch(MessageException e){
            throw new SharedContextSendException(e);
        }catch(MessageSendException e){
            throw new SharedContextSendException(e);
        }
        boolean result = lock == null;
        if(lock != null){
            result = lock.release(id, force);
        }
        return result;
    }
    
    @Override
    public Object getLockOwner(K key){
        Lock lock = null;
        synchronized(keyLockMap){
            lock = (Lock)keyLockMap.get(key);
        }
        return lock == null ? null : lock.getOwner();
    }
    
    @Override
    public Object put(K key, Object value) throws SharedContextSendException{
        return put(key, value, defaultTimeout);
    }
    
    @Override
    public Object put(K key, Object value, long timeout) throws SharedContextSendException, SharedContextTimeoutException{
        if(updateListeners != null){
            for(int i = 0; i < updateListeners.size(); i++){
                if(!updateListeners.get(i).onPut(this, true, key, value)){
                    return null;
                }
            }
        }
        Object result = null;
        try{
            Message message = serverConnection.createMessage(subject, key == null ? null : key.toString());
            message.setSubject(clientSubject, key == null ? null : key.toString());
            Set<Object> receiveClients = serverConnection.getReceiveClientIds(message);
            if(receiveClients.size() != 0){
                message.setObject(new SharedContextEvent(SharedContextEvent.EVENT_PUT, key, value));
                Message[] responses = serverConnection.request(
                    message,
                    isClient ? clientSubject : subject,
                    key == null ? null : key.toString(),
                    0,
                    timeout
                );
                if(responses != null && responses.length >= receiveClients.size()){
                    for(int i = 0; i < responses.length; i++){
                        if(responses[i].getObject() != null){
                            result = responses[i].getObject();
                            break;
                        }
                    }
                }else{
                    throw new SharedContextTimeoutException();
                }
            }
        }catch(MessageException e){
            throw new SharedContextSendException(e);
        }catch(MessageSendException e){
            throw new SharedContextSendException(e);
        }
        if(isClient){
            if(super.containsKey(key)){
                result = super.put(key, wrapCachedReference(key, value));
                result = unwrapCachedReference(result, false, true);
            }
        }else{
            result = super.put(key, wrapCachedReference(key, value));
            result = unwrapCachedReference(result, false, true);
        }
        return result;
    }
    
    @Override
    public void putAsynch(K key, Object value) throws SharedContextSendException{
        if(updateListeners != null){
            for(int i = 0; i < updateListeners.size(); i++){
                if(!updateListeners.get(i).onPut(this, true, key, value)){
                    return;
                }
            }
        }
        try{
            Message message = serverConnection.createMessage(subject, key == null ? null : key.toString());
            message.setSubject(clientSubject, key == null ? null : key.toString());
            message.setObject(new SharedContextEvent(SharedContextEvent.EVENT_PUT, key, value));
            serverConnection.sendAsynch(message);
        }catch(MessageException e){
            throw new SharedContextSendException(e);
        }catch(MessageSendException e){
            throw new SharedContextSendException(e);
        }
        Object removed = null;
        if(isClient){
            if(super.containsKey(key)){
                removed = super.put(key, wrapCachedReference(key, value));
            }
        }else{
            removed = super.put(key, wrapCachedReference(key, value));
        }
        unwrapCachedReference(removed, false, true);
    }
    
    @Override
    public void update(K key, SharedContextValueDifference diff) throws SharedContextSendException{
        update(key, diff, defaultTimeout);
    }
    
    @SuppressWarnings("unchecked")
    @Override
    public void update(K key, SharedContextValueDifference diff, long timeout) throws SharedContextSendException, SharedContextTimeoutException{
        if(diff == null){
            return;
        }
        if(updateListeners != null){
            for(int i = 0; i < updateListeners.size(); i++){
                if(!updateListeners.get(i).onUpdate(this, true, key, diff)){
                    return;
                }
            }
        }
        try{
            Message message = serverConnection.createMessage(subject, key == null ? null : key.toString());
            message.setSubject(clientSubject, key == null ? null : key.toString());
            Set<Object> receiveClients = serverConnection.getReceiveClientIds(message);
            if(receiveClients.size() != 0){
                message.setObject(new SharedContextEvent(SharedContextEvent.EVENT_UPDATE, key, diff));
                Message[] responses = serverConnection.request(
                    message,
                    isClient ? clientSubject : subject,
                    key == null ? null : key.toString(),
                    0,
                    timeout
                );
                if(responses != null && responses.length >= receiveClients.size()){
                    for(int i = 0; i < responses.length; i++){
                        Object ret = responses[i].getObject();
                        if(ret != null){
                            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);
        }
        if(!isClient || super.containsKey(key)){
            Object current = super.get(key);
            Object currentValue = unwrapCachedReference(current, false, false);
            if(currentValue == null){
                throw new SharedContextUpdateException("Current value is null. key=" + key);
            }else if(currentValue instanceof SharedContextValueDifferenceSupport){
                ((SharedContextValueDifferenceSupport)currentValue).update(diff);
            }else{
                throw new SharedContextUpdateException("Not support SharedContextValueDifference. key=" + key + ", value=" + currentValue);
            }
            if(current instanceof CachedReference){
                try{
                    ((CachedReference<Object>)current).set(this, currentValue);
                }catch(IllegalCachedReferenceException e){
                    throw new SharedContextUpdateException(e);
                }
            }
        }
    }
    
    @SuppressWarnings("unchecked")
    @Override
    public void updateAsynch(K key, SharedContextValueDifference diff) throws SharedContextSendException{
        if(diff == null){
            return;
        }
        if(updateListeners != null){
            for(int i = 0; i < updateListeners.size(); i++){
                if(!updateListeners.get(i).onUpdate(this, true, key, diff)){
                    return;
                }
            }
        }
        try{
            Message message = serverConnection.createMessage(subject, key == null ? null : key.toString());
            message.setSubject(clientSubject, key == null ? null : key.toString());
            message.setObject(new SharedContextEvent(SharedContextEvent.EVENT_UPDATE, key, diff));
            serverConnection.sendAsynch(message);
        }catch(MessageException e){
            throw new SharedContextSendException(e);
        }catch(MessageSendException e){
            throw new SharedContextSendException(e);
        }
        if(!isClient || super.containsKey(key)){
            Object current = super.get(key);
            Object currentValue = unwrapCachedReference(current, false, false);
            if(currentValue == null){
                throw new SharedContextUpdateException("Current value is null. key=" + key);
            }else if(currentValue instanceof SharedContextValueDifferenceSupport){
                ((SharedContextValueDifferenceSupport)currentValue).update(diff);
            }else{
                throw new SharedContextUpdateException("Not support SharedContextValueDifference. key=" + key + ", value=" + currentValue);
            }
            if(current instanceof CachedReference){
                try{
                    ((CachedReference<Object>)current).set(this, currentValue);
                }catch(IllegalCachedReferenceException e){
                    throw new SharedContextUpdateException(e);
                }
            }
        }
    }
    
    @Override
    public Object remove(Object key) throws SharedContextSendException{
        return remove(key, defaultTimeout);
    }
    
    @Override
    public Object remove(Object key, long timeout) throws SharedContextSendException, SharedContextTimeoutException{
        if(isMain() && !super.containsKey(key)){
            return null;
        }
        if(updateListeners != null){
            for(int i = 0; i < updateListeners.size(); i++){
                if(!updateListeners.get(i).onRemove(this, true, key)){
                    return null;
                }
            }
        }
        Object result = null;
        try{
            Message message = serverConnection.createMessage(subject, key == null ? null : key.toString());
            message.setSubject(clientSubject, key == null ? null : key.toString());
            Set<Object> receiveClients = serverConnection.getReceiveClientIds(message);
            if(receiveClients.size() != 0){
                message.setObject(new SharedContextEvent(SharedContextEvent.EVENT_REMOVE, key));
                Message[] responses = serverConnection.request(
                    message,
                    isClient ? clientSubject : subject,
                    key == null ? null : key.toString(),
                    0,
                    timeout
                );
                if(responses != null && responses.length >= receiveClients.size()){
                    if(isMain()){
                        result = super.remove(key);
                        result = unwrapCachedReference(result, false, true);
                    }else{
                        Object removed = super.remove(key);
                        unwrapCachedReference(removed, false, true);
                        for(int i = 0; i < responses.length; i++){
                            if(responses[i].getObject() != null){
                                result = responses[i].getObject();
                                break;
                            }
                        }
                    }
                }else{
                    throw new SharedContextTimeoutException();
                }
            }else{
                result = super.remove(key);
                result = unwrapCachedReference(result, false, true);
            }
        }catch(MessageException e){
            throw new SharedContextSendException(e);
        }catch(MessageSendException e){
            throw new SharedContextSendException(e);
        }
        return result;
    }
    
    @Override
    public void removeAsynch(Object key) throws SharedContextSendException{
        if(updateListeners != null){
            for(int i = 0; i < updateListeners.size(); i++){
                if(!updateListeners.get(i).onRemove(this, true, key)){
                    return;
                }
            }
        }
        try{
            Message message = serverConnection.createMessage(subject, key == null ? null : key.toString());
            message.setSubject(clientSubject, key == null ? null : key.toString());
            message.setObject(new SharedContextEvent(SharedContextEvent.EVENT_REMOVE, key));
            serverConnection.sendAsynch(message);
        }catch(MessageException e){
            throw new SharedContextSendException(e);
        }catch(MessageSendException e){
            throw new SharedContextSendException(e);
        }
        Object removed = super.remove(key);
        unwrapCachedReference(removed, false, true);
    }
    
    @Override
    public void putAll(Map<? extends K,? extends Object> t){
        putAll(t, defaultTimeout);
    }
    
    @Override
    public synchronized void putAll(Map<? extends K,? extends Object> t, long timeout) throws SharedContextSendException, SharedContextTimeoutException{
        if(t.size() == 0){
            return;
        }
        if(updateListeners != null){
            Map<K,Object> tmpMap = new LinkedHashMap<K,Object>();
            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();
                for(int i = 0; i < updateListeners.size(); i++){
                    if(!isClient || super.containsKey(entry.getKey())){
                        if(updateListeners.get(i).onPut(this, true, entry.getKey(), entry.getValue())){
                            tmpMap.put((K)entry.getKey(), entry.getValue());
                        }
                    }
                }
            }
            if(tmpMap.size() == 0){
                return;
            }
            t = tmpMap;
        }
        try{
            Message message = serverConnection.createMessage(subject, null);
            message.setSubject(clientSubject, null);
            Set<Object> receiveClients = serverConnection.getReceiveClientIds(message);
            if(receiveClients.size() != 0){
                message.setObject(new SharedContextEvent(SharedContextEvent.EVENT_PUT_ALL, null, t));
                Message[] responses = serverConnection.request(
                    message,
                    isClient ? clientSubject : subject,
                    null,
                    0,
                    timeout
                );
                if(responses == null || responses.length < receiveClients.size()){
                    throw new SharedContextTimeoutException();
                }
            }
        }catch(MessageException e){
            throw new SharedContextSendException(e);
        }catch(MessageSendException e){
            throw new SharedContextSendException(e);
        }
        Iterator<?> entries = t.entrySet().iterator();
        while(entries.hasNext()){
            @SuppressWarnings("unchecked")
            Map.Entry<K,Object> entry = (Map.Entry<K,Object>)entries.next();
            if(isClient){
                if(super.containsKey(entry.getKey())){
                    Object old = super.put(entry.getKey(), wrapCachedReference(entry.getKey(), entry.getValue()));
                    unwrapCachedReference(old, false, true);
                }
            }else{
                Object old = super.put(entry.getKey(), wrapCachedReference(entry.getKey(), entry.getValue()));
                unwrapCachedReference(old, false, true);
            }
        }
    }
    
    @Override
    public synchronized void putAllAsynch(Map<? extends K,? extends Object> t) throws SharedContextSendException{
        if(t.size() == 0){
            return;
        }
        if(updateListeners != null){
            Map<K,Object> tmpMap = new LinkedHashMap<K,Object>();
            Iterator<?> entries = t.entrySet().iterator();
            while(entries.hasNext()){
                @SuppressWarnings("unchecked")
                Map.Entry<K,Object> entry = (Map.Entry<K,Object>)entries.next();
                for(int i = 0; i < updateListeners.size(); i++){
                    if(!isClient || super.containsKey(entry.getKey())){
                        if(updateListeners.get(i).onPut(this, true, entry.getKey(), entry.getValue())){
                            tmpMap.put(entry.getKey(), entry.getValue());
                        }
                    }
                }
            }
            if(tmpMap.size() == 0){
                return;
            }
            t = tmpMap;
        }
        try{
            Message message = serverConnection.createMessage(subject, null);
            message.setSubject(clientSubject, null);
            message.setObject(new SharedContextEvent(SharedContextEvent.EVENT_PUT_ALL, null, t));
            serverConnection.sendAsynch(message);
        }catch(MessageException e){
            throw new SharedContextSendException(e);
        }catch(MessageSendException e){
            throw new SharedContextSendException(e);
        }
        Iterator<?> entries = t.entrySet().iterator();
        while(entries.hasNext()){
            @SuppressWarnings("unchecked")
            Map.Entry<? extends K,? extends Object> entry = (java.util.Map.Entry<? extends K, ? extends Object>)entries.next();
            if(isClient){
                if(super.containsKey(entry.getKey())){
                    Object old = super.put(entry.getKey(), wrapCachedReference(entry.getKey(), entry.getValue()));
                    unwrapCachedReference(old, false, true);
                }
            }else{
                Object old = super.put(entry.getKey(), wrapCachedReference(entry.getKey(), entry.getValue()));
                unwrapCachedReference(old, false, true);
            }
        }
    }
    
    @Override
    public void clear() throws SharedContextSendException{
        clear(defaultTimeout);
    }
    
    @Override
    public synchronized void clear(long timeout) throws SharedContextSendException, SharedContextTimeoutException{
        if(isMain() && size() == 0){
            return;
        }
        try{
            Message message = serverConnection.createMessage(subject, null);
            message.setSubject(clientSubject, null);
            Set<Object> receiveClients = serverConnection.getReceiveClientIds(message);
            if(receiveClients.size() != 0){
                message.setObject(new SharedContextEvent(SharedContextEvent.EVENT_CLEAR));
                Message[] responses = serverConnection.request(
                    message,
                    isClient ? clientSubject : subject,
                    null,
                    0,
                    timeout
                );
                if(responses == null || responses.length < receiveClients.size()){
                    throw new SharedContextTimeoutException();
                }
            }
        }catch(MessageException e){
            throw new SharedContextSendException(e);
        }catch(MessageSendException e){
            throw new SharedContextSendException(e);
        }
        Object[] keys = null;
        synchronized(context){
            keys = super.keySet().toArray();
        }
        for(int i = 0; i < keys.length; i++){
            if(updateListeners != null){
                boolean isRemove = true;
                for(int j = 0; j < updateListeners.size(); j++){
                    if(!updateListeners.get(j).onRemove(this, true, keys[i])){
                        isRemove = false;
                        break;
                    }
                }
                if(!isRemove){
                    continue;
                }
            }
            Object removed = super.remove(keys[i]);
            unwrapCachedReference(removed, false, true);
        }
    }
    
    @Override
    public synchronized void clearAsynch() throws SharedContextSendException{
        if(isMain() && size() == 0){
            return;
        }
        try{
            Message message = serverConnection.createMessage(subject, null);
            message.setSubject(clientSubject, null);
            message.setObject(new SharedContextEvent(SharedContextEvent.EVENT_CLEAR));
            serverConnection.sendAsynch(message);
        }catch(MessageException e){
            throw new SharedContextSendException(e);
        }catch(MessageSendException e){
            throw new SharedContextSendException(e);
        }
        Object[] keys = null;
        synchronized(context){
            keys = super.keySet().toArray();
        }
        for(int i = 0; i < keys.length; i++){
            if(updateListeners != null){
                boolean isRemove = true;
                for(int j = 0; j < updateListeners.size(); j++){
                    if(!updateListeners.get(j).onRemove(this, true, keys[i])){
                        isRemove = false;
                        break;
                    }
                }
                if(!isRemove){
                    continue;
                }
            }
            Object removed = super.remove(keys[i]);
            unwrapCachedReference(removed, false, true);
        }
    }
    
    @Override
    public Object get(Object key) throws SharedContextSendException{
        return get(key, defaultTimeout);
    }
    
    @SuppressWarnings("unchecked")
    @Override
    public Object get(Object key, long timeout) throws SharedContextSendException, SharedContextTimeoutException{
        Object result = null;
        if(isClient){
            if(super.containsKey(key)){
                result = super.get(key);
                result = unwrapCachedReference(result, true, false);
            }else{
                try{
                    Message message = serverConnection.createMessage(subject, key == null ? null : key.toString());
                    message.setObject(new SharedContextEvent(SharedContextEvent.EVENT_GET, key));
                    Message[] responses = serverConnection.request(
                        message,
                        clientSubject,
                        key == null ? null : key.toString(),
                        1,
                        timeout
                    );
                    if(responses != null && responses.length > 0){
                        result = responses[0].getObject();
                        super.put((K)key, wrapCachedReference((K)key, result));
                    }else{
                        throw new SharedContextTimeoutException();
                    }
                }catch(MessageException e){
                    throw new SharedContextSendException(e);
                }catch(MessageSendException e){
                    throw new SharedContextSendException(e);
                }
            }
        }else{
            result = super.get(key);
            result = unwrapCachedReference(result, true, false);
        }
        return result;
    }
    
    @Override
    public Set<K> keySet() throws SharedContextSendException{
        return keySet(defaultTimeout);
    }
    
    @SuppressWarnings("unchecked")
    @Override
    public synchronized Set<K> keySet(long timeout) throws SharedContextSendException, SharedContextTimeoutException{
        if(!isClient){
            return super.keySet();
        }else{
            try{
                Message message = serverConnection.createMessage(subject, null);
                message.setObject(new SharedContextEvent(SharedContextEvent.EVENT_KEY_SET));
                Message[] responses = serverConnection.request(
                    message,
                    clientSubject,
                    null,
                    1,
                    timeout
                );
                if(responses != null && responses.length > 0){
                    return (Set<K>)responses[0].getObject();
                }else{
                    throw new SharedContextTimeoutException();
                }
            }catch(MessageException e){
                throw new SharedContextSendException(e);
            }catch(MessageSendException e){
                throw new SharedContextSendException(e);
            }
        }
    }
    
    @Override
    public int size() throws SharedContextSendException{
        return size(defaultTimeout);
    }
    
    @Override
    public synchronized int size(long timeout) throws SharedContextSendException, SharedContextTimeoutException{
        if(!isClient){
            return super.size();
        }else{
            try{
                Message message = serverConnection.createMessage(subject, null);
                message.setObject(new SharedContextEvent(SharedContextEvent.EVENT_SIZE));
                Message[] responses = serverConnection.request(
                    message,
                    clientSubject,
                    null,
                    1,
                    timeout
                );
                if(responses != null && responses.length > 0){
                    return ((Integer)responses[0].getObject()).intValue();
                }else{
                    throw new SharedContextTimeoutException();
                }
            }catch(MessageException e){
                throw new SharedContextSendException(e);
            }catch(MessageSendException e){
                throw new SharedContextSendException(e);
            }
        }
    }
    
    @Override
    public boolean isEmpty() throws SharedContextSendException{
        return size() == 0 ? true : false;
    }
    
    @Override
    public boolean containsKey(Object key) throws SharedContextSendException{
        return containsKey(key, defaultTimeout);
    }
    
    @Override
    public boolean containsKey(Object key, long timeout) throws SharedContextSendException, SharedContextTimeoutException{
        if(!isClient){
            return super.containsKey(key);
        }else{
            try{
                Message message = serverConnection.createMessage(subject, key == null ? null : key.toString());
                message.setObject(new SharedContextEvent(SharedContextEvent.EVENT_CONTAINS_KEY, key));
                Message[] responses = serverConnection.request(
                    message,
                    clientSubject,
                    key == null ? null : key.toString(),
                    1,
                    timeout
                );
                if(responses != null && responses.length > 0){
                    return ((Boolean)responses[0].getObject()).booleanValue();
                }else{
                    throw new SharedContextTimeoutException();
                }
            }catch(MessageException e){
                throw new SharedContextSendException(e);
            }catch(MessageSendException e){
                throw new SharedContextSendException(e);
            }
        }
    }
    
    public boolean containsValue(Object value) throws SharedContextSendException{
        return containsValue(value, defaultTimeout);
    }
    
    public boolean containsValue(Object value, long timeout) throws SharedContextSendException, SharedContextTimeoutException{
        if(!isClient){
            if(cacheMap == null){
                return super.containsValue(value);
            }else{
                Object[] keys = null;
                synchronized(context){
                    keys = super.keySet().toArray();
                }
                for(int i = 0; i < keys.length; i++){
                    Object val = cacheMap.get(keys[i]);
                    if(val == null){
                        if(value == null){
                            return true;
                        }
                    }else if(val.equals(value)){
                        return true;
                    }
                }
                return false;
            }
        }else{
            try{
                Message message = serverConnection.createMessage(subject, null);
                message.setObject(new SharedContextEvent(SharedContextEvent.EVENT_CONTAINS_VALUE, null, value));
                Message[] responses = serverConnection.request(
                    message,
                    clientSubject,
                    null,
                    1,
                    timeout
                );
                if(responses != null && responses.length > 0){
                    return ((Boolean)responses[0].getObject()).booleanValue();
                }else{
                    throw new SharedContextTimeoutException();
                }
            }catch(MessageException e){
                throw new SharedContextSendException(e);
            }catch(MessageSendException e){
                throw new SharedContextSendException(e);
            }
        }
    }
    
    public Collection<Object> values(){
        if(isClient){
            throw new UnsupportedOperationException();
        }else{
            if(cacheMap == null){
                return super.values();
            }else{
                List<Object> result = new ArrayList<Object>();
                Object[] keys = null;
                synchronized(context){
                    keys = super.keySet().toArray();
                }
                for(int i = 0; i < keys.length; i++){
                    result.add(cacheMap.get(keys[i]));
                }
                return result;
            }
        }
    }
    
    @SuppressWarnings("unchecked")
    public Set<Map.Entry<K, Object>> entrySet(){
        if(isClient){
            throw new UnsupportedOperationException();
        }else{
            if(cacheMap == null){
                return super.entrySet();
            }else{
                Set<Map.Entry<K, Object>> reuslt = new HashSet<Map.Entry<K, Object>>();
                Map.Entry<K, Object>[] entries = null;
                synchronized(context){
                    entries = cacheMap.entrySet().toArray(new Map.Entry[cacheMap.size()]);
                }
                for(int i = 0; i < entries.length; i++){
                    if(super.containsKey(entries[i].getKey())){
                        reuslt.add(entries[i]);
                    }
                }
                return reuslt;
            }
        }
    }
    
    public boolean isMain(){
        return isMain(cluster.getMembers());
    }
    
    protected 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 addSharedContextUpdateListener(SharedContextUpdateListener<K> listener){
        if(updateListeners == null){
            updateListeners = Collections.synchronizedList(new ArrayList<SharedContextUpdateListener<K>>());
        }
        if(!updateListeners.contains(listener)){
            updateListeners.add(listener);
        }
    }
    
    public void removeSharedContextUpdateListener(SharedContextUpdateListener<K> listener){
        if(updateListeners == null){
            return;
        }
        updateListeners.remove(listener);
    }
    
    public void memberInit(Object myId, List<? extends Object> members){
    }
    
    public void memberChange(List<? extends Object> oldMembers, List<? extends Object> newMembers){
        if(isMain(newMembers)){
            Set<Object> deadMembers = new HashSet<Object>(oldMembers);
            deadMembers.removeAll(newMembers);
            if(deadMembers.size() != 0){
                Iterator<Object> ids = deadMembers.iterator();
                while(ids.hasNext()){
                    Object id = ids.next();
                    Set<K> keySet = idLocksMap.remove(id);
                    if(keySet == null || keySet.size() == 0){
                        continue;
                    }
                    synchronized(keySet){
                        Iterator<K> keys = keySet.iterator();
                        while(keys.hasNext()){
                            try{
                                unlock(keys.next(), true);
                            }catch(SharedContextSendException e){}
                        }
                    }
                }
            }
        }
    }
    
    public void changeMain() throws Exception{
    }
    
    public void changeSub(){
    }
    
    public void removed(CachedReference<Object> ref){
        if(ref == null){
            return;
        }
        @SuppressWarnings("unchecked")
        KeyCachedReference<K, Object> kcr = (KeyCachedReference<K, Object>)ref;
        super.remove(kcr.getKey());
    }
    
    public void onMessage(Message message){
        SharedContextEvent event = null;
        try{
            event = (SharedContextEvent)message.getObject();
        }catch(MessageException e){
            e.printStackTrace();
            return;
        }
        switch(event.type){
        case SharedContextEvent.EVENT_PUT:
            onPut(event, null, null);
            break;
        case SharedContextEvent.EVENT_PUT_ALL:
            onPutAll(event, null, null);
            break;
        case SharedContextEvent.EVENT_REMOVE:
            onRemove(event, null, null);
            break;
        case SharedContextEvent.EVENT_CLEAR:
            onClear(event, null, null);
            break;
        case SharedContextEvent.EVENT_RELEASE_LOCK:
            onReleaseLock(event, null, null);
            break;
        case SharedContextEvent.EVENT_UPDATE:
            onUpdate(event, null, null);
            break;
        default:
        }
    }
    
    @Override
    public Message onRequestMessage(Object sourceId, int sequence, Message message, String responseSubject, String responseKey){
        SharedContextEvent event = null;
        try{
            event = (SharedContextEvent)message.getObject();
        }catch(MessageException e){
            e.printStackTrace();
            return null;
        }
        Message result = null;
        switch(event.type){
        case SharedContextEvent.EVENT_PUT:
            result = onPut(event, responseSubject, responseKey);
            break;
        case SharedContextEvent.EVENT_PUT_ALL:
            result = onPutAll(event, responseSubject, responseKey);
            break;
        case SharedContextEvent.EVENT_REMOVE:
            result = onRemove(event, responseSubject, responseKey);
            break;
        case SharedContextEvent.EVENT_CLEAR:
            result = onClear(event, responseSubject, responseKey);
            break;
        case SharedContextEvent.EVENT_GET:
            result = onGet(event, responseSubject, responseKey);
            break;
        case SharedContextEvent.EVENT_GET_ALL:
            result = onGetAll(event, responseSubject, responseKey);
            break;
        case SharedContextEvent.EVENT_KEY_SET:
            result = onKeySet(event, responseSubject, responseKey);
            break;
        case SharedContextEvent.EVENT_SIZE:
            result = onSize(event, responseSubject, responseKey);
            break;
        case SharedContextEvent.EVENT_CONTAINS_KEY:
            result = onContainsKey(event, responseSubject, responseKey);
            break;
        case SharedContextEvent.EVENT_CONTAINS_VALUE:
            result = onContainsValue(event, responseSubject, responseKey);
            break;
        case SharedContextEvent.EVENT_SYNCH:
            result = onSynchronize(event, sourceId, sequence, responseSubject, responseKey);
            break;
        case SharedContextEvent.EVENT_GET_LOCK:
            result = onGetLock(event, responseSubject, responseKey, sourceId);
            break;
        case SharedContextEvent.EVENT_GOT_LOCK:
            result = onGotLock(event, responseSubject, responseKey);
            break;
        case SharedContextEvent.EVENT_SAVE:
            result = onSave(event, responseSubject, responseKey);
            break;
        case SharedContextEvent.EVENT_LOAD:
            result = onLoad(event, responseSubject, responseKey);
            break;
        case SharedContextEvent.EVENT_UPDATE:
            result = onUpdate(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;
    }
    
    @SuppressWarnings("unchecked")
    protected Message onPut(SharedContextEvent event, String responseSubject, String responseKey){
        Object result = null;
        if(isClient){
            if(super.containsKey(event.key)){
                if(updateListeners != null){
                    for(int i = 0; i < updateListeners.size(); i++){
                        if(!updateListeners.get(i).onPut(this, false, (K)event.key, event.value)){
                            return createResponseMessage(responseSubject, responseKey, result);
                        }
                    }
                }
                super.put((K)event.key, wrapCachedReference((K)event.key, event.value));
            }
            result = null;
        }else{
            if(updateListeners != null){
                for(int i = 0; i < updateListeners.size(); i++){
                    if(!updateListeners.get(i).onPut(this, false, (K)event.key, event.value)){
                        return createResponseMessage(responseSubject, responseKey, result);
                    }
                }
            }
            result = super.put((K)event.key, wrapCachedReference((K)event.key, event.value));
            if(!isMain()){
                result = null;
            }
        }
        return createResponseMessage(responseSubject, responseKey, result);
    }
    
    protected Message onPutAll(SharedContextEvent event, String responseSubject, String responseKey){
        @SuppressWarnings("unchecked")
        Map<K, Object> map = (Map<K, Object>)event.key;
        if(map != null){
            Iterator<Map.Entry<K, Object>> entries = map.entrySet().iterator();
            while(entries.hasNext()){
                Map.Entry<K, Object> entry = entries.next();
                if(isClient){
                    if(super.containsKey(entry.getKey())){
                        boolean isPut = true;
                        if(updateListeners != null){
                            for(int i = 0; i < updateListeners.size(); i++){
                                if(!updateListeners.get(i).onPut(this, false, entry.getKey(), entry.getValue())){
                                    isPut = false;
                                    break;
                                }
                            }
                        }
                        if(isPut){
                            Object old = super.put(entry.getKey(), wrapCachedReference(entry.getKey(), entry.getValue()));
                            unwrapCachedReference(old, false, true);
                        }
                    }
                }else{
                    boolean isPut = true;
                    if(updateListeners != null){
                        for(int i = 0; i < updateListeners.size(); i++){
                            if(!updateListeners.get(i).onPut(this, false, entry.getKey(), entry.getValue())){
                                isPut = false;
                                break;
                            }
                        }
                    }
                    if(isPut){
                        Object old = super.put(entry.getKey(), wrapCachedReference(entry.getKey(), entry.getValue()));
                        unwrapCachedReference(old, false, true);
                    }
                }
            }
        }
        return createResponseMessage(responseSubject, responseKey, null);
    }
    
    @SuppressWarnings("unchecked")
    protected Message onUpdate(SharedContextEvent event, String responseSubject, String responseKey){
        if(!isClient || super.containsKey(event.key)){
            Object current = super.get(event.key);
            Object currentValue = unwrapCachedReference(current, false, false);
            if(currentValue == null){
                return createResponseMessage(responseSubject, responseKey, new SharedContextUpdateException("Current value is null. key=" + event.key));
            }else if(currentValue instanceof SharedContextValueDifferenceSupport){
                if(updateListeners != null){
                    for(int i = 0; i < updateListeners.size(); i++){
                        if(!updateListeners.get(i).onUpdate(this, false, (K)event.key, (SharedContextValueDifference)event.value)){
                            return createResponseMessage(responseSubject, responseKey, null);
                        }
                    }
                }
                try{
                    ((SharedContextValueDifferenceSupport)currentValue).update((SharedContextValueDifference)event.value);
                }catch(Throwable th){
                    return createResponseMessage(responseSubject, responseKey, th);
                }
                if(current instanceof CachedReference){
                    try{
                        ((CachedReference<Object>)current).set(this, currentValue);
                    }catch(IllegalCachedReferenceException e){
                        throw new SharedContextUpdateException(e);
                    }
                }
            }else{
                return createResponseMessage(responseSubject, responseKey, new SharedContextUpdateException("Not support SharedContextValueDifference. key=" + event.key + ", value=" + currentValue));
            }
        }
        return createResponseMessage(responseSubject, responseKey, null);
    }
    
    protected Message onRemove(SharedContextEvent event, String responseSubject, String responseKey){
        if(updateListeners != null){
            for(int i = 0; i < updateListeners.size(); i++){
                if(!updateListeners.get(i).onRemove(this, false, event.key)){
                    return createResponseMessage(responseSubject, responseKey, null);
                }
            }
        }
        Object result = super.remove(event.key);
        result = unwrapCachedReference(result, false, true);
        return createResponseMessage(responseSubject, responseKey, isMain() ? result : null);
    }
    
    protected Message onClear(SharedContextEvent event, String responseSubject, String responseKey){
        Object[] keys = null;
        synchronized(context){
            keys = super.keySet().toArray();
        }
        for(int i = 0; i < keys.length; i++){
            if(updateListeners != null){
                boolean isRemove = true;
                for(int j = 0; j < updateListeners.size(); j++){
                    if(!updateListeners.get(j).onRemove(this, false, keys[i])){
                        isRemove = false;
                        break;
                    }
                }
                if(!isRemove){
                    continue;
                }
            }
            Object result = super.remove(keys[i]);
            result = unwrapCachedReference(result, false, true);
        }
        return createResponseMessage(responseSubject, responseKey, null);
    }
    
    protected Message onGet(SharedContextEvent event, String responseSubject, String responseKey){
        if(isMain()){
            return createResponseMessage(responseSubject, responseKey, unwrapCachedReference(super.get(event.key), true, false));
        }else{
            return null;
        }
    }
    
    @SuppressWarnings("unchecked")
    protected Message onGetAll(SharedContextEvent event, String responseSubject, String responseKey){
        if(isMain()){
            Map<K, Object> result = new HashMap<K, Object>();
            synchronized(context){
                if(cacheMap == null){
                    result.putAll(context);
                }else{
                    Object[] keys = null;
                    synchronized(context){
                        keys = super.keySet().toArray();
                    }
                    for(int i = 0; i < keys.length; i++){
                        if(cacheMap.containsKey(keys[i])){
                            result.put((K)keys[i], cacheMap.get(keys[i]));
                        }
                    }
                }
            }
            return createResponseMessage(responseSubject, responseKey, result);
        }else{
            return null;
        }
    }
    
    protected Message onKeySet(SharedContextEvent event, String responseSubject, String responseKey){
        if(isMain()){
            Set<K> result = new HashSet<K>();
            synchronized(context){
                result.addAll(context.keySet());
            }
            return createResponseMessage(responseSubject, responseKey, result);
        }else{
            return null;
        }
    }
    
    protected Message onSize(SharedContextEvent event, String responseSubject, String responseKey){
        if(isMain()){
            return createResponseMessage(responseSubject, responseKey, new Integer(size()));
        }else{
            return null;
        }
    }
    
    protected Message onContainsKey(SharedContextEvent event, String responseSubject, String responseKey){
        if(isMain()){
            return createResponseMessage(responseSubject, responseKey, containsKey(event.key) ? Boolean.TRUE : Boolean.FALSE);
        }else{
            return null;
        }
    }
    
    protected Message onContainsValue(SharedContextEvent event, String responseSubject, String responseKey){
        if(isMain()){
            return createResponseMessage(responseSubject, responseKey, containsValue(event.value) ? Boolean.TRUE : Boolean.FALSE);
        }else{
            return null;
        }
    }
    
    protected Message onSynchronize(final SharedContextEvent event, final Object sourceId, final int sequence, final String responseSubject, final String responseKey){
        Thread synchronizeThread = new Thread(){
            public void run(){
                Message response = null;
                try{
                    synchronize(((Long)event.value).longValue());
                    response = createResponseMessage(responseSubject, responseKey, Boolean.TRUE);
                }catch(SharedContextSendException e){
                    response = createResponseMessage(responseSubject, responseKey, Boolean.FALSE);
                }catch(SharedContextTimeoutException e){
                    response = createResponseMessage(responseSubject, responseKey, Boolean.FALSE);
                }
                try{
                    serverConnection.response(sourceId, sequence, response);
                }catch(MessageSendException e){
                    e.printStackTrace();
                    // TODO
                }
            }
        };
        synchronizeThread.start();
        return null;
    }
    
    @SuppressWarnings("unchecked")
    protected Message onGetLock(SharedContextEvent event, String responseSubject, String responseKey, Object sourceId){
        if(isMain()){
            Object result = null;
            Lock lock = null;
            synchronized(keyLockMap){
                lock = (Lock)keyLockMap.get(event.key);
                if(lock == null){
                    lock = new Lock((K)event.key);
                    keyLockMap.put(event.key, lock);
                }
            }
            long timeout = ((Long)event.value).longValue();
            final long start = System.currentTimeMillis();
            if(lock.acquire(sourceId, timeout)){
                final boolean isNoTimeout = timeout <= 0;
                timeout = isNoTimeout ? timeout : timeout - (System.currentTimeMillis() - start);
                if(!isNoTimeout && timeout < 0){
                    lock.release(sourceId, false);
                    result = Boolean.FALSE;
                }else{
                    try{
                        Message message = serverConnection.createMessage(subject, event.key == null ? null : event.key.toString());
                        message.setSubject(clientSubject, event.key == null ? null : event.key.toString());
                        Set<Object> receiveClients =  serverConnection.getReceiveClientIds(message);
                        receiveClients.remove(sourceId);
                        result = Boolean.TRUE;
                        if(receiveClients.size() != 0){
                            message.setDestinationIds(receiveClients);
                            message.setObject(new SharedContextEvent(SharedContextEvent.EVENT_GOT_LOCK, event.key, new Object[]{sourceId, new Long(timeout)}));
                            Message[] responses = serverConnection.request(message, 0, timeout);
                            if(responses != null && responses.length >= receiveClients.size()){
                                for(int i = 0; i < responses.length; i++){
                                    Object ret = responses[i].getObject();
                                    if(ret == null
                                        || ret instanceof Throwable
                                        || !((Boolean)ret).booleanValue()
                                    ){
                                        unlock((K)event.key);
                                        result = ret;
                                        break;
                                    }
                                }
                            }else{
                                unlock((K)event.key);
                                result = Boolean.FALSE;
                            }
                        }
                    }catch(Throwable th){
                        try{
                            unlock((K)event.key);
                        }catch(SharedContextSendException e){
                            // TODO
                        }
                        result = th;
                    }
                }
            }else{
                result = Boolean.FALSE;
            }
            return createResponseMessage(responseSubject, responseKey, result);
        }else{
            return null;
        }
    }
    
    @SuppressWarnings("unchecked")
    protected Message onGotLock(SharedContextEvent event, String responseSubject, String responseKey){
        Lock lock = null;
        synchronized(keyLockMap){
            lock = (Lock)keyLockMap.get(event.key);
            if(lock == null){
                lock = new Lock((K)event.key);
                keyLockMap.put(event.key, lock);
            }
        }
        final Object[] params = (Object[])event.value;
        final Object id = params[0];
        final long timeout = ((Long)params[1]).longValue();
        try{
            if(lock.acquire(id, timeout)){
                return createResponseMessage(responseSubject, responseKey, Boolean.TRUE);
            }else{
                return createResponseMessage(responseSubject, responseKey, Boolean.FALSE);
            }
        }catch(Throwable th){
            lock.release(id, false);
            return createResponseMessage(responseSubject, responseKey, th);
        }
    }
    
    protected Message onReleaseLock(SharedContextEvent event, String responseSubject, String responseKey){
        Lock lock = null;
        synchronized(keyLockMap){
            lock = (Lock)keyLockMap.get(event.key);
        }
        if(lock != null){
            lock.release(event.value, true);
        }
        return createResponseMessage(responseSubject, responseKey, null);
    }
    
    protected Message onSave(SharedContextEvent event, String responseSubject, String responseKey){
        if(!isClient && (isMain() || !isSaveOnlyMain)){
            try{
                if(sharedContextStore != null){
                    if(isClearBeforeSave){
                        sharedContextStore.clear();
                    }
                    sharedContextStore.save(this);
                }else{
                    throw new UnsupportedOperationException();
                }
            }catch(Throwable th){
                return createResponseMessage(responseSubject, responseKey, th);
            }
        }
        return createResponseMessage(responseSubject, responseKey, null);
    }
    
    protected Message onLoad(SharedContextEvent event, String responseSubject, String responseKey){
        if(isMain()){
            try{
                if(sharedContextStore != null){
                    sharedContextStore.load(this);
                }else{
                    throw new UnsupportedOperationException();
                }
            }catch(Throwable th){
                return createResponseMessage(responseSubject, responseKey, th);
            }
        }
        return createResponseMessage(responseSubject, responseKey, null);
    }
    
    protected static class SharedContextEvent implements java.io.Externalizable{
        
        public static final byte EVENT_PUT            = (byte)1;
        public static final byte EVENT_REMOVE         = (byte)2;
        public static final byte EVENT_CLEAR          = (byte)3;
        public static final byte EVENT_GET_ALL        = (byte)4;
        public static final byte EVENT_GET            = (byte)5;
        public static final byte EVENT_PUT_ALL        = (byte)6;
        public static final byte EVENT_KEY_SET        = (byte)7;
        public static final byte EVENT_SIZE           = (byte)8;
        public static final byte EVENT_CONTAINS_KEY   = (byte)9;
        public static final byte EVENT_CONTAINS_VALUE = (byte)10;
        public static final byte EVENT_SYNCH          = (byte)11;
        public static final byte EVENT_GET_LOCK       = (byte)12;
        public static final byte EVENT_GOT_LOCK       = (byte)13;
        public static final byte EVENT_RELEASE_LOCK   = (byte)14;
        public static final byte EVENT_SAVE           = (byte)15;
        public static final byte EVENT_LOAD           = (byte)16;
        public static final byte EVENT_UPDATE         = (byte)17;
        
        public byte type;
        public Object key;
        public Object value;
        
        public SharedContextEvent(){
        }
        
        public SharedContextEvent(byte type){
            this(type, null, null);
        }
        
        public SharedContextEvent(byte type, Object key){
            this(type, key, null);
        }
        
        public SharedContextEvent(byte type, Object key, Object value){
            this.type = type;
            this.key = key;
            this.value = value;
        }
        
        public void writeExternal(java.io.ObjectOutput out) throws java.io.IOException{
            out.write(type);
            out.writeObject(key);
            out.writeObject(value);
        }
        
        public void readExternal(java.io.ObjectInput in) throws java.io.IOException, ClassNotFoundException{
            type = (byte)in.read();
            key = in.readObject();
            value = in.readObject();
        }
        
        public String toString(){
            StringBuilder buf = new StringBuilder(super.toString());
            buf.append('{');
            buf.append("type=").append(type);
            buf.append(", key=").append(key);
            buf.append(", value=").append(value);
            buf.append('}');
            return buf.toString();
        }
    }
    
    protected class Lock{
        private K key;
        private Object owner;
        private Thread ownerThread;
        private final SynchronizeMonitor monitor = new WaitSynchronizeMonitor();
        
        public Lock(K key){
            this.key = key;
        }
        
        public boolean acquire(Object id, long timeout){
            final boolean isLocal = id.equals(getId());
            synchronized(Lock.this){
                if(owner == null){
                    owner = id;
                    if(isLocal){
                        ownerThread = Thread.currentThread();
                    }
                    synchronized(idLocksMap){
                        Set<K> keySet = idLocksMap.get(id);
                        if(keySet == null){
                            keySet = new HashSet<K>();
                            idLocksMap.put(id, keySet);
                        }
                        synchronized(keySet){
                            keySet.add(key);
                        }
                    }
                    return true;
                }else if(id.equals(owner)){
                    return isLocal ? Thread.currentThread().equals(ownerThread) : true;
                }
            }
            try{
                long start = System.currentTimeMillis();
                if(monitor.initAndWaitMonitor(timeout)){
                    timeout = timeout - (System.currentTimeMillis() - start);
                    if(timeout <= 0){
                        return false;
                    }
                    return acquire(id, timeout);
                }else{
                    return false;
                }
            }catch(InterruptedException e){
                return false;
            }finally{
                monitor.releaseMonitor();
            }
        }
        
        public Object getOwner(){
            return owner;
        }
        
        public boolean release(Object id, boolean force){
            final boolean isLocal = id.equals(getId());
            boolean result = false;
            synchronized(Lock.this){
                if(owner == null || (id.equals(owner) && (force || !isLocal || Thread.currentThread().equals(ownerThread)))){
                    owner = null;
                    ownerThread = null;
                    result = true;
                }
                synchronized(idLocksMap){
                    Set<K> keySet = idLocksMap.get(id);
                    if(keySet != null){
                        synchronized(keySet){
                            keySet.remove(key);
                        }
                    }
                }
            }
            monitor.notifyMonitor();
            return result;
        }
    }
}