diff --git a/spring-session-data-geode/src/integration-test/java/org/springframework/session/data/gemfire/ConcurrentSessionOperationsUsingClientCachingProxyRegionIntegrationTests.java b/spring-session-data-geode/src/integration-test/java/org/springframework/session/data/gemfire/ConcurrentSessionOperationsUsingClientCachingProxyRegionIntegrationTests.java index d9efbb2..0898065 100644 --- a/spring-session-data-geode/src/integration-test/java/org/springframework/session/data/gemfire/ConcurrentSessionOperationsUsingClientCachingProxyRegionIntegrationTests.java +++ b/spring-session-data-geode/src/integration-test/java/org/springframework/session/data/gemfire/ConcurrentSessionOperationsUsingClientCachingProxyRegionIntegrationTests.java @@ -17,17 +17,32 @@ package org.springframework.session.data.gemfire; import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.isA; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.springframework.data.gemfire.util.ArrayUtils.nullSafeArray; +import static org.springframework.data.gemfire.util.RuntimeExceptionFactory.newIllegalStateException; +import static org.springframework.session.data.gemfire.AbstractGemFireOperationsSessionRepository.GemFireSession; +import java.io.DataOutput; import java.io.IOException; +import java.lang.reflect.Method; +import java.util.Arrays; +import java.util.Objects; import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Predicate; import org.junit.BeforeClass; import org.junit.Test; import org.junit.runner.RunWith; +import org.mockito.Mockito; import edu.umd.cs.mtc.TestFramework; +import org.apache.geode.DataSerializer; +import org.apache.geode.cache.Region; import org.apache.geode.cache.client.ClientRegionShortcut; +import org.apache.geode.internal.InternalDataSerializer; import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.data.gemfire.config.annotation.CacheServerApplication; @@ -35,6 +50,7 @@ import org.springframework.data.gemfire.config.annotation.ClientCacheApplication import org.springframework.session.Session; import org.springframework.session.data.gemfire.config.annotation.web.http.EnableGemFireHttpSession; import org.springframework.session.data.gemfire.config.annotation.web.http.GemFireHttpSessionConfiguration; +import org.springframework.session.data.gemfire.serialization.SessionSerializer; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringRunner; @@ -48,20 +64,28 @@ import org.springframework.test.context.junit4.SpringRunner; @ContextConfiguration( classes = ConcurrentSessionOperationsUsingClientCachingProxyRegionIntegrationTests.GemFireClientConfiguration.class ) +@SuppressWarnings("unused") public class ConcurrentSessionOperationsUsingClientCachingProxyRegionIntegrationTests extends AbstractConcurrentSessionOperationsIntegrationTests { + private static final String GEMFIRE_LOG_LEVEL = "error"; + @Test - public void concurrentCachedSessionAccessIsCorrect() throws Throwable { - TestFramework.runOnce(new ConcurrentCachedSessionAccessTestCase(this)); + public void concurrentCachedSessionOperationsAreCorrect() throws Throwable { + TestFramework.runOnce(new ConcurrentCachedSessionOperationsTestCase(this)); } - @SuppressWarnings("unused") - public static class ConcurrentCachedSessionAccessTestCase extends AbstractConcurrentSessionOperationsTestCase { + @Test + public void regionPutWithNonDirtySessionResultsInInefficientIncorrectBehavior() throws Throwable { + TestFramework.runOnce(new RegionPutWithNonDirtySessionTestCase(this)); + } + + // Tests that 2 Threads share the same Session object reference and therefore see's each other's changes. + public static class ConcurrentCachedSessionOperationsTestCase extends AbstractConcurrentSessionOperationsTestCase { private final AtomicReference sessionId = new AtomicReference<>(null); - public ConcurrentCachedSessionAccessTestCase( + public ConcurrentCachedSessionOperationsTestCase( ConcurrentSessionOperationsUsingClientCachingProxyRegionIntegrationTests testInstance) { super(testInstance); @@ -115,14 +139,164 @@ public class ConcurrentSessionOperationsUsingClientCachingProxyRegionIntegration } } + // Tests that DataSerializer.toData(..) is called twice; once for when the Session is new + // and again when Region.put(..) is call with a Session having no delta/no changes. + public static class RegionPutWithNonDirtySessionTestCase extends AbstractConcurrentSessionOperationsTestCase { + + private static final String DATA_SERIALIZER_NOT_FOUND_EXCEPTION_MESSAGE = + "No DataSerializer was found capable of de/serializing Sessions"; + + private final AtomicReference sessionId = new AtomicReference<>(null); + + private final DataSerializer sessionSerializer; + + private final Region sessions; + + public RegionPutWithNonDirtySessionTestCase( + ConcurrentSessionOperationsUsingClientCachingProxyRegionIntegrationTests testInstance) { + + super(testInstance); + + this.sessions = testInstance.getSessionRegion(); + this.sessionSerializer = reregisterDataSerializer(resolveDataSerializer()); + } + + private DataSerializer resolveDataSerializer() { + + return Arrays.stream(nullSafeArray(InternalDataSerializer.getSerializers(), DataSerializer.class)) + .filter(this.sessionSerializerFilter()) + .findFirst() + .map(Mockito::spy) + .orElseThrow(() -> newIllegalStateException(DATA_SERIALIZER_NOT_FOUND_EXCEPTION_MESSAGE)); + } + + private Predicate sessionSerializerFilter() { + + return dataSerializer -> { + + boolean isSessionSerializer = dataSerializer instanceof SessionSerializer; + + if (!isSessionSerializer) { + isSessionSerializer = + Arrays.stream(nullSafeArray(dataSerializer.getSupportedClasses(), Class.class)) + .filter(Objects::nonNull) + .anyMatch(Session.class::isAssignableFrom); + } + + return isSessionSerializer; + }; + } + + private DataSerializer reregisterDataSerializer(DataSerializer dataSerializer) { + + try { + + if (dataSerializer != null) { + + InternalDataSerializer.unregister(dataSerializer.getId()); + + Method registerDataSerializer = InternalDataSerializer.class + .getDeclaredMethod("_register", DataSerializer.class, Boolean.TYPE); + + registerDataSerializer.setAccessible(true); + + registerDataSerializer.invoke(null, dataSerializer, false); + } + + return dataSerializer; + } + catch (Exception cause) { + throw new RuntimeException(cause); + } + } + + private Session get(String id) { + return this.sessions.get(id); + } + + private void put(Session session) { + + this.sessions.put(session.getId(), session); + + if (session instanceof GemFireSession) { + ((GemFireSession) session).commit(); + } + } + + public void thread1() { + + Thread.currentThread().setName("User Session One"); + + assertTick(0); + + Session session = newSession(); + + assertThat(session).isInstanceOf(GemFireSession.class); + assertThat(session.getId()).isNotEmpty(); + assertThat(session.isExpired()).isFalse(); + assertThat(session.getAttributeNames()).isEmpty(); + + session.setAttribute("attributeOne", "testOne"); + session.setAttribute("attributeTwo", "testTwo"); + + assertThat(session.getAttributeNames()).containsOnly("attributeOne", "attributeTwo"); + assertThat(((GemFireSession) session).hasDelta()).isTrue(); + + put(session); + + assertThat(((GemFireSession) session).hasDelta()).isFalse(); + + this.sessionId.set(session.getId()); + } + + public void thread2() { + + Thread.currentThread().setName("User Session Two"); + + waitForTick(1); + assertTick(1); + + Session session = get(this.sessionId.get()); + + assertThat(session).isInstanceOf(GemFireSession.class); + assertThat(session.getId()).isEqualTo(this.sessionId.get()); + assertThat(session.isExpired()).isFalse(); + assertThat(session.getAttributeNames()).containsOnly("attributeOne", "attributeTwo"); + assertThat(session.getAttribute("attributeOne")).isEqualTo("testOne"); + assertThat(session.getAttribute("attributeTwo")).isEqualTo("testTwo"); + assertThat(((GemFireSession) session).hasDelta()).isFalse(); + + put(session); + } + + @Override + public void finish() { + + Session session = get(this.sessionId.get()); + + assertThat(session).isNotNull(); + assertThat(session.getId()).isEqualTo(this.sessionId.get()); + assertThat(session.isExpired()).isFalse(); + assertThat(session.getAttributeNames()).containsOnly("attributeOne", "attributeTwo"); + assertThat(session.getAttribute("attributeOne")).isEqualTo("testOne"); + assertThat(session.getAttribute("attributeTwo")).isEqualTo("testTwo"); + + try { + verify(this.sessionSerializer, times(2)) + .toData(isA(GemFireSession.class), isA(DataOutput.class)); + } + catch (IOException ignore) { } + } + } + @BeforeClass public static void startGemFireServer() throws IOException { startGemFireServer(GemFireServerConfiguration.class); } - // Tests fail when copyOnRead is set to true. - //@ClientCacheApplication(copyOnRead = true, logLevel = "error", subscriptionEnabled = true) - @ClientCacheApplication(logLevel = "error", subscriptionEnabled = true) + // Tests fail when 'copyOnRead' is set to 'true'! + //@ClientCacheApplication(copyOnRead = true, logLevel = GEMFIRE_LOG_LEVEL, subscriptionEnabled = true) + @ClientCacheApplication(logLevel = GEMFIRE_LOG_LEVEL, subscriptionEnabled = true) @EnableGemFireHttpSession( clientRegionShortcut = ClientRegionShortcut.CACHING_PROXY, poolName = "DEFAULT", @@ -132,7 +306,7 @@ public class ConcurrentSessionOperationsUsingClientCachingProxyRegionIntegration @CacheServerApplication( name = "ConcurrentSessionOperationsUsingClientCachingProxyRegionIntegrationTests", - logLevel = "error" + logLevel = GEMFIRE_LOG_LEVEL ) @EnableGemFireHttpSession( sessionSerializerBeanName = GemFireHttpSessionConfiguration.SESSION_DATA_SERIALIZER_BEAN_NAME