FINERACT-1694: Bulk event support

fineract-avro-schemas/src/main/avro/BulkMessageItemV1.avsc

@@ -0,0 +1,32 @@
+{
+    "name": "BulkMessageItemV1",
+    "namespace": "org.apache.fineract.avro",
+    "type": "record",
+    "fields": [
+        {
+            "name": "id",
+            "doc": "The ID of the message to be sent",
+            "type": "int"
+        },
+        {
+            "name": "type",
+            "doc": "The type of event the payload refers to. For example LoanApprovedBusinessEvent",
+            "type": "string"
+        },
+        {
+            "name": "category",
+            "doc": "The category of event the payload refers to. For example LOAN",
+            "type": "string"
+        },
+        {
+            "name": "dataschema",
+            "doc": "The fully qualified name of the schema of the event payload. For example org.apache.fineract.avro.loan.v1.LoanAccountDataV1",
+            "type": "string"
+        },
+        {
+            "name": "data",
+            "doc": "The payload data serialized into Avro bytes",
+            "type": "bytes"
+        }
+    ]
+}

fineract-avro-schemas/src/main/avro/BulkMessagePayloadV1.avsc

@@ -0,0 +1,13 @@
+{
+    "name": "BulkMessagePayloadV1",
+    "namespace": "org.apache.fineract.avro",
+    "type": "record",
+    "fields": [{
+        "name": "datas",
+        "doc": "The individual messages within this bulk message",
+        "type": {
+            "type": "array",
+            "items": "org.apache.fineract.avro.BulkMessageItemV1"
+        }
+    }]
+}

fineract-avro-schemas/src/main/avro/BulkMessageV1.avsc

@@ -0,0 +1,42 @@
+{
+    "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"
+        }
+    ]
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/service/ExternalEventService.java

@@ -19,15 +19,22 @@
 package org.apache.fineract.infrastructure.event.external.service;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
 import javax.persistence.EntityManager;
 import javax.persistence.PersistenceContext;
 import lombok.RequiredArgsConstructor;
+import org.apache.fineract.avro.BulkMessageItemV1;
+import org.apache.fineract.avro.BulkMessagePayloadV1;
+import org.apache.fineract.infrastructure.event.business.domain.BulkBusinessEvent;
 import org.apache.fineract.infrastructure.event.business.domain.BusinessEvent;
 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.service.idempotency.ExternalEventIdempotencyKeyGenerator;
-import org.apache.fineract.infrastructure.event.external.service.serialization.BusinessEventSerializerFactory;
+import org.apache.fineract.infrastructure.event.external.service.message.BulkMessageItemFactory;
 import org.apache.fineract.infrastructure.event.external.service.serialization.serializer.BusinessEventSerializer;
+import org.apache.fineract.infrastructure.event.external.service.serialization.serializer.BusinessEventSerializerFactory;
+import org.apache.fineract.infrastructure.event.external.service.support.ByteBufferConverter;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
@@ -39,6 +46,8 @@ public class ExternalEventService {
     private final ExternalEventRepository repository;
     private final ExternalEventIdempotencyKeyGenerator idempotencyKeyGenerator;
     private final BusinessEventSerializerFactory serializerFactory;
+    private final ByteBufferConverter byteBufferConverter;
+    private final BulkMessageItemFactory bulkMessageItemFactory;
 
     private EntityManager entityManager;
 
@@ -47,15 +56,14 @@ public class ExternalEventService {
             throw new IllegalArgumentException("event cannot be null");
         }
 
-        String eventType = event.getType();
-        String idempotencyKey = idempotencyKeyGenerator.generate(event);
         try {
-            BusinessEventSerializer serializer = serializerFactory.create(event);
-            String schema = serializer.getSupportedSchema().getName();
             flushChangesBeforeSerialization();
-            byte[] data = serializer.serialize(event);
-            ExternalEvent externalEvent = new ExternalEvent(eventType, schema, data, idempotencyKey);
-
+            ExternalEvent externalEvent;
+            if (event instanceof BulkBusinessEvent) {
+                externalEvent = handleBulkBusinessEvent((BulkBusinessEvent) event);
+            } else {
+                externalEvent = handleRegularBusinessEvent(event);
+            }
             repository.save(externalEvent);
         } catch (IOException e) {
             throw new RuntimeException("Error while serializing event " + event.getClass().getSimpleName(), e);
@@ -63,6 +71,32 @@ public class ExternalEventService {
 
     }
 
+    private ExternalEvent handleBulkBusinessEvent(BulkBusinessEvent bulkBusinessEvent) throws IOException {
+        List<BulkMessageItemV1> messages = new ArrayList<>();
+        List<BusinessEvent<?>> events = bulkBusinessEvent.get();
+        for (int i = 0; i < events.size(); i++) {
+            BusinessEvent<?> event = events.get(i);
+            int id = i + 1;
+            BulkMessageItemV1 message = bulkMessageItemFactory.createBulkMessageItem(id, event);
+            messages.add(message);
+        }
+        String idempotencyKey = idempotencyKeyGenerator.generate(bulkBusinessEvent);
+        BulkMessagePayloadV1 avroDto = new BulkMessagePayloadV1(messages);
+        byte[] data = byteBufferConverter.convert(avroDto.toByteBuffer());
+
+        return new ExternalEvent(bulkBusinessEvent.getType(), BulkMessagePayloadV1.class.getName(), data, idempotencyKey);
+    }
+
+    private <T> ExternalEvent handleRegularBusinessEvent(BusinessEvent<T> event) throws IOException {
+        String eventType = event.getType();
+        String idempotencyKey = idempotencyKeyGenerator.generate(event);
+        BusinessEventSerializer serializer = serializerFactory.create(event);
+        String schema = serializer.getSupportedSchema().getName();
+        byte[] data = serializer.serialize(event);
+
+        return new ExternalEvent(eventType, schema, data, idempotencyKey);
+    }
+
     private void flushChangesBeforeSerialization() {
         entityManager.flush();
     }

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/service/message/BulkMessageItemFactory.java

@@ -0,0 +1,47 @@
+/**
+ * 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;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import lombok.RequiredArgsConstructor;
+import org.apache.fineract.avro.BulkMessageItemV1;
+import org.apache.fineract.infrastructure.event.business.domain.BusinessEvent;
+import org.apache.fineract.infrastructure.event.external.service.serialization.serializer.BusinessEventSerializer;
+import org.apache.fineract.infrastructure.event.external.service.serialization.serializer.BusinessEventSerializerFactory;
+import org.apache.fineract.infrastructure.event.external.service.support.ByteBufferConverter;
+import org.springframework.stereotype.Component;
+
+@Component
+@RequiredArgsConstructor
+public class BulkMessageItemFactory {
+
+    private final BusinessEventSerializerFactory serializerFactory;
+    private final ByteBufferConverter byteBufferConverter;
+
+    public BulkMessageItemV1 createBulkMessageItem(int id, BusinessEvent<?> event) throws IOException {
+        BusinessEventSerializer eventSerializer = serializerFactory.create(event);
+        byte[] serializedContent = eventSerializer.serialize(event);
+        String type = event.getType();
+        String category = "nocategory"; // TODO: switch this to the actual category when implemented
+        String schema = eventSerializer.getSupportedSchema().getName();
+        ByteBuffer data = byteBufferConverter.convert(serializedContent);
+        return new BulkMessageItemV1(id, type, category, schema, data);
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/service/message/MessageFactory.java

@@ -0,0 +1,76 @@
+/**
+ * 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;
+
+import java.time.OffsetDateTime;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+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.service.message.domain.BulkMessageData;
+import org.apache.fineract.infrastructure.event.external.service.message.domain.MessageCategory;
+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;
+import org.apache.fineract.infrastructure.event.external.service.message.domain.MessageIdempotencyKey;
+import org.apache.fineract.infrastructure.event.external.service.message.domain.MessageSource;
+import org.apache.fineract.infrastructure.event.external.service.message.domain.MessageType;
+import org.springframework.stereotype.Component;
+
+@Component
+public class MessageFactory {
+
+    public MessageV1 createMessage(MessageId id, MessageSource source, MessageType type, MessageCategory category,
+            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.setTenantId(getTenantId());
+        result.setIdempotencyKey(idempotencyKey.getIdempotencyKey());
+        result.setDataschema(dataSchema.getDataSchema());
+        result.setData(data.getData());
+        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;
+    }
+
+    private String getTenantId() {
+        return ThreadLocalContextUtil.getTenant().getName();
+    }
+
+    private String getMessageCreatedAt() {
+        OffsetDateTime createdAt = DateUtils.getOffsetDateTimeOfTenant();
+        return createdAt.withOffsetSameInstant(ZoneOffset.UTC).toLocalDateTime().format(DateTimeFormatter.ISO_LOCAL_DATE_TIME);
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/service/message/domain/BulkMessageData.java

@@ -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.util.Objects;
+import lombok.Getter;
+import org.apache.fineract.avro.BulkMessagePayloadV1;
+
+@Getter
+public class BulkMessageData {
+
+    private final BulkMessagePayloadV1 data;
+
+    public BulkMessageData(BulkMessagePayloadV1 data) {
+        this.data = Objects.requireNonNull(data, "data cannot be null");
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/service/message/domain/MessageCategory.java

@@ -0,0 +1,32 @@
+/**
+ * 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.util.Objects;
+import lombok.Getter;
+
+@Getter
+public class MessageCategory {
+
+    private final String category;
+
+    public MessageCategory(String category) {
+        this.category = Objects.requireNonNull(category, "category cannot be null");
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/service/message/domain/MessageData.java

@@ -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.nio.ByteBuffer;
+import java.util.Objects;
+import lombok.Getter;
+
+@Getter
+public class MessageData {
+
+    private final ByteBuffer data;
+
+    public MessageData(ByteBuffer data) {
+        this.data = Objects.requireNonNull(data, "data cannot be null");
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/service/message/domain/MessageDataSchema.java

@@ -0,0 +1,32 @@
+/**
+ * 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.util.Objects;
+import lombok.Getter;
+
+@Getter
+public class MessageDataSchema {
+
+    private final String dataSchema;
+
+    public MessageDataSchema(String dataSchema) {
+        this.dataSchema = Objects.requireNonNull(dataSchema, "dataSchema cannot be null");
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/service/message/domain/MessageId.java

@@ -0,0 +1,29 @@
+/**
+ * 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 lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+@RequiredArgsConstructor
+@Getter
+public class MessageId {
+
+    private final int id;
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/service/message/domain/MessageIdempotencyKey.java

@@ -0,0 +1,32 @@
+/**
+ * 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.util.Objects;
+import lombok.Getter;
+
+@Getter
+public class MessageIdempotencyKey {
+
+    private final String idempotencyKey;
+
+    public MessageIdempotencyKey(String idempotencyKey) {
+        this.idempotencyKey = Objects.requireNonNull(idempotencyKey, "idempotencyKey cannot be null");
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/service/message/domain/MessageSource.java

@@ -0,0 +1,32 @@
+/**
+ * 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.util.Objects;
+import lombok.Getter;
+
+@Getter
+public class MessageSource {
+
+    private final String source;
+
+    public MessageSource(String source) {
+        this.source = Objects.requireNonNull(source, "source cannot be null");
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/service/message/domain/MessageType.java

@@ -0,0 +1,32 @@
+/**
+ * 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.util.Objects;
+import lombok.Getter;
+
+@Getter
+public class MessageType {
+
+    private final String type;
+
+    public MessageType(String type) {
+        this.type = Objects.requireNonNull(type, "type cannot be null");
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/service/serialization/serializer/BusinessEventSerializerFactory.java

@@ -16,12 +16,11 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.infrastructure.event.external.service.serialization;
+package org.apache.fineract.infrastructure.event.external.service.serialization.serializer;
 
 import java.util.List;
 import lombok.RequiredArgsConstructor;
 import org.apache.fineract.infrastructure.event.business.domain.BusinessEvent;
-import org.apache.fineract.infrastructure.event.external.service.serialization.serializer.BusinessEventSerializer;
 import org.springframework.stereotype.Component;
 
 @Component

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/service/support/ByteBufferConverter.java

@@ -30,4 +30,8 @@ public class ByteBufferConverter {
         buffer.position(buffer.position() - bytes.length);
         return bytes;
     }
+
+    public ByteBuffer convert(byte[] buffer) {
+        return ByteBuffer.wrap(buffer);
+    }
 }

fineract-provider/src/main/java/org/apache/fineract/portfolio/client/service/ClientWritePlatformServiceJpaRepositoryImpl.java

@@ -223,6 +223,7 @@ public class ClientWritePlatformServiceJpaRepositoryImpl implements ClientWriteP
     public CommandProcessingResult createClient(final JsonCommand command) {
 
         try {
+            businessEventNotifierService.startExternalEventRecording();
             final AppUser currentUser = this.context.authenticatedUser();
 
             this.fromApiJsonDeserializer.validateForCreate(command.json());
@@ -339,6 +340,7 @@ public class ClientWritePlatformServiceJpaRepositoryImpl implements ClientWriteP
 
             entityDatatableChecksWritePlatformService.runTheCheck(newClient.getId(), EntityTables.CLIENT.getName(),
                     StatusEnum.CREATE.getCode().longValue(), EntityTables.CLIENT.getForeignKeyColumnNameOnDatatable());
+            businessEventNotifierService.stopExternalEventRecording();
             return new CommandProcessingResultBuilder() //
                     .withCommandId(command.commandId()) //
                     .withOfficeId(clientOffice.getId()) //

fineract-provider/src/test/java/org/apache/fineract/infrastructure/event/external/service/ExternalEventServiceTest.java

@@ -20,26 +20,33 @@ package org.apache.fineract.infrastructure.event.external.service;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.BDDMockito.given;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 
 import java.io.IOException;
+import java.nio.ByteBuffer;
 import java.time.LocalDate;
 import java.time.ZoneId;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import javax.persistence.EntityManager;
+import org.apache.fineract.avro.BulkMessageItemV1;
 import org.apache.fineract.avro.loan.v1.LoanAccountDataV1;
 import org.apache.fineract.infrastructure.businessdate.domain.BusinessDateType;
 import org.apache.fineract.infrastructure.core.domain.FineractPlatformTenant;
 import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
+import org.apache.fineract.infrastructure.event.business.domain.BulkBusinessEvent;
 import org.apache.fineract.infrastructure.event.business.domain.BusinessEvent;
 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.service.idempotency.ExternalEventIdempotencyKeyGenerator;
-import org.apache.fineract.infrastructure.event.external.service.serialization.BusinessEventSerializerFactory;
+import org.apache.fineract.infrastructure.event.external.service.message.BulkMessageItemFactory;
 import org.apache.fineract.infrastructure.event.external.service.serialization.serializer.BusinessEventSerializer;
+import org.apache.fineract.infrastructure.event.external.service.serialization.serializer.BusinessEventSerializerFactory;
+import org.apache.fineract.infrastructure.event.external.service.support.ByteBufferConverter;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -58,13 +65,18 @@ class ExternalEventServiceTest {
     @Mock
     private BusinessEventSerializerFactory serializerFactory;
     @Mock
+    private ByteBufferConverter byteBufferConverter;
+    @Mock
+    private BulkMessageItemFactory bulkMessageItemFactory;
+    @Mock
     private EntityManager entityManager;
 
     private ExternalEventService underTest;
 
     @BeforeEach
     public void setUp() {
-        underTest = new ExternalEventService(repository, idempotencyKeyGenerator, serializerFactory);
+        underTest = new ExternalEventService(repository, idempotencyKeyGenerator, serializerFactory, byteBufferConverter,
+                bulkMessageItemFactory);
         underTest.setEntityManager(entityManager);
         FineractPlatformTenant tenant = new FineractPlatformTenant(1L, "default", "Default Tenant", "Europe/Budapest", null);
         ThreadLocalContextUtil.setTenant(tenant);
@@ -94,7 +106,7 @@ class ExternalEventServiceTest {
     }
 
     @Test
-    public void testPostEventShouldWork() throws IOException {
+    public void testPostEventShouldWorkWithRegularEvent() throws IOException {
         // given
         ArgumentCaptor<ExternalEvent> externalEventArgumentCaptor = ArgumentCaptor.forClass(ExternalEvent.class);
 
@@ -120,4 +132,31 @@ class ExternalEventServiceTest {
         assertThat(externalEvent.getType()).isEqualTo(eventType);
         assertThat(externalEvent.getSchema()).isEqualTo(eventSchema);
     }
+
+    @Test
+    public void testPostEventShouldWorkWithBulkEvent() throws IOException {
+        // given
+        ArgumentCaptor<ExternalEvent> externalEventArgumentCaptor = ArgumentCaptor.forClass(ExternalEvent.class);
+        String eventType = "BulkBusinessEvent";
+        String schema = "org.apache.fineract.avro.BulkMessagePayloadV1";
+
+        String idempotencyKey = "key";
+        BusinessEvent event = mock(BusinessEvent.class);
+        BulkMessageItemV1 messageItem = new BulkMessageItemV1(1, "", "", "", ByteBuffer.wrap(new byte[0]));
+        BulkBusinessEvent bulkEvent = new BulkBusinessEvent(List.of(event));
+        byte[] data = new byte[0];
+
+        given(bulkMessageItemFactory.createBulkMessageItem(1, event)).willReturn(messageItem);
+        given(idempotencyKeyGenerator.generate(bulkEvent)).willReturn(idempotencyKey);
+        given(byteBufferConverter.convert(any(ByteBuffer.class))).willReturn(data);
+        // when
+        underTest.postEvent(bulkEvent);
+        // then
+        verify(repository).save(externalEventArgumentCaptor.capture());
+        ExternalEvent externalEvent = externalEventArgumentCaptor.getValue();
+        assertThat(externalEvent.getIdempotencyKey()).isEqualTo(idempotencyKey);
+        assertThat(externalEvent.getData()).isEqualTo(data);
+        assertThat(externalEvent.getType()).isEqualTo(eventType);
+        assertThat(externalEvent.getSchema()).isEqualTo(schema);
+    }
 }