Implement caching StoreClientFactory, which required adding equals/ha…

…shCode to the InconsistencyResolvers

src/java/voldemort/client/CachingStoreClientFactory.java

@@ -0,0 +1,85 @@
+/*
+ * 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);
+        StoreClient retVal = cache.get(key);
+        if(retVal == null) {
+            retVal = inner.getStoreClient(storeName);
+            cache.putIfAbsent(key, retVal);
+        }
+
+        return retVal;
+    }
+
+    @SuppressWarnings("unchecked")
+    public <K, V> StoreClient<K, V> getStoreClient(String storeName,
+                                                   InconsistencyResolver<Versioned<V>> resolver) {
+        Pair<String, Object> key = Pair.create(storeName, (Object) resolver);
+        StoreClient retVal = cache.get(key);
+        if(retVal == null) {
+            retVal = inner.getStoreClient(storeName, resolver);
+            cache.putIfAbsent(key, retVal);
+        }
+
+        return retVal;
+    }
+
+    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();
+    }
+}

src/java/voldemort/versioning/ArbitraryInconsistencyResolver.java

@@ -41,4 +41,15 @@ public List<T> resolveConflicts(List<T> values) {
             return Collections.singletonList(values.get(0));
     }
 
+    @Override
+    public boolean equals(Object o) {
+        if(o == this)
+            return true;
+        return (o instanceof ArbitraryInconsistencyResolver);
+    }
+
+    @Override
+    public int hashCode() {
+        return 31 * this.getClass().hashCode();
+    }
 }

src/java/voldemort/versioning/ChainedResolver.java

@@ -48,4 +48,21 @@ public List<T> resolveConflicts(List<T> items) {
         return items;
     }
 
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+
+        ChainedResolver that = (ChainedResolver) o;
+
+        if (resolvers != null ? !resolvers.equals(that.resolvers) : that.resolvers != null)
+            return false;
+
+        return true;
+    }
+
+    @Override
+    public int hashCode() {
+        return resolvers != null ? resolvers.hashCode() : 0;
+    }
 }

src/java/voldemort/versioning/FailingInconsistencyResolver.java

@@ -31,4 +31,17 @@ public List<T> resolveConflicts(List<T> items) {
         else
             return items;
     }
+
+
+    @Override
+    public boolean equals(Object o) {
+        if(o == this)
+            return true;
+        return (o instanceof FailingInconsistencyResolver);
+    }
+
+    @Override
+    public int hashCode() {
+        return 31 * this.getClass().hashCode();
+    }
 }

src/java/voldemort/versioning/MergingInconsistencyResolver.java

@@ -49,4 +49,17 @@ public List<Versioned<T>> resolveConflicts(List<Versioned<T>> items) {
             return Collections.singletonList(new Versioned<T>(merged, clock));
         }
     }
+
+
+    @Override
+    public boolean equals(Object o) {
+        if(o == this)
+            return true;
+        return (o instanceof MergingInconsistencyResolver);
+    }
+
+    @Override
+    public int hashCode() {
+        return 31 * this.getClass().hashCode();
+    }
 }

src/java/voldemort/versioning/TimeBasedInconsistencyResolver.java

@@ -43,4 +43,17 @@ public List<Versioned<T>> resolveConflicts(List<Versioned<T>> items) {
             return Collections.singletonList(max);
         }
     }
+
+
+    @Override
+    public boolean equals(Object o) {
+        if(o == this)
+            return true;
+        return (o instanceof TimeBasedInconsistencyResolver);
+    }
+
+    @Override
+    public int hashCode() {
+        return 31 * this.getClass().hashCode();
+    }
 }

src/java/voldemort/versioning/VectorClockInconsistencyResolver.java

@@ -54,4 +54,16 @@ public List<Versioned<T>> resolveConflicts(List<Versioned<T>> items) {
         }
         return newItems;
     }
+
+    @Override
+    public boolean equals(Object o) {
+        if(o == this)
+            return true;
+        return (o instanceof VectorClockInconsistencyResolver);
+    }
+
+    @Override
+    public int hashCode() {
+        return 31 * this.getClass().hashCode();
+    }
 }

test/unit/voldemort/client/CachingStoreClientFactoryTest.java

@@ -0,0 +1,49 @@
+/*
+ * 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 org.junit.Test;
+import voldemort.serialization.StringSerializer;
+import voldemort.versioning.TimeBasedInconsistencyResolver;
+
+import static org.junit.Assert.*;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.*;
+
+public class CachingStoreClientFactoryTest {
+
+    @Test
+    public void testCaching() {
+        StoreClientFactory inner = new MockStoreClientFactory(new StringSerializer(),
+                                                              new StringSerializer(),
+                                                              null);
+        StoreClientFactory spyFactory = spy(inner);
+        StoreClientFactory cachingFactory = new CachingStoreClientFactory(spyFactory);
+        TimeBasedInconsistencyResolver<Object> resolver = new TimeBasedInconsistencyResolver<Object>();
+
+        when(spyFactory.<Object, Object>getStoreClient(anyString())).thenCallRealMethod();
+        when(spyFactory.<Object, Object>getStoreClient(anyString(), eq(resolver))).thenCallRealMethod();
+
+        for(int i = 0; i < 10; i++) {
+            assertNotNull(cachingFactory.getStoreClient("foo"));
+            assertNotNull(cachingFactory.getStoreClient("foo", resolver));
+        }
+
+        verify(spyFactory, times(1)).getStoreClient("foo");
+        verify(spyFactory, times(2)).getStoreClient("foo", resolver);
+    }
+}