diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/CachedPreparedStatementCreator.java b/spring-cql/src/main/java/org/springframework/cassandra/core/CachedPreparedStatementCreator.java
index 5c9019a70..324fbbfbb 100644
--- a/spring-cql/src/main/java/org/springframework/cassandra/core/CachedPreparedStatementCreator.java
+++ b/spring-cql/src/main/java/org/springframework/cassandra/core/CachedPreparedStatementCreator.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2013-2014 the original author or authors.
+ * Copyright 2013-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -27,27 +27,32 @@ import com.datastax.driver.core.Session;
import com.datastax.driver.core.exceptions.DriverException;
/**
- * This Prepared Statement Creator maintains a cache of all prepared statements for the duration of this life of the
- * container. When preparing statements with Cassandra, each Statement should be prepared once and only once due to the
- * overhead of preparing the statement.
+ * This {@link PreparedStatementCreator} maintains a static cache of all prepared statements for the duration of the JVM
+ * runtime, more specific the lifecycle of the associated {@link ClassLoader}. When preparing statements with Cassandra,
+ * each Statement should be prepared once and only once due to the overhead of preparing the statement.
+ *
+ * {@link CachedPreparedStatementCreator} is thread-safe and does not require external synchronization when used by
+ * concurrent threads.
*
* @author David Webb
+ * @author Mark Paluch
*/
public class CachedPreparedStatementCreator implements PreparedStatementCreator {
private static final Logger log = LoggerFactory.getLogger(CachedPreparedStatementCreator.class);
+ private static final Map> CACHE = new ConcurrentHashMap>();
private final String cql;
- private static final Map> psMap = new ConcurrentHashMap>();
-
/**
- * Create a PreparedStatementCreator from the provided CQL.
+ * Create a {@link PreparedStatementCreator} from the provided CQL.
*
- * @param cql
+ * @param cql must not be empty and not {@literal null}.
*/
public CachedPreparedStatementCreator(String cql) {
- Assert.notNull(cql, "CQL is required to create a PreparedStatement");
+
+ Assert.hasText(cql, "CQL is required to create a PreparedStatement");
+
this.cql = cql;
}
@@ -55,29 +60,67 @@ public class CachedPreparedStatementCreator implements PreparedStatementCreator
return this.cql;
}
+ /* (non-Javadoc)
+ * @see org.springframework.cassandra.core.PreparedStatementCreator#createPreparedStatement(com.datastax.driver.core.Session)
+ */
@Override
public PreparedStatement createPreparedStatement(Session session) throws DriverException {
- StringBuilder keyspaceCQLKey = new StringBuilder().append(session.getLoggedKeyspace()).append("|").append(this.cql);
+ StringBuilder cacheKey = new StringBuilder().append(session.getLoggedKeyspace()).append("|").append(this.cql);
- log.debug(String.format("Cachable PreparedStatement in Keyspace [%s]", session.getLoggedKeyspace()));
+ log.debug("Cachable PreparedStatement in Keyspace {}", session.getLoggedKeyspace());
+
+ Map sessionCache = getOrCreateSessionLocalCache(session);
+ return getOrPrepareStatement(session, cacheKey.toString(), sessionCache);
+ }
+
+ private Map getOrCreateSessionLocalCache(Session session) {
+
+ Map sessionMap = CACHE.get(session);
- Map sessionMap = psMap.get(session);
if (sessionMap == null) {
- sessionMap = new ConcurrentHashMap();
- psMap.put(session, sessionMap);
+
+ synchronized (session) {
+
+ if (CACHE.containsKey(session)) {
+ sessionMap = CACHE.get(session);
+ } else {
+
+ sessionMap = new ConcurrentHashMap();
+ CACHE.put(session, sessionMap);
+ }
+ }
}
- PreparedStatement pstmt = sessionMap.get(keyspaceCQLKey.toString());
+ return sessionMap;
+ }
+
+ private PreparedStatement getOrPrepareStatement(Session session, String cacheKey,
+ Map sessionCache) {
+
+ PreparedStatement pstmt = sessionCache.get(cacheKey);
+
if (pstmt == null) {
- log.debug("No Cached PreparedStatement found...Creating and Caching");
- pstmt = session.prepare(this.cql);
- sessionMap.put(keyspaceCQLKey.toString(), pstmt);
- } else {
- log.debug("Found cached PreparedStatement");
+
+ synchronized (sessionCache) {
+
+ if (sessionCache.containsKey(cacheKey)) {
+
+ log.debug("Found cached PreparedStatement");
+ return sessionCache.get(cacheKey);
+ }
+
+ log.debug("No Cached PreparedStatement found...Creating and Caching");
+
+ pstmt = session.prepare(this.cql);
+ sessionCache.put(cacheKey, pstmt);
+
+ return pstmt;
+ }
}
+ log.debug("Found cached PreparedStatement");
+
return pstmt;
}
-
}
diff --git a/spring-cql/src/test/java/org/springframework/cassandra/core/CachedPreparedStatementCreatorUnitTests.java b/spring-cql/src/test/java/org/springframework/cassandra/core/CachedPreparedStatementCreatorUnitTests.java
new file mode 100644
index 000000000..64fc9a4c6
--- /dev/null
+++ b/spring-cql/src/test/java/org/springframework/cassandra/core/CachedPreparedStatementCreatorUnitTests.java
@@ -0,0 +1,187 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * 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 org.springframework.cassandra.core;
+
+import static edu.umd.cs.mtc.TestFramework.*;
+import static org.hamcrest.Matchers.*;
+import static org.junit.Assert.*;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.*;
+
+import java.lang.reflect.InvocationHandler;
+import java.lang.reflect.Method;
+import java.lang.reflect.Proxy;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnitRunner;
+
+import com.datastax.driver.core.PreparedStatement;
+import com.datastax.driver.core.Session;
+
+import edu.umd.cs.mtc.MultithreadedTestCase;
+
+/**
+ * Unit tests for {@link CachedPreparedStatementCreator}.
+ *
+ * @author Mark Paluch
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class CachedPreparedStatementCreatorUnitTests {
+
+ PreparedStatement preparedStatement;
+ @Mock Session sessionMock;
+
+ @Before
+ public void before() throws Exception {
+
+ preparedStatement = newProxy(PreparedStatement.class, new TestInvocationHandler());
+ when(sessionMock.prepare(anyString())).thenReturn(preparedStatement);
+ }
+
+ /**
+ * @see DATACASS-253
+ */
+ @Test(expected = IllegalArgumentException.class)
+ public void shouldRejectEmptyCql() {
+ new CachedPreparedStatementCreator("");
+ }
+
+ /**
+ * @see DATACASS-253
+ */
+ @Test(expected = IllegalArgumentException.class)
+ public void shouldRejectNullCql() {
+ new CachedPreparedStatementCreator(null);
+ }
+
+ /**
+ * @see DATACASS-253
+ */
+ @Test
+ public void shouldCreatePreparedStatement() {
+
+ CachedPreparedStatementCreator cachedPreparedStatementCreator = new CachedPreparedStatementCreator("my cql");
+
+ PreparedStatement result = cachedPreparedStatementCreator.createPreparedStatement(sessionMock);
+
+ assertThat(result, is(sameInstance(preparedStatement)));
+ verify(sessionMock).prepare("my cql");
+ }
+
+ /**
+ * @see DATACASS-253
+ */
+ @Test
+ public void shouldCacheCreatePreparedStatement() {
+
+ CachedPreparedStatementCreator cachedPreparedStatementCreator = new CachedPreparedStatementCreator("my cql");
+
+ cachedPreparedStatementCreator.createPreparedStatement(sessionMock);
+ cachedPreparedStatementCreator.createPreparedStatement(sessionMock);
+ PreparedStatement result = cachedPreparedStatementCreator.createPreparedStatement(sessionMock);
+
+ assertThat(result, is(sameInstance(preparedStatement)));
+ verify(sessionMock, times(1)).prepare("my cql");
+ }
+
+ /**
+ * @see DATACASS-253
+ * @throws Throwable
+ */
+ @Test
+ public void concurrentAccessToCreateStatementShouldBeSynchronized() throws Throwable {
+
+ CreatePreparedStatementIsThreadSafe concurrentPrepareStatement = new CreatePreparedStatementIsThreadSafe(
+ preparedStatement, new CachedPreparedStatementCreator("my cql"));
+
+ runManyTimes(concurrentPrepareStatement, 5);
+ }
+
+ @SuppressWarnings("unused")
+ private static class CreatePreparedStatementIsThreadSafe extends MultithreadedTestCase {
+
+ final CachedPreparedStatementCreator preparedStatementCreator;
+ final Session session;
+ final AtomicInteger atomicInteger = new AtomicInteger();
+
+ public CreatePreparedStatementIsThreadSafe(final PreparedStatement preparedStatement,
+ CachedPreparedStatementCreator preparedStatementCreator) {
+
+ this.preparedStatementCreator = preparedStatementCreator;
+
+ this.session = newProxy(Session.class, new TestInvocationHandler() {
+
+ @Override
+ public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
+
+ if (method.getName().equals("prepare") && args.length == 1) {
+
+ waitForTick(2);
+ atomicInteger.incrementAndGet();
+ return preparedStatement;
+ }
+
+ return super.invoke(proxy, method, args);
+ }
+ });
+ }
+
+ public void thread1() {
+
+ waitForTick(1);
+
+ preparedStatementCreator.createPreparedStatement(session);
+
+ assertThat(atomicInteger.get(), is(1));
+
+ }
+
+ public void thread2() {
+
+ waitForTick(1);
+
+ preparedStatementCreator.createPreparedStatement(session);
+
+ assertThat(atomicInteger.get(), is(1));
+ }
+ }
+
+ private static class TestInvocationHandler implements InvocationHandler {
+
+ @Override
+ public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
+
+ if (method.getName().equals("hashCode")) {
+ return hashCode();
+ }
+
+ if (method.getName().equals("equals") && args.length == 1) {
+ return equals(args[0]);
+ }
+
+ return null;
+ }
+ }
+
+ private static T newProxy(Class theClass, InvocationHandler invocationHandler) {
+ return (T) Proxy.newProxyInstance(CachedPreparedStatementCreatorUnitTests.class.getClassLoader(),
+ new Class[] { theClass }, invocationHandler);
+ }
+}