FINERACT-1694 Send Asynchronous Events

fineract-provider/src/test/java/org/apache/fineract/infrastructure/event/external/jobs/SendAsynchronousEventsTaskletTest.java

@@ -0,0 +1,144 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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
+ *
+ * http://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.apache.fineract.infrastructure.event.external.jobs;
+
+import static com.google.common.truth.Truth.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.time.LocalDate;
+import java.time.ZoneId;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.fineract.avro.MessageV1;
+import org.apache.fineract.infrastructure.businessdate.domain.BusinessDateType;
+import org.apache.fineract.infrastructure.core.config.FineractProperties;
+import org.apache.fineract.infrastructure.core.domain.FineractPlatformTenant;
+import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
+import org.apache.fineract.infrastructure.event.external.exception.AcknowledgementTimeoutException;
+import org.apache.fineract.infrastructure.event.external.producer.ExternalEventProducer;
+import org.apache.fineract.infrastructure.event.external.repository.ExternalEventRepository;
+import org.apache.fineract.infrastructure.event.external.repository.domain.ExternalEvent;
+import org.apache.fineract.infrastructure.event.external.repository.domain.ExternalEventStatus;
+import org.apache.fineract.infrastructure.event.external.service.message.MessageFactory;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.batch.core.StepContribution;
+import org.springframework.batch.core.scope.context.ChunkContext;
+import org.springframework.batch.repeat.RepeatStatus;
+
+@ExtendWith(MockitoExtension.class)
+class SendAsynchronousEventsTaskletTest {
+
+    @Mock
+    private FineractProperties fineractProperties;
+    @Mock
+    private ExternalEventRepository repository;
+    @Mock
+    private ExternalEventProducer eventProducer;
+    @Mock
+    private MessageFactory messageFactory;
+    @Mock
+    private StepContribution stepContribution;
+    @Mock
+    private ChunkContext chunkContext;
+    private SendAsynchronousEventsTasklet underTest;
+    private RepeatStatus resultStatus;
+
+    @BeforeEach
+    public void setUp() {
+        ThreadLocalContextUtil.setTenant(new FineractPlatformTenant(1L, "default", "Default", "Asia/Kolkata", null));
+        ThreadLocalContextUtil
+                .setBusinessDates(new HashMap<>(Map.of(BusinessDateType.BUSINESS_DATE, LocalDate.now(ZoneId.systemDefault()))));
+        configureExternalEventsProducerReadBatchSizeProperty(1000);
+        underTest = new SendAsynchronousEventsTasklet(fineractProperties, repository, eventProducer, messageFactory);
+    }
+
+    private void configureExternalEventsProducerReadBatchSizeProperty(int readBatchSize) {
+        FineractProperties.FineractEventsProperties eventsProperties = new FineractProperties.FineractEventsProperties();
+        FineractProperties.FineractExternalEventsProperties externalProperties = new FineractProperties.FineractExternalEventsProperties();
+        FineractProperties.FineractExternalEventsProducerProperties externalEventsProducerProperties = new FineractProperties.FineractExternalEventsProducerProperties();
+        externalProperties.setEnabled(true);
+        externalEventsProducerProperties.setReadBatchSize(readBatchSize);
+        externalProperties.setProducer(externalEventsProducerProperties);
+        eventsProperties.setExternal(externalProperties);
+        when(fineractProperties.getEvents()).thenReturn(eventsProperties);
+    }
+
+    @Test
+    public void givenBatchSize2WhenTaskExecutionThenSend2Events() throws Exception {
+        // given
+        List<ExternalEvent> events = Arrays.asList(new ExternalEvent("aType", "aSchema", new byte[0], "aIdemtpotencyKey"),
+                new ExternalEvent("aType", "aSchema", new byte[0], "aIdemtpotencyKey"));
+        when(repository.findByStatusOrderById(Mockito.any(), Mockito.any())).thenReturn(events);
+        when(messageFactory.createMessage(Mockito.any())).thenReturn(new MessageV1());
+        doNothing().when(eventProducer).sendEvent(Mockito.any(MessageV1.class));
+        // when
+        resultStatus = this.underTest.execute(stepContribution, chunkContext);
+        // then
+        verify(eventProducer, times(2)).sendEvent(Mockito.any(MessageV1.class));
+        verify(repository, times(2)).save(Mockito.any(ExternalEvent.class));
+        assertEquals(RepeatStatus.FINISHED, resultStatus);
+    }
+
+    @Test
+    public void givenBatchSize2WhenEventSendFailsThenExecutionStops() throws Exception {
+        // given
+        List<ExternalEvent> events = Arrays.asList(new ExternalEvent("aType", "aSchema", new byte[0], "aIdemtpotencyKey"),
+                new ExternalEvent("aType", "aSchema", new byte[0], "aIdemtpotencyKey"));
+        when(repository.findByStatusOrderById(Mockito.any(), Mockito.any())).thenReturn(events);
+        when(messageFactory.createMessage(Mockito.any())).thenReturn(new MessageV1());
+        doThrow(new AcknowledgementTimeoutException("Event Send Exception", new RuntimeException())).when(eventProducer)
+                .sendEvent(Mockito.any(MessageV1.class));
+        // when
+        resultStatus = this.underTest.execute(stepContribution, chunkContext);
+        // then
+        verify(repository, times(0)).save(Mockito.any(ExternalEvent.class));
+        assertEquals(RepeatStatus.FINISHED, resultStatus);
+    }
+
+    @Test
+    public void givenOneEventWhenEventSentThenEventStatusUpdates() throws Exception {
+        // given
+        ArgumentCaptor<ExternalEvent> externalEventArgumentCaptor = ArgumentCaptor.forClass(ExternalEvent.class);
+        List<ExternalEvent> events = Arrays.asList(new ExternalEvent("aType", "aSchema", new byte[0], "aIdemtpotencyKey"));
+        when(repository.findByStatusOrderById(Mockito.any(), Mockito.any())).thenReturn(events);
+        when(messageFactory.createMessage(Mockito.any())).thenReturn(new MessageV1());
+        doNothing().when(eventProducer).sendEvent(Mockito.any(MessageV1.class));
+        // when
+        resultStatus = this.underTest.execute(stepContribution, chunkContext);
+        // then
+        verify(repository).save(externalEventArgumentCaptor.capture());
+        ExternalEvent externalEvent = externalEventArgumentCaptor.getValue();
+        assertThat(externalEvent.getStatus()).isEqualTo(ExternalEventStatus.SENT);
+        assertEquals(RepeatStatus.FINISHED, resultStatus);
+    }
+
+}

fineract-provider/src/test/resources/application-test.properties

@@ -46,6 +46,9 @@ fineract.remote-job-message-handler.spring-events.enabled=${FINERACT_REMOTE_JOB_
 fineract.remote-job-message-handler.jms.enabled=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_JMS_ENABLED:false}
 fineract.remote-job-message-handler.jms.request-queue-name=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_JMS_QUEUE_NAME:JMS-request-queue}
 fineract.events.external.enabled=${FINERACT_EXTERNAL_EVENTS_ENABLED:false}
+fineract.events.external.producer.read-batch-size=${FINERACT_EXTERNAL_EVENTS_PRODUCER_READ_BATCH_SIZE:1000}
+fineract.events.external.producer.jms.enabled=${FINERACT_EXTERNAL_EVENTS_PRODUCER_JMS_ENABLED:false}
+fineract.events.external.producer.jms.event-queue-name=${FINERACT_EXTERNAL_EVENTS_PRODUCER_JMS_QUEUE_NAME:JMS-event-queue}
 
 management.health.jms.enabled=false