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.model.streams.dmaap.MessageRouterSink;
32 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api.MessageRouterPublisher;
33 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api.MessageRouterSubscriber;
34 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterPublishRequest;
35 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterPublishResponse;
36 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterSubscribeRequest;
37 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterSubscribeResponse;
38 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishRequest;
39 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishResponse;
40 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterSubscribeRequest;
41 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterSubscribeResponse;
42 import reactor.core.publisher.Flux;
45 public final class MessageRouterTestsUtils {
46 private MessageRouterTestsUtils() {}
48 public static MessageRouterPublishRequest createPublishRequest(String topicUrl){
49 return createPublishRequest(topicUrl, ContentType.APPLICATION_JSON);
52 public static MessageRouterPublishRequest createPublishRequest(String topicUrl, ContentType contentType){
53 MessageRouterSink sinkDefinition = ImmutableMessageRouterSink.builder()
58 return ImmutableMessageRouterPublishRequest.builder()
59 .sinkDefinition(sinkDefinition)
60 .contentType(contentType)
64 public static MessageRouterSubscribeRequest createMRSubscribeRequest(String topicUrl,
65 String consumerGroup, String consumerId) {
66 ImmutableMessageRouterSource sourceDefinition = ImmutableMessageRouterSource.builder()
71 return ImmutableMessageRouterSubscribeRequest
73 .sourceDefinition(sourceDefinition)
74 .consumerGroup(consumerGroup)
75 .consumerId(consumerId)
79 public static List<JsonElement> getAsJsonElements(List<String> messages){
80 return messages.map(JsonParser::parseString);
83 public static List<JsonObject> getAsJsonObjects(List<String> messages){
84 return getAsJsonElements(messages).map(JsonElement::getAsJsonObject);
87 public static List<JsonPrimitive> getAsJsonPrimitives(List<String> messages){
88 return getAsJsonElements(messages).map(JsonElement::getAsJsonPrimitive);
91 public static JsonObject getAsJsonObject(String item){
92 return new Gson().fromJson(item, JsonObject.class);
95 public static Flux<JsonElement> plainBatch(List<String> messages){
96 return Flux.fromIterable(getAsJsonElements(messages));
99 public static Flux<JsonObject> jsonBatch(List<String> messages){
100 return Flux.fromIterable(getAsJsonObjects(messages));
103 public static MessageRouterSubscribeResponse errorSubscribeResponse(String failReasonFormat, Object... formatArgs){
104 return ImmutableMessageRouterSubscribeResponse
106 .failReason(String.format(failReasonFormat, formatArgs))
110 public static MessageRouterSubscribeResponse successSubscribeResponse(List<JsonElement> items){
111 return ImmutableMessageRouterSubscribeResponse
117 public static MessageRouterPublishResponse errorPublishResponse(String failReasonFormat, Object... formatArgs){
118 return ImmutableMessageRouterPublishResponse
120 .failReason(String.format(failReasonFormat, formatArgs))
124 public static MessageRouterPublishResponse successPublishResponse(List<JsonElement> items){
125 return ImmutableMessageRouterPublishResponse
131 public static void registerTopic(MessageRouterPublisher publisher, MessageRouterPublishRequest publishRequest,
132 MessageRouterSubscriber subscriber, MessageRouterSubscribeRequest subscribeRequest) {
133 final List<String> sampleJsonMessages = List.of("{\"message\":\"message1\"}",
134 "{\"differentMessage\":\"message2\"}");
135 final Flux<JsonObject> jsonMessageBatch = MessageRouterTestsUtils.jsonBatch(sampleJsonMessages);
137 publisher.put(publishRequest, jsonMessageBatch).blockLast();
138 subscriber.get(subscribeRequest).block();