CPS-821 Spike: Support Async read-write operations on CPS-NCMP interface

CPS-821 - Getting issue details... STATUS

Description/Scope

The scope of this spike is to ascertain:

• How to use messaging (producer, agree topic etc))
• Using existing rest endpoint with additional flag indicating async response
• Also consider asynchronous request option using messaging in the proposal

Associated Jira

Issues/Decisions

Proposed Diagram

Rest Endpoint with Async Flag

Kafka config & Implementation

Example of Existing Consumer (Java)

The below code snippet taken from cps-temporal can be used in the same way in NCMP to listen to message from DMI.

 /**
     * Consume the specified event.
     *
     * @param cpsDataUpdatedEvent the data updated event to be consumed and persisted.
     */
    @KafkaListener(topics = "${app.listener.data-updated.topic}", errorHandler = "dataUpdatedEventListenerErrorHandler")
    public void consume(final CpsDataUpdatedEvent cpsDataUpdatedEvent) {

        log.debug("Receiving {} ...", cpsDataUpdatedEvent);

        // Validate event envelop
        validateEventEnvelop(cpsDataUpdatedEvent);

        // Map event to entity
        final var networkData = this.cpsDataUpdatedEventMapper.eventToEntity(cpsDataUpdatedEvent);
        log.debug("Persisting {} ...", networkData);

        // Persist entity
        final var persistedNetworkData = this.networkDataService.addNetworkData(networkData);
        log.debug("Persisted {}", persistedNetworkData);

    }

Example of Existing Consumer (Config)

Example of Existing Producer (Java)

Example of Existing Producer (Config)

Response types for passthrough

Will we be able to handle everything "as-is"?

Future or alternative

What are Futures?

CompletableFuture (Java8+)

Java 8 introduced the CompletableFuture class. Along with the Future interface, it also implemented the CompletionStage interface. This interface defines the contract for an asynchronous computation step that we can combine with other steps.

CompletableFuture is at the same time a building block and a framework, with about 50 different methods for composing, combining, and executing asynchronous computation steps and handling errors.

public Future<String> calculateAsync() throws InterruptedException {
    CompletableFuture<String> completableFuture = new CompletableFuture<>();

    Executors.newCachedThreadPool().submit(() -> {
        Thread.sleep(500);
        completableFuture.complete("Hello");
        return null;
    });

    return completableFuture;
}

source: https://www.baeldung.com/java-completablefuture

RequestID Generation

String uniqueID = UUID.randomUUID().toString();

Async Request Option using Messaging

Demo/Test