Enhance dmaap-client v2

* include sent messages in dmaap mr response
* split publisher and subscriber
* write some unit tests

Issue-ID: DCAEGEN2-1421
Change-Id: Ie71e9344efd4a520406b87923413a45d51c28225
Signed-off-by: Piotr Jaszczyk <piotr.jaszczyk@nokia.com>

10 files changed, 397 insertions, 118 deletions

diff --git a/rest-services/common-dependency/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/RxHttpClient.java b/rest-services/common-dependency/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/RxHttpClient.java
index 709f5e59..3c010ef2 100644
--- a/rest-services/common-dependency/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/RxHttpClient.java
+++ b/rest-services/common-dependency/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/RxHttpClient.java
@@ -22,6 +22,7 @@ package org.onap.dcaegen2.services.sdk.rest.services.adapters.http;
 import io.netty.handler.ssl.SslContext;
 import io.vavr.collection.Stream;
 import java.util.stream.Collectors;
+import org.jetbrains.annotations.NotNull;
 import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnosticContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -44,7 +45,8 @@ public class RxHttpClient {
     }
 
     // TODO: hide netty from public api (io.netty.handler.ssl.SslContext)
-    public static RxHttpClient create(SslContext sslContext) {
+    public static RxHttpClient create(@NotNull
+            SslContext sslContext) {
         return new RxHttpClient(HttpClient.create().secure(sslContextSpec -> sslContextSpec.sslContext(sslContext)));
     }
 
diff --git a/rest-services/dmaap-client/pom.xml b/rest-services/dmaap-client/pom.xml
index 703d337a..c465cef4 100644
--- a/rest-services/dmaap-client/pom.xml
+++ b/rest-services/dmaap-client/pom.xml
@@ -46,7 +46,6 @@
     <dependency>
       <groupId>org.junit.jupiter</groupId>
       <artifactId>junit-jupiter-engine</artifactId>
-      <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>io.projectreactor</groupId>
@@ -56,7 +55,10 @@
     <dependency>
       <groupId>org.mockito</groupId>
       <artifactId>mockito-core</artifactId>
-      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.assertj</groupId>
+      <artifactId>assertj-core</artifactId>
     </dependency>
     <dependency>
       <groupId>org.onap.dcaegen2.services.sdk.rest.services</groupId>
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 7eb72f7c..0ac2d0bd 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
@@ -21,11 +21,15 @@
 package org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api;
 
 import com.google.gson.Gson;
+import io.netty.handler.ssl.SslContext;
 import io.vavr.Lazy;
+import java.time.Duration;
 import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.RxHttpClient;
 import org.onap.dcaegen2.services.sdk.rest.services.annotations.ExperimentalApi;
-import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.impl.MessageRouterClientImpl;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.impl.MessageRouterPublisherImpl;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.impl.MessageRouterSubscriberImpl;
 
 /**
  * <b>WARNING</b>: This is a proof-of-concept. It is untested. API may change or be removed.  Use at your own risk.
@@ -37,17 +41,33 @@ import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.impl.MessageRou
 @ExperimentalApi
 public final class DmaapClientFactory {
 
-    private static final Lazy<MessageRouterClientImpl> THE_CLIENT = Lazy.of(() ->
-            new MessageRouterClientImpl(RxHttpClient.create(), new Gson()));
+    private static final Duration DEFAULT_MAX_BATCH_DURATION = Duration.ofSeconds(1);
+    private static final int DEFAULT_MAX_BATCH_SIZE = 512;
 
     private DmaapClientFactory() {
     }
 
     public static @NotNull MessageRouterPublisher createMessageRouterPublisher() {
-        return THE_CLIENT.get();
+        return new MessageRouterPublisherImpl(
+                RxHttpClient.create(),
+                DEFAULT_MAX_BATCH_SIZE,
+                DEFAULT_MAX_BATCH_DURATION);
+    }
+
+    public static @NotNull MessageRouterPublisher createMessageRouterPublisher(@NotNull SslContext sslContext) {
+        return new MessageRouterPublisherImpl(
+                RxHttpClient.create(sslContext),
+                DEFAULT_MAX_BATCH_SIZE,
+                DEFAULT_MAX_BATCH_DURATION);
     }
 
     public static @NotNull MessageRouterSubscriber createMessageRouterSubscriber() {
-        return THE_CLIENT.get();
+        return new MessageRouterSubscriberImpl(RxHttpClient.create(), new Gson());
+    }
+
+    public static @NotNull MessageRouterSubscriber createMessageRouterSubscriber(@NotNull SslContext sslContext) {
+        return new MessageRouterSubscriberImpl(
+                RxHttpClient.create(sslContext),
+                new Gson());
     }
 }
diff --git a/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/Commons.java b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/Commons.java
new file mode 100644
index 00000000..5211807a
--- /dev/null
+++ b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/Commons.java
@@ -0,0 +1,38 @@
+/*
+ * ============LICENSE_START====================================
+ * DCAEGEN2-SERVICES-SDK
+ * =========================================================
+ * Copyright (C) 2019 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.dcaegen2.services.sdk.rest.services.dmaap.client.impl;
+
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpResponse;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ * @since April 2019
+ */
+final class Commons {
+
+    private Commons() {
+    }
+
+    static String extractFailReason(HttpResponse httpResponse) {
+        return String.format("%d %s%n%s", httpResponse.statusCode(), httpResponse.statusReason(),
+                httpResponse.bodyAsString());
+    }
+}
diff --git a/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterPublisherImpl.java b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterPublisherImpl.java
new file mode 100644
index 00000000..f09c5397
--- /dev/null
+++ b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterPublisherImpl.java
@@ -0,0 +1,101 @@
+/*
+ * ============LICENSE_START====================================
+ * DCAEGEN2-SERVICES-SDK
+ * =========================================================
+ * Copyright (C) 2019 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.dcaegen2.services.sdk.rest.services.dmaap.client.impl;
+
+import static org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.impl.Commons.extractFailReason;
+
+import com.google.gson.JsonArray;
+import com.google.gson.JsonElement;
+import io.vavr.collection.HashMap;
+import io.vavr.collection.List;
+import java.time.Duration;
+import org.jetbrains.annotations.NotNull;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpHeaders;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpMethod;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpRequest;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpResponse;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.ImmutableHttpRequest;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.RequestBody;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.RxHttpClient;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api.MessageRouterPublisher;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterPublishResponse;
+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 org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ * @since March 2019
+ */
+// TODO: This is a PoC. It's untested.
+public class MessageRouterPublisherImpl implements MessageRouterPublisher {
+    private final RxHttpClient httpClient;
+    private final int maxBatchSize;
+    private final Duration maxBatchDuration;
+
+    public MessageRouterPublisherImpl(RxHttpClient httpClient, int maxBatchSize, Duration maxBatchDuration) {
+        this.httpClient = httpClient;
+        this.maxBatchSize = maxBatchSize;
+        this.maxBatchDuration = maxBatchDuration;
+    }
+
+    @Override
+    public Flux<MessageRouterPublishResponse> put(
+            MessageRouterPublishRequest request,
+            Flux<? extends JsonElement> items) {
+        return items.bufferTimeout(maxBatchSize, maxBatchDuration)
+                .flatMap(subItems -> subItems.isEmpty() ? Mono.empty() : pushBatchToMr(request, List.ofAll(subItems)));
+    }
+
+    private Publisher<? extends MessageRouterPublishResponse> pushBatchToMr(
+            MessageRouterPublishRequest request,
+            List<JsonElement> batch) {
+        return httpClient.call(buildHttpRequest(request, asJsonBody(batch)))
+                .map(httpResponse -> buildResponse(httpResponse, batch));
+    }
+
+    private @NotNull RequestBody asJsonBody(List<? extends JsonElement> subItems) {
+        final JsonArray elements = new JsonArray(subItems.size());
+        subItems.forEach(elements::add);
+        return RequestBody.fromJson(elements);
+    }
+
+    private @NotNull HttpRequest buildHttpRequest(MessageRouterPublishRequest request, RequestBody body) {
+        return ImmutableHttpRequest.builder()
+                .method(HttpMethod.POST)
+                .url(request.sinkDefinition().topicUrl())
+                .diagnosticContext(request.diagnosticContext())
+                .customHeaders(HashMap.of(HttpHeaders.CONTENT_TYPE, request.contentType()))
+                .body(body)
+                .build();
+    }
+
+    private MessageRouterPublishResponse buildResponse(
+            HttpResponse httpResponse, List<JsonElement> batch) {
+        final ImmutableMessageRouterPublishResponse.Builder builder =
+                ImmutableMessageRouterPublishResponse.builder();
+        return httpResponse.successful()
+                ? builder.items(batch).build()
+                : builder.failReason(extractFailReason(httpResponse)).build();
+    }
+}
diff --git a/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterClientImpl.java b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterSubscriberImpl.java
index 0ef06201..e91a77fa 100644
--- a/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterClientImpl.java
+++ b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterSubscriberImpl.java
@@ -20,31 +20,21 @@
 
 package org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.impl;
 
+import static org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.impl.Commons.extractFailReason;
+
 import com.google.gson.Gson;
 import com.google.gson.JsonArray;
-import com.google.gson.JsonElement;
-import io.netty.buffer.ByteBuf;
-import io.vavr.collection.HashMap;
 import java.nio.charset.StandardCharsets;
-import java.time.Duration;
 import org.jetbrains.annotations.NotNull;
-import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpHeaders;
 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpMethod;
 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpRequest;
 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpResponse;
 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.ImmutableHttpRequest;
-import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.RequestBody;
 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.RxHttpClient;
-import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api.MessageRouterPublisher;
 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api.MessageRouterSubscriber;
-import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterPublishResponse;
 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterSubscribeResponse;
-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 org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterSubscribeRequest;
 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterSubscribeResponse;
-import org.reactivestreams.Publisher;
-import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 /**
@@ -52,42 +42,20 @@ import reactor.core.publisher.Mono;
  * @since March 2019
  */
 // TODO: This is a PoC. It's untested.
-public class MessageRouterClientImpl implements MessageRouterPublisher, MessageRouterSubscriber {
-
-    private static final Duration WINDOW_MAX_TIME = Duration.ofSeconds(1);
-    private static final int WINDOW_MAX_SIZE = 512;
+public class MessageRouterSubscriberImpl implements MessageRouterSubscriber {
     private final RxHttpClient httpClient;
     private final Gson gson;
 
-    public MessageRouterClientImpl(RxHttpClient httpClient, Gson gson) {
+    public MessageRouterSubscriberImpl(RxHttpClient httpClient, Gson gson) {
         this.httpClient = httpClient;
         this.gson = gson;
     }
 
     @Override
-    public Flux<MessageRouterPublishResponse> put(
-            MessageRouterPublishRequest request,
-            Flux<? extends JsonElement> items) {
-        return items.windowTimeout(WINDOW_MAX_SIZE, WINDOW_MAX_TIME).flatMap(subItems ->
-                subItems.collect(JsonArray::new, JsonArray::add)
-                        .filter(arr -> arr.size() > 0)
-                        .map(RequestBody::fromJson)
-                        .flatMap(body -> httpClient.call(buildPostHttpRequest(request, body)))
-                        .map(this::buildPutResponse));
-    }
-
-    @Override
     public Mono<MessageRouterSubscribeResponse> get(MessageRouterSubscribeRequest request) {
         return httpClient.call(buildGetHttpRequest(request)).map(this::buildGetResponse);
     }
 
-    private @NotNull MessageRouterPublishResponse buildPutResponse(HttpResponse httpResponse) {
-        final ImmutableMessageRouterPublishResponse.Builder builder =
-                ImmutableMessageRouterPublishResponse.builder();
-        return httpResponse.successful()
-                ? builder.build()
-                : builder.failReason(extractFailReason(httpResponse)).build();
-    }
 
     private @NotNull MessageRouterSubscribeResponse buildGetResponse(HttpResponse httpResponse) {
         final ImmutableMessageRouterSubscribeResponse.Builder builder =
@@ -97,20 +65,6 @@ public class MessageRouterClientImpl implements MessageRouterPublisher, MessageR
                 : builder.failReason(extractFailReason(httpResponse)).build();
     }
 
-    private String extractFailReason(HttpResponse httpResponse) {
-        return String.format("%d %s%n%s", httpResponse.statusCode(), httpResponse.statusReason(),
-                httpResponse.bodyAsString());
-    }
-
-    private @NotNull HttpRequest buildPostHttpRequest(MessageRouterPublishRequest request, RequestBody body) {
-        return ImmutableHttpRequest.builder()
-                .method(HttpMethod.POST)
-                .url(request.sinkDefinition().topicUrl())
-                .diagnosticContext(request.diagnosticContext())
-                .customHeaders(HashMap.of(HttpHeaders.CONTENT_TYPE, request.contentType()))
-                .body(body)
-                .build();
-    }
 
     private @NotNull HttpRequest buildGetHttpRequest(MessageRouterSubscribeRequest request) {
         return ImmutableHttpRequest.builder()
@@ -121,6 +75,7 @@ public class MessageRouterClientImpl implements MessageRouterPublisher, MessageR
     }
 
     private String buildSubscribeUrl(MessageRouterSubscribeRequest request) {
-        return String.format("%s/%s/%s", request.sourceDefinition().topicUrl(), request.consumerGroup(), request.consumerId());
+        return String.format("%s/%s/%s", request.sourceDefinition().topicUrl(), request.consumerGroup(),
+                request.consumerId());
     }
 }
diff --git a/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/model/MessageRouterPublishResponse.java b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/model/MessageRouterPublishResponse.java
index 62175e09..cc038a6e 100644
--- a/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/model/MessageRouterPublishResponse.java
+++ b/rest-services/dmaap-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/model/MessageRouterPublishResponse.java
@@ -21,6 +21,8 @@
 package org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model;
 
 
+import com.google.gson.JsonElement;
+import io.vavr.collection.List;
 import org.immutables.value.Value;
 import org.onap.dcaegen2.services.sdk.rest.services.annotations.ExperimentalApi;
 
@@ -32,4 +34,8 @@ import org.onap.dcaegen2.services.sdk.rest.services.annotations.ExperimentalApi;
 @Value.Immutable
 public interface MessageRouterPublishResponse extends DmaapResponse {
 
+    @Value.Default
+    default List<JsonElement> items() {
+        return List.empty();
+    }
 }
diff --git a/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterPublisherTest.java b/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterPublisherTest.java
deleted file mode 100644
index 9656ae86..00000000
--- a/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterPublisherTest.java
+++ /dev/null
@@ -1,57 +0,0 @@
-/*
- * ============LICENSE_START====================================
- * DCAEGEN2-SERVICES-SDK
- * =========================================================
- * Copyright (C) 2019 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.dcaegen2.services.sdk.rest.services.dmaap.client.api;
-
-import static org.mockito.Mockito.mock;
-
-import com.google.gson.JsonPrimitive;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.Disabled;
-import org.onap.dcaegen2.services.sdk.model.streams.dmaap.MessageRouterSink;
-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 reactor.core.publisher.Flux;
-
-/**
- * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
- * @since March 2019
- */
-@Disabled
-class MessageRouterPublisherTest {
-
-    private final MessageRouterPublisher cut = mock(MessageRouterPublisher.class);
-    private final MessageRouterSink sinkDefinition = mock(MessageRouterSink.class);
-    private final MessageRouterPublishRequest request = ImmutableMessageRouterPublishRequest.builder()
-            .sinkDefinition(sinkDefinition).build();
-
-    @Test
-    void apiShouldBeUsableWithTransform() {
-        Flux.just(1, 2, 3)
-                .map(JsonPrimitive::new)
-                .transform(input -> cut.put(request, input));
-    }
-
-    @Test
-    void apiShouldBeUsableWithSingleCall() {
-        final Flux<JsonPrimitive> input = Flux.just(1, 2, 3).map(JsonPrimitive::new);
-        cut.put(request, input);
-    }
-}
\ No newline at end of file
diff --git a/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterPublisherImplTest.java b/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterPublisherImplTest.java
new file mode 100644
index 00000000..103b4807
--- /dev/null
+++ b/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterPublisherImplTest.java
@@ -0,0 +1,210 @@
+/*
+ * ============LICENSE_START====================================
+ * DCAEGEN2-SERVICES-SDK
+ * =========================================================
+ * Copyright (C) 2019 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.dcaegen2.services.sdk.rest.services.dmaap.client.impl;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import com.google.gson.Gson;
+import com.google.gson.JsonArray;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonPrimitive;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.CompositeByteBuf;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.List;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.verification.VerificationMode;
+import org.onap.dcaegen2.services.sdk.model.streams.dmaap.ImmutableMessageRouterSink;
+import org.onap.dcaegen2.services.sdk.model.streams.dmaap.MessageRouterSink;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpMethod;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpRequest;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpResponse;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.ImmutableHttpResponse;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.RxHttpClient;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api.MessageRouterPublisher;
+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.MessageRouterPublishResponse;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ * @since April 2019
+ */
+class MessageRouterPublisherImplTest {
+
+    private static final Duration TIMEOUT = Duration.ofSeconds(5);
+    private final RxHttpClient httpClient = mock(RxHttpClient.class);
+    private final MessageRouterPublisher cut = new MessageRouterPublisherImpl(httpClient, 3, Duration.ofMinutes(1));
+
+    private final ArgumentCaptor<HttpRequest> httpRequestArgumentCaptor = ArgumentCaptor.forClass(HttpRequest.class);
+    private final MessageRouterSink sinkDefinition = ImmutableMessageRouterSink.builder()
+            .name("the topic")
+            .topicUrl("https://dmaap-mr/TOPIC")
+            .build();
+    private final MessageRouterPublishRequest mrRequest = ImmutableMessageRouterPublishRequest.builder()
+            .sinkDefinition(sinkDefinition)
+            .build();
+    private final HttpResponse httpResponse = ImmutableHttpResponse.builder()
+            .statusCode(200)
+            .statusReason("OK")
+            .url(sinkDefinition.topicUrl())
+            .rawBody("{}".getBytes())
+            .build();
+
+    @Test
+    void puttingElementsShouldYieldNonChunkedHttpRequest() {
+        // given
+        given(httpClient.call(any(HttpRequest.class))).willReturn(Mono.just(httpResponse));
+
+        // when
+        final Flux<MessageRouterPublishResponse> responses = cut
+                .put(mrRequest, Flux.just("I", "like", "cookies").map(JsonPrimitive::new));
+        responses.then().block();
+
+        // then
+        verify(httpClient).call(httpRequestArgumentCaptor.capture());
+        final HttpRequest httpRequest = httpRequestArgumentCaptor.getValue();
+        assertThat(httpRequest.method()).isEqualTo(HttpMethod.POST);
+        assertThat(httpRequest.url()).isEqualTo(sinkDefinition.topicUrl());
+        assertThat(httpRequest.body()).isNotNull();
+        assertThat(httpRequest.body().length()).isGreaterThan(0);
+    }
+
+    @Test
+    void puttingLowNumberOfElementsShouldYieldSingleHttpRequest() {
+        // given
+        given(httpClient.call(any(HttpRequest.class))).willReturn(Mono.just(httpResponse));
+
+        // when
+        final Flux<MessageRouterPublishResponse> responses = cut
+                .put(mrRequest, Flux.just("I", "like", "cookies").map(JsonPrimitive::new));
+        responses.then().block();
+
+        // then
+        verify(httpClient).call(httpRequestArgumentCaptor.capture());
+        final HttpRequest httpRequest = httpRequestArgumentCaptor.getValue();
+        final JsonArray elementsInRequest = extractNonEmptyRequestBody(httpRequest);
+        assertThat(elementsInRequest.size()).isEqualTo(3);
+        assertThat(elementsInRequest.get(0).getAsString()).isEqualTo("I");
+        assertThat(elementsInRequest.get(1).getAsString()).isEqualTo("like");
+        assertThat(elementsInRequest.get(2).getAsString()).isEqualTo("cookies");
+    }
+
+    @Test
+    void puttingLowNumberOfElementsShouldReturnSingleResponse() {
+        // given
+        given(httpClient.call(any(HttpRequest.class))).willReturn(Mono.just(httpResponse));
+
+        // when
+        final Flux<MessageRouterPublishResponse> responses = cut
+                .put(mrRequest, Flux.just("I", "like", "cookies").map(JsonPrimitive::new));
+
+        // then
+        StepVerifier.create(responses)
+                .consumeNextWith(response -> {
+                    assertThat(response.successful()).describedAs("successful").isTrue();
+                    assertThat(response.items()).containsExactly(
+                            new JsonPrimitive("I"),
+                            new JsonPrimitive("like"),
+                            new JsonPrimitive("cookies"));
+                })
+                .expectComplete()
+                .verify(TIMEOUT);
+    }
+
+
+    @Test
+    void puttingHighNumberOfElementsShouldYieldMultipleHttpRequests() {
+        // given
+        given(httpClient.call(any(HttpRequest.class))).willReturn(Mono.just(httpResponse));
+
+        // when
+        final Flux<MessageRouterPublishResponse> responses = cut
+                .put(mrRequest, Flux.just("I", "like", "cookies", "and", "pierogi").map(JsonPrimitive::new));
+
+        // then
+        responses.then().block();
+
+        verify(httpClient, times(2)).call(httpRequestArgumentCaptor.capture());
+        final List<HttpRequest> httpRequests = httpRequestArgumentCaptor.getAllValues();
+        assertThat(httpRequests.size()).describedAs("number of requests").isEqualTo(2);
+
+        final JsonArray firstRequest = extractNonEmptyRequestBody(httpRequests.get(0));
+        assertThat(firstRequest.size()).isEqualTo(3);
+        assertThat(firstRequest.get(0).getAsString()).isEqualTo("I");
+        assertThat(firstRequest.get(1).getAsString()).isEqualTo("like");
+        assertThat(firstRequest.get(2).getAsString()).isEqualTo("cookies");
+
+        final JsonArray secondRequest = extractNonEmptyRequestBody(httpRequests.get(1));
+        assertThat(secondRequest.size()).isEqualTo(2);
+        assertThat(secondRequest.get(0).getAsString()).isEqualTo("and");
+        assertThat(secondRequest.get(1).getAsString()).isEqualTo("pierogi");
+    }
+
+    @Test
+    void puttingHighNumberOfElementsShouldReturnMoreResponses() {
+        // given
+        given(httpClient.call(any(HttpRequest.class))).willReturn(Mono.just(httpResponse));
+
+        // when
+        final Flux<MessageRouterPublishResponse> responses = cut
+                .put(mrRequest, Flux.just("I", "like", "cookies", "and", "pierogi").map(JsonPrimitive::new));
+
+        // then
+        StepVerifier.create(responses)
+                .consumeNextWith(response -> {
+                    assertThat(response.successful()).describedAs("successful").isTrue();
+                    assertThat(response.items()).containsExactly(
+                            new JsonPrimitive("I"),
+                            new JsonPrimitive("like"),
+                            new JsonPrimitive("cookies"));
+                })
+                .consumeNextWith(response -> {
+                    assertThat(response.successful()).describedAs("successful").isTrue();
+                    assertThat(response.items()).containsExactly(
+                            new JsonPrimitive("and"),
+                            new JsonPrimitive("pierogi"));
+                })
+                .expectComplete()
+                .verify(TIMEOUT);
+    }
+
+    private JsonArray extractNonEmptyRequestBody(HttpRequest httpRequest) {
+        final String body = Flux.from(httpRequest.body().contents())
+                .collect(ByteBufAllocator.DEFAULT::compositeBuffer,
+                        (byteBufs, buffer) -> byteBufs.addComponent(true, buffer))
+                .map(byteBufs -> byteBufs.toString(StandardCharsets.UTF_8))
+                .block();
+        assertThat(body).describedAs("request body").isNotBlank();
+        return new Gson().fromJson(body, JsonArray.class);
+    }
+}
\ No newline at end of file
diff --git a/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterSubscriberTest.java b/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterSubscriberImplTest.java
index b8bcde92..c9587206 100644
--- a/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterSubscriberTest.java
+++ b/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/impl/MessageRouterSubscriberImplTest.java
@@ -18,7 +18,7 @@
  * ============LICENSE_END=====================================
  */
 
-package org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api;
+package org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.impl;
 
 import static org.mockito.Mockito.mock;
 
@@ -27,6 +27,7 @@ import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.onap.dcaegen2.services.sdk.model.streams.dmaap.MessageRouterSink;
 import org.onap.dcaegen2.services.sdk.model.streams.dmaap.MessageRouterSource;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api.MessageRouterSubscriber;
 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.DmaapResponse;
 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterPublishRequest;
 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterSubscribeRequest;
@@ -39,8 +40,9 @@ import reactor.core.publisher.Flux;
  * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
  * @since March 2019
  */
+// TODO: Write proper unit tests
 @Disabled
-class MessageRouterSubscriberTest {
+class MessageRouterSubscriberImplTest {
 
     private final MessageRouterSubscriber cut = mock(MessageRouterSubscriber.class);
     private final MessageRouterSource sinkDefinition = mock(MessageRouterSource.class);