Spring/spring-session-data-geode
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add additional test case covering the serialization behavior of a Region.put(key, value) when value is new and when value.hasDelta() returns false.

Browse Source
This commit is contained in:
John Blum
2018-12-05 01:58:20 -08:00
parent b269bd23b1
commit d56604276e
1 changed files with 183 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

192
spring-session-data-geode/src/integration-test/java/org/springframework/session/data/gemfire/ConcurrentSessionOperationsUsingClientCachingProxyRegionIntegrationTests.java
View File

@@ -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<String> 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<String> sessionId = new AtomicReference<>(null);
private final DataSerializer sessionSerializer;
private final Region<Object, Session> 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<? super DataSerializer> 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.<String>getAttribute("attributeOne")).isEqualTo("testOne");
assertThat(session.<String>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.<String>getAttribute("attributeOne")).isEqualTo("testOne");
assertThat(session.<String>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