FINERACT-1694: Extending MessageV1 and BulkMessageV1 with businessDate + fixing bulk event handling for publication + mapstruct mapper fixes to be more strict

This commit is contained in:
Arnold Galovics
2022-10-21 11:43:27 +02:00
committed by Arnold Gálovics
parent d1f3ffe490
commit 81d9e4a961
20 changed files with 172 additions and 116 deletions
@@ -1,42 +0,0 @@
{
"name": "BulkMessageV1",
"namespace": "org.apache.fineract.avro",
"type": "record",
"fields": [
{
"name": "id",
"doc": "The ID of the message to be sent",
"type": "int"
},
{
"name": "source",
"doc": "A unique identifier of the source service",
"type": "string"
},
{
"name": "type",
"doc": "The type of event the payload refers to. For example LoanApprovedBusinessEvent",
"type": "string"
},
{
"name": "createdAt",
"doc": "The UTC time of when the event has been raised; in ISO_LOCAL_DATE_TIME format. For example 2011-12-03T10:15:30",
"type": "string"
},
{
"name": "tenantId",
"doc": "The tenantId that the event has been sent from. For example default",
"type": "string"
},
{
"name": "idempotencyKey",
"doc": "The idempotency key for this particular event for consumer de-duplication",
"type": "string"
},
{
"name": "data",
"doc": "The payload for this bulk message",
"type": "org.apache.fineract.avro.BulkMessagePayloadV1"
}
]
}
@@ -28,6 +28,11 @@
"doc": "The UTC time of when the event has been raised; in ISO_LOCAL_DATE_TIME format. For example 2011-12-03T10:15:30",
"type": "string"
},
{
"name": "businessDate",
"doc": "The business date when the event has been raised; in ISO_LOCAL_DATE format. For example 2011-12-03",
"type": "string"
},
{
"name": "tenantId",
"doc": "The tenantId that the event has been sent from. For example default",
@@ -21,11 +21,12 @@ package org.apache.fineract.cob.service;
import java.util.List;
import org.apache.fineract.cob.data.BusinessStep;
import org.apache.fineract.cob.domain.BatchBusinessStep;
import org.apache.fineract.infrastructure.core.config.MapstructMapperConfig;
import org.mapstruct.Mapper;
import org.mapstruct.Mapping;
import org.mapstruct.Mappings;
@Mapper(componentModel = "spring")
@Mapper(config = MapstructMapperConfig.class)
public interface BusinessStepMapper {
@Mappings({ @Mapping(target = "order", source = "source.stepOrder") })
@@ -21,14 +21,15 @@ package org.apache.fineract.infrastructure.businessdate.mapper;
import java.util.List;
import org.apache.fineract.infrastructure.businessdate.data.BusinessDateData;
import org.apache.fineract.infrastructure.businessdate.domain.BusinessDate;
import org.apache.fineract.infrastructure.core.config.MapstructMapperConfig;
import org.mapstruct.Mapper;
import org.mapstruct.Mapping;
import org.mapstruct.Mappings;
@Mapper(componentModel = "spring")
@Mapper(config = MapstructMapperConfig.class)
public interface BusinessDateMapper {
@Mappings({ @Mapping(target = "type", source = "source.type"), @Mapping(target = "date", source = "source.date") })
@Mappings({ @Mapping(target = "description", source = "source.type.description") })
BusinessDateData map(BusinessDate source);
List<BusinessDateData> map(List<BusinessDate> sources);
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
package org.apache.fineract.infrastructure.event.external.service.support;
package org.apache.fineract.infrastructure.core.config;
import org.mapstruct.Builder;
import org.mapstruct.MapperConfig;
@@ -24,4 +24,4 @@ import org.mapstruct.MappingConstants;
import org.mapstruct.ReportingPolicy;
@MapperConfig(componentModel = MappingConstants.ComponentModel.SPRING, unmappedTargetPolicy = ReportingPolicy.ERROR, builder = @Builder(disableBuilder = true))
public class ExternalEventMapperConfig {}
public class MapstructMapperConfig {}
@@ -23,7 +23,7 @@ import java.util.List;
public class BulkBusinessEvent extends AbstractBusinessEvent<List<BusinessEvent<?>>> {
private static final String CATEGORY = "Bulk";
private static final String TYPE = "BulkBusinessEvent";
public static final String TYPE = "BulkBusinessEvent";
public BulkBusinessEvent(List<BusinessEvent<?>> value) {
super(value);
@@ -139,8 +139,17 @@ public class BusinessEventNotifierServiceImpl implements BusinessEventNotifierSe
try {
List<BusinessEvent<?>> recordedBusinessEvents = recordedEvents.get();
if (isExternalEventPostingEnabled()) {
log.debug("Posting the BulkBusinessEvent for the recorded {} events", recordedBusinessEvents.size());
externalEventService.postEvent(new BulkBusinessEvent(recordedBusinessEvents));
if (recordedBusinessEvents.isEmpty()) {
log.debug("Not posting a BulkBusinessEvent since there were no events recorded");
} else {
if (recordedBusinessEvents.size() == 1) {
log.debug("Posting a singular event instead of a BulkBusinessEvent since there was only a single event recorded");
externalEventService.postEvent(recordedBusinessEvents.get(0));
} else {
log.debug("Posting the BulkBusinessEvent for the recorded {} events", recordedBusinessEvents.size());
externalEventService.postEvent(new BulkBusinessEvent(recordedBusinessEvents));
}
}
}
} finally {
recordedEvents.remove();
@@ -70,14 +70,14 @@ public class SendAsynchronousEventsTasklet implements Tasklet {
private List<ExternalEvent> getQueuedEventsBatch() {
int readBatchSize = getBatchSize();
Pageable batchSize = PageRequest.ofSize(readBatchSize);
List<ExternalEvent> events = repository.findByStatusOrderById(ExternalEventStatus.TO_BE_SENT, batchSize);
return events;
return repository.findByStatusOrderById(ExternalEventStatus.TO_BE_SENT, batchSize);
}
private void processEvents(List<ExternalEvent> queuedEvents) throws IOException {
for (ExternalEvent event : queuedEvents) {
MessageV1 message = messageFactory.createMessage(event);
eventProducer.sendEvent(byteBufferConverter.convert(message.toByteBuffer()));
byte[] byteMessage = byteBufferConverter.convert(message.toByteBuffer());
eventProducer.sendEvent(byteMessage);
event.setStatus(ExternalEventStatus.SENT);
event.setSentAt(DateUtils.getOffsetDateTimeOfTenant());
repository.save(event);
@@ -19,12 +19,12 @@
package org.apache.fineract.infrastructure.event.external.service;
import java.util.List;
import org.apache.fineract.infrastructure.core.config.MapstructMapperConfig;
import org.apache.fineract.infrastructure.event.external.data.ExternalEventConfigurationItemData;
import org.apache.fineract.infrastructure.event.external.repository.domain.ExternalEventConfiguration;
import org.apache.fineract.infrastructure.event.external.service.support.ExternalEventMapperConfig;
import org.mapstruct.Mapper;
@Mapper(config = ExternalEventMapperConfig.class)
@Mapper(config = MapstructMapperConfig.class)
public interface ExternalEventsConfigurationMapper {
List<ExternalEventConfigurationItemData> map(List<ExternalEventConfiguration> source);
@@ -18,19 +18,19 @@
*/
package org.apache.fineract.infrastructure.event.external.service.message;
import java.time.LocalDate;
import java.time.OffsetDateTime;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;
import java.util.UUID;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.fineract.avro.BulkMessageV1;
import org.apache.fineract.avro.MessageV1;
import org.apache.fineract.infrastructure.core.service.DateUtils;
import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
import org.apache.fineract.infrastructure.event.external.repository.domain.ExternalEvent;
import org.apache.fineract.infrastructure.event.external.service.message.domain.BulkMessageData;
import org.apache.fineract.infrastructure.event.external.service.message.domain.MessageBusinessDate;
import org.apache.fineract.infrastructure.event.external.service.message.domain.MessageCategory;
import org.apache.fineract.infrastructure.event.external.service.message.domain.MessageCreatedAt;
import org.apache.fineract.infrastructure.event.external.service.message.domain.MessageData;
import org.apache.fineract.infrastructure.event.external.service.message.domain.MessageDataSchema;
import org.apache.fineract.infrastructure.event.external.service.message.domain.MessageId;
@@ -50,13 +50,15 @@ public class MessageFactory implements InitializingBean {
private static final String SOURCE_UUID = UUID.randomUUID().toString();
public MessageV1 createMessage(MessageId id, MessageSource source, MessageType type, MessageCategory category,
MessageIdempotencyKey idempotencyKey, MessageDataSchema dataSchema, MessageData data) {
MessageCreatedAt createdAt, MessageBusinessDate businessDate, MessageIdempotencyKey idempotencyKey,
MessageDataSchema dataSchema, MessageData data) {
MessageV1 result = new MessageV1();
result.setId(id.getId());
result.setSource(source.getSource());
result.setType(type.getType());
result.setCategory(category.getCategory());
result.setCreatedAt(getMessageCreatedAt());
result.setCreatedAt(getMessageCreatedAt(createdAt.getCreatedAt()));
result.setBusinessDate(getMessageBusinessDate(businessDate.getBusinessDate()));
result.setTenantId(getTenantId());
result.setIdempotencyKey(idempotencyKey.getIdempotencyKey());
result.setDataschema(dataSchema.getDataSchema());
@@ -64,39 +66,31 @@ public class MessageFactory implements InitializingBean {
return result;
}
public BulkMessageV1 createBulkMessage(MessageId id, MessageSource source, MessageType type, BulkMessageData data) {
BulkMessageV1 result = new BulkMessageV1();
result.setId(id.getId());
result.setSource(source.getSource());
result.setType(type.getType());
result.setCreatedAt(getMessageCreatedAt());
result.setTenantId(getTenantId());
result.setData(data.getData());
return result;
}
public MessageV1 createMessage(ExternalEvent event) {
MessageId id = new MessageId(event.getId().intValue());
MessageSource source = new MessageSource(SOURCE_UUID);
MessageType type = new MessageType(event.getType());
MessageCategory category = new MessageCategory(event.getCategory());
MessageCreatedAt createdAt = new MessageCreatedAt(event.getCreatedAt());
MessageBusinessDate businessDate = new MessageBusinessDate(event.getBusinessDate());
MessageIdempotencyKey idempotencyKey = new MessageIdempotencyKey(event.getIdempotencyKey());
MessageDataSchema dataSchema = new MessageDataSchema(event.getSchema());
MessageData data = new MessageData(byteBufferConverter.convert(event.getData()));
MessageV1 message = createMessage(id, source, type, category, idempotencyKey, dataSchema, data);
return message;
return createMessage(id, source, type, category, createdAt, businessDate, idempotencyKey, dataSchema, data);
}
private String getTenantId() {
return ThreadLocalContextUtil.getTenant().getTenantIdentifier();
}
private String getMessageCreatedAt() {
OffsetDateTime createdAt = DateUtils.getOffsetDateTimeOfTenant();
private String getMessageCreatedAt(OffsetDateTime createdAt) {
return createdAt.withOffsetSameInstant(ZoneOffset.UTC).toLocalDateTime().format(DateTimeFormatter.ISO_LOCAL_DATE_TIME);
}
private String getMessageBusinessDate(LocalDate businessDate) {
return businessDate.format(DateTimeFormatter.ISO_LOCAL_DATE);
}
@Override
public void afterPropertiesSet() throws Exception {
log.info("Message source set to {}", SOURCE_UUID);
@@ -0,0 +1,33 @@
/**
* 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.service.message.domain;
import java.time.LocalDate;
import java.util.Objects;
import lombok.Getter;
@Getter
public class MessageBusinessDate {
private LocalDate businessDate;
public MessageBusinessDate(LocalDate businessDate) {
this.businessDate = Objects.requireNonNull(businessDate, "businessDate cannot be null");
}
}
@@ -0,0 +1,33 @@
/**
* 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.service.message.domain;
import java.time.OffsetDateTime;
import java.util.Objects;
import lombok.Getter;
@Getter
public class MessageCreatedAt {
private OffsetDateTime createdAt;
public MessageCreatedAt(OffsetDateTime createdAt) {
this.createdAt = Objects.requireNonNull(createdAt, "createdAt cannot be null");
}
}
@@ -19,16 +19,14 @@
package org.apache.fineract.portfolio.delinquency.mapper;
import java.util.List;
import org.apache.fineract.infrastructure.core.config.MapstructMapperConfig;
import org.apache.fineract.portfolio.delinquency.data.DelinquencyBucketData;
import org.apache.fineract.portfolio.delinquency.domain.DelinquencyBucket;
import org.mapstruct.Mapper;
import org.mapstruct.Mapping;
import org.mapstruct.Mappings;
@Mapper(componentModel = "spring")
@Mapper(config = MapstructMapperConfig.class)
public interface DelinquencyBucketMapper {
@Mappings({ @Mapping(target = "id", source = "source.id"), @Mapping(target = "name", source = "source.name") })
DelinquencyBucketData map(DelinquencyBucket source);
List<DelinquencyBucketData> map(List<DelinquencyBucket> sources);
@@ -19,18 +19,14 @@
package org.apache.fineract.portfolio.delinquency.mapper;
import java.util.List;
import org.apache.fineract.infrastructure.core.config.MapstructMapperConfig;
import org.apache.fineract.portfolio.delinquency.data.DelinquencyRangeData;
import org.apache.fineract.portfolio.delinquency.domain.DelinquencyRange;
import org.mapstruct.Mapper;
import org.mapstruct.Mapping;
import org.mapstruct.Mappings;
@Mapper(componentModel = "spring")
@Mapper(config = MapstructMapperConfig.class)
public interface DelinquencyRangeMapper {
@Mappings({ @Mapping(target = "id", source = "source.id"), @Mapping(target = "classification", source = "source.classification"),
@Mapping(target = "minimumAgeDays", source = "source.minimumAgeDays"),
@Mapping(target = "maximumAgeDays", source = "source.maximumAgeDays") })
DelinquencyRangeData map(DelinquencyRange source);
List<DelinquencyRangeData> map(List<DelinquencyRange> sources);
@@ -19,19 +19,16 @@
package org.apache.fineract.portfolio.delinquency.mapper;
import java.util.List;
import org.apache.fineract.infrastructure.core.config.MapstructMapperConfig;
import org.apache.fineract.portfolio.delinquency.data.LoanDelinquencyTagHistoryData;
import org.apache.fineract.portfolio.delinquency.domain.LoanDelinquencyTagHistory;
import org.mapstruct.Mapper;
import org.mapstruct.Mapping;
@Mapper(componentModel = "spring", uses = DelinquencyRangeMapper.class)
@Mapper(config = MapstructMapperConfig.class, uses = DelinquencyRangeMapper.class)
public interface LoanDelinquencyTagMapper {
@Mapping(target = "id", source = "source.id")
@Mapping(target = "loanId", source = "source.loan.id")
@Mapping(target = "delinquencyRange", source = "source.delinquencyRange")
@Mapping(target = "addedOnDate", source = "source.addedOnDate")
@Mapping(target = "liftedOnDate", source = "source.liftedOnDate")
LoanDelinquencyTagHistoryData map(LoanDelinquencyTagHistory source);
List<LoanDelinquencyTagHistoryData> map(List<LoanDelinquencyTagHistory> sources);
@@ -19,17 +19,17 @@
package org.apache.fineract.portfolio.loanaccount.mapper;
import java.util.List;
import org.apache.fineract.infrastructure.core.config.MapstructMapperConfig;
import org.apache.fineract.portfolio.loanaccount.data.LoanTransactionRelationData;
import org.apache.fineract.portfolio.loanaccount.domain.LoanTransactionRelation;
import org.mapstruct.Mapper;
import org.mapstruct.Mapping;
@Mapper(componentModel = "spring")
@Mapper(config = MapstructMapperConfig.class)
public interface LoanTransactionRelationMapper {
@Mapping(target = "fromLoanTransaction", source = "source.fromTransaction.id")
@Mapping(target = "toLoanTransaction", source = "source.toTransaction.id")
@Mapping(target = "relationType", source = "source.relationType")
@Mapping(target = "amount", source = "source.toTransaction.amount")
@Mapping(target = "paymentType", source = "source.toTransaction.paymentDetail.paymentType.name")
LoanTransactionRelationData map(LoanTransactionRelation source);
@@ -19,20 +19,14 @@
package org.apache.fineract.portfolio.paymenttype.mapper;
import java.util.List;
import org.apache.fineract.infrastructure.core.config.MapstructMapperConfig;
import org.apache.fineract.portfolio.paymenttype.data.PaymentTypeData;
import org.apache.fineract.portfolio.paymenttype.domain.PaymentType;
import org.mapstruct.Mapper;
import org.mapstruct.Mapping;
@Mapper(componentModel = "spring")
@Mapper(config = MapstructMapperConfig.class)
public interface PaymentTypeMapper {
@Mapping(target = "id", source = "source.id")
@Mapping(target = "name", source = "source.name")
@Mapping(target = "description", source = "source.description")
@Mapping(target = "isCashPayment", source = "source.isCashPayment")
@Mapping(target = "codeName", source = "source.codeName")
@Mapping(target = "isSystemDefined", source = "source.isSystemDefined")
PaymentTypeData map(PaymentType source);
List<PaymentTypeData> map(List<PaymentType> sources);
@@ -109,7 +109,7 @@ public class BusinessDateValidatorTest {
}
@Test
public void validateAndParseUpdateWithRighDate() {
public void validateAndParseUpdateWithRightDate() {
String json = "{\"type\":\"COB_DATE\", \"locale\":\"hu\",\"dateFormat\":\"yyyy-MM-dd\",\"date\":\"2022-06-11\"}";
JsonCommand command = JsonCommand.from(json);
BusinessDateData result = businessDateDataParserAndValidator.validateAndParseUpdate(command);
@@ -19,8 +19,10 @@
package org.apache.fineract.infrastructure.event.business.service;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.BDDMockito.given;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
@@ -93,7 +95,20 @@ class BusinessEventNotifierServiceImplTest {
}
@Test
public void testNotifyPostBusinessEventShouldNotifyPostListenersAndPostAnBulkExternalEventWhenRecordingEnabled() {
public void testNotifyPostBusinessEventShouldNotPostAnythingWhenNoEventWasRaisedExternalEventWhenRecordingEnabled() {
// given
configureExternalEventsProperties(true);
when(externalEventConfigurationRepository.findExternalEventConfigurationByTypeWithNotFoundDetection(Mockito.any()))
.thenReturn(new ExternalEventConfiguration("aType", true));
underTest.startExternalEventRecording();
// when
underTest.stopExternalEventRecording();
// then
verify(externalEventService, never()).postEvent(any());
}
@Test
public void testNotifyPostBusinessEventShouldNotifyPostListenersAndPostARegularExternalEventWhenRecordingEnabled() {
// given
configureExternalEventsProperties(true);
when(externalEventConfigurationRepository.findExternalEventConfigurationByTypeWithNotFoundDetection(Mockito.any()))
@@ -107,12 +122,34 @@ class BusinessEventNotifierServiceImplTest {
underTest.stopExternalEventRecording();
// then
verify(postListener).onBusinessEvent(event);
verify(externalEventService).postEvent(event);
}
@Test
public void testNotifyPostBusinessEventShouldNotifyPostListenersAndPostAnBulkExternalEventWhenRecordingEnabled() {
// given
configureExternalEventsProperties(true);
when(externalEventConfigurationRepository.findExternalEventConfigurationByTypeWithNotFoundDetection(Mockito.any()))
.thenReturn(new ExternalEventConfiguration("aType", true));
MockBusinessEvent event = new MockBusinessEvent();
MockBusinessEvent event2 = new MockBusinessEvent();
BusinessEventListener<MockBusinessEvent> postListener = mockListener();
underTest.addPostBusinessEventListener(MockBusinessEvent.class, postListener);
underTest.startExternalEventRecording();
underTest.notifyPostBusinessEvent(event);
underTest.notifyPostBusinessEvent(event2);
// when
underTest.stopExternalEventRecording();
// then
verify(postListener).onBusinessEvent(event);
verify(postListener).onBusinessEvent(event2);
ArgumentCaptor<BulkBusinessEvent> argumentCaptor = ArgumentCaptor.forClass(BulkBusinessEvent.class);
verify(externalEventService).postEvent(argumentCaptor.capture());
BulkBusinessEvent capturedEvent = argumentCaptor.getValue();
assertThat(capturedEvent.get()).hasSize(1);
assertThat(capturedEvent.get()).hasSize(2);
assertThat(capturedEvent.get().get(0)).isEqualTo(event);
assertThat(capturedEvent.get().get(1)).isEqualTo(event2);
}
@Test
@@ -102,11 +102,11 @@ class SendAsynchronousEventsTaskletTest {
@Test
public void givenBatchSize2WhenTaskExecutionThenSend2Events() throws Exception {
// given
List<ExternalEvent> events = Arrays.asList(new ExternalEvent("aType", "aCategory", "aSchema", new byte[0], "aIdemtpotencyKey"),
new ExternalEvent("aType", "aCategory", "aSchema", new byte[0], "aIdemtpotencyKey"));
List<ExternalEvent> events = Arrays.asList(new ExternalEvent("aType", "aCategory", "aSchema", new byte[0], "aIdempotencyKey"),
new ExternalEvent("aType", "aCategory", "aSchema", new byte[0], "aIdempotencyKey"));
// Dummy Message
MessageV1 dummyMessage = new MessageV1(1, "aSource", "aType", "nocategory", "aCreateDate", "aTennantId", "anidempotencyKey",
"aSchema", Mockito.mock(ByteBuffer.class));
MessageV1 dummyMessage = new MessageV1(1, "aSource", "aType", "nocategory", "aCreateDate", "aBusinessDate", "aTenantId",
"anidempotencyKey", "aSchema", Mockito.mock(ByteBuffer.class));
when(repository.findByStatusOrderById(Mockito.any(), Mockito.any())).thenReturn(events);
when(messageFactory.createMessage(Mockito.any())).thenReturn(dummyMessage);
@@ -123,10 +123,10 @@ class SendAsynchronousEventsTaskletTest {
@Test
public void givenBatchSize2WhenEventSendFailsThenExecutionStops() throws Exception {
// given
List<ExternalEvent> events = Arrays.asList(new ExternalEvent("aType", "aCategory", "aSchema", new byte[0], "aIdemtpotencyKey"),
new ExternalEvent("aType", "aCategory", "aSchema", new byte[0], "aIdemtpotencyKey"));
MessageV1 dummyMessage = new MessageV1(1, "aSource", "aType", "nocategory", "aCreateDate", "aTennantId", "anidempotencyKey",
"aSchema", Mockito.mock(ByteBuffer.class));
List<ExternalEvent> events = Arrays.asList(new ExternalEvent("aType", "aCategory", "aSchema", new byte[0], "aIdempotencyKey"),
new ExternalEvent("aType", "aCategory", "aSchema", new byte[0], "aIdempotencyKey"));
MessageV1 dummyMessage = new MessageV1(1, "aSource", "aType", "nocategory", "aCreateDate", "aBusinessDate", "aTenantId",
"anidempotencyKey", "aSchema", Mockito.mock(ByteBuffer.class));
when(repository.findByStatusOrderById(Mockito.any(), Mockito.any())).thenReturn(events);
when(messageFactory.createMessage(Mockito.any())).thenReturn(dummyMessage);
when(byteBufferConverter.convert(Mockito.any(ByteBuffer.class))).thenReturn(new byte[0]);
@@ -143,9 +143,9 @@ class SendAsynchronousEventsTaskletTest {
public void givenOneEventWhenEventSentThenEventStatusUpdates() throws Exception {
// given
ArgumentCaptor<ExternalEvent> externalEventArgumentCaptor = ArgumentCaptor.forClass(ExternalEvent.class);
List<ExternalEvent> events = Arrays.asList(new ExternalEvent("aType", "aCategory", "aSchema", new byte[0], "aIdemtpotencyKey"));
MessageV1 dummyMessage = new MessageV1(1, "aSource", "aType", "nocategory", "aCreateDate", "aTennantId", "anidempotencyKey",
"aSchema", Mockito.mock(ByteBuffer.class));
List<ExternalEvent> events = Arrays.asList(new ExternalEvent("aType", "aCategory", "aSchema", new byte[0], "aIdempotencyKey"));
MessageV1 dummyMessage = new MessageV1(1, "aSource", "aType", "nocategory", "aCreateDate", "aBusinessDate", "aTenantId",
"anidempotencyKey", "aSchema", Mockito.mock(ByteBuffer.class));
when(repository.findByStatusOrderById(Mockito.any(), Mockito.any())).thenReturn(events);
when(messageFactory.createMessage(Mockito.any())).thenReturn(dummyMessage);
when(byteBufferConverter.convert(Mockito.any(ByteBuffer.class))).thenReturn(new byte[0]);
@@ -153,10 +153,10 @@ class SendAsynchronousEventsTaskletTest {
// when
resultStatus = this.underTest.execute(stepContribution, chunkContext);
// then
verify(messageFactory).createMessage(Mockito.any());
verify(repository).save(externalEventArgumentCaptor.capture());
ExternalEvent externalEvent = externalEventArgumentCaptor.getValue();
assertThat(externalEvent.getStatus()).isEqualTo(ExternalEventStatus.SENT);
assertEquals(RepeatStatus.FINISHED, resultStatus);
}
}