Create SDG RegionConfigurer implementation to configure a target Region with Asynchronous, Inline Caching functionality.
Resolves gh-58.
This commit is contained in:
John Blum
501
spring-geode/src/main/java/org/springframework/geode/cache/AsyncInlineCachingRegionConfigurer.java
vendored
Normal file
501
spring-geode/src/main/java/org/springframework/geode/cache/AsyncInlineCachingRegionConfigurer.java
vendored
Normal file
@@ -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<T, ID> 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 <T> {@link Class type} of the entity.
|
||||
* @param <ID> {@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 <T, ID> AsyncInlineCachingRegionConfigurer<T, ID> create(@NonNull CrudRepository<T, ID> repository,
|
||||
@Nullable Predicate<String> 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 <T> {@link Class type} of the entity.
|
||||
* @param <ID> {@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 <T, ID> AsyncInlineCachingRegionConfigurer<T, ID> create(@NonNull CrudRepository<T, ID> 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<T, ID> repository;
|
||||
|
||||
private Integer batchSize;
|
||||
private Integer batchTimeInterval;
|
||||
private Integer dispatcherThreads;
|
||||
private Integer maximumQueueMemory;
|
||||
|
||||
@SuppressWarnings("rawtypes")
|
||||
private GatewayEventSubstitutionFilter gatewayEventSubstitutionFilter;
|
||||
|
||||
private GatewaySender.OrderPolicy orderPolicy;
|
||||
|
||||
private List<GatewayEventFilter> gatewayEventFilters;
|
||||
|
||||
private final Predicate<String> 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<T, ID> repository,
|
||||
@Nullable Predicate<String> 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<String> 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<T, ID> 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<T, ID> 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<T, ID> 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<T, ID> 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<T, ID> 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<T, ID> 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<T, ID> 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<T, ID> 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<T, ID> 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<T, ID> 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<T, ID> withQueueEventFilters(List<GatewayEventFilter> 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<T, ID> withQueueEventSubstitutionFilter(
|
||||
@Nullable GatewayEventSubstitutionFilter<ID, T> 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<T, ID> 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<T, ID> 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<T, ID> withQueueOrderPolicy(
|
||||
@Nullable GatewaySender.OrderPolicy orderPolicy) {
|
||||
|
||||
this.orderPolicy = orderPolicy;
|
||||
|
||||
return this;
|
||||
}
|
||||
}
|
||||
@@ -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<String> 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<String> 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);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user