/*
 * Copyright 2008-2010 LinkedIn, Inc
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */

package voldemort.client;

import voldemort.cluster.failuredetector.FailureDetector;
import voldemort.store.Store;
import voldemort.utils.Pair;
import voldemort.versioning.InconsistencyResolver;
import voldemort.versioning.Versioned;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * A wrapper for a store {@link StoreClientFactory} which caches requests
 * to <code>getStoreClient</code>
 *
 */
public class CachingStoreClientFactory implements StoreClientFactory {

    private final StoreClientFactory inner;
    private final ConcurrentMap<Pair<String, Object>, StoreClient> cache;

    public CachingStoreClientFactory(StoreClientFactory inner) {
        this.inner = inner;
        this.cache = new ConcurrentHashMap<Pair<String, Object>, StoreClient>();
    }


    @SuppressWarnings("unchecked")
    public <K, V> StoreClient<K, V> getStoreClient(String storeName) {
        Pair<String, Object> key = Pair.create(storeName, null);
        if(!cache.containsKey(key)) {
            StoreClient<K, V> result = inner.getStoreClient(storeName);
            cache.putIfAbsent(key, result);
        }

        return cache.get(key);
    }

    @SuppressWarnings("unchecked")
    public <K, V> StoreClient<K, V> getStoreClient(String storeName,
                                                   InconsistencyResolver<Versioned<V>> resolver) {
        Pair<String, Object> key = Pair.create(storeName, (Object) resolver);
        if(!cache.containsKey(key)) {
            StoreClient<K, V> result = inner.getStoreClient(storeName, resolver);
            cache.putIfAbsent(key, result);
        }

        return cache.get(key);
    }

    public <K, V, T> Store<K, V, T> getRawStore(String storeName,
                                                InconsistencyResolver<Versioned<V>> resolver) {
        return inner.getRawStore(storeName, resolver);
    }

    public void close() {
        try {
            cache.clear();
        } finally {
            inner.close();
        }
    }

    public FailureDetector getFailureDetector() {
        return inner.getFailureDetector();
    }
}

@@ -619,16 +619,188 @@
 lient;%0A%0A
+import com.google.common.collect.ImmutableList;%0Aimport org.apache.log4j.Logger;%0Aimport voldemort.annotations.jmx.JmxManaged;%0Aimport voldemort.annotations.jmx.JmxOperation;%0A
 import v
@@ -849,16 +849,16 @@
 tector;%0A
-
 import v
@@ -999,16 +999,39 @@
 ioned;%0A%0A
+import java.util.List;%0A
 import j
@@ -1236,78 +1236,242 @@
  */%0A
-public class CachingStoreClientFactory implements StoreClientFactory %7B
+@JmxManaged(description = %22A StoreClientFactory which caches clients%22)%0Apublic class CachingStoreClientFactory implements StoreClientFactory %7B%0A%0A    private final static Logger logger = Logger.getLogger(CachingStoreClientFactory.class);
 %0A%0A  
@@ -3090,11 +3090,756 @@
 );%0A    %7D
+%0A%0A    @JmxOperation(description = %22Clear the cache%22)%0A    public synchronized void clear() %7B%0A        try %7B%0A            cache.clear();%0A        %7D catch(Exception e) %7B%0A            logger.warn(%22Exception when clearing the cache%22, e);%0A        %7D%0A    %7D%0A%0A    @JmxOperation(description = %22Bootstrap all clients in the cache%22)%0A    public void bootstrapAllClients() %7B%0A        List%3CStoreClient%3E allClients = ImmutableList.copyOf(cache.values());%0A        try %7B%0A            for(StoreClient client: allClients) %7B%0A                if(client instanceof DefaultStoreClient)%0A                    ((DefaultStoreClient) client).bootStrap();%0A            %7D%0A        %7D catch(Exception e) %7B%0A            logger.warn(%22Exception during bootstrapAllClients%22, e);%0A        %7D%0A    %7D
 %0A%7D%0A

Make CachingStoreClientFactory JMX managed, add ability to rebootstrap all clients.