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 new file mode 100644 index 00000000..f1a8d338 --- /dev/null +++ b/spring-geode/src/main/java/org/springframework/geode/cache/AsyncInlineCachingRegionConfigurer.java @@ -0,0 +1,501 @@ +/* + * Copyright 2020 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 + * + * https://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.geode.cache; + +import java.time.Duration; +import java.util.List; +import java.util.Objects; +import java.util.Optional; +import java.util.UUID; +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.AsyncEventListener; +import org.apache.geode.cache.asyncqueue.AsyncEventQueue; +import org.apache.geode.cache.asyncqueue.AsyncEventQueueFactory; +import org.apache.geode.cache.wan.GatewayEventFilter; +import org.apache.geode.cache.wan.GatewayEventSubstitutionFilter; +import org.apache.geode.cache.wan.GatewaySender; + +import org.springframework.data.gemfire.PeerRegionFactoryBean; +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.lang.NonNull; +import org.springframework.lang.Nullable; +import org.springframework.util.Assert; +import org.springframework.util.StringUtils; + +/** + * A Spring Data for Apache Geode {@link RegionConfigurer} implementation used to configure a target {@link Region} + * to use {@literal Asynchronous Inline Caching} based on the Spring Data {@link CrudRepository Repositories} + * abstraction. + * + * @author John Blum + * @see java.util.function.Predicate + * @see org.apache.geode.cache.Cache + * @see org.apache.geode.cache.Region + * @see org.apache.geode.cache.asyncqueue.AsyncEventListener + * @see org.apache.geode.cache.asyncqueue.AsyncEventQueue + * @see org.apache.geode.cache.asyncqueue.AsyncEventQueueFactory + * @see org.springframework.data.gemfire.PeerRegionFactoryBean + * @see org.springframework.data.gemfire.config.annotation.RegionConfigurer + * @see org.springframework.data.repository.CrudRepository + * @since 1.4.0 + */ +public class AsyncInlineCachingRegionConfigurer implements RegionConfigurer { + + /** + * 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} + * on which to configure {@literal Asynchronous Inline Caching}. + * + * @param {@link Class type} of the entity. + * @param {@link Class type} of the identifier, or {@link Region} key. + * @param repository {@link CrudRepository} used to perform data access operations on an external data source + * triggered by cache events and operations on the identified {@link Region}; must not be {@literal null}. + * @param regionBeanName {@link Predicate} used to identify the {@link Region} by {@link String name} on which + * {@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 java.util.function.Predicate + */ + public static AsyncInlineCachingRegionConfigurer create(@NonNull CrudRepository repository, + @Nullable Predicate regionBeanName) { + + return new AsyncInlineCachingRegionConfigurer<>(repository, regionBeanName); + } + + /** + * Factory method used to construct a new instance of {@link AsyncInlineCachingRegionConfigurer} initialized with + * the given Spring Data {@link CrudRepository} and {@link String} identifying the target {@link Region} + * on which to configure {@literal Asynchronous Inline Caching}. + * + * @param {@link Class type} of the entity. + * @param {@link Class type} of the identifier, or {@link Region} key. + * @param repository {@link CrudRepository} used to perform data access operations on an external data source + * triggered by cache events and operations on the identified {@link Region}; must not be {@literal null}. + * @param regionBeanName {@link String} used to identify the {@link Region} by {@link String name} on which + * {@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 java.lang.String + */ + public static AsyncInlineCachingRegionConfigurer create(@NonNull CrudRepository repository, + @Nullable String regionBeanName) { + + return new AsyncInlineCachingRegionConfigurer<>(repository, Predicate.isEqual(regionBeanName)); + } + + private Boolean batchConflationEnabled; + private Boolean diskSynchronous; + private Boolean forwardExpirationDestroy; + private Boolean parallel; + private Boolean persistent; + private Boolean pauseEventDispatching; + + private final CrudRepository repository; + + private Integer batchSize; + private Integer batchTimeInterval; + private Integer dispatcherThreads; + private Integer maximumQueueMemory; + + @SuppressWarnings("rawtypes") + private GatewayEventSubstitutionFilter gatewayEventSubstitutionFilter; + + private GatewaySender.OrderPolicy orderPolicy; + + private List gatewayEventFilters; + + private final Predicate regionBeanName; + + private String diskStoreName; + + /** + * Constructs a new instance of {@link AsyncInlineCachingRegionConfigurer} initialized with the given + * {@link CrudRepository} and {@link Predicate} identifying the {@link Region} on which + * {@literal Asynchronous Inline Caching} will be configured. + * + * @param repository {@link CrudRepository} used to perform data access operations on an external data source + * triggered by cache events and operations on the identified {@link Region}; must not be {@literal null}. + * @param regionBeanName {@link Predicate} used to identify the {@link Region} by {@link String name} on which + * {@literal Asynchronous Inline Caching} will be configured. + * @throws IllegalArgumentException if {@link CrudRepository} is {@literal null}. + * @see org.springframework.data.repository.CrudRepository + * @see java.util.function.Predicate + */ + public AsyncInlineCachingRegionConfigurer(@NonNull CrudRepository repository, + @Nullable Predicate regionBeanName) { + + Assert.notNull(repository, "CrudRepository must not be null"); + + this.repository = repository; + this.regionBeanName = regionBeanName != null ? regionBeanName : beanName -> false; + } + + /** + * Gets the {@link Predicate} identifying the {@link Region} on which {@literal Asynchronous Inline Caching} + * will be configured. + * + * @return the {@link Predicate} used to match the {@link Region} by {@link String name} on which + * {@literal Asynchronous Inline Caching} will be configured; never {@literal null}. + * @see java.util.function.Predicate + */ + protected @NonNull Predicate getRegionBeanName() { + return this.regionBeanName; + } + + /** + * Gets the Spring Data {@link CrudRepository} used to perform data access operations on an external data source + * triggered cache events and operations on the target {@link Region}. + * + * @return the Spring Data {@link CrudRepository} used to perform data access operations on an external data source + * triggered cache events and operations on the target {@link Region}; never {@literal null}. + * @see org.springframework.data.repository.CrudRepository + */ + protected @NonNull CrudRepository getRepository() { + return this.repository; + } + + /** + * Configures the target {@link Region} by {@link String name} with {@literal Asynchronous Inline Caching} + * functionality. + * + * Effectively, this Configurer creates an {@link AsyncEventQueue} attached to the target {@link Region} with a + * registered {@link RepositoryAsyncEventListener} to perform asynchronous data access operations triggered by + * cache operations to an external, backend data source using a Spring Data {@link CrudRepository}. + * + * @param beanName {@link String} specifying the name of the target {@link Region} and bean name + * in the Spring container. + * @param bean {@link PeerRegionFactoryBean} containing the configuration of the target {@link Region} + * in the Spring container. + * @see org.springframework.data.gemfire.PeerRegionFactoryBean + */ + @Override + public void configure(String beanName, PeerRegionFactoryBean bean) { + + if (getRegionBeanName().test(beanName)) { + + AsyncEventQueue queue = newAsyncEventQueue((Cache) bean.getCache(), beanName); + + // TODO: change; PeerRegionFactoryBean.setAsyncEventQueues(..) overwrites the existing user-configured AEQs. + bean.setAsyncEventQueues(ArrayUtils.asArray(queue)); + } + } + + /** + * Generates a new {@link String ID} for the {@link AsyncEventQueue}. + * + * @param regionBeanName {@link String name} of the target {@link Region}. + * @return a new {@link String ID} for the {@link AsyncEventQueue}. + */ + protected @NonNull String generateId(@NonNull String regionBeanName) { + + Assert.hasText(regionBeanName, () -> String.format("Region bean name [%s] must be specified", regionBeanName)); + + return regionBeanName.concat(String.format("-AEQ:%s", UUID.randomUUID().toString())); + } + + /** + * Constructs a new instance of an {@link AsyncEventQueue} to attach to the target {@link Region} configured for + * {@literal Asynchronous Inline Caching}. + * + * @param peerCache reference to the {@link Cache peer cache}; must not be {@literal null}. + * @param regionBeanName {@link String name} of the target {@link Region}; must not be {@literal null}. + * @return a new {@link AsyncEventQueue}. + * @see org.apache.geode.cache.asyncqueue.AsyncEventQueue + * @see org.apache.geode.cache.Cache + * @see #newRepositoryAsyncEventListener() + * @see #generateId(String) + */ + protected AsyncEventQueue newAsyncEventQueue(@NonNull Cache peerCache, @NonNull String regionBeanName) { + + AsyncEventQueueFactory asyncEventQueueFactory = peerCache.createAsyncEventQueueFactory(); + + Optional.ofNullable(this.batchConflationEnabled).ifPresent(asyncEventQueueFactory::setBatchConflationEnabled); + Optional.ofNullable(this.batchSize).ifPresent(asyncEventQueueFactory::setBatchSize); + Optional.ofNullable(this.batchTimeInterval).ifPresent(asyncEventQueueFactory::setBatchTimeInterval); + Optional.ofNullable(this.diskStoreName).filter(StringUtils::hasText).ifPresent(asyncEventQueueFactory::setDiskStoreName); + Optional.ofNullable(this.diskSynchronous).ifPresent(asyncEventQueueFactory::setDiskSynchronous); + Optional.ofNullable(this.dispatcherThreads).ifPresent(asyncEventQueueFactory::setDispatcherThreads); + Optional.ofNullable(this.forwardExpirationDestroy).ifPresent(asyncEventQueueFactory::setForwardExpirationDestroy); + Optional.ofNullable(this.gatewayEventSubstitutionFilter).ifPresent(asyncEventQueueFactory::setGatewayEventSubstitutionListener); + Optional.ofNullable(this.maximumQueueMemory).ifPresent(asyncEventQueueFactory::setMaximumQueueMemory); + Optional.ofNullable(this.orderPolicy).ifPresent(asyncEventQueueFactory::setOrderPolicy); + Optional.ofNullable(this.parallel).ifPresent(asyncEventQueueFactory::setParallel); + Optional.ofNullable(this.persistent).ifPresent(asyncEventQueueFactory::setPersistent); + + CollectionUtils.nullSafeList(this.gatewayEventFilters).stream() + .filter(Objects::nonNull) + .forEach(asyncEventQueueFactory::addGatewayEventFilter); + + if (Boolean.TRUE.equals(this.pauseEventDispatching)) { + asyncEventQueueFactory.pauseEventDispatching(); + } + + return asyncEventQueueFactory.create(generateId(regionBeanName), newRepositoryAsyncEventListener()); + } + + /** + * 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}. + * + * @return a new {@link RepositoryAsyncEventListener}. + * @see org.springframework.geode.cache.RepositoryAsyncEventListener + * @see #getRepository() + */ + protected @NonNull AsyncEventListener newRepositoryAsyncEventListener() { + return new RepositoryAsyncEventListener<>(getRepository()); + } + + /** + * Builder method used to enable all {@link AsyncEventQueue AEQs} attached to {@link Region Regions} hosted + * and distributed across the cache cluster to process cache events. + * + * Default is {@literal false}. + * + * @return this {@link AsyncInlineCachingRegionConfigurer}. + */ + public AsyncInlineCachingRegionConfigurer withParallelQueue() { + this.parallel = true; + return this; + } + + /** + * Builder method used to enable the {@link AsyncEventQueue} to persist cache events to disk in order to + * preserve unprocessed cache events while offline. + * + * Keep in mind that the {@link AsyncEventQueue} must be persistent if the data {@link Region} + * to which the AEQ is attached is persistent. + * + * Default is {@literal false}. + * + * @return this {@link AsyncInlineCachingRegionConfigurer}. + */ + public AsyncInlineCachingRegionConfigurer withPersistentQueue() { + this.persistent = true; + return this; + } + + /** + * Builder method used to configure the {@link AsyncEventQueue} to conflate cache events in the queue. + * + * When conflation is enabled, the AEQ listener will only receive the latest update in the AEQ for cache entry + * based on key. + * + * Defaults to {@literal false}. + * + * @return this {@link AsyncInlineCachingRegionConfigurer}. + */ + public AsyncInlineCachingRegionConfigurer withQueueBatchConflationEnabled() { + this.batchConflationEnabled = true; + return this; + } + + /** + * Builder method used to configure the {@link AsyncEventQueue} {@link Integer batch size}, which determines + * the number (i.e. threshold) of cache events that will trigger the AEQ listener, before any set period of time. + * + * The batch size is often used in tandem with the batch time interval, which determines when the AEQ listener + * will be invoked after a period of time if the batch size is not reached within the period so that cache events + * can also be processed in a timely manner if they are occurring infrequently. + * + * Defaults to {@literal 100}. + * + * @param batchSize the {@link Integer number} of cache events in the queue before the AEQ listener is called. + * @return this {@link AsyncInlineCachingRegionConfigurer}. + * @see #withQueueBatchTimeInterval(Duration) + */ + public AsyncInlineCachingRegionConfigurer withQueueBatchSize(int batchSize) { + this.batchSize = batchSize; + return this; + } + + /** + * Builder method used to configure the {@link AsyncEventQueue} {@link Duration batch time interval} determining + * when the AEQ listener will be trigger before the number of cache events reaches any set size. + * + * The {@link Duration} is converted to milliseconds (ms), as expected by the configuration + * of the {@link AsyncEventQueue}. + * + * The batch time interval is often used in tandem with batch size, which determines for how many cache events + * in the queue will trigger the AEQ listener. If cache events are occurring rather frequently, then the batch size + * can help reduce memory consumption by processing the cache events before the batch time interval expires. + * + * Defaults to {@link 5 ms}. + * + * @param batchTimeInterval {@link Duration} of time to determine when the AEQ listener should be invoked with + * any existing cache events in the queue. + * @return this {@link AsyncInlineCachingRegionConfigurer}. + */ + public AsyncInlineCachingRegionConfigurer withQueueBatchTimeInterval(Duration batchTimeInterval) { + + this.batchTimeInterval = batchTimeInterval != null + ? Long.valueOf(batchTimeInterval.toMillis()).intValue() + : null; + + return this; + } + + /** + * Builder method used to configure the {@link String name} of the {@link DiskStore} used by + * the {@link AsyncEventQueue} to persist or overflow cache events. + * + * By default, the AEQ will write cache events to the {@literal DEFAULT} {@link DiskStore}. + * + * @param diskStoreName {@link String name} of the {@link DiskStore}. + * @return this {@link AsyncInlineCachingRegionConfigurer}. + */ + public AsyncInlineCachingRegionConfigurer withQueueDiskStore(String diskStoreName) { + this.diskStoreName = diskStoreName; + return this; + } + + /** + * Builder method used to configure the {@link AsyncEventQueue} to perform all disk write operations synchronously. + * + * Default is {@literal true}. + * + * @return this {@link AsyncInlineCachingRegionConfigurer}. + */ + public AsyncInlineCachingRegionConfigurer withQueueDiskSynchronizationEnabled() { + this.diskSynchronous = true; + return this; + } + + /** + * Builder method to configure the number of {@link Thread Threads} to process the cache events (contents) + * in the {@link AsyncEventQueue} when the queue is parallel. + * + * When a queue is parallel, the total number of queues is determined by the number of Geode members + * hosting the {@link Region} to which the queue is attached. + * + * When a queue is serial and multiple dispatcher threads are configured, Geode creates an additional copy of + * the queue for each thread on each Geode member that hosts the queue. When the queue is serial and multiple + * dispatcher threads are configure, then you can use the {@link GatewaySender.OrderPolicy} to control + * the distribution of cache events from the queue by the threads. + * + * Default is {@literal 5}. + * + * @param dispatcherThreadCount {@link Integer number} of dispatcher {@link Thread Threads} processing cache events + * in the queue. + * @return this {@link AsyncInlineCachingRegionConfigurer}. + */ + public AsyncInlineCachingRegionConfigurer withQueueDispatcherThreadCount(int dispatcherThreadCount) { + this.dispatcherThreads = dispatcherThreadCount; + return this; + } + + /** + * Builder method used to configure whether the {@link AsyncEventQueue} is currently processing cache events + * or is paused. + * + * When paused, cache events will not be dispatched to the AEQ listener for processing. Call the + * {@link AsyncEventQueue#resumeEventDispatching()} to resume cache event processing and AEQ listener callbacks. + * + * @return this {@link AsyncInlineCachingRegionConfigurer}. + */ + public AsyncInlineCachingRegionConfigurer withQueueEventDispatchingPaused() { + this.pauseEventDispatching = true; + return this; + } + + /** + * Builder method to configure the {@link AsyncEventQueue} with a {@link List} of + * {@link GatewayEventFilter GatewayEventFilters} to filter cache events sent to the configured AEQ listener. + * + * @param eventFilters {@link List} of {@link GatewayEventFilter GatewayEventFilters} used to control and filter + * the cache events sent to the configured AEQ listener. + * @return this {@link AsyncInlineCachingRegionConfigurer}. + * @see org.apache.geode.cache.wan.GatewayEventFilter + * @see java.util.List + */ + public AsyncInlineCachingRegionConfigurer withQueueEventFilters(List eventFilters) { + this.gatewayEventFilters = eventFilters; + return this; + } + + /** + * Builder method used to configure the {@link AsyncEventQueue} with a + * {@link GatewayEventSubstitutionFilter cache event substitution filter} used to replace (or "substitute") + * the original cache entry event value enqueued in the AEQ. + * + * @param eventSubstitutionFilter {@link GatewayEventSubstitutionFilter} used to replace/substitute the value + * in the enqueued cache entry event. + * @return this {@link AsyncInlineCachingRegionConfigurer}. + * @see org.apache.geode.cache.wan.GatewayEventSubstitutionFilter + */ + public AsyncInlineCachingRegionConfigurer withQueueEventSubstitutionFilter( + @Nullable GatewayEventSubstitutionFilter eventSubstitutionFilter) { + + this.gatewayEventSubstitutionFilter = eventSubstitutionFilter; + + return this; + } + + /** + * Builder method used to configure whether cache {@link Region} entry destroyed events due to expiration + * are forwarded to the {@link AsyncEventQueue}. + * + * @return this {@link AsyncInlineCachingRegionConfigurer}. + */ + public AsyncInlineCachingRegionConfigurer withQueueForwardedExpirationDestroyEvents() { + this.forwardExpirationDestroy = true; + return this; + } + + /** + * Builder method used to configure the maximum JVM Heap memory in megabytes used by the {@link AsyncEventQueue}. + * + * After the maximum memory threshold is reached then the AEQ overflows cache events to disk. + * + * Default to {@literal 100 MB}. + * + * @param maximumMemory {@link Integer} value specifying the maximum amount of memory in megabytes used by the AEQ + * to capture cache events. + * @return this {@link AsyncInlineCachingRegionConfigurer}. + */ + public AsyncInlineCachingRegionConfigurer withQueueMaxMemory(int maximumMemory) { + this.maximumQueueMemory = maximumMemory; + return this; + } + + /** + * Builder method used to configure the {@link AsyncEventQueue} order of processing for cache events when the AEQ + * is serial and the AEQ is using multiple dispatcher threads. + * + * @param orderPolicy {@link GatewaySender.OrderPolicy} used to determine the order of processing for cache events + * when the AEQ is serial and uses multiple dispatcher threads. + * @return this {@link AsyncInlineCachingRegionConfigurer}. + * @see org.apache.geode.cache.wan.GatewaySender.OrderPolicy + */ + public AsyncInlineCachingRegionConfigurer withQueueOrderPolicy( + @Nullable GatewaySender.OrderPolicy orderPolicy) { + + this.orderPolicy = orderPolicy; + + 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 new file mode 100644 index 00000000..cd48d431 --- /dev/null +++ b/spring-geode/src/test/java/org/springframework/geode/cache/AsyncInlineCachingRegionConfigurerUnitTests.java @@ -0,0 +1,368 @@ +/* + * Copyright 2020 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 + * + * https://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.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.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.verifyNoMoreInteractions; + +import java.time.Duration; +import java.util.Arrays; +import java.util.function.Predicate; + +import org.junit.Test; + +import org.apache.geode.cache.Cache; +import org.apache.geode.cache.asyncqueue.AsyncEventListener; +import org.apache.geode.cache.asyncqueue.AsyncEventQueue; +import org.apache.geode.cache.asyncqueue.AsyncEventQueueFactory; +import org.apache.geode.cache.wan.GatewayEventFilter; +import org.apache.geode.cache.wan.GatewayEventSubstitutionFilter; +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; + +/** + * Unit Tests for {@link AsyncInlineCachingRegionConfigurer}. + * + * @author John Blum + * @see java.util.function.Predicate + * @see org.junit.Test + * @see org.mockito.Mockito + * @see org.apache.geode.cache.Cache + * @see org.apache.geode.cache.asyncqueue.AsyncEventListener + * @see org.apache.geode.cache.asyncqueue.AsyncEventQueue + * @see org.apache.geode.cache.asyncqueue.AsyncEventQueueFactory + * @see org.springframework.data.gemfire.PeerRegionFactoryBean + * @see org.springframework.data.repository.CrudRepository + * @see org.springframework.geode.cache.AsyncInlineCachingRegionConfigurer + * @since 1.4.0 + */ +public class AsyncInlineCachingRegionConfigurerUnitTests { + + @Test + public void constructAsyncInlineCachingRegionConfigurer() { + + CrudRepository mockRepository = mock(CrudRepository.class); + + Predicate regionBeanName = Predicate.isEqual("TestRegion"); + + AsyncInlineCachingRegionConfigurer regionConfigurer = + new AsyncInlineCachingRegionConfigurer<>(mockRepository, regionBeanName); + + assertThat(regionConfigurer).isNotNull(); + assertThat(regionConfigurer.getRegionBeanName()).isEqualTo(regionBeanName); + assertThat(regionConfigurer.getRepository()).isEqualTo(mockRepository); + + verifyNoInteractions(mockRepository); + } + + @Test + public void constructAsyncInlineCachingRegionConfigurerWithNullRegionBeanNamePredicate() { + + CrudRepository mockRepository = mock(CrudRepository.class); + + AsyncInlineCachingRegionConfigurer regionConfigurer = + new AsyncInlineCachingRegionConfigurer<>(mockRepository, null); + + assertThat(regionConfigurer).isNotNull(); + assertThat(regionConfigurer.getRegionBeanName()).isNotNull(); + assertThat(regionConfigurer.getRepository()).isEqualTo(mockRepository); + + verifyNoInteractions(mockRepository); + } + + @Test(expected = IllegalArgumentException.class) + public void constructAsyncInlineCachingRegionConfigurerWithNullRepositoryThrowsIllegalArgumentException() { + + try { + new AsyncInlineCachingRegionConfigurer<>(null, Predicate.isEqual("MockRegion")); + } + catch (IllegalArgumentException expected) { + + assertThat(expected).hasMessage("CrudRepository must not be null"); + assertThat(expected).hasNoCause(); + + throw expected; + } + } + + @Test + public void createAsyncInlineCachingRegionConfigurerFromCrudRepositoryAndPredicate() { + + CrudRepository mockRepository = mock(CrudRepository.class); + + Predicate regionBeanName = Predicate.isEqual("TestRegion"); + + AsyncInlineCachingRegionConfigurer regionConfigurer = + AsyncInlineCachingRegionConfigurer.create(mockRepository, regionBeanName); + + assertThat(regionConfigurer).isNotNull(); + assertThat(regionConfigurer.getRepository()).isEqualTo(mockRepository); + assertThat(regionConfigurer.getRegionBeanName()).isEqualTo(regionBeanName); + + verifyNoInteractions(mockRepository); + } + + @Test + public void createAsyncInlineCachingRegionConfigurerFromCrudRepositoryAndString() { + + CrudRepository mockRepository = mock(CrudRepository.class); + + AsyncInlineCachingRegionConfigurer regionConfigurer = + 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("TestRegion")).isFalse(); + + verifyNoInteractions(mockRepository); + } + + @Test(expected = IllegalArgumentException.class) + public void createAsyncInlineCachingRegionConfigurerFromNullCrudRepositoryAndPredicateThrowsIllegalArgumentException() { + + try { + AsyncInlineCachingRegionConfigurer.create(null, Predicate.isEqual("TestRegion")); + } + catch (IllegalArgumentException expected) { + + assertThat(expected).hasMessage("CrudRepository must not be null"); + assertThat(expected).hasNoCause(); + + throw expected; + } + } + + @Test(expected = IllegalArgumentException.class) + public void createAsyncInlineCachingRegionConfigurerFromNullCrudRepositoryAndStringThrowsIllegalArgumentException() { + + try { + AsyncInlineCachingRegionConfigurer.create(null, "MockRegion"); + } + catch (IllegalArgumentException expected) { + + assertThat(expected).hasMessage("CrudRepository must not be null"); + assertThat(expected).hasNoCause(); + + throw expected; + } + } + + @Test + public void configureConfiguresRegionForAsyncInlineCachingFunctionality() { + + AsyncEventQueue mockAsyncEventQueue = mock(AsyncEventQueue.class, "Mock AEQ"); + + Cache mockCache = mock(Cache.class, "Mock Peer Cache"); + + CrudRepository mockRepository = mock(CrudRepository.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")); + + regionConfigurer.configure("TestRegion", peerRegionFactoryBean); + + verify(regionConfigurer, times(1)) + .configure(eq("TestRegion"), eq(peerRegionFactoryBean)); + verify(regionConfigurer, times(1)).getRegionBeanName(); + verify(regionConfigurer, times(1)) + .newAsyncEventQueue(eq(mockCache), eq("TestRegion")); + verify(peerRegionFactoryBean, times(1)).getCache(); + verify(peerRegionFactoryBean, times(1)) + .setAsyncEventQueues(eq(ArrayUtils.asArray(mockAsyncEventQueue))); + + verifyNoMoreInteractions(regionConfigurer, peerRegionFactoryBean); + verifyNoInteractions(mockAsyncEventQueue, mockCache, mockRepository); + } + + @Test + public void configureDoesNothingWhenRegionBeanNameDoesNotMatch() { + + CrudRepository mockRepository = mock(CrudRepository.class); + + AsyncInlineCachingRegionConfigurer regionConfigurer = + spy(new AsyncInlineCachingRegionConfigurer<>(mockRepository, Predicate.isEqual("TestRegion"))); + + PeerRegionFactoryBean peerRegionFactoryBean = mock(PeerRegionFactoryBean.class); + + regionConfigurer.configure("MockRegion", peerRegionFactoryBean); + + verify(regionConfigurer, times(1)) + .configure(eq("MockRegion"), eq(peerRegionFactoryBean)); + verify(regionConfigurer, times(1)).getRegionBeanName(); + verifyNoMoreInteractions(regionConfigurer); + verifyNoInteractions(peerRegionFactoryBean); + } + + @Test + public void generateIdIsSuccessful() { + + CrudRepository mockRepository = mock(CrudRepository.class); + + AsyncInlineCachingRegionConfigurer regionConfigurer = + new AsyncInlineCachingRegionConfigurer<>(mockRepository, Predicate.isEqual("TestRegion")); + + assertThat(regionConfigurer.generateId("MockRegion")).startsWith("MockRegion-AEQ:"); + + verifyNoInteractions(mockRepository); + } + + private void testGenerateIdWithInvalidRegionBeanName(String regionBeanName) { + + CrudRepository mockRepository = mock(CrudRepository.class); + + AsyncInlineCachingRegionConfigurer regionConfigurer = + new AsyncInlineCachingRegionConfigurer<>(mockRepository, Predicate.isEqual("TestRegion")); + + try { + regionConfigurer.generateId(regionBeanName); + } + catch (IllegalArgumentException expected) { + + assertThat(expected).hasMessage("Region bean name [%s] must be specified", regionBeanName); + assertThat(expected).hasNoCause(); + + throw expected; + } + finally { + verifyNoInteractions(mockRepository); + } + } + + @Test(expected = IllegalArgumentException.class) + public void generateIdWhenRegionBeanNameIsBlank() { + testGenerateIdWithInvalidRegionBeanName(" "); + } + + @Test(expected = IllegalArgumentException.class) + public void generateIdWhenRegionBeanNameIsEmpty() { + testGenerateIdWithInvalidRegionBeanName(""); + } + + @Test(expected = IllegalArgumentException.class) + public void generateIdWhenRegionBeanNameIsNull() { + testGenerateIdWithInvalidRegionBeanName(null); + } + + @Test + @SuppressWarnings({ "rawtypes", "unchecked" }) + public void newAsyncEventQueueCreatesAsyncEventQueueFromCacheInitializedWithRegionConfigurer() { + + AsyncEventListener mockAsyncEventListener = mock(AsyncEventListener.class, "Mock AEQ Listener"); + + AsyncEventQueue mockAsyncEventQueue = mock(AsyncEventQueue.class, "Mock AEQ"); + + AsyncEventQueueFactory mockAsyncEventQueueFactory = mock(AsyncEventQueueFactory.class, "Mock AEQ Factory"); + + Cache mockCache = mock(Cache.class, "Mock Cache"); + + CrudRepository mockRepository = mock(CrudRepository.class); + + Duration batchTimeInterval = Duration.ofSeconds(15); + + GatewayEventFilter mockEventFilterOne = mock(GatewayEventFilter.class); + GatewayEventFilter mockEventFilterTwo = mock(GatewayEventFilter.class); + + GatewayEventSubstitutionFilter mockEventSubstitutionFilter = mock(GatewayEventSubstitutionFilter.class); + + AsyncInlineCachingRegionConfigurer regionConfigurer = + spy(new AsyncInlineCachingRegionConfigurer<>(mockRepository, Predicate.isEqual("TestRegion"))); + + doReturn(mockAsyncEventQueueFactory).when(mockCache).createAsyncEventQueueFactory(); + doReturn(mockAsyncEventQueue).when(mockAsyncEventQueueFactory).create(eq("123"), eq(mockAsyncEventListener)); + doReturn("123").when(regionConfigurer).generateId(eq("TestRegion")); + doReturn(mockAsyncEventListener).when(regionConfigurer).newRepositoryAsyncEventListener(); + + assertThat(regionConfigurer.withParallelQueue()).isSameAs(regionConfigurer); + assertThat(regionConfigurer.withPersistentQueue()).isSameAs(regionConfigurer); + assertThat(regionConfigurer.withQueueBatchConflationEnabled()).isSameAs(regionConfigurer); + assertThat(regionConfigurer.withQueueBatchSize(224)).isSameAs(regionConfigurer); + assertThat(regionConfigurer.withQueueBatchTimeInterval(batchTimeInterval)).isSameAs(regionConfigurer); + assertThat(regionConfigurer.withQueueDiskStore("TestDiskStore")).isSameAs(regionConfigurer); + assertThat(regionConfigurer.withQueueDiskSynchronizationEnabled()).isSameAs(regionConfigurer); + assertThat(regionConfigurer.withQueueDispatcherThreadCount(8)).isSameAs(regionConfigurer); + assertThat(regionConfigurer.withQueueEventDispatchingPaused()).isSameAs(regionConfigurer); + assertThat(regionConfigurer.withQueueEventFilters(Arrays.asList(mockEventFilterOne, mockEventFilterTwo))) + .isSameAs(regionConfigurer); + assertThat(regionConfigurer.withQueueEventSubstitutionFilter(mockEventSubstitutionFilter)) + .isSameAs(regionConfigurer); + assertThat(regionConfigurer.withQueueForwardedExpirationDestroyEvents()).isSameAs(regionConfigurer); + assertThat(regionConfigurer.withQueueMaxMemory(51)).isSameAs(regionConfigurer); + assertThat(regionConfigurer.withQueueOrderPolicy(GatewaySender.OrderPolicy.THREAD)).isSameAs(regionConfigurer); + + // Create the AsyncEventQueue (AEQ) + 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(); + + verifyNoMoreInteractions(mockCache, mockAsyncEventQueueFactory); + + verifyNoInteractions(mockAsyncEventListener, mockAsyncEventQueue, mockRepository, mockEventFilterOne, + mockEventFilterTwo, mockEventSubstitutionFilter); + } + + @Test + public void newRepositoryAsyncEventListener() { + + 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)).getRepository(); + verifyNoInteractions(mockRepository); + } +}