FINERACT-2242: New command processing - External Event Configuration (org.apache.fineract.infrastructure.event.external)

This commit is contained in:
viktorpavlenko
2025-06-26 15:37:53 +02:00
committed by Aleksandar Vidakovic
parent 86024b83cc
commit ba06234f26
34 changed files with 340 additions and 316 deletions
@@ -19,27 +19,25 @@
package org.apache.fineract.infrastructure.event.external.api;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.Parameter;
import io.swagger.v3.oas.annotations.media.Content;
import io.swagger.v3.oas.annotations.media.Schema;
import io.swagger.v3.oas.annotations.parameters.RequestBody;
import io.swagger.v3.oas.annotations.tags.Tag;
import jakarta.validation.Valid;
import jakarta.ws.rs.Consumes;
import jakarta.ws.rs.GET;
import jakarta.ws.rs.HeaderParam;
import jakarta.ws.rs.PUT;
import jakarta.ws.rs.Path;
import jakarta.ws.rs.Produces;
import jakarta.ws.rs.core.MediaType;
import java.util.UUID;
import java.util.function.Supplier;
import lombok.RequiredArgsConstructor;
import org.apache.fineract.commands.domain.CommandWrapper;
import org.apache.fineract.commands.service.CommandWrapperBuilder;
import org.apache.fineract.commands.service.PortfolioCommandSourceWritePlatformService;
import org.apache.fineract.infrastructure.core.data.CommandProcessingResult;
import org.apache.fineract.infrastructure.core.serialization.DefaultToApiJsonSerializer;
import org.apache.fineract.infrastructure.event.external.command.ExternalEventConfigurationCommand;
import org.apache.fineract.infrastructure.event.external.data.ExternalEventConfigurationData;
import org.apache.fineract.command.core.CommandPipeline;
import org.apache.fineract.infrastructure.core.service.DateUtils;
import org.apache.fineract.infrastructure.event.external.command.ExternalConfigurationsUpdateCommand;
import org.apache.fineract.infrastructure.event.external.data.ExternalEventConfigurationResponse;
import org.apache.fineract.infrastructure.event.external.data.ExternalEventConfigurationUpdateRequest;
import org.apache.fineract.infrastructure.event.external.data.ExternalEventConfigurationUpdateResponse;
import org.apache.fineract.infrastructure.event.external.service.ExternalEventConfigurationReadPlatformService;
import org.apache.fineract.infrastructure.security.service.PlatformSecurityContext;
import org.springframework.stereotype.Component;
@RequiredArgsConstructor
@@ -48,19 +46,14 @@ import org.springframework.stereotype.Component;
@Tag(name = "External event configuration", description = "External event configuration enables user to enable/disable event posting to downstream message channel")
public class ExternalEventConfigurationApiResource {
private static final String RESOURCE_NAME_FOR_PERMISSIONS = "EXTERNAL_EVENT_CONFIGURATION";
private final PlatformSecurityContext context;
private final PortfolioCommandSourceWritePlatformService commandWritePlatformService;
private final DefaultToApiJsonSerializer<ExternalEventConfigurationData> jsonSerializer;
private final ExternalEventConfigurationReadPlatformService readPlatformService;
private final CommandPipeline commandPipeline;
@GET
@Consumes({ MediaType.APPLICATION_JSON })
@Produces({ MediaType.APPLICATION_JSON })
@Operation(summary = "List all external event configurations", description = "")
public ExternalEventConfigurationData retrieveExternalEventConfiguration() {
context.authenticatedUser().validateHasReadPermission(RESOURCE_NAME_FOR_PERMISSIONS);
public ExternalEventConfigurationResponse getExternalEventConfigurations() {
return readPlatformService.findAllExternalEventConfigurations();
}
@@ -68,12 +61,17 @@ public class ExternalEventConfigurationApiResource {
@Consumes({ MediaType.APPLICATION_JSON })
@Produces({ MediaType.APPLICATION_JSON })
@Operation(summary = "Enable/Disable external events posting", description = "")
@RequestBody(required = true, content = @Content(schema = @Schema(implementation = ExternalEventConfigurationCommand.class)))
public CommandProcessingResult updateExternalEventConfigurationsDetails(
@Parameter(hidden = true) ExternalEventConfigurationCommand command) {
context.authenticatedUser().validateHasUpdatePermission(RESOURCE_NAME_FOR_PERMISSIONS);
final CommandWrapper commandRequest = new CommandWrapperBuilder().updateExternalEventConfigurations()
.withJson(jsonSerializer.serialize(command)).build();
return commandWritePlatformService.logCommandSource(commandRequest);
public ExternalEventConfigurationUpdateResponse updateExternalEventConfigurations(@HeaderParam("Idempotency-Key") String idempotencyKey,
@Valid ExternalEventConfigurationUpdateRequest request) {
final var command = new ExternalConfigurationsUpdateCommand();
command.setId(UUID.randomUUID());
command.setIdempotencyKey(idempotencyKey);
command.setCreatedAt(DateUtils.getAuditOffsetDateTime());
command.setPayload(request);
final Supplier<ExternalEventConfigurationUpdateResponse> response = commandPipeline.send(command);
return response.get();
}
}
@@ -18,7 +18,6 @@
*/
package org.apache.fineract.infrastructure.event.external.api;
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
import jakarta.ws.rs.Consumes;
import jakarta.ws.rs.DELETE;
import jakarta.ws.rs.GET;
@@ -30,51 +29,36 @@ import java.util.List;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.fineract.infrastructure.core.boot.FineractProfiles;
import org.apache.fineract.infrastructure.core.serialization.DefaultToApiJsonSerializer;
import org.apache.fineract.infrastructure.event.external.data.ExternalEventResponse;
import org.apache.fineract.infrastructure.event.external.service.InternalExternalEventService;
import org.apache.fineract.infrastructure.event.external.service.validation.ExternalEventDTO;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.annotation.Profile;
import org.springframework.stereotype.Component;
@Slf4j
// TODO: can't we test this differently without creating boilerplate code that's only available during testing?
@Profile(FineractProfiles.TEST)
@Component
@Path("/v1/internal/externalevents")
@RequiredArgsConstructor
@Slf4j
public class InternalExternalEventsApiResource implements InitializingBean {
public class InternalExternalEventsApiResource {
private final InternalExternalEventService internalExternalEventService;
private final DefaultToApiJsonSerializer<List<ExternalEventDTO>> jsonSerializer;
@Override
@SuppressFBWarnings("SLF4J_SIGN_ONLY_FORMAT")
public void afterPropertiesSet() throws Exception {
log.warn("------------------------------------------------------------");
log.warn(" ");
log.warn("DO NOT USE THIS IN PRODUCTION!");
log.warn("Internal client services mode is enabled");
log.warn("DO NOT USE THIS IN PRODUCTION!");
log.warn(" ");
log.warn("------------------------------------------------------------");
}
@GET
@Consumes({ MediaType.APPLICATION_JSON })
@Produces({ MediaType.APPLICATION_JSON })
public String getAllExternalEvents(@QueryParam("idempotencyKey") final String idempotencyKey, @QueryParam("type") final String type,
@QueryParam("category") final String category, @QueryParam("aggregateRootId") final Long aggregateRootId) {
log.debug("getAllExternalEvents called with params idempotencyKey:{}, type:{}, category:{}, aggregateRootId:{} ", idempotencyKey,
type, category, aggregateRootId);
List<ExternalEventDTO> allExternalEvents = internalExternalEventService.getAllExternalEvents(idempotencyKey, type, category,
aggregateRootId);
return jsonSerializer.serialize(allExternalEvents);
public List<ExternalEventResponse> getAllExternalEvents(@QueryParam("idempotencyKey") final String idempotencyKey,
@QueryParam("type") final String type, @QueryParam("category") final String category,
@QueryParam("aggregateRootId") final Long aggregateRootId) {
// TODO: authorization constraints?
return internalExternalEventService.getAllExternalEvents(idempotencyKey, type, category, aggregateRootId);
}
@DELETE
@Consumes({ MediaType.APPLICATION_JSON })
@Produces({ MediaType.APPLICATION_JSON })
public void deleteAllExternalEvents() {
log.debug("deleteAllExternalEvents called");
// TODO: authorization constraints?
internalExternalEventService.deleteAllExternalEvents();
}
}
@@ -18,13 +18,11 @@
*/
package org.apache.fineract.infrastructure.event.external.command;
import java.io.Serial;
import java.io.Serializable;
import java.util.Map;
import lombok.Data;
import lombok.EqualsAndHashCode;
import org.apache.fineract.command.core.Command;
import org.apache.fineract.infrastructure.event.external.data.ExternalEventConfigurationUpdateRequest;
public record ExternalEventConfigurationCommand(Map<String, Boolean> externalEventConfigurations) implements Serializable {
@Serial
private static final long serialVersionUID = 1L;
}
@Data
@EqualsAndHashCode(callSuper = true)
public class ExternalConfigurationsUpdateCommand extends Command<ExternalEventConfigurationUpdateRequest> {}
@@ -18,14 +18,21 @@
*/
package org.apache.fineract.infrastructure.event.external.data;
import java.io.Serial;
import java.io.Serializable;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
@Builder
@Data
@NoArgsConstructor
@AllArgsConstructor
public class ExternalEventConfigurationItemData {
public class ExternalEventConfigurationItemResponse implements Serializable {
@Serial
private static final long serialVersionUID = 1L;
private String type;
private boolean enabled;
@@ -0,0 +1,40 @@
/**
* 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.data;
import java.io.Serial;
import java.io.Serializable;
import java.util.List;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
@Builder
@Data
@NoArgsConstructor
@AllArgsConstructor
public class ExternalEventConfigurationResponse implements Serializable {
@Serial
private static final long serialVersionUID = 1L;
// TODO: why wrap things in this useless class?!? Just more boilerplate! Keeping for compatibility...
private List<ExternalEventConfigurationItemResponse> externalEventConfiguration;
}
@@ -16,26 +16,26 @@
* specific language governing permissions and limitations
* under the License.
*/
package org.apache.fineract.infrastructure.event.external.service.validation;
package org.apache.fineract.infrastructure.event.external.data;
import java.time.LocalDate;
import java.time.OffsetDateTime;
import jakarta.validation.constraints.NotNull;
import java.io.Serial;
import java.io.Serializable;
import java.util.Map;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.ToString;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
@Getter
@Builder
@Data
@NoArgsConstructor
@AllArgsConstructor
@ToString
public class ExternalEventDTO {
public class ExternalEventConfigurationUpdateRequest implements Serializable {
private final Long eventId;
private final String type;
private final String category;
private final OffsetDateTime createdAt;
private final Map<String, Object> payLoad;
private final LocalDate businessDate;
private final String schema;
private final Long aggregateRootId;
@Serial
private static final long serialVersionUID = 1L;
@NotNull(message = "{org.apache.fineract.externalevent.configurations.not-null}")
private Map<String, Boolean> externalEventConfigurations;
}
@@ -18,13 +18,22 @@
*/
package org.apache.fineract.infrastructure.event.external.data;
import java.util.List;
import java.io.Serial;
import java.io.Serializable;
import java.util.HashMap;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
@Builder
@Data
@NoArgsConstructor
public class ExternalEventConfigurationData {
@AllArgsConstructor
public class ExternalEventConfigurationUpdateResponse implements Serializable {
private List<ExternalEventConfigurationItemData> externalEventConfiguration;
@Serial
private static final long serialVersionUID = 1L;
private HashMap<String, Object> changes;
}
@@ -0,0 +1,48 @@
/**
* 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.data;
import java.io.Serial;
import java.io.Serializable;
import java.time.LocalDate;
import java.time.OffsetDateTime;
import java.util.Map;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
@Builder
@Data
@NoArgsConstructor
@AllArgsConstructor
public class ExternalEventResponse implements Serializable {
@Serial
private static final long serialVersionUID = 1L;
private Long eventId;
private String type;
private String category;
private OffsetDateTime createdAt;
private Map<String, Object> payLoad;
private LocalDate businessDate;
private String schema;
private Long aggregateRootId;
}
@@ -18,12 +18,12 @@
*/
package org.apache.fineract.infrastructure.event.external.handler;
import jakarta.validation.constraints.NotNull;
import lombok.RequiredArgsConstructor;
import org.apache.fineract.command.core.Command;
import org.apache.fineract.command.core.CommandHandler;
import org.apache.fineract.commands.annotation.CommandType;
import org.apache.fineract.commands.handler.NewCommandSourceHandler;
import org.apache.fineract.infrastructure.core.api.JsonCommand;
import org.apache.fineract.infrastructure.core.data.CommandProcessingResult;
import org.apache.fineract.infrastructure.event.external.data.ExternalEventConfigurationUpdateRequest;
import org.apache.fineract.infrastructure.event.external.data.ExternalEventConfigurationUpdateResponse;
import org.apache.fineract.infrastructure.event.external.service.ExternalEventConfigurationWritePlatformService;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
@@ -31,13 +31,14 @@ import org.springframework.transaction.annotation.Transactional;
@RequiredArgsConstructor
@Service
@CommandType(entity = "EXTERNAL_EVENT_CONFIGURATION", action = "UPDATE")
public class ExternalEventConfigurationUpdateHandler implements NewCommandSourceHandler {
public class ExternalEventConfigurationUpdateHandler
implements CommandHandler<ExternalEventConfigurationUpdateRequest, ExternalEventConfigurationUpdateResponse> {
private final ExternalEventConfigurationWritePlatformService writePlatformService;
@Transactional
@Override
public CommandProcessingResult processCommand(@NotNull final JsonCommand command) {
return writePlatformService.updateConfigurations(command);
public ExternalEventConfigurationUpdateResponse handle(Command<ExternalEventConfigurationUpdateRequest> command) {
return writePlatformService.updateConfigurations(command.getPayload());
}
}
@@ -1,55 +0,0 @@
/**
* 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.serialization;
import com.google.gson.reflect.TypeToken;
import java.lang.reflect.Type;
import java.util.Arrays;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import lombok.AllArgsConstructor;
import org.apache.commons.lang3.StringUtils;
import org.apache.fineract.infrastructure.core.exception.InvalidJsonException;
import org.apache.fineract.infrastructure.core.serialization.AbstractFromApiJsonDeserializer;
import org.apache.fineract.infrastructure.core.serialization.FromJsonHelper;
import org.apache.fineract.infrastructure.event.external.command.ExternalEventConfigurationCommand;
import org.springframework.stereotype.Component;
@Component
@AllArgsConstructor
public class ExternalEventConfigurationCommandFromApiJsonDeserializer
extends AbstractFromApiJsonDeserializer<ExternalEventConfigurationCommand> {
private static final String EXTERNAL_EVENT_CONFIGURATIONS = "externalEventConfigurations";
private final Set<String> supportedParameters = new HashSet<>(Arrays.asList(EXTERNAL_EVENT_CONFIGURATIONS));
private final FromJsonHelper fromApiJsonHelper;
@Override
public ExternalEventConfigurationCommand commandFromApiJson(String json) {
if (StringUtils.isBlank(json)) {
throw new InvalidJsonException();
}
final Type typeOfMap = new TypeToken<Map<String, Object>>() {}.getType();
fromApiJsonHelper.checkForUnsupportedParameters(typeOfMap, json, supportedParameters);
return fromApiJsonHelper.fromJson(json, ExternalEventConfigurationCommand.class);
}
}
@@ -18,9 +18,9 @@
*/
package org.apache.fineract.infrastructure.event.external.service;
import org.apache.fineract.infrastructure.event.external.data.ExternalEventConfigurationData;
import org.apache.fineract.infrastructure.event.external.data.ExternalEventConfigurationResponse;
public interface ExternalEventConfigurationReadPlatformService {
ExternalEventConfigurationData findAllExternalEventConfigurations();
ExternalEventConfigurationResponse findAllExternalEventConfigurations();
}
@@ -20,7 +20,7 @@ package org.apache.fineract.infrastructure.event.external.service;
import java.util.List;
import lombok.RequiredArgsConstructor;
import org.apache.fineract.infrastructure.event.external.data.ExternalEventConfigurationData;
import org.apache.fineract.infrastructure.event.external.data.ExternalEventConfigurationResponse;
import org.apache.fineract.infrastructure.event.external.repository.ExternalEventConfigurationRepository;
import org.apache.fineract.infrastructure.event.external.repository.domain.ExternalEventConfiguration;
import org.springframework.stereotype.Service;
@@ -33,8 +33,8 @@ public class ExternalEventConfigurationReadPlatformServiceImpl implements Extern
private final ExternalEventsConfigurationMapper mapper;
@Override
public ExternalEventConfigurationData findAllExternalEventConfigurations() {
ExternalEventConfigurationData configurationData = new ExternalEventConfigurationData();
public ExternalEventConfigurationResponse findAllExternalEventConfigurations() {
ExternalEventConfigurationResponse configurationData = new ExternalEventConfigurationResponse();
List<ExternalEventConfiguration> eventConfigurations = repository.findAll();
configurationData.setExternalEventConfiguration(mapper.map(eventConfigurations));
return configurationData;
@@ -18,10 +18,10 @@
*/
package org.apache.fineract.infrastructure.event.external.service;
import org.apache.fineract.infrastructure.core.api.JsonCommand;
import org.apache.fineract.infrastructure.core.data.CommandProcessingResult;
import org.apache.fineract.infrastructure.event.external.data.ExternalEventConfigurationUpdateRequest;
import org.apache.fineract.infrastructure.event.external.data.ExternalEventConfigurationUpdateResponse;
public interface ExternalEventConfigurationWritePlatformService {
CommandProcessingResult updateConfigurations(JsonCommand command);
ExternalEventConfigurationUpdateResponse updateConfigurations(ExternalEventConfigurationUpdateRequest request);
}
@@ -20,16 +20,11 @@ package org.apache.fineract.infrastructure.event.external.service;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import lombok.AllArgsConstructor;
import org.apache.fineract.infrastructure.core.api.JsonCommand;
import org.apache.fineract.infrastructure.core.data.CommandProcessingResult;
import org.apache.fineract.infrastructure.core.data.CommandProcessingResultBuilder;
import org.apache.fineract.infrastructure.event.external.command.ExternalEventConfigurationCommand;
import org.apache.fineract.infrastructure.event.external.data.ExternalEventConfigurationUpdateRequest;
import org.apache.fineract.infrastructure.event.external.data.ExternalEventConfigurationUpdateResponse;
import org.apache.fineract.infrastructure.event.external.repository.ExternalEventConfigurationRepository;
import org.apache.fineract.infrastructure.event.external.repository.domain.ExternalEventConfiguration;
import org.apache.fineract.infrastructure.event.external.serialization.ExternalEventConfigurationCommandFromApiJsonDeserializer;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
@@ -38,33 +33,30 @@ import org.springframework.transaction.annotation.Transactional;
public class ExternalEventConfigurationWritePlatformServiceImpl implements ExternalEventConfigurationWritePlatformService {
private final ExternalEventConfigurationRepository repository;
private final ExternalEventConfigurationCommandFromApiJsonDeserializer fromApiJsonDeserializer;
@Transactional
@Override
public CommandProcessingResult updateConfigurations(final JsonCommand command) {
final ExternalEventConfigurationCommand configurationCommand = fromApiJsonDeserializer.commandFromApiJson(command.json());
final Map<String, Boolean> commandConfigurations = configurationCommand.externalEventConfigurations();
final Map<String, Object> changes = new HashMap<>();
final Map<String, Boolean> changedConfigurations = new HashMap<>();
final List<ExternalEventConfiguration> modifiedConfigurations = new ArrayList<>();
public ExternalEventConfigurationUpdateResponse updateConfigurations(final ExternalEventConfigurationUpdateRequest request) {
final var commandConfigurations = request.getExternalEventConfigurations();
final var changes = new HashMap<String, Object>();
final var changedConfigurations = new HashMap<String, Boolean>();
final var modifiedConfigurations = new ArrayList<ExternalEventConfiguration>();
for (Map.Entry<String, Boolean> entry : commandConfigurations.entrySet()) {
final ExternalEventConfiguration configuration = repository
.findExternalEventConfigurationByTypeWithNotFoundDetection(entry.getKey());
for (var entry : commandConfigurations.entrySet()) {
final var configuration = repository.findExternalEventConfigurationByTypeWithNotFoundDetection(entry.getKey());
configuration.setEnabled(entry.getValue());
changedConfigurations.put(entry.getKey(), entry.getValue());
modifiedConfigurations.add(configuration);
}
if (!modifiedConfigurations.isEmpty()) {
this.repository.saveAll(modifiedConfigurations);
repository.saveAll(modifiedConfigurations);
}
if (!changedConfigurations.isEmpty()) {
changes.put("externalEventConfigurations", changedConfigurations);
}
return new CommandProcessingResultBuilder().withCommandId(command.commandId()).with(changes).build();
return ExternalEventConfigurationUpdateResponse.builder().changes(changes).build();
}
}
@@ -63,7 +63,7 @@ public class ExternalEventService {
}
try {
flushChangesBeforeSerialization();
entityManager.flush();
ExternalEvent externalEvent;
if (event instanceof BulkBusinessEvent) {
externalEvent = handleBulkBusinessEvent((BulkBusinessEvent) event);
@@ -79,6 +79,11 @@ public class ExternalEventService {
}
@PersistenceContext
public void setEntityManager(EntityManager entityManager) {
this.entityManager = entityManager;
}
private ExternalEvent handleBulkBusinessEvent(BulkBusinessEvent bulkBusinessEvent) throws IOException {
List<BulkMessageItemV1> messages = new ArrayList<>();
List<BusinessEvent<?>> events = bulkBusinessEvent.get();
@@ -109,13 +114,4 @@ public class ExternalEventService {
return new ExternalEvent(eventType, eventCategory, schema, data, idempotencyKey, aggregateRootId);
}
private void flushChangesBeforeSerialization() {
entityManager.flush();
}
@PersistenceContext
public void setEntityManager(EntityManager entityManager) {
this.entityManager = entityManager;
}
}
@@ -20,12 +20,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.data.ExternalEventConfigurationItemResponse;
import org.apache.fineract.infrastructure.event.external.repository.domain.ExternalEventConfiguration;
import org.mapstruct.Mapper;
@Mapper(config = MapstructMapperConfig.class)
public interface ExternalEventsConfigurationMapper {
List<ExternalEventConfigurationItemData> map(List<ExternalEventConfiguration> source);
List<ExternalEventConfigurationItemResponse> map(List<ExternalEventConfiguration> source);
}
@@ -22,7 +22,6 @@ import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
@@ -32,9 +31,9 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.fineract.avro.BulkMessageItemV1;
import org.apache.fineract.infrastructure.core.boot.FineractProfiles;
import org.apache.fineract.infrastructure.event.external.data.ExternalEventResponse;
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.validation.ExternalEventDTO;
import org.springframework.context.annotation.Profile;
import org.springframework.data.jpa.domain.Specification;
import org.springframework.stereotype.Service;
@@ -45,14 +44,15 @@ import org.springframework.stereotype.Service;
@AllArgsConstructor
public class InternalExternalEventService {
private final ObjectMapper mapper;
private final ExternalEventRepository externalEventRepository;
public void deleteAllExternalEvents() {
externalEventRepository.deleteAll();
}
public List<ExternalEventDTO> getAllExternalEvents(String idempotencyKey, String type, String category, Long aggregateRootId) {
List<Specification<ExternalEvent>> specifications = new ArrayList<>();
public List<ExternalEventResponse> getAllExternalEvents(String idempotencyKey, String type, String category, Long aggregateRootId) {
var specifications = new ArrayList<Specification<ExternalEvent>>();
if (StringUtils.isNotEmpty(idempotencyKey)) {
specifications.add(hasIdempotencyKey(idempotencyKey));
@@ -70,9 +70,9 @@ public class InternalExternalEventService {
specifications.add(hasAggregateRootId(aggregateRootId));
}
Specification<ExternalEvent> reducedSpecification = specifications.stream().reduce(Specification::and)
var reducedSpecification = specifications.stream().reduce(Specification::and)
.orElse((Specification<ExternalEvent>) (root, query, criteriaBuilder) -> null);
List<ExternalEvent> externalEvents = externalEventRepository.findAll(reducedSpecification);
var externalEvents = externalEventRepository.findAll(reducedSpecification);
try {
return convertToReadableFormat(externalEvents);
@@ -98,29 +98,29 @@ public class InternalExternalEventService {
return (root, query, cb) -> cb.equal(root.get("aggregateRootId"), aggregateRootId);
}
private List<ExternalEventDTO> convertToReadableFormat(List<ExternalEvent> externalEvents) throws ClassNotFoundException,
private List<ExternalEventResponse> convertToReadableFormat(List<ExternalEvent> externalEvents) throws ClassNotFoundException,
NoSuchMethodException, InvocationTargetException, IllegalAccessException, JsonProcessingException {
List<ExternalEventDTO> eventMessages = new ArrayList<>();
for (ExternalEvent externalEvent : externalEvents) {
Class<?> payLoadClass = Class.forName(externalEvent.getSchema());
ByteBuffer byteBuffer = ByteBuffer.wrap(externalEvent.getData());
Method method = payLoadClass.getMethod("fromByteBuffer", ByteBuffer.class);
Object payLoad = method.invoke(null, byteBuffer);
var eventMessages = new ArrayList<ExternalEventResponse>();
for (var externalEvent : externalEvents) {
var payLoadClass = Class.forName(externalEvent.getSchema());
var byteBuffer = ByteBuffer.wrap(externalEvent.getData());
var method = payLoadClass.getMethod("fromByteBuffer", ByteBuffer.class);
var payLoad = method.invoke(null, byteBuffer);
if (externalEvent.getType().equalsIgnoreCase("BulkBusinessEvent")) {
Method methodToGetDatas = payLoad.getClass().getMethod("getDatas", (Class<?>) null);
List<BulkMessageItemV1> bulkMessages = (List<BulkMessageItemV1>) methodToGetDatas.invoke(payLoad);
StringBuilder bulkMessagePayload = new StringBuilder();
for (BulkMessageItemV1 bulkMessage : bulkMessages) {
ExternalEventDTO bulkMessageData = retrieveBulkMessage(bulkMessage, externalEvent);
var methodToGetDatas = payLoad.getClass().getMethod("getDatas", (Class<?>) null);
var bulkMessages = (List<BulkMessageItemV1>) methodToGetDatas.invoke(payLoad);
var bulkMessagePayload = new StringBuilder();
for (var bulkMessage : bulkMessages) {
var bulkMessageData = retrieveBulkMessage(bulkMessage, externalEvent);
bulkMessagePayload.append(bulkMessageData);
bulkMessagePayload.append(System.lineSeparator());
}
eventMessages.add(new ExternalEventDTO(externalEvent.getId(), externalEvent.getType(), externalEvent.getCategory(),
eventMessages.add(new ExternalEventResponse(externalEvent.getId(), externalEvent.getType(), externalEvent.getCategory(),
externalEvent.getCreatedAt(), toJsonMap(bulkMessagePayload.toString()), externalEvent.getBusinessDate(),
externalEvent.getSchema(), externalEvent.getAggregateRootId()));
} else {
eventMessages.add(new ExternalEventDTO(externalEvent.getId(), externalEvent.getType(), externalEvent.getCategory(),
eventMessages.add(new ExternalEventResponse(externalEvent.getId(), externalEvent.getType(), externalEvent.getCategory(),
externalEvent.getCreatedAt(), toJsonMap(payLoad.toString()), externalEvent.getBusinessDate(),
externalEvent.getSchema(), externalEvent.getAggregateRootId()));
}
@@ -129,19 +129,19 @@ public class InternalExternalEventService {
return eventMessages;
}
private ExternalEventDTO retrieveBulkMessage(BulkMessageItemV1 messageItem, ExternalEvent externalEvent) throws ClassNotFoundException,
InvocationTargetException, IllegalAccessException, NoSuchMethodException, JsonProcessingException {
Class<?> messageBulkMessagePayLoad = Class.forName(messageItem.getDataschema());
Method methodForPayLoad = messageBulkMessagePayLoad.getMethod("fromByteBuffer", ByteBuffer.class);
Object payLoadBulkItem = methodForPayLoad.invoke(null, messageItem.getData());
return new ExternalEventDTO((long) messageItem.getId(), messageItem.getType(), messageItem.getCategory(),
private ExternalEventResponse retrieveBulkMessage(BulkMessageItemV1 messageItem, ExternalEvent externalEvent)
throws ClassNotFoundException, InvocationTargetException, IllegalAccessException, NoSuchMethodException,
JsonProcessingException {
var messageBulkMessagePayLoad = Class.forName(messageItem.getDataschema());
var methodForPayLoad = messageBulkMessagePayLoad.getMethod("fromByteBuffer", ByteBuffer.class);
var payLoadBulkItem = methodForPayLoad.invoke(null, messageItem.getData());
return new ExternalEventResponse(messageItem.getId(), messageItem.getType(), messageItem.getCategory(),
externalEvent.getCreatedAt(), toJsonMap(payLoadBulkItem.toString()), externalEvent.getBusinessDate(),
externalEvent.getSchema(), externalEvent.getAggregateRootId());
}
private Map<String, Object> toJsonMap(String json) throws JsonProcessingException {
ObjectMapper objectMapper = new ObjectMapper();
return objectMapper.readValue(json, new TypeReference<>() {});
return mapper.readValue(json, new TypeReference<>() {});
}
}