-
Notifications
You must be signed in to change notification settings - Fork 20
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
MODINVSTOR-1245: Implement synchronization operation for service poin…
…t events (#1088) * MODINVSTOR-1245: Implement synchronization operation for service point event * MODINVSTOR-1245: Fix from code review * MODINVSTOR-1245: Comply with check style * MODINVSTOR-1245: Comply with check style * MODINVSTOR-1245: Add api test * MODINVSTOR-1245: Rename service point event * MODINVSTOR-1245: Rename tests * MODINVSTOR-1245: Replace service point creation with service implementation * MODINVSTOR-1245: Add validation for hold shelf expiry and pick up location * MODINVSTOR-1245: Replace system-stubs-junit4 instead of hard-coded env variable * MODINVSTOR-1245: Fix from code review * MODINVSTOR-1245: Fix from code review * MODINVSTOR-1245: Add logs
- Loading branch information
1 parent
6101067
commit f7f9680
Showing
18 changed files
with
920 additions
and
16 deletions.
There are no files selected for viewing
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
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
111 changes: 111 additions & 0 deletions
111
src/main/java/org/folio/services/consortium/ServicePointSynchronizationVerticle.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,111 @@ | ||
| package org.folio.services.consortium; | ||
|
|
||
| import static org.folio.rest.tools.utils.ModuleName.getModuleName; | ||
| import static org.folio.rest.tools.utils.ModuleName.getModuleVersion; | ||
| import static org.folio.services.domainevent.ServicePointEventType.SERVICE_POINT_CREATED; | ||
| import static org.folio.services.domainevent.ServicePointEventType.SERVICE_POINT_DELETED; | ||
| import static org.folio.services.domainevent.ServicePointEventType.SERVICE_POINT_UPDATED; | ||
|
|
||
| import io.vertx.core.AbstractVerticle; | ||
| import io.vertx.core.Future; | ||
| import io.vertx.core.Promise; | ||
| import io.vertx.core.http.HttpClient; | ||
| 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.kafka.services.KafkaEnvironmentProperties; | ||
| import org.folio.kafka.services.KafkaTopic; | ||
| import org.folio.services.caches.ConsortiumDataCache; | ||
| import org.folio.services.consortium.handler.ServicePointSynchronizationCreateHandler; | ||
| import org.folio.services.consortium.handler.ServicePointSynchronizationDeleteHandler; | ||
| import org.folio.services.consortium.handler.ServicePointSynchronizationUpdateHandler; | ||
| import org.folio.services.domainevent.ServicePointEventType; | ||
|
|
||
| public class ServicePointSynchronizationVerticle extends AbstractVerticle { | ||
|
|
||
| private static final Logger log = LogManager.getLogger(ServicePointSynchronizationVerticle.class); | ||
| private static final String TENANT_PATTERN = "\\w{1,}"; | ||
| private static final String MODULE_ID = getModuleId(); | ||
| private static final int DEFAULT_LOAD_LIMIT = 5; | ||
| private final ConsortiumDataCache consortiumDataCache; | ||
|
|
||
| private final List<KafkaConsumerWrapper<String, String>> consumers = new ArrayList<>(); | ||
|
|
||
| public ServicePointSynchronizationVerticle(final ConsortiumDataCache consortiumDataCache) { | ||
| this.consortiumDataCache = consortiumDataCache; | ||
| } | ||
|
|
||
| @Override | ||
| public void start(Promise<Void> startPromise) throws Exception { | ||
| var httpClient = vertx.createHttpClient(); | ||
|
|
||
| createConsumers(httpClient) | ||
| .onSuccess(v -> log.info("start:: verticle started")) | ||
| .onFailure(t -> log.error("start:: verticle start failed", t)) | ||
| .onComplete(startPromise); | ||
| } | ||
|
|
||
| private Future<Void> createConsumers(HttpClient httpClient) { | ||
| final var config = getKafkaConfig(); | ||
|
|
||
| return createEventConsumer(SERVICE_POINT_CREATED, config, | ||
| new ServicePointSynchronizationCreateHandler(consortiumDataCache, httpClient, vertx)) | ||
| .compose(r -> createEventConsumer(SERVICE_POINT_UPDATED, config, | ||
| new ServicePointSynchronizationUpdateHandler(consortiumDataCache, httpClient, vertx))) | ||
| .compose(r -> createEventConsumer(SERVICE_POINT_DELETED, config, | ||
| new ServicePointSynchronizationDeleteHandler(consortiumDataCache, httpClient, vertx))) | ||
| .mapEmpty(); | ||
| } | ||
|
|
||
| private Future<KafkaConsumerWrapper<String, String>> createEventConsumer( | ||
| ServicePointEventType eventType, KafkaConfig kafkaConfig, | ||
| AsyncRecordHandler<String, String> handler) { | ||
|
|
||
| var subscriptionDefinition = SubscriptionDefinition.builder() | ||
| .eventType(eventType.name()) | ||
| .subscriptionPattern(buildSubscriptionPattern(eventType.getKafkaTopic(), kafkaConfig)) | ||
| .build(); | ||
|
|
||
| return createConsumer(kafkaConfig, subscriptionDefinition, handler); | ||
| } | ||
|
|
||
| private Future<KafkaConsumerWrapper<String, String>> createConsumer(KafkaConfig kafkaConfig, | ||
| SubscriptionDefinition subscriptionDefinition, | ||
| AsyncRecordHandler<String, String> recordHandler) { | ||
|
|
||
| var consumer = KafkaConsumerWrapper.<String, String>builder() | ||
| .context(context) | ||
| .vertx(vertx) | ||
| .kafkaConfig(kafkaConfig) | ||
| .loadLimit(DEFAULT_LOAD_LIMIT) | ||
| .globalLoadSensor(new GlobalLoadSensor()) | ||
| .subscriptionDefinition(subscriptionDefinition) | ||
| .build(); | ||
|
|
||
| return consumer.start(recordHandler, MODULE_ID) | ||
| .onSuccess(v -> consumers.add(consumer)) | ||
| .map(consumer); | ||
| } | ||
|
|
||
| private static String buildSubscriptionPattern(KafkaTopic kafkaTopic, KafkaConfig kafkaConfig) { | ||
| return kafkaTopic.fullTopicName(kafkaConfig, TENANT_PATTERN); | ||
| } | ||
|
|
||
| private static String getModuleId() { | ||
| return getModuleName().replace("_", "-") + "-" + getModuleVersion(); | ||
| } | ||
|
|
||
| private KafkaConfig getKafkaConfig() { | ||
| return KafkaConfig.builder() | ||
| .envId(KafkaEnvironmentProperties.environment()) | ||
| .kafkaHost(KafkaEnvironmentProperties.host()) | ||
| .kafkaPort(KafkaEnvironmentProperties.port()) | ||
| .build(); | ||
| } | ||
| } |
26 changes: 26 additions & 0 deletions
26
.../java/org/folio/services/consortium/handler/ServicePointSynchronizationCreateHandler.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,26 @@ | ||
| package org.folio.services.consortium.handler; | ||
|
|
||
| import io.vertx.core.Vertx; | ||
| import io.vertx.core.http.HttpClient; | ||
| import org.folio.rest.jaxrs.model.Servicepoint; | ||
| import org.folio.services.caches.ConsortiumDataCache; | ||
| import org.folio.services.consortium.processor.ServicePointSynchronizationCreateEventProcessor; | ||
| import org.folio.services.consortium.processor.ServicePointSynchronizationEventProcessor; | ||
| import org.folio.services.domainevent.DomainEvent; | ||
|
|
||
| public class ServicePointSynchronizationCreateHandler extends ServicePointSynchronizationHandler { | ||
|
|
||
| public ServicePointSynchronizationCreateHandler(ConsortiumDataCache consortiumDataCache, | ||
| HttpClient httpClient, Vertx vertx) { | ||
|
|
||
| super(consortiumDataCache, httpClient, vertx); | ||
| } | ||
|
|
||
| @Override | ||
| protected ServicePointSynchronizationEventProcessor getServicePointSynchronizationProcessor( | ||
| DomainEvent<Servicepoint> domainEvent) { | ||
|
|
||
| return new ServicePointSynchronizationCreateEventProcessor(domainEvent); | ||
| } | ||
|
|
||
| } |
26 changes: 26 additions & 0 deletions
26
.../java/org/folio/services/consortium/handler/ServicePointSynchronizationDeleteHandler.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,26 @@ | ||
| package org.folio.services.consortium.handler; | ||
|
|
||
| import io.vertx.core.Vertx; | ||
| import io.vertx.core.http.HttpClient; | ||
| import org.folio.rest.jaxrs.model.Servicepoint; | ||
| import org.folio.services.caches.ConsortiumDataCache; | ||
| import org.folio.services.consortium.processor.ServicePointSynchronizationDeleteEventProcessor; | ||
| import org.folio.services.consortium.processor.ServicePointSynchronizationEventProcessor; | ||
| import org.folio.services.domainevent.DomainEvent; | ||
|
|
||
| public class ServicePointSynchronizationDeleteHandler extends ServicePointSynchronizationHandler { | ||
|
|
||
| public ServicePointSynchronizationDeleteHandler(ConsortiumDataCache consortiumDataCache, | ||
| HttpClient httpClient, Vertx vertx) { | ||
|
|
||
| super(consortiumDataCache, httpClient, vertx); | ||
| } | ||
|
|
||
| @Override | ||
| protected ServicePointSynchronizationEventProcessor getServicePointSynchronizationProcessor( | ||
| DomainEvent<Servicepoint> domainEvent) { | ||
|
|
||
| return new ServicePointSynchronizationDeleteEventProcessor(domainEvent); | ||
| } | ||
|
|
||
| } |
73 changes: 73 additions & 0 deletions
73
src/main/java/org/folio/services/consortium/handler/ServicePointSynchronizationHandler.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,73 @@ | ||
| package org.folio.services.consortium.handler; | ||
|
|
||
| import static io.vertx.core.Future.succeededFuture; | ||
|
|
||
| import io.vertx.core.Future; | ||
| import io.vertx.core.Vertx; | ||
| import io.vertx.core.http.HttpClient; | ||
| import io.vertx.core.json.Json; | ||
| import io.vertx.kafka.client.consumer.KafkaConsumerRecord; | ||
| import java.util.Optional; | ||
| import org.apache.commons.collections4.map.CaseInsensitiveMap; | ||
| import org.apache.logging.log4j.LogManager; | ||
| import org.apache.logging.log4j.Logger; | ||
| import org.folio.kafka.AsyncRecordHandler; | ||
| import org.folio.kafka.KafkaHeaderUtils; | ||
| import org.folio.rest.jaxrs.model.Servicepoint; | ||
| import org.folio.services.caches.ConsortiumData; | ||
| import org.folio.services.caches.ConsortiumDataCache; | ||
| import org.folio.services.consortium.SynchronizationContext; | ||
| import org.folio.services.consortium.processor.ServicePointSynchronizationEventProcessor; | ||
| import org.folio.services.domainevent.DomainEvent; | ||
|
|
||
| public abstract class ServicePointSynchronizationHandler | ||
| implements AsyncRecordHandler<String, String> { | ||
|
|
||
| private static final Logger log = LogManager.getLogger( | ||
| ServicePointSynchronizationHandler.class); | ||
|
|
||
| private final ConsortiumDataCache consortiumDataCache; | ||
| private final HttpClient httpClient; | ||
| private final Vertx vertx; | ||
|
|
||
| protected ServicePointSynchronizationHandler(ConsortiumDataCache consortiumDataCache, | ||
| HttpClient httpClient, Vertx vertx) { | ||
|
|
||
| this.consortiumDataCache = consortiumDataCache; | ||
| this.httpClient = httpClient; | ||
| this.vertx = vertx; | ||
| } | ||
|
|
||
| @Override | ||
| public Future<String> handle(KafkaConsumerRecord<String, String> kafkaConsumerRecord) { | ||
| log.info("handle:: Processing event {}", kafkaConsumerRecord.topic()); | ||
| var headers = new CaseInsensitiveMap<>(KafkaHeaderUtils.kafkaHeadersToMap( | ||
| kafkaConsumerRecord.headers())); | ||
| return consortiumDataCache.getConsortiumData(headers) | ||
| .compose(consortiumData -> processConsortiumData(kafkaConsumerRecord, consortiumData, | ||
| headers)); | ||
| } | ||
|
|
||
| private Future<String> processConsortiumData( | ||
| KafkaConsumerRecord<String, String> kafkaConsumerRecord, | ||
| Optional<ConsortiumData> consortiumData, CaseInsensitiveMap<String, String> headers) { | ||
|
|
||
| log.info("processConsortiumData:: {}", consortiumData); | ||
| return consortiumData.map(data -> processConsortiumDataByEvent(data, kafkaConsumerRecord, | ||
| headers)).orElseGet(() -> succeededFuture(kafkaConsumerRecord.key())); | ||
| } | ||
|
|
||
| private Future<String> processConsortiumDataByEvent(ConsortiumData data, | ||
| KafkaConsumerRecord<String, String> kafkaConsumerRecord, | ||
| CaseInsensitiveMap<String, String> headers) { | ||
|
|
||
| var event = Json.decodeValue(kafkaConsumerRecord.value(), DomainEvent.class); | ||
| var servicePointSynchronizationProcessor = getServicePointSynchronizationProcessor(event); | ||
| return servicePointSynchronizationProcessor.process(kafkaConsumerRecord.key(), | ||
| new SynchronizationContext(data, headers, vertx, httpClient)); | ||
| } | ||
|
|
||
| protected abstract ServicePointSynchronizationEventProcessor getServicePointSynchronizationProcessor( | ||
| DomainEvent<Servicepoint> domainEvent); | ||
|
|
||
| } |
26 changes: 26 additions & 0 deletions
26
.../java/org/folio/services/consortium/handler/ServicePointSynchronizationUpdateHandler.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,26 @@ | ||
| package org.folio.services.consortium.handler; | ||
|
|
||
| import io.vertx.core.Vertx; | ||
| import io.vertx.core.http.HttpClient; | ||
| import org.folio.rest.jaxrs.model.Servicepoint; | ||
| import org.folio.services.caches.ConsortiumDataCache; | ||
| import org.folio.services.consortium.processor.ServicePointSynchronizationEventProcessor; | ||
| import org.folio.services.consortium.processor.ServicePointSynchronizationUpdateEventProcessor; | ||
| import org.folio.services.domainevent.DomainEvent; | ||
|
|
||
| public class ServicePointSynchronizationUpdateHandler extends ServicePointSynchronizationHandler { | ||
|
|
||
| public ServicePointSynchronizationUpdateHandler(ConsortiumDataCache consortiumDataCache, | ||
| HttpClient httpClient, Vertx vertx) { | ||
|
|
||
| super(consortiumDataCache, httpClient, vertx); | ||
| } | ||
|
|
||
| @Override | ||
| protected ServicePointSynchronizationEventProcessor getServicePointSynchronizationProcessor( | ||
| DomainEvent<Servicepoint> domainEvent) { | ||
|
|
||
| return new ServicePointSynchronizationUpdateEventProcessor(domainEvent); | ||
| } | ||
|
|
||
| } |
Oops, something went wrong.