2 * ============LICENSE_START====================================
3 * DCAEGEN2-SERVICES-SDK
4 * =========================================================
5 * Copyright (C) 2019-2020 Nokia. All rights reserved.
6 * =========================================================
7 * Licensed under the Apache License, Version 2.0 (the "License");
8 * you may not use this file except in compliance with the License.
9 * You may obtain a copy of the License at
11 * http://www.apache.org/licenses/LICENSE-2.0
13 * Unless required by applicable law or agreed to in writing, software
14 * distributed under the License is distributed on an "AS IS" BASIS,
15 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16 * See the License for the specific language governing permissions and
17 * limitations under the License.
18 * ============LICENSE_END=====================================
21 package org.onap.dcaegen2.services.sdk.rest.services.dmaap.client;
23 import com.google.gson.Gson;
24 import com.google.gson.JsonElement;
25 import com.google.gson.JsonObject;
26 import com.google.gson.JsonParser;
27 import com.google.gson.JsonPrimitive;
28 import io.vavr.collection.List;
29 import org.onap.dcaegen2.services.sdk.model.streams.dmaap.ImmutableMessageRouterSink;
30 import org.onap.dcaegen2.services.sdk.model.streams.dmaap.ImmutableMessageRouterSource;
31 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api.MessageRouterPublisher;
32 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api.MessageRouterSubscriber;
33 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterPublishRequest;
34 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterPublishResponse;
35 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterSubscribeRequest;
36 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterSubscribeResponse;
37 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishRequest;
38 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishResponse;
39 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterSubscribeRequest;
40 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterSubscribeResponse;
41 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.config.ImmutableTimeoutConfig;
42 import reactor.core.publisher.Flux;
44 import java.time.Duration;
47 public final class MessageRouterTestsUtils {
48 private MessageRouterTestsUtils() {
51 public static MessageRouterPublishRequest createPublishRequest(String topicUrl) {
52 return createPublishRequest(topicUrl, ContentType.APPLICATION_JSON);
55 public static MessageRouterPublishRequest createPublishRequest(String topicUrl, Duration timeout) {
56 return ImmutableMessageRouterPublishRequest.builder()
57 .sinkDefinition(createMessageRouterSink(topicUrl))
58 .contentType(ContentType.APPLICATION_JSON)
59 .timeoutConfig(ImmutableTimeoutConfig.builder()
65 public static MessageRouterPublishRequest createPublishRequest(String topicUrl, ContentType contentType) {
66 return ImmutableMessageRouterPublishRequest.builder()
67 .sinkDefinition(createMessageRouterSink(topicUrl))
68 .contentType(contentType)
72 public static MessageRouterSubscribeRequest createMRSubscribeRequest(String topicUrl,
73 String consumerGroup, String consumerId) {
74 ImmutableMessageRouterSource sourceDefinition = ImmutableMessageRouterSource.builder()
79 return ImmutableMessageRouterSubscribeRequest
81 .sourceDefinition(sourceDefinition)
82 .consumerGroup(consumerGroup)
83 .consumerId(consumerId)
87 public static List<JsonElement> getAsJsonElements(List<String> messages) {
88 return messages.map(JsonParser::parseString);
91 public static List<JsonObject> getAsJsonObjects(List<String> messages) {
92 return getAsJsonElements(messages).map(JsonElement::getAsJsonObject);
95 public static List<JsonPrimitive> getAsJsonPrimitives(List<String> messages) {
96 return getAsJsonElements(messages).map(JsonElement::getAsJsonPrimitive);
99 public static JsonObject getAsJsonObject(String item) {
100 return new Gson().fromJson(item, JsonObject.class);
103 public static Flux<JsonElement> plainBatch(List<String> messages) {
104 return Flux.fromIterable(getAsJsonElements(messages));
107 public static Flux<JsonObject> jsonBatch(List<String> messages) {
108 return Flux.fromIterable(getAsJsonObjects(messages));
111 public static MessageRouterSubscribeResponse errorSubscribeResponse(String failReasonFormat, Object... formatArgs) {
112 return ImmutableMessageRouterSubscribeResponse
114 .failReason(String.format(failReasonFormat, formatArgs))
118 public static MessageRouterSubscribeResponse successSubscribeResponse(List<JsonElement> items) {
119 return ImmutableMessageRouterSubscribeResponse
125 public static MessageRouterPublishResponse errorPublishResponse(String failReasonFormat, Object... formatArgs) {
126 String failReason = formatArgs.length == 0 ? failReasonFormat : String.format(failReasonFormat, formatArgs);
127 return ImmutableMessageRouterPublishResponse
129 .failReason(failReason)
133 public static MessageRouterPublishResponse successPublishResponse(List<JsonElement> items) {
134 return ImmutableMessageRouterPublishResponse
140 public static void registerTopic(MessageRouterPublisher publisher, MessageRouterPublishRequest publishRequest,
141 MessageRouterSubscriber subscriber, MessageRouterSubscribeRequest subscribeRequest) {
142 final List<String> sampleJsonMessages = List.of("{\"message\":\"message1\"}",
143 "{\"differentMessage\":\"message2\"}");
144 final Flux<JsonObject> jsonMessageBatch = MessageRouterTestsUtils.jsonBatch(sampleJsonMessages);
146 publisher.put(publishRequest, jsonMessageBatch).blockLast();
147 subscriber.get(subscribeRequest).block();
150 private static ImmutableMessageRouterSink createMessageRouterSink(String topicUrl) {
151 return ImmutableMessageRouterSink.builder()