diff options
Diffstat (limited to 'rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api')
3 files changed, 41 insertions, 6 deletions
diff --git a/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/DmaapClientFactory.java b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/DmaapClientFactory.java index ee4f6d38..6e7f6049 100644 --- a/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/DmaapClientFactory.java +++ b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/DmaapClientFactory.java @@ -3,6 +3,7 @@ * DCAEGEN2-SERVICES-SDK * ========================================================= * Copyright (C) 2019-2021 Nokia. All rights reserved. + * Copyright (C) 2023 Deutsche Telekom AG. All rights reserved. * ========================================================= * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -35,6 +36,8 @@ import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.impl.MessageRou import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.config.DmaapClientConfiguration; import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.config.MessageRouterPublisherConfig; import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.config.MessageRouterSubscriberConfig; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.time.Duration; @@ -46,26 +49,37 @@ import static org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.co * @since 1.1.4 */ public final class DmaapClientFactory { - + private static final Logger LOGGER = LoggerFactory.getLogger(DmaapClientFactory.class); private DmaapClientFactory() { } public static @NotNull MessageRouterPublisher createMessageRouterPublisher( @NotNull MessageRouterPublisherConfig clientConfiguration) { - return new MessageRouterPublisherImpl( + try { + return new MessageRouterPublisherImpl( createHttpClient(clientConfiguration), clientConfiguration.maxBatchSize(), clientConfiguration.maxBatchDuration(), new ClientErrorReasonPresenter()); + } catch (Exception e) { + LOGGER.error("Error while creating the Message Router Publisher."); + return null; + } } public static @NotNull MessageRouterSubscriber createMessageRouterSubscriber( @NotNull MessageRouterSubscriberConfig clientConfiguration) { - return new MessageRouterSubscriberImpl( - createHttpClient(clientConfiguration), - clientConfiguration.gsonInstance(), - new ClientErrorReasonPresenter()); + try { + return new MessageRouterSubscriberImpl( + createHttpClient(clientConfiguration), + clientConfiguration.gsonInstance(), + new ClientErrorReasonPresenter()); + } catch (Exception e) { + LOGGER.error("Error while creating the Message Router Subscriber."); + return null; + } + } private static @NotNull RxHttpClient createHttpClient(DmaapClientConfiguration config) { diff --git a/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterPublisher.java b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterPublisher.java index 08825b4c..f98e8198 100644 --- a/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterPublisher.java +++ b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterPublisher.java @@ -3,6 +3,7 @@ * DCAEGEN2-SERVICES-SDK * ========================================================= * Copyright (C) 2019 Nokia. All rights reserved. + * Copyright (C) 2023 Deutsche Telekom AG. All rights reserved. * ========================================================= * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -21,6 +22,13 @@ package org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api; import com.google.gson.JsonElement; + +import io.vavr.collection.List; + +import java.util.Properties; + +import org.apache.kafka.clients.producer.Producer; +import org.onap.dcaegen2.services.sdk.model.streams.dmaap.KafkaSink; import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishRequest; import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishResponse; import reactor.core.publisher.Flux; @@ -30,5 +38,7 @@ import reactor.core.publisher.Flux; * @since 1.1.4 */ public interface MessageRouterPublisher { + void close(); + void setKafkaProducer(Producer<String, String> kafkaProducer); Flux<MessageRouterPublishResponse> put(MessageRouterPublishRequest request, Flux<? extends JsonElement> items); } diff --git a/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterSubscriber.java b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterSubscriber.java index d91535d3..fbf90d9e 100644 --- a/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterSubscriber.java +++ b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterSubscriber.java @@ -3,6 +3,7 @@ * DCAEGEN2-SERVICES-SDK * ========================================================= * Copyright (C) 2019 Nokia. All rights reserved. + * Copyright (C) 2023 Deutsche Telekom AG. All rights reserved. * ========================================================= * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -21,7 +22,14 @@ package org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api; import com.google.gson.JsonElement; + +import io.vavr.collection.List; + import java.time.Duration; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.onap.dcaegen2.services.sdk.model.streams.dmaap.KafkaSource; import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterSubscribeRequest; import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterSubscribeResponse; import reactor.core.publisher.Flux; @@ -33,6 +41,8 @@ import reactor.core.publisher.Mono; */ public interface MessageRouterSubscriber { + void setConsumer(Consumer<String, String> consumer); + void close(); Mono<MessageRouterSubscribeResponse> get(MessageRouterSubscribeRequest request); default Flux<JsonElement> getElements(MessageRouterSubscribeRequest request) { @@ -49,4 +59,5 @@ public interface MessageRouterSubscriber { default Flux<JsonElement> subscribeForElements(MessageRouterSubscribeRequest request, Duration period) { return Flux.interval(period).concatMap(i->getElements(request)); } + } |