Add streams parsing integration tests

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

16 files changed, 715 insertions, 0 deletions

diff --git a/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/AafCredentials.java b/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/AafCredentials.java
new file mode 100644
index 00000000..565efa10
--- /dev/null
+++ b/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/AafCredentials.java
@@ -0,0 +1,44 @@
+/*
+ * ============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.model.streams;
+
+
+import com.google.gson.annotations.SerializedName;
+import org.immutables.gson.Gson;
+import org.immutables.value.Value;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Represents the AAF Credentials. Currently it contains only user name and password.
+ *
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ * @since 1.1.4
+ */
+@Value.Immutable
+@Gson.TypeAdapters
+public interface AafCredentials {
+
+    @SerializedName(value = "username", alternate = "aaf_username")
+    @Nullable String username();
+
+    @SerializedName(value = "password", alternate = "aaf_password")
+    @Nullable String password();
+}
diff --git a/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/DataStream.java b/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/DataStream.java
new file mode 100644
index 00000000..06fabccd
--- /dev/null
+++ b/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/DataStream.java
@@ -0,0 +1,36 @@
+/*
+ * ============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.model.streams;
+
+import org.immutables.value.Value;
+
+/**
+ * Represents a named data stream.
+ *
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ * @since 1.1.4
+ */
+public interface DataStream {
+    @Value.Default
+    default String name() {
+        return "";
+    }
+}
diff --git a/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/DataStreamDirection.java b/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/DataStreamDirection.java
new file mode 100644
index 00000000..240a4c82
--- /dev/null
+++ b/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/DataStreamDirection.java
@@ -0,0 +1,48 @@
+/*
+ * ============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.model.streams;
+
+/**
+ * The direction of the stream, ie. whether it's input ({@code SOURCE}) or output ({@code SINK}) stream.
+ *
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ * @since 1.1.4
+ */
+public enum DataStreamDirection {
+
+    SINK("streams_publishes"),
+    SOURCE("streams_subscribes");
+
+    private final String configurationKey;
+
+    DataStreamDirection(String configurationKey) {
+        this.configurationKey = configurationKey;
+    }
+
+    /**
+     * The configuration key under which the single stream definitions should reside.
+     *
+     * @return the configuration key
+     */
+    public String configurationKey() {
+        return configurationKey;
+    }
+}
diff --git a/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/RawDataStream.java b/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/RawDataStream.java
new file mode 100644
index 00000000..7f6040ee
--- /dev/null
+++ b/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/RawDataStream.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.model.streams;
+
+import org.immutables.value.Value;
+
+/**
+ * Represents a raw/uninterpreted data stream.
+ *
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ * @since 1.1.4
+ * @param <T> type of raw data, eg. JsonObject
+ */
+@Value.Immutable
+public interface RawDataStream<T> {
+    String name();
+    StreamType type();
+    DataStreamDirection direction();
+    T descriptor();
+}
diff --git a/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/SinkStream.java b/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/SinkStream.java
new file mode 100644
index 00000000..5d1d5873
--- /dev/null
+++ b/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/SinkStream.java
@@ -0,0 +1,32 @@
+/*
+ * ============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.model.streams;
+
+/**
+ * Represents an output stream, ie. one of objects in <em>streams_publishes</em> array from application configuration.
+ * Application can put data to this stream.
+ *
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ * @since 1.1.4
+ */
+public interface SinkStream extends DataStream {
+
+}
diff --git a/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/SourceStream.java b/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/SourceStream.java
new file mode 100644
index 00000000..9b68c785
--- /dev/null
+++ b/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/SourceStream.java
@@ -0,0 +1,32 @@
+/*
+ * ============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.model.streams;
+
+/**
+ * Represents an input stream, ie. one of objects in <em>streams_subscribes</em> array from application configuration.
+ * Application can read data from this stream.
+ *
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ * @since 1.1.4
+ */
+public interface SourceStream extends DataStream {
+
+}
diff --git a/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/StreamType.java b/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/StreamType.java
new file mode 100644
index 00000000..2e08c82b
--- /dev/null
+++ b/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/StreamType.java
@@ -0,0 +1,52 @@
+/*
+ * ============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.model.streams;
+
+import io.vavr.collection.Stream;
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ * @since 1.1.4
+ */
+public enum StreamType {
+    MESSAGE_ROUTER("message_router"),
+    DATA_ROUTER("data_router"),
+    KAFKA("kafka"),
+    UNKNOWN("unknown");
+
+    private final String rawType;
+
+    StreamType(String rawType) {
+        this.rawType = rawType;
+    }
+
+    public static StreamType parse(@NotNull String rawType) {
+        return Stream.of(StreamType.values())
+                .find(type -> type.rawType.equals(rawType))
+                .getOrElse(UNKNOWN);
+    }
+
+    @Override
+    public String toString() {
+        return rawType;
+    }
+}
diff --git a/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/dmaap/DataRouter.java b/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/dmaap/DataRouter.java
new file mode 100644
index 00000000..38adb197
--- /dev/null
+++ b/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/dmaap/DataRouter.java
@@ -0,0 +1,57 @@
+/*
+ * ============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.model.streams.dmaap;
+
+
+import com.google.gson.annotations.SerializedName;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ * @since 1.1.4
+ */
+public interface DataRouter {
+
+    /**
+     * DCAE location for the publisher, used to set up routing.
+     */
+    @SerializedName("location")
+    @Nullable String location();
+
+    /**
+     * Username
+     * <ul>
+     * <li>Data Router uses to authenticate to the subscriber when delivering files OR</li>
+     * <li>the publisher uses to authenticate to Data Router.</li>
+     * </ul>
+     */
+    @SerializedName("username")
+    @Nullable String username();
+
+    /**
+     * Password
+     * <ul>
+     * <li>Data Router uses to authenticate to the subscriber when delivering files OR</li>
+     * <li>the publisher uses to authenticate to Data Router.</li>
+     * </ul>
+     */
+    @SerializedName("password")
+    @Nullable String password();
+}
diff --git a/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/dmaap/DataRouterSink.java b/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/dmaap/DataRouterSink.java
new file mode 100644
index 00000000..bfe31182
--- /dev/null
+++ b/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/dmaap/DataRouterSink.java
@@ -0,0 +1,55 @@
+/*
+ * ============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.model.streams.dmaap;
+
+
+import com.google.gson.annotations.SerializedName;
+import org.immutables.gson.Gson;
+import org.immutables.value.Value;
+import org.jetbrains.annotations.Nullable;
+import org.onap.dcaegen2.services.sdk.model.streams.SinkStream;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ * @since 1.1.4
+ */
+@Gson.TypeAdapters
+@Value.Immutable
+public interface DataRouterSink extends DataRouter, SinkStream {
+
+    /**
+     * URL to which the publisher makes Data Router publish requests.
+     */
+    @SerializedName("publish_url")
+    String publishUrl();
+
+    /**
+     * Publisher id in Data Router
+     */
+    @SerializedName("publisher_id")
+    @Nullable String publisherId();
+
+    /**
+     * URL from which log data for the feed can be obtained.
+     */
+    @SerializedName("log_url")
+    @Nullable String logUrl();
+
+}
diff --git a/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/dmaap/DataRouterSource.java b/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/dmaap/DataRouterSource.java
new file mode 100644
index 00000000..4ba81acb
--- /dev/null
+++ b/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/dmaap/DataRouterSource.java
@@ -0,0 +1,49 @@
+/*
+ * ============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.model.streams.dmaap;
+
+
+import com.google.gson.annotations.SerializedName;
+import org.immutables.gson.Gson;
+import org.immutables.value.Value;
+import org.jetbrains.annotations.Nullable;
+import org.onap.dcaegen2.services.sdk.model.streams.SourceStream;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ * @since 1.1.4
+ */
+@Gson.TypeAdapters
+@Value.Immutable
+public interface DataRouterSource extends DataRouter, SourceStream {
+
+    /**
+     * URL to which the Data Router should deliver files.
+     */
+    // TODO: since crucial, we need to verify if it should be non-null
+    @SerializedName("delivery_url")
+    @Nullable String deliveryUrl();
+
+    /**
+     * Subscriber id in Data Router.
+     */
+    @SerializedName("subscriber_id")
+    @Nullable String subscriberId();
+}
diff --git a/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/dmaap/Kafka.java b/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/dmaap/Kafka.java
new file mode 100644
index 00000000..df2cee6d
--- /dev/null
+++ b/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/dmaap/Kafka.java
@@ -0,0 +1,77 @@
+/*
+ * ============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.model.streams.dmaap;
+
+import static io.vavr.Predicates.not;
+
+import io.vavr.collection.List;
+import org.immutables.value.Value;
+import org.jetbrains.annotations.Nullable;
+import org.onap.dcaegen2.services.sdk.model.streams.AafCredentials;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ * @since 1.1.4
+ */
+public interface Kafka {
+
+    /**
+     * Kafka bootstrap servers as defined in Kafka client documentation under <em>bootstrap.servers</em> configuration
+     * key.
+     */
+    String bootstrapServers();
+
+    /**
+     * The name of the topic where application should publish or subscribe for the messages.
+     */
+    String topicName();
+
+    /**
+     * The credentials to use when authenticating to Kafka cluster or null when connection should be unauthenticated.
+     */
+    @Nullable AafCredentials aafCredentials();
+
+    /**
+     * AAF client role that’s requesting publish or subscribe access to the topic.
+     */
+    @Nullable String clientRole();
+
+    /**
+     * Client id for given AAF client.
+     */
+    @Nullable String clientId();
+
+    /**
+     * The limit on the size of message published to/subscribed from the topic. Can be used to set Kafka client
+     * <em>max.request.size</em> configuration property.
+     */
+    @Value.Default
+    default int maxPayloadSizeBytes() {
+        return 1024 * 1024;
+    }
+
+    /**
+     * The {@code bootstrapServers} converted to the list of servers' addresses.
+     */
+    @Value.Derived
+    default List<String> bootstrapServerList() {
+        return List.of(bootstrapServers().split(",")).filter(not(String::isEmpty));
+    }
+}
diff --git a/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/dmaap/KafkaSink.java b/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/dmaap/KafkaSink.java
new file mode 100644
index 00000000..2c397615
--- /dev/null
+++ b/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/dmaap/KafkaSink.java
@@ -0,0 +1,32 @@
+/*
+ * ============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.model.streams.dmaap;
+
+import org.immutables.value.Value;
+import org.onap.dcaegen2.services.sdk.model.streams.SinkStream;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ * @since 1.1.4
+ */
+@Value.Immutable
+public interface KafkaSink extends Kafka, SinkStream {
+
+}
diff --git a/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/dmaap/KafkaSource.java b/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/dmaap/KafkaSource.java
new file mode 100644
index 00000000..799d3af5
--- /dev/null
+++ b/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/dmaap/KafkaSource.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.model.streams.dmaap;
+
+import org.immutables.value.Value;
+import org.jetbrains.annotations.Nullable;
+import org.onap.dcaegen2.services.sdk.model.streams.SourceStream;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ * @since 1.1.4
+ */
+@Value.Immutable
+public interface KafkaSource extends Kafka, SourceStream {
+
+    /**
+     * A unique string that identifies the consumer group this consumer belongs to as defined in Kafka consumer
+     * configuration key <em>group.id</em>.
+     */
+    @Nullable String consumerGroupId();
+}
diff --git a/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/dmaap/MessageRouter.java b/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/dmaap/MessageRouter.java
new file mode 100644
index 00000000..3a6ba0f6
--- /dev/null
+++ b/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/dmaap/MessageRouter.java
@@ -0,0 +1,61 @@
+/*
+ * ============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.model.streams.dmaap;
+
+import com.google.gson.annotations.SerializedName;
+import org.jetbrains.annotations.Nullable;
+import org.onap.dcaegen2.services.sdk.model.streams.AafCredentials;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ * @since 1.1.4
+ */
+public interface MessageRouter {
+
+    /**
+     * URL for accessing the topic to publish or receive events.
+     */
+    @SerializedName("topic_url")
+    String topicUrl();
+
+    /**
+     * AAF client role that’s requesting publish or subscribe access to the topic.
+     */
+    @SerializedName("client_role")
+    @Nullable String clientRole();
+
+    /**
+     * Client id for given AAF client.
+     */
+    @SerializedName("client_id")
+    @Nullable String clientId();
+
+    /**
+     * DCAE location for the publisher or subscriber, used to set up routing.
+     */
+    @SerializedName("location")
+    @Nullable String location();
+
+    /**
+     * The AAF credentials.
+     */
+    @SerializedName("aaf_credentials")
+    @Nullable AafCredentials aafCredentials();
+}
diff --git a/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/dmaap/MessageRouterSink.java b/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/dmaap/MessageRouterSink.java
new file mode 100644
index 00000000..1820775b
--- /dev/null
+++ b/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/dmaap/MessageRouterSink.java
@@ -0,0 +1,32 @@
+/*
+ * ============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.model.streams.dmaap;
+
+import org.immutables.value.Value;
+import org.onap.dcaegen2.services.sdk.model.streams.SinkStream;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ * @since 1.1.4
+ */
+@Value.Immutable
+public interface MessageRouterSink extends MessageRouter, SinkStream {
+
+}
diff --git a/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/dmaap/MessageRouterSource.java b/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/dmaap/MessageRouterSource.java
new file mode 100644
index 00000000..b92dff1f
--- /dev/null
+++ b/rest-services/model/src/main/java/org/onap/dcaegen2/services/sdk/model/streams/dmaap/MessageRouterSource.java
@@ -0,0 +1,32 @@
+/*
+ * ============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.model.streams.dmaap;
+
+import org.immutables.value.Value;
+import org.onap.dcaegen2.services.sdk.model.streams.SourceStream;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ * @since 1.1.4
+ */
+@Value.Immutable
+public interface MessageRouterSource extends MessageRouter, SourceStream {
+
+}