From 800a0ee4340b65e222197a6e9f0cb56f48938b10 Mon Sep 17 00:00:00 2001 From: John Blum Date: Wed, 19 Aug 2020 19:36:14 -0700 Subject: [PATCH] Add Integration Tests asserting the configuration and behavior of the AsyncInlineCachingRepositoryConfigurer using a custom AsyncEventErrorHandler and user-defined AsyncEventOperationRepositoryFunction. Resolves gh-58. --- ...syncEventErrorHandlerIntegrationTests.java | 272 ++++++++++++++++++ 1 file changed, 272 insertions(+) create mode 100644 spring-geode/src/test/java/org/springframework/geode/cache/AsyncInlineCachingRegionConfigurerWithUserDefinedAsyncEventOperationRepositoryFunctionAndCustomAsyncEventErrorHandlerIntegrationTests.java diff --git a/spring-geode/src/test/java/org/springframework/geode/cache/AsyncInlineCachingRegionConfigurerWithUserDefinedAsyncEventOperationRepositoryFunctionAndCustomAsyncEventErrorHandlerIntegrationTests.java b/spring-geode/src/test/java/org/springframework/geode/cache/AsyncInlineCachingRegionConfigurerWithUserDefinedAsyncEventOperationRepositoryFunctionAndCustomAsyncEventErrorHandlerIntegrationTests.java new file mode 100644 index 00000000..af739589 --- /dev/null +++ b/spring-geode/src/test/java/org/springframework/geode/cache/AsyncInlineCachingRegionConfigurerWithUserDefinedAsyncEventOperationRepositoryFunctionAndCustomAsyncEventErrorHandlerIntegrationTests.java @@ -0,0 +1,272 @@ +/* + * 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.anyLong; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.isA; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; +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.util.Collections; +import java.util.Set; +import java.util.concurrent.atomic.AtomicReference; + +import javax.annotation.Resource; + +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.apache.geode.cache.Cache; +import org.apache.geode.cache.Operation; +import org.apache.geode.cache.Region; +import org.apache.geode.cache.RegionAttributes; +import org.apache.geode.cache.asyncqueue.AsyncEvent; +import org.apache.geode.cache.asyncqueue.AsyncEventListener; +import org.apache.geode.cache.asyncqueue.AsyncEventQueue; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.context.annotation.Bean; +import org.springframework.data.annotation.Id; +import org.springframework.data.gemfire.config.annotation.EnableEntityDefinedRegions; +import org.springframework.data.gemfire.config.annotation.PeerCacheApplication; +import org.springframework.data.gemfire.tests.integration.IntegrationTestsSupport; +import org.springframework.data.gemfire.tests.mock.annotation.EnableGemFireMockObjects; +import org.springframework.data.repository.CrudRepository; +import org.springframework.geode.cache.RepositoryAsyncEventListener.AbstractAsyncEventOperationRepositoryFunction; +import org.springframework.geode.cache.RepositoryAsyncEventListener.AsyncEventError; +import org.springframework.geode.cache.RepositoryAsyncEventListener.AsyncEventErrorHandler; +import org.springframework.geode.cache.RepositoryAsyncEventListener.AsyncEventOperationRepositoryFunction; +import org.springframework.lang.NonNull; +import org.springframework.lang.Nullable; +import org.springframework.test.context.junit4.SpringRunner; + +import lombok.AllArgsConstructor; +import lombok.EqualsAndHashCode; +import lombok.Getter; +import lombok.ToString; + +/** + * Integration Tests for {@link AsyncInlineCachingRegionConfigurer} and {@link RepositoryAsyncEventListener} + * with custom {@link AsyncEventErrorHandler} and user-defined {@link AsyncEventOperationRepositoryFunction}. + * + * @author John Blum + * @see org.junit.Test + * @see org.mockito.Mockito + * @see org.apache.geode.cache.Cache + * @see org.apache.geode.cache.Operation + * @see org.apache.geode.cache.Region + * @see org.apache.geode.cache.asyncqueue.AsyncEvent + * @see org.apache.geode.cache.asyncqueue.AsyncEventListener + * @see org.apache.geode.cache.asyncqueue.AsyncEventQueue + * @see org.springframework.data.gemfire.config.annotation.PeerCacheApplication + * @see org.springframework.data.gemfire.tests.mock.annotation.EnableGemFireMockObjects + * @see org.springframework.data.gemfire.tests.integration.IntegrationTestsSupport + * @see org.springframework.data.repository.CrudRepository + * @see org.springframework.geode.cache.AsyncInlineCachingRegionConfigurer + * @see org.springframework.geode.cache.RepositoryAsyncEventListener + * @see org.springframework.geode.cache.RepositoryAsyncEventListener.AbstractAsyncEventOperationRepositoryFunction + * @see org.springframework.geode.cache.RepositoryAsyncEventListener.AsyncEventErrorHandler + * @see org.springframework.geode.cache.RepositoryAsyncEventListener.AsyncEventOperationRepositoryFunction + * @since 1.4.0 + */ +@RunWith(SpringRunner.class) +@SuppressWarnings({ "unused" }) +public class AsyncInlineCachingRegionConfigurerWithUserDefinedAsyncEventOperationRepositoryFunctionAndCustomAsyncEventErrorHandlerIntegrationTests + extends IntegrationTestsSupport { + + private final AtomicReference asyncEventErrorReference = new AtomicReference<>(null); + + @Autowired + private AsyncEventErrorHandler errorHandler; + + @Autowired + private Cache peerCache; + + @Autowired + @Qualifier("processRepository") + private CrudRepository processRepository; + + @Resource(name = "Processes") + private Region processes; + + @Before + @SuppressWarnings("unchecked") + public void forwardRegionOperationsToAsyncEventQueueListener() { + + assertThat(this.errorHandler).isNotNull(); + assertThat(this.peerCache).isNotNull(); + assertThat(this.processes).isNotNull(); + assertThat(this.processes.getName()).isEqualTo("Processes"); + assertThat(this.processRepository).isNotNull(); + + RegionAttributes processesAttributes = this.processes.getAttributes(); + + assertThat(processesAttributes).isNotNull(); + + Set asyncEventQueueIds = processesAttributes.getAsyncEventQueueIds(); + + assertThat(asyncEventQueueIds).isNotNull(); + assertThat(asyncEventQueueIds).hasSize(1); + + String asyncEventQueueId = asyncEventQueueIds.iterator().next(); + + assertThat(asyncEventQueueId).isNotBlank(); + + AsyncEventQueue asyncEventQueue = this.peerCache.getAsyncEventQueue(asyncEventQueueId); + + assertThat(asyncEventQueue).isNotNull(); + assertThat(asyncEventQueue.getId()).isEqualTo(asyncEventQueueId); + + AsyncEventListener listener = asyncEventQueue.getAsyncEventListener(); + + assertThat(listener).isInstanceOf(RepositoryAsyncEventListener.class); + assertThat(((RepositoryAsyncEventListener) listener).getAsyncEventErrorHandler()) + .isEqualTo(this.errorHandler); + assertThat(((RepositoryAsyncEventListener) listener).getRepository()) + .isEqualTo(this.processRepository); + + doAnswer(invocation -> { + + Long key = invocation.getArgument(0); + + Process process = this.processes.get(key); + + AsyncEvent mockEvent = mock(AsyncEvent.class); + + doReturn(key).when(mockEvent).getKey(); + doReturn(process).when(mockEvent).getDeserializedValue(); + doReturn(Operation.CONTAINS_KEY).when(mockEvent).getOperation(); + doReturn(this.processes).when(mockEvent).getRegion(); + + listener.processEvents(Collections.singletonList(mockEvent)); + + return process != null; + + }).when(this.processes).containsKey(anyLong()); + + doAnswer(invocation -> { + + AsyncEventError eventError = invocation.getArgument(0); + + assertThat(eventError).isNotNull(); + assertThat(eventError.getCause()).isInstanceOf(IllegalStateException.class); + assertThat(eventError.getCause().getMessage()).isEqualTo("The entity (deserialized value) was null"); + assertThat(eventError.getCause()).hasNoCause(); + assertThat(eventError.getEvent()).isNotNull(); + + this.asyncEventErrorReference.set(eventError); + + return false; + + }).when(this.errorHandler).apply(isA(AsyncEventError.class)); + } + + @Test + public void asyncEventQueueEventsProcessedByListener() { + + Process process = Process.newProcess(21L, "Test Process"); + + assertThat(this.processes.containsKey(process.getId())).isFalse(); + assertThat(this.asyncEventErrorReference.get()).isNotNull(); + + verify(this.errorHandler, times(1)) + .apply(eq(this.asyncEventErrorReference.getAndUpdate(value -> null))); + verifyNoInteractions(this.processRepository); + + assertThat(this.processes.put(process.getId(), process)).isNull(); + assertThat(this.processes.containsKey(process.getId())).isTrue(); + assertThat(this.asyncEventErrorReference.get()).isNull(); + + verify(this.processRepository, times(1)).existsById(eq(process.getId())); + verifyNoMoreInteractions(this.errorHandler, this.processRepository); + } + + @PeerCacheApplication + @EnableGemFireMockObjects + @EnableEntityDefinedRegions + static class GeodeConfiguration { + + @Bean + AsyncEventErrorHandler mockAsyncEventErrorHandler() { + return mock(AsyncEventErrorHandler.class); + } + + @Bean + @SuppressWarnings("unchecked") + AsyncInlineCachingRegionConfigurer asyncInlineCachingProcessesRegionConfigurer( + AsyncEventErrorHandler asyncEventErrorHandler, + @Qualifier("processRepository") CrudRepository processRepository + ) { + + return AsyncInlineCachingRegionConfigurer.create(processRepository, "Processes") + .withAsyncEventErrorHandler(asyncEventErrorHandler) + .applyToListener(listener -> { + + if (listener instanceof RepositoryAsyncEventListener) { + + RepositoryAsyncEventListener repositoryListener = + (RepositoryAsyncEventListener) listener; + + repositoryListener.register( + new AbstractAsyncEventOperationRepositoryFunction(repositoryListener) { + + @Override + public boolean canProcess(@Nullable AsyncEvent event) { + return event != null && Operation.CONTAINS_KEY.equals(event.getOperation()); + } + + @Override + protected Object doRepositoryOp(@NonNull Process process) { + return getRepository().existsById(process.getId()); + } + }); + } + + return listener; + }); + } + + @Bean + @SuppressWarnings("unchecked") + CrudRepository processRepository() { + return mock(CrudRepository.class); + } + } + + @Getter + @EqualsAndHashCode + @ToString(of = "name") + @AllArgsConstructor(staticName = "newProcess") + @org.springframework.data.gemfire.mapping.annotation.Region("Processes") + static class Process { + + @Id + private Long id; + + @NonNull + private String name; + + } +}