Implement StreamParser

Change-Id: I3ee50ad70bb107fa39f227af7dab67948b7dbe4b
Issue-ID: DCAEGEN2-1342
Signed-off-by: kjaniak <kornel.janiak@nokia.com>

4 files changed, 442 insertions, 0 deletions

diff --git a/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/streams/gson/DataRouterSinkParserTest.java b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/streams/gson/DataRouterSinkParserTest.java
new file mode 100644
index 00000000..398ebcd9
--- /dev/null
+++ b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/streams/gson/DataRouterSinkParserTest.java
@@ -0,0 +1,110 @@
+/*
+ * ============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.cbs.client.impl.streams.gson;
+
+import com.google.gson.Gson;
+import com.google.gson.JsonObject;
+import io.vavr.control.Either;
+import org.junit.jupiter.api.Test;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.exceptions.StreamParserError;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.streams.StreamFromGsonParser;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.streams.StreamFromGsonParsers;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.streams.dmaap.DataRouterSink;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.streams.dmaap.ImmutableDataRouterSink;
+
+import java.io.IOException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl.streams.gson.StreamsConstants.DATA_ROUTER_TYPE;
+import static org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl.streams.gson.StreamsConstants.MESSAGE_ROUTER_TYPE;
+
+
+class DataRouterSinkParserTest {
+    private static final String SAMPLE_LOCATION = "mtc00";
+    private static final String SAMPLE_PUBLISH_URL = "https://we-are-data-router.us/feed/xyz";
+    private static final String SAMPLE_LOG_URL = "https://we-are-data-router.us/feed/xyz/logs";
+    private static final String SAMPLE_USER = "some-user";
+    private static final String SAMPLE_PASSWORD = "some-password";
+    private static final String SAMPLE_PUBLISHER_ID = "123456";
+
+    private static final Gson gson = new Gson();
+
+    private final StreamFromGsonParser<DataRouterSink> streamParser = StreamFromGsonParsers.dataRouterSinkParser();
+
+    private static final DataRouterSink fullConfigurationStream = ImmutableDataRouterSink.builder()
+            .location(SAMPLE_LOCATION)
+            .publishUrl(SAMPLE_PUBLISH_URL)
+            .logUrl(SAMPLE_LOG_URL)
+            .username(SAMPLE_USER)
+            .password(SAMPLE_PASSWORD)
+            .publisherId(SAMPLE_PUBLISHER_ID)
+            .build();
+
+    private static final DataRouterSink minimalConfigurationStream = ImmutableDataRouterSink.builder()
+            .publishUrl(SAMPLE_PUBLISH_URL)
+            .build();
+
+    @Test
+    void fullConfiguration_shouldGenerateDataRouterSinkObject() throws IOException {
+        // given
+        JsonObject input = GsonUtils.readObjectFromResource("/streams/data_router_sink_full.json");
+        // when
+        DataRouterSink result = streamParser.unsafeParse(input);
+        // then
+        assertThat(result).isEqualTo(fullConfigurationStream);
+    }
+
+    @Test
+    void minimalConfiguration_shouldGenerateDataRouterSinkObject() throws IOException {
+        //given
+        JsonObject input = GsonUtils.readObjectFromResource("/streams/data_router_sink_minimal.json");
+        // when
+        DataRouterSink result = streamParser.unsafeParse(input);
+        // then
+        assertThat(result).isEqualTo(minimalConfigurationStream);
+    }
+
+    @Test
+    void incorrectConfiguration_shouldParseToStreamParserError() throws IOException {
+        // given
+        JsonObject input = GsonUtils.readObjectFromResource("/streams/message_router_full.json");
+        // when
+        Either<StreamParserError, DataRouterSink> result = streamParser.parse(input);
+        // then
+        assertThat(result.getLeft()).isInstanceOf(StreamParserError.class);
+        result.peekLeft(error -> {
+                    assertThat(error.message()).contains("Invalid stream type");
+                    assertThat(error.message()).contains("Expected '" + DATA_ROUTER_TYPE + "', but was '"
+                            + MESSAGE_ROUTER_TYPE + "'");
+                }
+        );
+    }
+
+    @Test
+    void emptyConfiguration_shouldParseToStreamParserError() {
+        // given
+        JsonObject input = gson.fromJson("{}", JsonObject.class);
+        // when
+        Either<StreamParserError, DataRouterSink> result = streamParser.parse(input);
+        // then
+        assertThat(result.getLeft()).isInstanceOf(StreamParserError.class);
+    }
+
+}
\ No newline at end of file
diff --git a/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/streams/gson/DataRouterSourceParserTest.java b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/streams/gson/DataRouterSourceParserTest.java
new file mode 100644
index 00000000..681fa147
--- /dev/null
+++ b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/streams/gson/DataRouterSourceParserTest.java
@@ -0,0 +1,108 @@
+/*
+ * ============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.cbs.client.impl.streams.gson;
+
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import com.google.gson.JsonObject;
+import io.vavr.control.Either;
+import org.junit.jupiter.api.Test;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.exceptions.StreamParserError;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.streams.StreamFromGsonParser;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.streams.StreamFromGsonParsers;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.streams.StreamParser;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.streams.DataStream;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.streams.dmaap.*;
+
+import java.io.IOException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl.streams.gson.StreamsConstants.DATA_ROUTER_TYPE;
+import static org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl.streams.gson.StreamsConstants.MESSAGE_ROUTER_TYPE;
+
+public class DataRouterSourceParserTest {
+    private static final String SAMPLE_LOCATION = "mtc00";
+    private static final String SAMPLE_DELIVERY_URL = "https://my-subscriber-app.dcae:8080/target-path";
+    private static final String SAMPLE_USER = "some-user";
+    private static final String SAMPLE_PASSWORD = "some-password";
+    private static final String SAMPLE_SUBSCRIBER_ID = "789012";
+
+    private static final Gson gson = new Gson();
+
+    private static final DataRouterSource fullConfigurationStream = ImmutableDataRouterSource.builder()
+            .location(SAMPLE_LOCATION)
+            .deliveryUrl(SAMPLE_DELIVERY_URL)
+            .username(SAMPLE_USER)
+            .password(SAMPLE_PASSWORD)
+            .subscriberId(SAMPLE_SUBSCRIBER_ID)
+            .build();
+
+    private static final DataRouterSource minimalConfigurationStream = ImmutableDataRouterSource.builder()
+            .build();
+
+
+    private final StreamFromGsonParser<DataRouterSource> streamParser = StreamFromGsonParsers.dataRouterSourceParser();
+
+    @Test
+    void fullConfiguration_shouldGenerateDataRouterSinkObject() throws IOException {
+        // given
+        JsonObject input = GsonUtils.readObjectFromResource("/streams/data_router_source_full.json");
+        // when
+        DataRouterSource result = streamParser.unsafeParse(input);
+        // then
+        assertThat(result).isEqualTo(fullConfigurationStream);
+    }
+
+    @Test
+    void minimalConfiguration_shouldGenerateDataRouterSinkObject() throws IOException {
+        // given
+        JsonObject input = GsonUtils.readObjectFromResource("/streams/data_router_source_minimal.json");
+        // when
+        DataRouterSource result = streamParser.unsafeParse(input);
+        // then
+        assertThat(result).isEqualTo(minimalConfigurationStream);
+    }
+
+    @Test
+    void incorrectConfiguration_shouldParseToStreamParserError() throws IOException {
+        // given
+        JsonObject input = GsonUtils.readObjectFromResource("/streams/message_router_full.json");
+        // when
+        Either<StreamParserError, DataRouterSource> result = streamParser.parse(input);
+        // then
+        assertThat(result.getLeft()).isInstanceOf(StreamParserError.class);
+        result.peekLeft(error -> {
+                    assertThat(error.message()).contains("Invalid stream type");
+                    assertThat(error.message()).contains("Expected '" + DATA_ROUTER_TYPE + "', but was '"
+                            + MESSAGE_ROUTER_TYPE + "'");
+                }
+        );
+    }
+
+    @Test
+    void emptyConfiguration_shouldBeParsedToStreamParserError() {
+        // given
+        JsonObject input = gson.fromJson("{}", JsonObject.class);
+        // when
+        Either<StreamParserError, DataRouterSource> result = streamParser.parse(input);
+        // then
+        assertThat(result.getLeft()).isInstanceOf(StreamParserError.class);
+    }
+}
diff --git a/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/streams/gson/MessageRouterSinkParserTest.java b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/streams/gson/MessageRouterSinkParserTest.java
new file mode 100644
index 00000000..63b04d1d
--- /dev/null
+++ b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/streams/gson/MessageRouterSinkParserTest.java
@@ -0,0 +1,113 @@
+/*
+ * ============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.cbs.client.impl.streams.gson;
+
+import com.google.gson.Gson;
+import com.google.gson.JsonObject;
+import io.vavr.control.Either;
+import org.junit.jupiter.api.Test;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.exceptions.StreamParserError;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.streams.StreamFromGsonParser;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.streams.StreamFromGsonParsers;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.streams.dmaap.DataRouterSink;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.streams.dmaap.MessageRouterSink;
+
+import java.io.IOException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl.streams.gson.StreamsConstants.DATA_ROUTER_TYPE;
+import static org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl.streams.gson.StreamsConstants.MESSAGE_ROUTER_TYPE;
+
+/**
+ * @author <a href="mailto:kornel.janiak@nokia.com">Kornel Janiak</a>
+ */
+
+public class MessageRouterSinkParserTest {
+
+    private static final String SAMPLE_AAF_USERNAME = "some-user";
+    private static final String SAMPLE_AAF_PASSWORD = "some-password";
+    private static final String SAMPLE_LOCATION = "mtc00";
+    private static final String SAMPLE_CLIENT_ROLE = "com.dcae.member";
+    private static final String SAMPLE_CLIENT_ID = "1500462518108";
+    private static final String SAMPLE_TOPIC_URL = "https://we-are-message-router.us:3905/events/some-topic";
+
+    private static final Gson gson = new Gson();
+
+    private final StreamFromGsonParser<MessageRouterSink> streamParser = StreamFromGsonParsers.messageRouterSinkParser();
+
+    @Test
+    void fullConfiguration_shouldGenerateDataRouterSinkObject() throws IOException {
+        // given
+        JsonObject input = GsonUtils.readObjectFromResource("/streams/message_router_full.json");
+        // when
+        MessageRouterSink result = streamParser.unsafeParse(input);
+        // then
+        assertThat(result).isInstanceOf(MessageRouterSink.class);
+        assertThat(result.aafCredentials().username()).isEqualTo(SAMPLE_AAF_USERNAME);
+        assertThat(result.aafCredentials().password()).isEqualTo(SAMPLE_AAF_PASSWORD);
+        assertThat(result.location()).isEqualTo(SAMPLE_LOCATION);
+        assertThat(result.clientRole()).isEqualTo(SAMPLE_CLIENT_ROLE);
+        assertThat(result.clientId()).isEqualTo(SAMPLE_CLIENT_ID);
+        assertThat(result.topicUrl()).isEqualTo(SAMPLE_TOPIC_URL);
+    }
+
+    @Test
+    void minimalConfiguration_shouldGenerateDataRouterSinkObject() throws IOException {
+        // given
+        JsonObject input = GsonUtils.readObjectFromResource("/streams/message_router_minimal.json");
+
+        // when
+        MessageRouterSink result = streamParser.unsafeParse(input);
+        // then
+        assertThat(result).isInstanceOf(MessageRouterSink.class);
+        assertThat(result.topicUrl()).isEqualTo(SAMPLE_TOPIC_URL);
+        assertThat(result.aafCredentials().username()).isNull();
+        assertThat(result.aafCredentials().password()).isNull();
+        assertThat(result.clientId()).isNull();
+    }
+
+    @Test
+    void incorrectConfiguration_shouldParseToStreamParserError() throws IOException {
+        // given
+        JsonObject input = GsonUtils.readObjectFromResource("/streams/data_router_sink_full.json");
+        // when
+        Either<StreamParserError, MessageRouterSink> result = streamParser.parse(input);
+        // then
+        assertThat(result.getLeft()).isInstanceOf(StreamParserError.class);
+        result.peekLeft(error -> {
+                    assertThat(error.message()).contains("Invalid stream type");
+                    assertThat(error.message()).contains("Expected '" + MESSAGE_ROUTER_TYPE + "', but was '"
+                            + DATA_ROUTER_TYPE + "'");
+                }
+        );
+    }
+
+    @Test
+    void emptyConfiguration_shouldParseToStreamParserError() {
+        // given
+        JsonObject input = gson.fromJson("{}", JsonObject.class);
+        // when
+        Either<StreamParserError, MessageRouterSink> result = streamParser.parse(input);
+        // then
+        assertThat(result.getLeft()).isInstanceOf(StreamParserError.class);
+    }
+
+
+}
\ No newline at end of file
diff --git a/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/streams/gson/MessageRouterSourceParserTest.java b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/streams/gson/MessageRouterSourceParserTest.java
new file mode 100644
index 00000000..fea63d66
--- /dev/null
+++ b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/streams/gson/MessageRouterSourceParserTest.java
@@ -0,0 +1,111 @@
+/*
+ * ============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.cbs.client.impl.streams.gson;
+
+import com.google.gson.Gson;
+import com.google.gson.JsonObject;
+import io.vavr.control.Either;
+import org.junit.jupiter.api.Test;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.exceptions.StreamParserError;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.streams.StreamFromGsonParser;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.streams.StreamFromGsonParsers;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.streams.dmaap.MessageRouterSink;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.streams.dmaap.MessageRouterSource;
+
+import java.io.IOException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl.streams.gson.StreamsConstants.DATA_ROUTER_TYPE;
+import static org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl.streams.gson.StreamsConstants.MESSAGE_ROUTER_TYPE;
+
+/**
+ * @author <a href="mailto:kornel.janiak@nokia.com">Kornel Janiak</a>
+ */
+
+public class MessageRouterSourceParserTest {
+
+    private static final String SAMPLE_AAF_USERNAME = "some-user";
+    private static final String SAMPLE_AAF_PASSWORD = "some-password";
+    private static final String SAMPLE_LOCATION = "mtc00";
+    private static final String SAMPLE_CLIENT_ROLE = "com.dcae.member";
+    private static final String SAMPLE_CLIENT_ID = "1500462518108";
+    private static final String SAMPLE_TOPIC_URL = "https://we-are-message-router.us:3905/events/some-topic";
+
+    private static final Gson gson = new Gson();
+
+    private final StreamFromGsonParser<MessageRouterSource> streamParser = StreamFromGsonParsers.messageRouterSourceParser();
+
+    @Test
+    void fullConfiguration_shouldGenerateDataRouterSourceObject() throws IOException {
+        // given
+        JsonObject input = GsonUtils.readObjectFromResource("/streams/message_router_full.json");
+        // when
+        MessageRouterSource result = streamParser.unsafeParse(input);
+        // then
+        assertThat(result.aafCredentials().username()).isEqualTo(SAMPLE_AAF_USERNAME);
+        assertThat(result.aafCredentials().password()).isEqualTo(SAMPLE_AAF_PASSWORD);
+        assertThat(result.location()).isEqualTo(SAMPLE_LOCATION);
+        assertThat(result.clientRole()).isEqualTo(SAMPLE_CLIENT_ROLE);
+        assertThat(result.clientId()).isEqualTo(SAMPLE_CLIENT_ID);
+        assertThat(result.topicUrl()).isEqualTo(SAMPLE_TOPIC_URL);
+    }
+
+    @Test
+    void minimalConfiguration_shouldGenerateDataRouterSourceObject() throws IOException {
+        // given
+        JsonObject input = GsonUtils.readObjectFromResource("/streams/message_router_minimal.json");
+
+        // when
+        MessageRouterSource result = streamParser.unsafeParse(input);
+        // then
+        assertThat(result.topicUrl()).isEqualTo(SAMPLE_TOPIC_URL);
+        assertThat(result.aafCredentials().username()).isNull();
+        assertThat(result.aafCredentials().password()).isNull();
+        assertThat(result.clientId()).isNull();
+    }
+
+    @Test
+    void incorrectConfiguration_shouldParseToStreamParserError() throws IOException {
+        // given
+        JsonObject input = GsonUtils.readObjectFromResource("/streams/data_router_sink_full.json");
+        // when
+        Either<StreamParserError, MessageRouterSource> result = streamParser.parse(input);
+        // then
+        assertThat(result.getLeft()).isInstanceOf(StreamParserError.class);
+        result.peekLeft(error -> {
+                    assertThat(error.message()).contains("Invalid stream type");
+                    assertThat(error.message()).contains("Expected '" + MESSAGE_ROUTER_TYPE + "', but was '"
+                            + DATA_ROUTER_TYPE + "'");
+                }
+        );
+    }
+
+    @Test
+    void emptyConfiguration_shouldParseToStreamParserError() {
+        // given
+        JsonObject input = gson.fromJson("{}", JsonObject.class);
+        // when
+        Either<StreamParserError, MessageRouterSource> result = streamParser.parse(input);
+        // then
+        assertThat(result.getLeft()).isInstanceOf(StreamParserError.class);
+    }
+
+
+}
\ No newline at end of file