diff --git a/spring-geode/src/main/java/org/springframework/geode/cache/AsyncInlineCachingRegionConfigurer.java b/spring-geode/src/main/java/org/springframework/geode/cache/AsyncInlineCachingRegionConfigurer.java index 1c62bab8..abd0260d 100644 --- a/spring-geode/src/main/java/org/springframework/geode/cache/AsyncInlineCachingRegionConfigurer.java +++ b/spring-geode/src/main/java/org/springframework/geode/cache/AsyncInlineCachingRegionConfigurer.java @@ -20,11 +20,13 @@ import java.util.List; import java.util.Objects; import java.util.Optional; import java.util.UUID; +import java.util.function.Function; import java.util.function.Predicate; import org.apache.geode.cache.Cache; import org.apache.geode.cache.DiskStore; import org.apache.geode.cache.Region; +import org.apache.geode.cache.asyncqueue.AsyncEvent; import org.apache.geode.cache.asyncqueue.AsyncEventListener; import org.apache.geode.cache.asyncqueue.AsyncEventQueue; import org.apache.geode.cache.asyncqueue.AsyncEventQueueFactory; @@ -38,6 +40,7 @@ import org.springframework.data.gemfire.config.annotation.RegionConfigurer; import org.springframework.data.gemfire.util.ArrayUtils; import org.springframework.data.gemfire.util.CollectionUtils; import org.springframework.data.repository.CrudRepository; +import org.springframework.geode.cache.RepositoryAsyncEventListener.AsyncEventErrorHandler; import org.springframework.lang.NonNull; import org.springframework.lang.Nullable; import org.springframework.util.Assert; @@ -49,6 +52,7 @@ import org.springframework.util.StringUtils; * abstraction. * * @author John Blum + * @see java.util.function.Function * @see java.util.function.Predicate * @see org.apache.geode.cache.Cache * @see org.apache.geode.cache.Region @@ -61,10 +65,13 @@ import org.springframework.util.StringUtils; * @see org.springframework.data.gemfire.PeerRegionFactoryBean * @see org.springframework.data.gemfire.config.annotation.RegionConfigurer * @see org.springframework.data.repository.CrudRepository + * @see org.springframework.geode.cache.RepositoryAsyncEventListener.AsyncEventErrorHandler * @since 1.4.0 */ public class AsyncInlineCachingRegionConfigurer implements RegionConfigurer { + protected static final Predicate DEFAULT_REGION_BEAN_NAME_PREDICATE = beanName -> false; + /** * Factory method used to construct a new instance of {@link AsyncInlineCachingRegionConfigurer} initialized with * the given Spring Data {@link CrudRepository} and {@link Predicate} identifying the target {@link Region} @@ -101,16 +108,18 @@ public class AsyncInlineCachingRegionConfigurer implements RegionConfigur * {@literal Asynchronous Inline Caching} will be configured. * @return a new {@link AsyncInlineCachingRegionConfigurer}. * @throws IllegalArgumentException if {@link CrudRepository} is {@literal null}. - * @see #AsyncInlineCachingRegionConfigurer(CrudRepository, Predicate) * @see org.springframework.data.repository.CrudRepository + * @see #create(CrudRepository, Predicate) * @see java.lang.String */ public static AsyncInlineCachingRegionConfigurer create(@NonNull CrudRepository repository, @Nullable String regionBeanName) { - return new AsyncInlineCachingRegionConfigurer<>(repository, Predicate.isEqual(regionBeanName)); + return create(repository, Predicate.isEqual(regionBeanName)); } + private AsyncEventErrorHandler asyncEventErrorHandler; + private Boolean batchConflationEnabled; private Boolean diskSynchronous; private Boolean forwardExpirationDestroy; @@ -120,6 +129,12 @@ public class AsyncInlineCachingRegionConfigurer implements RegionConfigur private final CrudRepository repository; + private Function asyncEventListenerPostProcessor; + + private Function asyncEventQueuePostProcessor; + + private Function asyncEventQueueFactoryPostProcessor; + private Integer batchSize; private Integer batchTimeInterval; private Integer dispatcherThreads; @@ -155,7 +170,7 @@ public class AsyncInlineCachingRegionConfigurer implements RegionConfigur Assert.notNull(repository, "CrudRepository must not be null"); this.repository = repository; - this.regionBeanName = regionBeanName != null ? regionBeanName : beanName -> false; + this.regionBeanName = regionBeanName != null ? regionBeanName : DEFAULT_REGION_BEAN_NAME_PREDICATE; } /** @@ -195,6 +210,7 @@ public class AsyncInlineCachingRegionConfigurer implements RegionConfigur * @param bean {@link PeerRegionFactoryBean} containing the configuration of the target {@link Region} * in the Spring container. * @see org.springframework.data.gemfire.PeerRegionFactoryBean + * @see #newAsyncEventQueue(Cache, String) */ @Override public void configure(String beanName, PeerRegionFactoryBean bean) { @@ -230,12 +246,17 @@ public class AsyncInlineCachingRegionConfigurer implements RegionConfigur * @return a new {@link AsyncEventQueue}. * @see org.apache.geode.cache.asyncqueue.AsyncEventQueue * @see org.apache.geode.cache.Cache - * @see #newRepositoryAsyncEventListener() * @see #generateId(String) + * @see #newAsyncEventQueueFactory(Cache) + * @see #newAsyncEventQueue(AsyncEventQueueFactory, String, AsyncEventListener) + * @see #newRepositoryAsyncEventListener() + * @see #postProcess(AsyncEventListener) + * @see #postProcess(AsyncEventQueue) + * @see #postProcess(AsyncEventQueueFactory) */ protected AsyncEventQueue newAsyncEventQueue(@NonNull Cache peerCache, @NonNull String regionBeanName) { - AsyncEventQueueFactory asyncEventQueueFactory = peerCache.createAsyncEventQueueFactory(); + AsyncEventQueueFactory asyncEventQueueFactory = newAsyncEventQueueFactory(peerCache); Optional.ofNullable(this.batchConflationEnabled).ifPresent(asyncEventQueueFactory::setBatchConflationEnabled); Optional.ofNullable(this.batchSize).ifPresent(asyncEventQueueFactory::setBatchSize); @@ -258,20 +279,236 @@ public class AsyncInlineCachingRegionConfigurer implements RegionConfigur asyncEventQueueFactory.pauseEventDispatching(); } - return asyncEventQueueFactory.create(generateId(regionBeanName), newRepositoryAsyncEventListener()); + String asyncEventQueueId = generateId(regionBeanName); + + AsyncEventListener asyncEventListener = newRepositoryAsyncEventListener(); + + asyncEventListener = postProcess(asyncEventListener); + asyncEventQueueFactory = postProcess(asyncEventQueueFactory); + + AsyncEventQueue asyncEventQueue = + newAsyncEventQueue(asyncEventQueueFactory, asyncEventQueueId, asyncEventListener); + + asyncEventQueue = postProcess(asyncEventQueue); + + return asyncEventQueue; + } + + /** + * Constructs (creates) a new instance of {@link AsyncEventQueue} using the given {@link AsyncEventQueueFactory} + * with the given {@link String AEQ ID} and {@link AsyncEventListener}. + * + * @param factory {@link AsyncEventQueueFactory} used to create the {@link AsyncEventQueue}; + * must not be {@literal null}. + * @param asyncEventQueueId {@link String} containing the {@literal ID} for the {@link AsyncEventQueue}; + * must not be {@literal null}. + * @param listener {@link AsyncEventListener} registered with the {@link AsyncEventQueue} to process cache events + * from the {@link AsyncEventQueue} attached to the {@link Region}. + * @return a new {@link AsyncEventQueue} with the {@link String ID} and registered {@link AsyncEventListener}. + * @see org.apache.geode.cache.asyncqueue.AsyncEventQueueFactory + * @see org.apache.geode.cache.asyncqueue.AsyncEventQueue + * @see org.apache.geode.cache.asyncqueue.AsyncEventListener + */ + protected @NonNull AsyncEventQueue newAsyncEventQueue(@NonNull AsyncEventQueueFactory factory, + @NonNull String asyncEventQueueId, @NonNull AsyncEventListener listener) { + + return factory.create(asyncEventQueueId, listener); + } + + /** + * Constructs (creates) a new instance of the {@link AsyncEventQueueFactory} from the given {@literal peer} + * {@link Cache}. + * + * @param peerCache {@literal Peer} {@link Cache} instance used to create an instance of + * the {@link AsyncEventQueueFactory}; must not be {@literal null}. + * @return a new instance of {@link AsyncEventQueueFactory} to create a {@link AsyncEventQueue}. + * @see org.apache.geode.cache.asyncqueue.AsyncEventQueueFactory + * @see org.apache.geode.cache.Cache + */ + protected @NonNull AsyncEventQueueFactory newAsyncEventQueueFactory(@NonNull Cache peerCache) { + return peerCache.createAsyncEventQueueFactory(); } /** * Constructs a new Apache Geode {@link AsyncEventListener} to register on an {@link AsyncEventQueue} attached to * the target {@link Region}, which uses the {@link CrudRepository} to perform data access operations on an external - * data source asynchronously when cache events and operations occur on the target {@link Region}. + * backend data source asynchronously when cache events and operations occur on the target {@link Region}. * * @return a new {@link RepositoryAsyncEventListener}. * @see org.springframework.geode.cache.RepositoryAsyncEventListener + * @see org.apache.geode.cache.asyncqueue.AsyncEventListener + * @see #newRepositoryAsyncEventListener(CrudRepository) * @see #getRepository() */ protected @NonNull AsyncEventListener newRepositoryAsyncEventListener() { - return new RepositoryAsyncEventListener<>(getRepository()); + return newRepositoryAsyncEventListener(getRepository()); + } + + /** + * Constructs a new Apache Geode {@link AsyncEventListener} to register on an {@link AsyncEventQueue} attached to + * the target {@link Region}, which uses the given {@link CrudRepository} to perform data access operations on an + * external, backend data source asynchronously when cache events and operations occur on the target {@link Region}. + * + * @param repository Spring Data {@link CrudRepository} used to perform data access operations on the external, + * backend data source; must not be {@literal null}. + * @return a new {@link RepositoryAsyncEventListener}. + * @see org.springframework.geode.cache.RepositoryAsyncEventListener + * @see org.apache.geode.cache.asyncqueue.AsyncEventListener + * @see #getRepository() + */ + protected @NonNull AsyncEventListener newRepositoryAsyncEventListener(@NonNull CrudRepository repository) { + return new RepositoryAsyncEventListener<>(repository); + } + + /** + * Applies the user-defined {@link Function} to the framework constructed/provided {@link AsyncEventListener} + * for post processing. + * + * @param asyncEventListener {@link AsyncEventListener} constructed by the framework and post processed by + * end-user code encapsulated in the {@link #applyToListener(Function) configured} {@link Function}. + * @return the post-processed {@link AsyncEventListener}. + * @see org.apache.geode.cache.asyncqueue.AsyncEventListener + * @see #applyToListener(Function) + */ + protected @NonNull AsyncEventListener postProcess(@NonNull AsyncEventListener asyncEventListener) { + return resolveAsyncEventListenerPostProcessor().apply(asyncEventListener); + } + + /** + * Applies the user-defined {@link Function} to the framework constructed/provided {@link AsyncEventQueue} + * for post processing. + * + * @param asyncEventQueue {@link AsyncEventQueue} constructed by the framework and post processed by + * end-user code encapsulated in the {@link #applyToQueue(Function) configured} {@link Function}. + * @return the post-processed {@link AsyncEventQueue}. + * @see org.apache.geode.cache.asyncqueue.AsyncEventQueue + * @see #applyToQueue(Function) + */ + protected @NonNull AsyncEventQueue postProcess(@NonNull AsyncEventQueue asyncEventQueue) { + + Function asyncEventQueuePostProcessor = + this.asyncEventQueuePostProcessor; + + return asyncEventQueuePostProcessor != null + ? asyncEventQueuePostProcessor.apply(asyncEventQueue) + : asyncEventQueue; + } + + /** + * Applies the user-defined {@link Function} to the framework constructed/provided {@link AsyncEventQueueFactory} + * for post processing. + * + * @param asyncEventQueueFactory {@link AsyncEventQueueFactory} constructed by the framework and post processed by + * end-user code encapsulated in the {@link #applyToQueueFactory(Function) configured} {@link Function}. + * @return the post-processed {@link AsyncEventQueueFactory}. + * @see org.apache.geode.cache.asyncqueue.AsyncEventQueueFactory + * @see #applyToQueueFactory(Function) + */ + protected @NonNull AsyncEventQueueFactory postProcess(@NonNull AsyncEventQueueFactory asyncEventQueueFactory) { + + Function asyncEventQueueFactoryPostProcessor = + this.asyncEventQueueFactoryPostProcessor; + + return asyncEventQueueFactoryPostProcessor != null + ? asyncEventQueueFactoryPostProcessor.apply(asyncEventQueueFactory) + : asyncEventQueueFactory; + } + + @SuppressWarnings("unchecked") + private @NonNull Function resolveAsyncEventListenerPostProcessor() { + + AsyncEventErrorHandler asyncEventErrorHandler = this.asyncEventErrorHandler; + + Function resolvedListenerPostProcessor = asyncEventErrorHandler != null + ? listener -> { + + if (listener instanceof RepositoryAsyncEventListener) { + ((RepositoryAsyncEventListener) listener).setAsyncEventErrorHandler(asyncEventErrorHandler); + } + + return listener; + } + : Function.identity(); + + Function asyncEventListenerPostProcessor = + this.asyncEventListenerPostProcessor; + + if (asyncEventListenerPostProcessor != null) { + resolvedListenerPostProcessor = resolvedListenerPostProcessor.andThen(asyncEventListenerPostProcessor); + } + + return resolvedListenerPostProcessor; + } + + /** + * Builder method used to configure the given user-defined {@link Function} applied to the framework constructed + * and provided {@link AsyncEventListener} for post processing. + * + * @param asyncEventListenerPostProcessor user-defined {@link Function} encapsulating the logic applied to + * the framework constructed/provided {@link AsyncEventListener} for post-processing. + * @return this {@link AsyncInlineCachingRegionConfigurer}. + * @see org.apache.geode.cache.asyncqueue.AsyncEventListener + * @see java.util.function.Function + */ + public AsyncInlineCachingRegionConfigurer applyToListener( + @Nullable Function asyncEventListenerPostProcessor) { + + this.asyncEventListenerPostProcessor = asyncEventListenerPostProcessor; + + return this; + } + + /** + * Builder method used to configure the given user-defined {@link Function} applied to the framework constructed + * and provided {@link AsyncEventQueue} for post processing. + * + * @param asyncEventQueuePostProcessor user-defined {@link Function} encapsulating the logic applied to + * the framework constructed {@link AsyncEventQueue} for post-processing. + * @return this {@link AsyncInlineCachingRegionConfigurer}. + * @see org.apache.geode.cache.asyncqueue.AsyncEventQueue + * @see java.util.function.Function + */ + public AsyncInlineCachingRegionConfigurer applyToQueue( + @Nullable Function asyncEventQueuePostProcessor) { + + this.asyncEventQueuePostProcessor = asyncEventQueuePostProcessor; + + return this; + } + + /** + * Builder method used to configure the given user-defined {@link Function} applied to the framework constructed + * and provided {@link AsyncEventQueueFactory} for post processing. + * + * @param asyncEventQueueFactoryPostProcessor user-defined {@link Function} encapsulating the logic applied to + * the framework constructed {@link AsyncEventQueueFactory} for post-processing. + * @return this {@link AsyncInlineCachingRegionConfigurer}. + * @see org.apache.geode.cache.asyncqueue.AsyncEventQueueFactory + * @see java.util.function.Function + */ + public AsyncInlineCachingRegionConfigurer applyToQueueFactory( + @Nullable Function asyncEventQueueFactoryPostProcessor) { + + this.asyncEventQueueFactoryPostProcessor = asyncEventQueueFactoryPostProcessor; + + return this; + } + + /** + * Builder method used to configure a {@link AsyncEventErrorHandler} to handle errors thrown while processing + * {@link AsyncEvent AsyncEvents} in the {@link AsyncEventListener}. + * + * @param errorHandler {@link AsyncEventErrorHandler} used to handle errors thrown while processing + * {@link AsyncEvent AsyncEvents} in the {@link AsyncEventListener}. + * @return this {@link AsyncInlineCachingRegionConfigurer}. + * @see org.springframework.geode.cache.RepositoryAsyncEventListener.AsyncEventErrorHandler + */ + public AsyncInlineCachingRegionConfigurer withAsyncEventErrorHandler( + @Nullable AsyncEventErrorHandler errorHandler) { + + this.asyncEventErrorHandler = errorHandler; + + return this; } /** diff --git a/spring-geode/src/test/java/org/springframework/geode/cache/AsyncInlineCachingRegionConfigurerUnitTests.java b/spring-geode/src/test/java/org/springframework/geode/cache/AsyncInlineCachingRegionConfigurerUnitTests.java index cd48d431..9945cc43 100644 --- a/spring-geode/src/test/java/org/springframework/geode/cache/AsyncInlineCachingRegionConfigurerUnitTests.java +++ b/spring-geode/src/test/java/org/springframework/geode/cache/AsyncInlineCachingRegionConfigurerUnitTests.java @@ -18,6 +18,7 @@ package org.springframework.geode.cache; import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; @@ -27,9 +28,11 @@ import static org.mockito.Mockito.verifyNoMoreInteractions; import java.time.Duration; import java.util.Arrays; +import java.util.function.Function; import java.util.function.Predicate; import org.junit.Test; +import org.mockito.InOrder; import org.apache.geode.cache.Cache; import org.apache.geode.cache.asyncqueue.AsyncEventListener; @@ -42,6 +45,7 @@ import org.apache.geode.cache.wan.GatewaySender; import org.springframework.data.gemfire.PeerRegionFactoryBean; import org.springframework.data.gemfire.util.ArrayUtils; import org.springframework.data.repository.CrudRepository; +import org.springframework.geode.cache.RepositoryAsyncEventListener.AsyncEventErrorHandler; /** * Unit Tests for {@link AsyncInlineCachingRegionConfigurer}. @@ -54,9 +58,12 @@ import org.springframework.data.repository.CrudRepository; * @see org.apache.geode.cache.asyncqueue.AsyncEventListener * @see org.apache.geode.cache.asyncqueue.AsyncEventQueue * @see org.apache.geode.cache.asyncqueue.AsyncEventQueueFactory + * @see org.apache.geode.cache.wan.GatewayEventFilter + * @see org.apache.geode.cache.wan.GatewayEventSubstitutionFilter * @see org.springframework.data.gemfire.PeerRegionFactoryBean * @see org.springframework.data.repository.CrudRepository * @see org.springframework.geode.cache.AsyncInlineCachingRegionConfigurer + * @see org.springframework.geode.cache.RepositoryAsyncEventListener.AsyncEventErrorHandler * @since 1.4.0 */ public class AsyncInlineCachingRegionConfigurerUnitTests { @@ -66,13 +73,13 @@ public class AsyncInlineCachingRegionConfigurerUnitTests { CrudRepository mockRepository = mock(CrudRepository.class); - Predicate regionBeanName = Predicate.isEqual("TestRegion"); + Predicate testRegionBeanName = Predicate.isEqual("TestRegion"); AsyncInlineCachingRegionConfigurer regionConfigurer = - new AsyncInlineCachingRegionConfigurer<>(mockRepository, regionBeanName); + new AsyncInlineCachingRegionConfigurer<>(mockRepository, testRegionBeanName); assertThat(regionConfigurer).isNotNull(); - assertThat(regionConfigurer.getRegionBeanName()).isEqualTo(regionBeanName); + assertThat(regionConfigurer.getRegionBeanName()).isEqualTo(testRegionBeanName); assertThat(regionConfigurer.getRepository()).isEqualTo(mockRepository); verifyNoInteractions(mockRepository); @@ -87,7 +94,8 @@ public class AsyncInlineCachingRegionConfigurerUnitTests { new AsyncInlineCachingRegionConfigurer<>(mockRepository, null); assertThat(regionConfigurer).isNotNull(); - assertThat(regionConfigurer.getRegionBeanName()).isNotNull(); + assertThat(regionConfigurer.getRegionBeanName()) + .isEqualTo(AsyncInlineCachingRegionConfigurer.DEFAULT_REGION_BEAN_NAME_PREDICATE); assertThat(regionConfigurer.getRepository()).isEqualTo(mockRepository); verifyNoInteractions(mockRepository); @@ -113,14 +121,30 @@ public class AsyncInlineCachingRegionConfigurerUnitTests { CrudRepository mockRepository = mock(CrudRepository.class); - Predicate regionBeanName = Predicate.isEqual("TestRegion"); + Predicate testRegionBeanName = Predicate.isEqual("TestRegion"); AsyncInlineCachingRegionConfigurer regionConfigurer = - AsyncInlineCachingRegionConfigurer.create(mockRepository, regionBeanName); + AsyncInlineCachingRegionConfigurer.create(mockRepository, testRegionBeanName); + + assertThat(regionConfigurer).isNotNull(); + assertThat(regionConfigurer.getRegionBeanName()).isEqualTo(testRegionBeanName); + assertThat(regionConfigurer.getRepository()).isEqualTo(mockRepository); + + verifyNoInteractions(mockRepository); + } + + @Test + public void createAsyncInlineCachingRegionConfigurerFromCrudRepositoryAndNullPredicate() { + + CrudRepository mockRepository = mock(CrudRepository.class); + + AsyncInlineCachingRegionConfigurer regionConfigurer = + AsyncInlineCachingRegionConfigurer.create(mockRepository, (Predicate) null); assertThat(regionConfigurer).isNotNull(); assertThat(regionConfigurer.getRepository()).isEqualTo(mockRepository); - assertThat(regionConfigurer.getRegionBeanName()).isEqualTo(regionBeanName); + assertThat(regionConfigurer.getRegionBeanName()) + .isEqualTo(AsyncInlineCachingRegionConfigurer.DEFAULT_REGION_BEAN_NAME_PREDICATE); verifyNoInteractions(mockRepository); } @@ -134,10 +158,30 @@ public class AsyncInlineCachingRegionConfigurerUnitTests { AsyncInlineCachingRegionConfigurer.create(mockRepository, "MockRegion"); assertThat(regionConfigurer).isNotNull(); - assertThat(regionConfigurer.getRepository()).isEqualTo(mockRepository); assertThat(regionConfigurer.getRegionBeanName()).isNotNull(); assertThat(regionConfigurer.getRegionBeanName().test("MockRegion")).isTrue(); + assertThat(regionConfigurer.getRegionBeanName().test("MOCKREGION")).isFalse(); + assertThat(regionConfigurer.getRegionBeanName().test("mockregion")).isFalse(); assertThat(regionConfigurer.getRegionBeanName().test("TestRegion")).isFalse(); + assertThat(regionConfigurer.getRepository()).isEqualTo(mockRepository); + + verifyNoInteractions(mockRepository); + } + + @Test + public void createAsyncInlineCachingRegionConfigurerFromCrudRepositoryAndNullString() { + + CrudRepository mockRepository = mock(CrudRepository.class); + + AsyncInlineCachingRegionConfigurer regionConfigurer = + AsyncInlineCachingRegionConfigurer.create(mockRepository, (String) null); + + assertThat(regionConfigurer).isNotNull(); + assertThat(regionConfigurer.getRegionBeanName()).isNotNull(); + assertThat(regionConfigurer.getRegionBeanName().test(null)).isTrue(); + assertThat(regionConfigurer.getRegionBeanName().test("MockRegion")).isFalse(); + assertThat(regionConfigurer.getRegionBeanName().test("TestRegion")).isFalse(); + assertThat(regionConfigurer.getRepository()).isEqualTo(mockRepository); verifyNoInteractions(mockRepository); } @@ -181,11 +225,11 @@ public class AsyncInlineCachingRegionConfigurerUnitTests { CrudRepository mockRepository = mock(CrudRepository.class); + PeerRegionFactoryBean peerRegionFactoryBean = mock(PeerRegionFactoryBean.class); + AsyncInlineCachingRegionConfigurer regionConfigurer = spy(new AsyncInlineCachingRegionConfigurer<>(mockRepository, Predicate.isEqual("TestRegion"))); - PeerRegionFactoryBean peerRegionFactoryBean = mock(PeerRegionFactoryBean.class); - doReturn(mockCache).when(peerRegionFactoryBean).getCache(); doReturn(mockAsyncEventQueue).when(regionConfigurer).newAsyncEventQueue(eq(mockCache), eq("TestRegion")); @@ -194,9 +238,9 @@ public class AsyncInlineCachingRegionConfigurerUnitTests { verify(regionConfigurer, times(1)) .configure(eq("TestRegion"), eq(peerRegionFactoryBean)); verify(regionConfigurer, times(1)).getRegionBeanName(); + verify(peerRegionFactoryBean, times(1)).getCache(); verify(regionConfigurer, times(1)) .newAsyncEventQueue(eq(mockCache), eq("TestRegion")); - verify(peerRegionFactoryBean, times(1)).getCache(); verify(peerRegionFactoryBean, times(1)) .setAsyncEventQueues(eq(ArrayUtils.asArray(mockAsyncEventQueue))); @@ -210,17 +254,20 @@ public class AsyncInlineCachingRegionConfigurerUnitTests { CrudRepository mockRepository = mock(CrudRepository.class); AsyncInlineCachingRegionConfigurer regionConfigurer = - spy(new AsyncInlineCachingRegionConfigurer<>(mockRepository, Predicate.isEqual("TestRegion"))); + spy(new AsyncInlineCachingRegionConfigurer<>(mockRepository, Predicate.isEqual("NoRegion"))); PeerRegionFactoryBean peerRegionFactoryBean = mock(PeerRegionFactoryBean.class); regionConfigurer.configure("MockRegion", peerRegionFactoryBean); + regionConfigurer.configure("TestRegion", peerRegionFactoryBean); verify(regionConfigurer, times(1)) .configure(eq("MockRegion"), eq(peerRegionFactoryBean)); - verify(regionConfigurer, times(1)).getRegionBeanName(); + verify(regionConfigurer, times(1)) + .configure(eq("TestRegion"), eq(peerRegionFactoryBean)); + verify(regionConfigurer, times(2)).getRegionBeanName(); verifyNoMoreInteractions(regionConfigurer); - verifyNoInteractions(peerRegionFactoryBean); + verifyNoInteractions(mockRepository, peerRegionFactoryBean); } @Test @@ -275,9 +322,9 @@ public class AsyncInlineCachingRegionConfigurerUnitTests { @Test @SuppressWarnings({ "rawtypes", "unchecked" }) - public void newAsyncEventQueueCreatesAsyncEventQueueFromCacheInitializedWithRegionConfigurer() { + public void newAsyncEventQueueCreatesQueueFromCacheInitializedWithRegionConfigurer() { - AsyncEventListener mockAsyncEventListener = mock(AsyncEventListener.class, "Mock AEQ Listener"); + AsyncEventErrorHandler mockAsyncEventErrorHandler = mock(AsyncEventErrorHandler.class); AsyncEventQueue mockAsyncEventQueue = mock(AsyncEventQueue.class, "Mock AEQ"); @@ -289,11 +336,18 @@ public class AsyncInlineCachingRegionConfigurerUnitTests { Duration batchTimeInterval = Duration.ofSeconds(15); + Function mockAsyncEventListenerFunction = mock(Function.class); + Function mockAsyncEventQueueFunction = mock(Function.class); + Function mockAsyncEventQueueFactoryFunction = mock(Function.class); + GatewayEventFilter mockEventFilterOne = mock(GatewayEventFilter.class); GatewayEventFilter mockEventFilterTwo = mock(GatewayEventFilter.class); GatewayEventSubstitutionFilter mockEventSubstitutionFilter = mock(GatewayEventSubstitutionFilter.class); + RepositoryAsyncEventListener mockAsyncEventListener = + mock(RepositoryAsyncEventListener.class, "Mock AEQ Listener"); + AsyncInlineCachingRegionConfigurer regionConfigurer = spy(new AsyncInlineCachingRegionConfigurer<>(mockRepository, Predicate.isEqual("TestRegion"))); @@ -301,7 +355,14 @@ public class AsyncInlineCachingRegionConfigurerUnitTests { doReturn(mockAsyncEventQueue).when(mockAsyncEventQueueFactory).create(eq("123"), eq(mockAsyncEventListener)); doReturn("123").when(regionConfigurer).generateId(eq("TestRegion")); doReturn(mockAsyncEventListener).when(regionConfigurer).newRepositoryAsyncEventListener(); + doReturn(mockAsyncEventListener).when(mockAsyncEventListenerFunction).apply(eq(mockAsyncEventListener)); + doReturn(mockAsyncEventQueue).when(mockAsyncEventQueueFunction).apply(eq(mockAsyncEventQueue)); + doReturn(mockAsyncEventQueueFactory).when(mockAsyncEventQueueFactoryFunction).apply(eq(mockAsyncEventQueueFactory)); + assertThat(regionConfigurer.applyToListener(mockAsyncEventListenerFunction)).isSameAs(regionConfigurer); + assertThat(regionConfigurer.applyToQueue(mockAsyncEventQueueFunction)).isSameAs(regionConfigurer); + assertThat(regionConfigurer.applyToQueueFactory(mockAsyncEventQueueFactoryFunction)).isSameAs(regionConfigurer); + assertThat(regionConfigurer.withAsyncEventErrorHandler(mockAsyncEventErrorHandler)).isSameAs(regionConfigurer); assertThat(regionConfigurer.withParallelQueue()).isSameAs(regionConfigurer); assertThat(regionConfigurer.withPersistentQueue()).isSameAs(regionConfigurer); assertThat(regionConfigurer.withQueueBatchConflationEnabled()).isSameAs(regionConfigurer); @@ -323,45 +384,103 @@ public class AsyncInlineCachingRegionConfigurerUnitTests { assertThat(regionConfigurer.newAsyncEventQueue(mockCache, "TestRegion")) .isEqualTo(mockAsyncEventQueue); - verify(mockCache, times(1)).createAsyncEventQueueFactory(); - verify(mockAsyncEventQueueFactory, times(1)).setBatchConflationEnabled(eq(true)); - verify(mockAsyncEventQueueFactory, times(1)).setBatchSize(eq(224)); - verify(mockAsyncEventQueueFactory, times(1)).setBatchTimeInterval(eq((int) batchTimeInterval.toMillis())); - verify(mockAsyncEventQueueFactory, times(1)).setDiskStoreName(eq("TestDiskStore")); - verify(mockAsyncEventQueueFactory, times(1)).setDiskSynchronous(eq(true)); - verify(mockAsyncEventQueueFactory, times(1)).setDispatcherThreads(eq(8)); - verify(mockAsyncEventQueueFactory, times(1)).setForwardExpirationDestroy(eq(true)); - verify(mockAsyncEventQueueFactory, times(1)).setGatewayEventSubstitutionListener(eq(mockEventSubstitutionFilter)); - verify(mockAsyncEventQueueFactory, times(1)).setMaximumQueueMemory(eq(51)); - verify(mockAsyncEventQueueFactory, times(1)).setOrderPolicy(eq(GatewaySender.OrderPolicy.THREAD)); - verify(mockAsyncEventQueueFactory, times(1)).setParallel(eq(true)); - verify(mockAsyncEventQueueFactory, times(1)).setPersistent(eq(true)); - verify(mockAsyncEventQueueFactory, times(1)).addGatewayEventFilter(eq(mockEventFilterOne)); - verify(mockAsyncEventQueueFactory, times(1)).addGatewayEventFilter(eq(mockEventFilterTwo)); - verify(mockAsyncEventQueueFactory, times(1)).pauseEventDispatching(); - verify(mockAsyncEventQueueFactory, times(1)).create(eq("123"), eq(mockAsyncEventListener)); - verify(regionConfigurer, times(1)).generateId(eq("TestRegion")); - verify(regionConfigurer, times(1)).newRepositoryAsyncEventListener(); + InOrder order = inOrder(regionConfigurer, mockAsyncEventListener, mockAsyncEventQueueFactory, mockCache, + mockAsyncEventListenerFunction, mockAsyncEventQueueFunction, mockAsyncEventQueueFactoryFunction); - verifyNoMoreInteractions(mockCache, mockAsyncEventQueueFactory); + order.verify(regionConfigurer, times(1)).newAsyncEventQueueFactory(eq(mockCache)); + order.verify(mockCache, times(1)).createAsyncEventQueueFactory(); + order.verify(mockAsyncEventQueueFactory, times(1)).setBatchConflationEnabled(eq(true)); + order.verify(mockAsyncEventQueueFactory, times(1)).setBatchSize(eq(224)); + order.verify(mockAsyncEventQueueFactory, times(1)).setBatchTimeInterval(eq((int) batchTimeInterval.toMillis())); + order.verify(mockAsyncEventQueueFactory, times(1)).setDiskStoreName(eq("TestDiskStore")); + order.verify(mockAsyncEventQueueFactory, times(1)).setDiskSynchronous(eq(true)); + order.verify(mockAsyncEventQueueFactory, times(1)).setDispatcherThreads(eq(8)); + order.verify(mockAsyncEventQueueFactory, times(1)).setForwardExpirationDestroy(eq(true)); + order.verify(mockAsyncEventQueueFactory, times(1)).setGatewayEventSubstitutionListener(eq(mockEventSubstitutionFilter)); + order.verify(mockAsyncEventQueueFactory, times(1)).setMaximumQueueMemory(eq(51)); + order.verify(mockAsyncEventQueueFactory, times(1)).setOrderPolicy(eq(GatewaySender.OrderPolicy.THREAD)); + order.verify(mockAsyncEventQueueFactory, times(1)).setParallel(eq(true)); + order.verify(mockAsyncEventQueueFactory, times(1)).setPersistent(eq(true)); + order.verify(mockAsyncEventQueueFactory, times(1)).addGatewayEventFilter(eq(mockEventFilterOne)); + order.verify(mockAsyncEventQueueFactory, times(1)).addGatewayEventFilter(eq(mockEventFilterTwo)); + order.verify(mockAsyncEventQueueFactory, times(1)).pauseEventDispatching(); + order.verify(regionConfigurer, times(1)).generateId(eq("TestRegion")); + order.verify(regionConfigurer, times(1)).newRepositoryAsyncEventListener(); + order.verify(regionConfigurer, times(1)).postProcess(eq(mockAsyncEventListener)); + order.verify(mockAsyncEventListener, times(1)).setAsyncEventErrorHandler(eq(mockAsyncEventErrorHandler)); + order.verify(mockAsyncEventListenerFunction, times(1)).apply(eq(mockAsyncEventListener)); + order.verify(regionConfigurer, times(1)).postProcess(eq(mockAsyncEventQueueFactory)); + order.verify(mockAsyncEventQueueFactoryFunction, times(1)).apply(eq(mockAsyncEventQueueFactory)); + order.verify(mockAsyncEventQueueFactory, times(1)).create(eq("123"), eq(mockAsyncEventListener)); + order.verify(regionConfigurer, times(1)).postProcess(eq(mockAsyncEventQueue)); + order.verify(mockAsyncEventQueueFunction, times(1)).apply(eq(mockAsyncEventQueue)); - verifyNoInteractions(mockAsyncEventListener, mockAsyncEventQueue, mockRepository, mockEventFilterOne, - mockEventFilterTwo, mockEventSubstitutionFilter); + verifyNoMoreInteractions(mockCache, mockAsyncEventListener, mockAsyncEventListenerFunction, + mockAsyncEventQueueFactory, mockAsyncEventQueueFunction, mockAsyncEventQueueFactoryFunction); + + verifyNoInteractions(mockAsyncEventErrorHandler, mockAsyncEventQueue, mockRepository, + mockEventFilterOne, mockEventFilterTwo, mockEventSubstitutionFilter); } @Test - public void newRepositoryAsyncEventListener() { + public void newAsyncEventQueueCreatesQueueFromFactoryWithIdAndListener() { + + AsyncEventListener mockAsyncEventListener = mock(AsyncEventListener.class, "Mock AEQ Listener"); + + AsyncEventQueue mockAsyncEventQueue = mock(AsyncEventQueue.class, "Mock AEQ"); + + AsyncEventQueueFactory mockAsyncEventQueueFactory = mock(AsyncEventQueueFactory.class, "Mock AEQ Factory"); + + CrudRepository mockRepository = mock(CrudRepository.class); + + String asyncEventQueueId = "abc123"; + + doReturn(mockAsyncEventQueue).when(mockAsyncEventQueueFactory) + .create(eq(asyncEventQueueId), eq(mockAsyncEventListener)); + + AsyncInlineCachingRegionConfigurer regionConfigurer = + new AsyncInlineCachingRegionConfigurer<>(mockRepository, Predicate.isEqual("TestRegion")); + + assertThat(regionConfigurer.newAsyncEventQueue(mockAsyncEventQueueFactory, asyncEventQueueId, mockAsyncEventListener)) + .isEqualTo(mockAsyncEventQueue); + + verify(mockAsyncEventQueueFactory, times(1)) + .create(eq(asyncEventQueueId), eq(mockAsyncEventListener)); + verifyNoMoreInteractions(mockAsyncEventQueueFactory); + verifyNoInteractions(mockAsyncEventListener, mockAsyncEventQueue, mockRepository); + } + + @Test + public void newAsyncEventQueueFactoryCallsCacheCreateAsyncEventQueueFactory() { + + Cache mockCache = mock(Cache.class, "Mock Peer Cache"); CrudRepository mockRepository = mock(CrudRepository.class); AsyncInlineCachingRegionConfigurer regionConfigurer = - spy(new AsyncInlineCachingRegionConfigurer<>(mockRepository, Predicate.isEqual("TestRegion"))); + new AsyncInlineCachingRegionConfigurer<>(mockRepository, Predicate.isEqual("TestRegion")); + + regionConfigurer.newAsyncEventQueueFactory(mockCache); + + verify(mockCache, times(1)).createAsyncEventQueueFactory(); + verifyNoMoreInteractions(mockCache); + } + + @Test + @SuppressWarnings({ "rawtypes", "unchecked" }) + public void newRepositoryAsyncEventListenerReturnsNewListener() { + + CrudRepository mockRepository = mock(CrudRepository.class); + + AsyncInlineCachingRegionConfigurer regionConfigurer = + spy(new AsyncInlineCachingRegionConfigurer(mockRepository, Predicate.isEqual("TestRegion"))); AsyncEventListener listener = regionConfigurer.newRepositoryAsyncEventListener(); assertThat(listener).isInstanceOf(RepositoryAsyncEventListener.class); assertThat(((RepositoryAsyncEventListener) listener).getRepository()).isEqualTo(mockRepository); + verify(regionConfigurer, times(1)).newRepositoryAsyncEventListener(eq(mockRepository)); verify(regionConfigurer, times(1)).getRepository(); verifyNoInteractions(mockRepository); }