-
Notifications
You must be signed in to change notification settings - Fork 6
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Created separate verticle for each type
- Loading branch information
Showing
9 changed files
with
254 additions
and
187 deletions.
There are no files selected for viewing
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,19 @@ | ||
| package org.folio.event; | ||
|
|
||
| import lombok.AllArgsConstructor; | ||
| import lombok.Getter; | ||
| import lombok.RequiredArgsConstructor; | ||
|
|
||
| @Getter | ||
| @RequiredArgsConstructor | ||
| @AllArgsConstructor | ||
| public enum InventoryEventType { | ||
| INVENTORY_ITEM_CREATE("inventory.item", PayloadType.CREATE); | ||
|
|
||
| private String topicName; | ||
| private PayloadType payloadType; | ||
|
|
||
| public enum PayloadType { | ||
| UPDATE, DELETE, CREATE, DELETE_ALL | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
108 changes: 108 additions & 0 deletions
108
src/main/java/org/folio/verticles/AbstractConsumersVerticle.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,108 @@ | ||
| package org.folio.verticles; | ||
|
|
||
| import io.vertx.core.AbstractVerticle; | ||
| import io.vertx.core.Context; | ||
| import io.vertx.core.Future; | ||
| import io.vertx.core.Promise; | ||
| import io.vertx.core.Vertx; | ||
| import java.util.ArrayList; | ||
| import java.util.List; | ||
| import org.apache.logging.log4j.LogManager; | ||
| import org.apache.logging.log4j.Logger; | ||
| import org.folio.kafka.AsyncRecordHandler; | ||
| import org.folio.kafka.GlobalLoadSensor; | ||
| import org.folio.kafka.KafkaConfig; | ||
| import org.folio.kafka.KafkaConsumerWrapper; | ||
| import org.folio.kafka.SubscriptionDefinition; | ||
| import org.folio.rest.tools.utils.ModuleName; | ||
| import org.springframework.beans.factory.annotation.Value; | ||
| import org.springframework.context.support.AbstractApplicationContext; | ||
|
|
||
| public abstract class AbstractConsumersVerticle<T> extends AbstractVerticle { | ||
|
|
||
| protected static final Logger log = LogManager.getLogger(); | ||
| protected static final String MODULE_ID = getModuleId(); | ||
| protected static final String TENANT_ID_PATTERN = "\\w{1,}"; | ||
|
|
||
| @Value("${kafka.export.orders.loadLimit:5}") | ||
| protected int loadLimit; | ||
|
|
||
| protected final List<KafkaConsumerWrapper<String, String>> consumers = new ArrayList<>(); | ||
| protected final AbstractApplicationContext springContext; | ||
| protected final KafkaConfig kafkaConfig; | ||
|
|
||
| public AbstractConsumersVerticle(KafkaConfig kafkaConfig, | ||
| AbstractApplicationContext springContext) { | ||
| this.springContext = springContext; | ||
| this.kafkaConfig = kafkaConfig; | ||
| } | ||
|
|
||
| @Override | ||
| public void init(Vertx vertx, Context context) { | ||
| super.init(vertx, context); | ||
| context.put("springContext", springContext); | ||
| } | ||
|
|
||
| @Override | ||
| public void start(Promise<Void> startPromise) { | ||
| log.info("start:: starting kafka consumer verticle. kafka config: {}", kafkaConfig); | ||
|
|
||
| createConsumers() | ||
| .onSuccess(v -> log.info("start:: kafka consumer verticle started successfully")) | ||
| .onFailure(t -> log.error("start:: failed to start kafka consumer verticle", t)) | ||
| .onComplete(startPromise); | ||
| } | ||
|
|
||
| @Override | ||
| public void stop(Promise<Void> stopPromise) { | ||
| log.info("stop:: stopping verticle"); | ||
|
|
||
| stopConsumers() | ||
| .onSuccess(v -> log.info("stop:: kafka consumer verticle stopped successfully")) | ||
| .onFailure(t -> log.error("stop:: failed to stop kafka consumer verticle", t)) | ||
| .onComplete(stopPromise); | ||
| } | ||
|
|
||
| /** | ||
| * Create consumers for the specific event types. | ||
| * | ||
| * @return future with the result of the operation | ||
| */ | ||
| protected abstract Future<Void> createConsumers(); | ||
|
|
||
| protected Future<KafkaConsumerWrapper<String, String>> createConsumer(T eventType, | ||
| SubscriptionDefinition subscriptionDefinition, | ||
| AsyncRecordHandler<String, String> handler) { | ||
| log.info("createConsumer:: creating consumer for event type: {}", eventType); | ||
| var consumerWrapper = KafkaConsumerWrapper.<String, String>builder() | ||
| .context(context) | ||
| .vertx(vertx) | ||
| .kafkaConfig(kafkaConfig) | ||
| .loadLimit(loadLimit) | ||
| .globalLoadSensor(new GlobalLoadSensor()) | ||
| .subscriptionDefinition(subscriptionDefinition) | ||
| .processRecordErrorHandler((t, r) -> log.error("Failed to process event: {}", r, t)) | ||
| .build(); | ||
|
|
||
| log.info("createConsumer:: moduleId={}", MODULE_ID); | ||
| return consumerWrapper.start(handler, MODULE_ID) | ||
| .map(consumerWrapper) | ||
| .onSuccess(consumers::add); | ||
| } | ||
|
|
||
| private Future<Void> stopConsumers() { | ||
| var stopFutures = consumers.stream() | ||
| .map(KafkaConsumerWrapper::stop) | ||
| .toList(); | ||
|
|
||
| return Future.all(stopFutures) | ||
| .onSuccess(v -> log.info("stop:: event consumers stopped successfully")) | ||
| .onFailure(t -> log.error("stop:: failed to stop event consumers", t)) | ||
| .mapEmpty(); | ||
| } | ||
|
|
||
| private static String getModuleId() { | ||
| return ModuleName.getModuleName().replace("_", "-") + "-" + ModuleName.getModuleVersion(); | ||
| } | ||
| } | ||
|
|
45 changes: 45 additions & 0 deletions
45
src/main/java/org/folio/verticles/EdiExportOrdersHistoryConsumersVerticle.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,45 @@ | ||
| package org.folio.verticles; | ||
|
|
||
| import io.vertx.core.Future; | ||
| import org.folio.event.EdiExportHistoryEventType; | ||
| import org.folio.event.handler.EdiExportOrdersHistoryAsyncRecordHandler; | ||
| import org.folio.kafka.AsyncRecordHandler; | ||
| import org.folio.kafka.KafkaConfig; | ||
| import org.folio.kafka.KafkaConsumerWrapper; | ||
| import org.folio.kafka.KafkaTopicNameHelper; | ||
| import org.springframework.beans.factory.annotation.Autowired; | ||
| import org.springframework.context.support.AbstractApplicationContext; | ||
|
|
||
| public class EdiExportOrdersHistoryConsumersVerticle extends AbstractConsumersVerticle<EdiExportHistoryEventType> { | ||
|
|
||
| @Autowired | ||
| public EdiExportOrdersHistoryConsumersVerticle(KafkaConfig kafkaConfig, | ||
| AbstractApplicationContext springContext) { | ||
| super(kafkaConfig, springContext); | ||
| } | ||
|
|
||
| @Override | ||
| protected Future<Void> createConsumers() { | ||
| log.info("createConsumers:: creating EDI Export Order History consumers"); | ||
| return createEdiExportConsumer(EdiExportHistoryEventType.EXPORT_HISTORY_CREATE, | ||
| new EdiExportOrdersHistoryAsyncRecordHandler(context, vertx)) | ||
| .mapEmpty(); | ||
| } | ||
|
|
||
| /** | ||
| * This method creates a consumer for the given event type. | ||
| * Note: The method is using specific subscription pattern for edifact export topics: | ||
| * {envId}.Default.{tenant}.{eventType} -> e.g. 'folio.Default.diku.edi-export-history.create' | ||
| * | ||
| * @param eventType - the event type | ||
| * @param handler - the handler to process the records | ||
| * @return future with the created consumer | ||
| */ | ||
| private Future<KafkaConsumerWrapper<String, String>> createEdiExportConsumer(EdiExportHistoryEventType eventType, | ||
| AsyncRecordHandler<String, String> handler) { | ||
| var subscriptionDefinition = KafkaTopicNameHelper.createSubscriptionDefinition( | ||
| kafkaConfig.getEnvId(), KafkaTopicNameHelper.getDefaultNameSpace(), eventType.getTopicName()); | ||
|
|
||
| return createConsumer(eventType, subscriptionDefinition, handler); | ||
| } | ||
| } |
56 changes: 56 additions & 0 deletions
56
src/main/java/org/folio/verticles/InventoryItemConsumersVerticle.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,56 @@ | ||
| package org.folio.verticles; | ||
|
|
||
| import io.vertx.core.Future; | ||
| import java.util.List; | ||
| import org.folio.event.InventoryEventType; | ||
| import org.folio.event.handler.ItemCreateAsyncRecordHandler; | ||
| import org.folio.kafka.AsyncRecordHandler; | ||
| import org.folio.kafka.KafkaConfig; | ||
| import org.folio.kafka.KafkaConsumerWrapper; | ||
| import org.folio.kafka.KafkaTopicNameHelper; | ||
| import org.folio.kafka.SubscriptionDefinition; | ||
| import org.springframework.beans.factory.annotation.Autowired; | ||
| import org.springframework.context.support.AbstractApplicationContext; | ||
|
|
||
| public class InventoryItemConsumersVerticle extends AbstractConsumersVerticle<InventoryEventType> { | ||
|
|
||
| @Autowired | ||
| public InventoryItemConsumersVerticle(KafkaConfig kafkaConfig, | ||
| AbstractApplicationContext springContext) { | ||
| super(kafkaConfig, springContext); | ||
| } | ||
|
|
||
| @Override | ||
| protected Future<Void> createConsumers() { | ||
| log.info("createConsumers:: creating Item Event consumers"); | ||
| return Future.all( | ||
| List.of( | ||
| createInventoryConsumer(InventoryEventType.INVENTORY_ITEM_CREATE, | ||
| new ItemCreateAsyncRecordHandler(context, vertx)) | ||
| ) | ||
| ).mapEmpty(); | ||
| } | ||
|
|
||
| /** | ||
| * This method creates a consumer for the given event type. | ||
| * Note: The method is using specific subscription pattern for inventory topics: | ||
| * {envId}.{tenant}.{eventType} -> e.g. 'folio.diku.inventory.item' | ||
| * | ||
| * @param eventType - the event type | ||
| * @param handler - the handler to process the records | ||
| * @return future with the created consumer | ||
| */ | ||
| private Future<KafkaConsumerWrapper<String, String>> createInventoryConsumer(InventoryEventType eventType, | ||
| AsyncRecordHandler<String, String> handler) { | ||
| var subscriptionPattern = KafkaTopicNameHelper.formatTopicName( | ||
| kafkaConfig.getEnvId(), TENANT_ID_PATTERN, eventType.getTopicName()); | ||
| log.info("createInventoryConsumer:: subscriptionPattern: {}, for evenType: {}", subscriptionPattern, eventType.name()); | ||
|
|
||
| var subscriptionDefinition = SubscriptionDefinition.builder() | ||
| .eventType(eventType.name()) | ||
| .subscriptionPattern(subscriptionPattern) | ||
| .build(); | ||
|
|
||
| return createConsumer(eventType, subscriptionDefinition, handler); | ||
| } | ||
| } |
Oops, something went wrong.