diff options
author | 2021-06-08 09:04:48 +0200 | |
---|---|---|
committer | 2021-06-17 10:03:49 +0200 | |
commit | 74b598291ed2461e0e482f556baf2943a97a54f2 (patch) | |
tree | 22e1140bc0f27680be4d47d40b0c94f7205b45b3 /src/main/java/org/onap/dcae/common/publishing/DmaapRequestConfiguration.java | |
parent | 26be283f4a7044aea4ee0ca480fde20eb5233ee2 (diff) |
Replace cambria with DmaaP client
- remove cambria, add DmaaP client
- sending event for many topics at once is no longer supported
- add backward compatibility status codes
- add additional validation for batchEvent
Issue-ID: DCAEGEN2-1483
Signed-off-by: Maciej Malewski <maciej.malewski@nokia.com>
Change-Id: I945c38b4ab04b697ecfabd5ce38502f83fa70d1a
Diffstat (limited to 'src/main/java/org/onap/dcae/common/publishing/DmaapRequestConfiguration.java')
-rw-r--r-- | src/main/java/org/onap/dcae/common/publishing/DmaapRequestConfiguration.java | 101 |
1 files changed, 101 insertions, 0 deletions
diff --git a/src/main/java/org/onap/dcae/common/publishing/DmaapRequestConfiguration.java b/src/main/java/org/onap/dcae/common/publishing/DmaapRequestConfiguration.java new file mode 100644 index 00000000..2eaeab6a --- /dev/null +++ b/src/main/java/org/onap/dcae/common/publishing/DmaapRequestConfiguration.java @@ -0,0 +1,101 @@ +/*- + * ============LICENSE_START======================================================= + * VES Collector + * ================================================================================ + * Copyright (C) 2021 Nokia. 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. + * 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. + * ============LICENSE_END========================================================= + */ +package org.onap.dcae.common.publishing; + +import com.google.gson.JsonElement; +import com.google.gson.JsonObject; +import com.google.gson.JsonParser; +import io.vavr.collection.List; +import io.vavr.control.Option; +import org.jetbrains.annotations.NotNull; +import org.onap.dcaegen2.services.sdk.model.streams.dmaap.ImmutableMessageRouterSink; +import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.ContentType; +import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterPublishRequest; +import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishRequest; +import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.config.ImmutableDmaapConnectionPoolConfig; +import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.config.ImmutableDmaapRetryConfig; +import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.config.ImmutableDmaapTimeoutConfig; +import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.config.ImmutableMessageRouterPublisherConfig; +import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.config.MessageRouterPublisherConfig; +import reactor.core.publisher.Flux; + +import java.time.Duration; + +public class DmaapRequestConfiguration { + + private static final Long TIMEOUT_SECONDS = 10L; + private static final int RETRY_INTERVAL_IN_SECONDS = 1; + private static final int RETRY_COUNT = 1; + + private DmaapRequestConfiguration() { + } + + static MessageRouterPublishRequest createPublishRequest(Option<PublisherConfig> publisherConfig, Long timeout) { + String topicUrl = createUrl(publisherConfig); + return ImmutableMessageRouterPublishRequest.builder() + .sinkDefinition(createMessageRouterSink(topicUrl)) + .contentType(ContentType.APPLICATION_JSON) + .timeoutConfig(timeOutConfiguration(timeout)) + .build(); + } + + static MessageRouterPublishRequest createPublishRequest(Option<PublisherConfig> publisherConfig) { + return createPublishRequest(publisherConfig, TIMEOUT_SECONDS); + } + + static Flux<JsonObject> jsonBatch(List<String> messages) { + return Flux.fromIterable(getAsJsonObjects(messages)); + } + + static MessageRouterPublisherConfig retryConfiguration() { + return ImmutableMessageRouterPublisherConfig.builder() + .retryConfig(ImmutableDmaapRetryConfig.builder() + .retryIntervalInSeconds(RETRY_INTERVAL_IN_SECONDS) + .retryCount(RETRY_COUNT) + .build()) + .build(); + } + + private static String createUrl(Option<PublisherConfig> publisherConfig) { + String hostAndPort = publisherConfig.get().getHostAndPort(); + String topicName = publisherConfig.get().topic(); + return String.format("http://%s/events/%s/",hostAndPort,topicName); + } + + private static List<JsonObject> getAsJsonObjects(List<String> messages) { + return getAsJsonElements(messages).map(JsonElement::getAsJsonObject); + } + + static List<JsonElement> getAsJsonElements(List<String> messages) { + return messages.map(JsonParser::parseString); + } + + static ImmutableMessageRouterSink createMessageRouterSink(String topicUrl) { + return ImmutableMessageRouterSink.builder() + .name("the topic") + .topicUrl(topicUrl) + .build(); + } + + @NotNull + private static ImmutableDmaapTimeoutConfig timeOutConfiguration(Long timeout) { + return ImmutableDmaapTimeoutConfig.builder().timeout(Duration.ofSeconds(timeout)).build(); + } +} |