Move avcnsimulator to nf-simulator/avcn-manager

Issue-ID: INT-1869
Signed-off-by: Edyta Krukowska <edyta.krukowska@nokia.com>
Change-Id: I5c0dcdf7bf67d0dac9ff7e77829890920f426b7d

38 files changed, 2205 insertions, 0 deletions

diff --git a/README.md b/README.md
new file mode 100644
index 0000000..45d8576
--- /dev/null
+++ b/README.md
@@ -0,0 +1,128 @@
+# AVCN Simulator
+Simulator that fetches changes of configuration from kafka and sends them to VES client.
+
+## What does it do?
+The simulator processes notifications from NETCONF server. It does this by being a subscriber of a Kafka topic that is fed 
+with NETCONF notifications. The topic name is "config". Incoming notifications are then processed and output of this
+ processing is sent to VES client.
+
+### Processing details
+1. When last element contain value (equal sign) then cut it together with `'/'` from path and fill attributeList object with key equal to name of element before `'='` and value equal to element after `'='` keeping key and value quoted (`""`),
+2. Convert remaining `'/'` into `' , '` ,
+3. Convert all container types (without `[key='value']`) to `<container name>=<container name>`
+4. Convert all list types (with `[key='value']`) to `<list name>=<value>`
+````
+{containerA}/{listB}[{keyB}={valueB}]/{containerC}/{listD}[{keyD}={valueD}]/{[leaf, leaf-list]} = {value of leaf}
+
+{
+  "variables": {
+    "dn": " {containerA}={containerA} , {listB}={valueB} , {containerC}={containerC} , {listD}={valued}",
+    "attributesList": {
+      "{leaf, leaf-list}": "{value of leaf}"
+    }
+  }
+} 
+````
+
+### VES client Request
+The resulting request to VES client looks like the following (the "dn" and "attributesList"
+ parameters are dependent on the actual NETCONF notification's value):
+````
+{
+  "simulatorParams": {
+    "repeatCount": 1,
+    "repeatInterval": 1,
+    "vesServerUrl": ""
+  },
+  "templateName": "notification.json",
+  "patch": {},
+  "variables": {
+    "dn": "",
+    "attributesList": {}
+  }
+}
+````
+
+### Examples
+````
+/example-sports:sports/person[name='name 1'] (list instance)
+
+{
+  "simulatorParams": { },
+  "templateName": "cmNotify.json",
+  "patch": { },
+  "variables": {
+    "dn":" example-sports:sports= example-sports:sports , person=name 1",
+    "attributesList": { }
+  }        
+}
+````
+
+````
+/example-sports:sports/person[name='name 1']/name = name 1
+
+{
+  "simulatorParams": { },
+  "templateName": "cmNotify.json",
+  "patch": { },
+  "variables": {
+    "dn":" example-sports:sports= example-sports:sports , person=name 1",
+    "attributesList": { "name": "name 2" }
+  }        
+}
+````
+
+````
+/example-sports:sports/person[name='name 2'] (list instance)
+
+{
+  "simulatorParams": { },
+  "templateName": "cmNotify.json",
+  "patch": { },
+  "variables": {
+    "dn":" example-sports:sports= example-sports:sports , person=name 2",
+    "attributesList": { }
+  }        
+}
+````
+
+````
+/example-sports:sports/team[name='team 1']/player[name='player 1'] (list instance)
+
+{
+  "simulatorParams": { },
+  "templateName": "cmNotify.json",
+  "patch": { },
+  "variables": {
+    "dn":" example-sports:sports= example-sports:sports , player=player 1",
+    "attributesList": { }
+  }        
+}
+````
+
+````
+/example-sports:sports/team[name='team 1']/player[name='player 1']/name = player 1
+{
+  "simulatorParams": { },
+  "templateName": "cmNotify.json",
+  "patch": { },
+   "variables": {
+     "dn":" example-sports:sports= example-sports:sports , player=player 1",
+     "attributesList": { "name": "player 1"”" }
+   }        
+}
+````
+## Simulator configuration
+It's possible to override default configuration. Following environment variables can be set 
+
+    KAFKA_BOOTSTRAP_SERVERS - Kafka host, by default kafka1:9092 
+    
+    KAFKA_APPLICATION_ID - An identifier for the stream processing application. Must be unique within the Kafka cluster. 
+                            By default avcn-simulator  
+    
+    KAFKA_SOURCE_TOPIC - Kafka topic, where Netconf simulator pushes notification, by default config
+        
+    REST_CLIENT_PNF_SIMULATOR_ENDPOINT -  VES client's URL, by default pnf-simulator:5000/simulator/start
+    
+    REST_CLIENT_VES_ENDPOINT - VES URL, AVNC events will be send via VES client to this address 
+    
diff --git a/checkstyle-suppressions.xml b/checkstyle-suppressions.xml
new file mode 100644
index 0000000..8d95607
--- /dev/null
+++ b/checkstyle-suppressions.xml
@@ -0,0 +1,46 @@
+<?xml version="1.0"?>
+<!--
+  ============LICENSE_START=======================================================
+  Simulator
+  ================================================================================
+  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=========================================================
+  -->
+
+
+<!DOCTYPE suppressions PUBLIC
+        "-//Checkstyle//DTD SuppressionFilter Configuration 1.0//EN"
+        "https://checkstyle.org/dtds/suppressions_1_0.dtd">
+
+<suppressions>
+    <suppress checks=".*" files="\.java" lines="1-20"
+    />
+    <suppress checks=".*" files="\.properties"
+    />
+    <suppress checks="javadoc" files="\.java"
+    />
+    <suppress checks="LineLength" files="\.java"
+    />
+    <suppress checks="HiddenField" files="\.java"
+    />
+    <suppress checks="FinalClass" files="\.java"
+    />
+    <suppress checks="FinalParameters" files="\.java"
+    />
+    <suppress checks="AvoidInlineConditionals" files="\.java"
+    />
+    <suppress checks="DesignForExtension" files="\.java"
+    />
+</suppressions>
diff --git a/docker/Dockerfile b/docker/Dockerfile
new file mode 100644
index 0000000..f21d0f1
--- /dev/null
+++ b/docker/Dockerfile
@@ -0,0 +1,6 @@
+FROM docker.io/openjdk:11-jre-slim
+
+ARG VERSION=${version}
+
+ADD target/avcn-manager-${VERSION}.jar /app/avcn-manager.jar
+CMD java -jar /app/avcn-manager.jar
diff --git a/pom.xml b/pom.xml
new file mode 100644
index 0000000..cc0520d
--- /dev/null
+++ b/pom.xml
@@ -0,0 +1,166 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ============LICENSE_START=======================================================
+  Simulator
+  ================================================================================
+  Copyright (C) 2021 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=========================================================
+  -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>org.onap.oparent</groupId>
+        <artifactId>oparent</artifactId>
+        <version>3.2.0</version>
+    </parent>
+
+    <groupId>org.onap.integration.nfsimulator</groupId>
+    <artifactId>avcn-manager</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+    <name>avcn-manager</name>
+
+    <properties>
+        <nexusproxy>https://nexus.onap.org</nexusproxy>
+        <snapshotNexusPath>/content/repositories/snapshots/</snapshotNexusPath>
+        <releaseNexusPath>/content/repositories/releases/</releaseNexusPath>
+        <stagingNexusPath>/content/repositories/staging/</stagingNexusPath>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+        <maven.compiler.source>11</maven.compiler.source>
+        <maven.compiler.target>11</maven.compiler.target>
+
+        <maven.build.timestamp.format>yyyyMMdd'T'HHmmss</maven.build.timestamp.format>
+        <docker.image.tag>latest</docker.image.tag>
+        <docker.image.name>onap/${project.artifactId}</docker.image.name>
+        <dependency.directory.name>libs</dependency.directory.name>
+        <dependency.directory.location>${project.build.directory}/${dependency.directory.name}
+        </dependency.directory.location>
+
+        <spring.boot.version>2.1.6.RELEASE</spring.boot.version>
+        <spring.kafka.streams.version>2.3.0</spring.kafka.streams.version>
+        <commons-io.version>2.6</commons-io.version>
+        <mockito-core.version>2.18.3</mockito-core.version>
+        <junit-jupiter-engine.version>5.3.1</junit-jupiter-engine.version>
+        <assertj-core.version>3.13.2</assertj-core.version>
+        <docker-maven-plugin.version>0.31.0</docker-maven-plugin.version>
+    </properties>
+
+
+    <dependencies>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter</artifactId>
+            <version>${spring.boot.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-web</artifactId>
+            <version>${spring.boot.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-streams</artifactId>
+            <version>${spring.kafka.streams.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>commons-io</groupId>
+            <artifactId>commons-io</artifactId>
+            <version>${commons-io.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <version>${mockito-core.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-engine</artifactId>
+            <version>${junit-jupiter-engine.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <version>${assertj-core.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+    </dependencies>
+    <profiles>
+        <profile>
+            <id>docker</id>
+            <activation>
+                <activeByDefault>false</activeByDefault>
+            </activation>
+            <build>
+                <plugins>
+                    <plugin>
+                        <groupId>io.fabric8</groupId>
+                        <artifactId>docker-maven-plugin</artifactId>
+                        <version>${docker-maven-plugin.version}</version>
+                        <executions>
+                            <execution>
+                                <id>build-image</id>
+                                <goals>
+                                    <goal>build</goal>
+                                    <goal>push</goal>
+                                </goals>
+                            </execution>
+                        </executions>
+                        <configuration>
+                            <images>
+                                <image>
+                                    <name>${onap.nexus.dockerregistry.daily}/${docker.image.name}</name>
+                                    <build>
+                                        <contextDir>${project.basedir}</contextDir>
+                                        <dockerFile>${project.basedir}/docker/Dockerfile</dockerFile>
+                                    </build>
+                                </image>
+                            </images>
+
+                        </configuration>
+                    </plugin>
+                </plugins>
+            </build>
+        </profile>
+    </profiles>
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-maven-plugin</artifactId>
+                <version>${spring.boot.version}</version>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>repackage</goal>
+                        </goals>
+                    </execution>
+                </executions>
+                <configuration>
+                    <mainClass>org.onap.avcnmanager.Main</mainClass>
+                    <layout>ZIP</layout>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>
diff --git a/src/main/java/org/onap/avcnmanager/Main.java b/src/main/java/org/onap/avcnmanager/Main.java
new file mode 100644
index 0000000..c0c1af9
--- /dev/null
+++ b/src/main/java/org/onap/avcnmanager/Main.java
@@ -0,0 +1,31 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Simulator
+ * ================================================================================
+ * Copyright (C) 2021 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.avcnmanager;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
+public class Main {
+    public static void main(String[] args) {
+        SpringApplication.run(Main.class, args);
+    }
+}
diff --git a/src/main/java/org/onap/avcnmanager/config/AVCNConfig.java b/src/main/java/org/onap/avcnmanager/config/AVCNConfig.java
new file mode 100644
index 0000000..35273ff
--- /dev/null
+++ b/src/main/java/org/onap/avcnmanager/config/AVCNConfig.java
@@ -0,0 +1,89 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Simulator
+ * ================================================================================
+ * Copyright (C) 2021 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.avcnmanager.config;
+
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.Topology;
+import org.apache.kafka.streams.kstream.KStream;
+import org.onap.avcnmanager.message.handlers.MessageHandler;
+import org.onap.avcnmanager.message.data.ChangePack;
+import org.onap.avcnmanager.message.serializers.ChangePackDeserializer;
+import org.onap.avcnmanager.message.serializers.ChangePackSerializer;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.client.RestTemplate;
+
+import java.util.Properties;
+
+@Configuration
+public class AVCNConfig {
+
+    @Bean(name = "AVCNProperties")
+    public Properties getKafkaStreamProperties(
+            @Value("${kafka.bootstrap-servers}") String bootstrapServer,
+            @Value("${kafka.application.id}") String applicationId
+    ) {
+        Properties properties = new Properties();
+        properties.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId);
+        properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer);
+        properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
+        properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, ChangePackSerde.class.getName());
+        return properties;
+    }
+
+    @Bean(name = "AVCNTopology")
+    public Topology getKafkaStreamTopology(
+            @Value("${kafka.source.topic}") String inTopic,
+            MessageHandler messageHandler
+    ) {
+        StreamsBuilder builder = new StreamsBuilder();
+        KStream<String, ChangePack> stream = builder.stream(inTopic);
+        stream.foreach(messageHandler::handleMessage);
+        return builder.build();
+    }
+
+
+    @Bean
+    public KafkaStreams getKafkaStream(
+            @Qualifier("AVCNTopology") Topology topology,
+            @Qualifier("AVCNProperties") Properties properties
+    ) throws InterruptedException {
+        Thread.sleep(10000);
+        return new KafkaStreams(topology, properties);
+    }
+
+
+    @Bean
+    public RestTemplate getRestTemplate() {
+        return new RestTemplate();
+    }
+
+    static public final class ChangePackSerde extends Serdes.WrapperSerde<ChangePack> {
+        public ChangePackSerde() {
+            super(new ChangePackSerializer(), new ChangePackDeserializer());
+        }
+    }
+}
\ No newline at end of file
diff --git a/src/main/java/org/onap/avcnmanager/kafka/stream/AVCNKafkaStream.java b/src/main/java/org/onap/avcnmanager/kafka/stream/AVCNKafkaStream.java
new file mode 100644
index 0000000..dfc7912
--- /dev/null
+++ b/src/main/java/org/onap/avcnmanager/kafka/stream/AVCNKafkaStream.java
@@ -0,0 +1,61 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Simulator
+ * ================================================================================
+ * Copyright (C) 2021 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.avcnmanager.kafka.stream;
+
+import org.apache.kafka.streams.KafkaStreams;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+
+
+@Component
+public class AVCNKafkaStream {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(AVCNKafkaStream.class);
+
+    private final KafkaStreams streams;
+
+    @Autowired
+    AVCNKafkaStream(KafkaStreams streams) {
+        this.streams = streams;
+        streams.setUncaughtExceptionHandler(this::handleExceptionInStreams);
+    }
+
+    @PostConstruct
+    void startKafkaStream() {
+        streams.start();
+    }
+
+    private void handleExceptionInStreams(Thread thread,Throwable throwable) {
+        LOGGER.warn("Exception occurred int kafka stream: " + thread);
+        LOGGER.debug(throwable.getMessage());
+        if(!streams.state().isRunning()) {
+            LOGGER.error("Kafka stream stop running, state: " + streams.state());
+            streams.close();
+            System.exit(1);
+        }
+    }
+}
+
+
diff --git a/src/main/java/org/onap/avcnmanager/message/data/Change.java b/src/main/java/org/onap/avcnmanager/message/data/Change.java
new file mode 100644
index 0000000..5c6402c
--- /dev/null
+++ b/src/main/java/org/onap/avcnmanager/message/data/Change.java
@@ -0,0 +1,68 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Simulator
+ * ================================================================================
+ * Copyright (C) 2021 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.avcnmanager.message.data;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.Objects;
+
+public class Change {
+    private final String path;
+    private final String value;
+
+    @JsonCreator(mode = JsonCreator.Mode.PROPERTIES)
+    public Change(@JsonProperty("path") String path,
+                  @JsonProperty("value") String value) {
+        this.path = path;
+        this.value = value;
+    }
+
+    public String getPath() {
+        return path;
+    }
+
+    public String getValue() {
+        return value;
+    }
+
+    @Override
+    public String toString() {
+        return "{" +
+                "path='" + path + '\'' +
+                ", value='" + value + '\'' +
+                '}';
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        Change change = (Change) o;
+        return Objects.equals(getPath(), change.getPath()) &&
+                Objects.equals(getValue(), change.getValue());
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(getPath(), getValue());
+    }
+}
diff --git a/src/main/java/org/onap/avcnmanager/message/data/ChangePack.java b/src/main/java/org/onap/avcnmanager/message/data/ChangePack.java
new file mode 100644
index 0000000..f7d4c3a
--- /dev/null
+++ b/src/main/java/org/onap/avcnmanager/message/data/ChangePack.java
@@ -0,0 +1,79 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Simulator
+ * ================================================================================
+ * Copyright (C) 2021 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.avcnmanager.message.data;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.Objects;
+
+public class ChangePack {
+
+    private final Change aNew;
+    private final Change old;
+
+    private final String type;
+
+    @JsonCreator(mode = JsonCreator.Mode.PROPERTIES)
+    public ChangePack(@JsonProperty("new") Change aNew,
+                      @JsonProperty("old") Change old,
+                      @JsonProperty("type") String type) {
+        this.aNew = aNew;
+        this.old = old;
+        this.type = type;
+    }
+
+    public Change getNew() {
+        return aNew;
+    }
+
+    public Change getOld() {
+        return old;
+    }
+
+    public String getType() {
+        return type;
+    }
+
+    @Override
+    public String toString() {
+        return "{" +
+                "new=" + aNew +
+                ", old=" + old +
+                ", type='" + type + '\'' +
+                '}';
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        ChangePack that = (ChangePack) o;
+        return Objects.equals(aNew, that.aNew) &&
+                Objects.equals(getOld(), that.getOld()) &&
+                Objects.equals(getType(), that.getType());
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(aNew, getOld(), getType());
+    }
+}
diff --git a/src/main/java/org/onap/avcnmanager/message/forwarders/Forwarder.java b/src/main/java/org/onap/avcnmanager/message/forwarders/Forwarder.java
new file mode 100644
index 0000000..fdf28ee
--- /dev/null
+++ b/src/main/java/org/onap/avcnmanager/message/forwarders/Forwarder.java
@@ -0,0 +1,27 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Simulator
+ * ================================================================================
+ * Copyright (C) 2021 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.avcnmanager.message.forwarders;
+
+import org.onap.avcnmanager.message.processing.ParsingResult;
+
+public interface Forwarder {
+    void send(ParsingResult processed);
+}
diff --git a/src/main/java/org/onap/avcnmanager/message/forwarders/RestForwarder.java b/src/main/java/org/onap/avcnmanager/message/forwarders/RestForwarder.java
new file mode 100644
index 0000000..6f1e6a8
--- /dev/null
+++ b/src/main/java/org/onap/avcnmanager/message/forwarders/RestForwarder.java
@@ -0,0 +1,86 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Simulator
+ * ================================================================================
+ * Copyright (C) 2021 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.avcnmanager.message.forwarders;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.onap.avcnmanager.message.processing.ParsingResult;
+import org.onap.avcnmanager.utils.JsonUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.http.HttpEntity;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.MediaType;
+import org.springframework.stereotype.Component;
+import org.springframework.web.client.RestTemplate;
+
+@Component
+public class RestForwarder implements Forwarder {
+    private static final Logger LOGGER = LoggerFactory.getLogger(RestForwarder.class);
+    private static final String PNF_PAYLOAD_TEMPLATE = "pnfsim_payload_notification.json";
+    private final String pnfSimulatorEndpoint;
+    private final RestTemplate restTemplate;
+    private final String vesEndpoint;
+    private final HttpHeaders headers;
+    private final JsonNode pnfSimJson;
+    private final JsonUtils jsonUtils;
+
+    public RestForwarder(@Value("${rest.client.pnfsimulator.endpoint}") String pnfSimulatorEndpoint,
+                         @Value("${rest.client.ves.endpoint}") String vesEndpoint,
+                         RestTemplate restTemplate,
+                         JsonUtils jsonUtils) {
+        this.pnfSimulatorEndpoint = pnfSimulatorEndpoint;
+        this.restTemplate = restTemplate;
+        this.vesEndpoint = vesEndpoint;
+        this.jsonUtils = jsonUtils;
+        this.headers = new HttpHeaders();
+        this.pnfSimJson = jsonUtils.jsonNodeFromResourceFile(PNF_PAYLOAD_TEMPLATE);
+        headers.setContentType(MediaType.APPLICATION_JSON);
+    }
+
+    @Override
+    public void send(ParsingResult message) {
+        String resultAsJson = jsonUtils.convertToJsonString(message);
+        JsonNode node = jsonUtils.convertToJsonNode(resultAsJson);
+        String payload = fillSimulatorRequest(node);
+        LOGGER.info("Will send payload: " + payload);
+        sendPayload(payload);
+    }
+
+
+    private String fillSimulatorRequest(JsonNode node) {
+        ObjectNode root = pnfSimJson.deepCopy();
+        ((ObjectNode)root.get("simulatorParams")).put("vesServerUrl", vesEndpoint);
+        root.set("variables", node);
+        return root.toString();
+    }
+
+    private void sendPayload(String payload) {
+        try {
+            HttpEntity<String> requestEntity = new HttpEntity<>(payload, headers);
+            restTemplate.postForObject(pnfSimulatorEndpoint, requestEntity, String.class);
+        } catch (Throwable e) {
+            LOGGER.error("Exception occurred: " + e.getMessage() + ", while sending: " + payload + ", the payload is skipped");
+            LOGGER.debug("Exception details: ", e);
+        }
+    }
+}
diff --git a/src/main/java/org/onap/avcnmanager/message/handlers/MessageHandler.java b/src/main/java/org/onap/avcnmanager/message/handlers/MessageHandler.java
new file mode 100644
index 0000000..fcc0691
--- /dev/null
+++ b/src/main/java/org/onap/avcnmanager/message/handlers/MessageHandler.java
@@ -0,0 +1,28 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Simulator
+ * ================================================================================
+ * Copyright (C) 2021 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.avcnmanager.message.handlers;
+
+import org.onap.avcnmanager.message.data.ChangePack;
+
+public interface MessageHandler {
+
+    void handleMessage(String key, ChangePack massage);
+}
diff --git a/src/main/java/org/onap/avcnmanager/message/handlers/NewMessageHandler.java b/src/main/java/org/onap/avcnmanager/message/handlers/NewMessageHandler.java
new file mode 100644
index 0000000..9a8fb7c
--- /dev/null
+++ b/src/main/java/org/onap/avcnmanager/message/handlers/NewMessageHandler.java
@@ -0,0 +1,61 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Simulator
+ * ================================================================================
+ * Copyright (C) 2021 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.avcnmanager.message.handlers;
+
+import java.util.Objects;
+import org.onap.avcnmanager.message.forwarders.Forwarder;
+import org.onap.avcnmanager.message.data.ChangePack;
+import org.onap.avcnmanager.message.processing.TextProcessor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+import java.util.stream.Stream;
+
+@Component
+public class NewMessageHandler implements MessageHandler {
+    private static final Logger LOGGER = LoggerFactory.getLogger(NewMessageHandler.class);
+
+    private final TextProcessor processor;
+    private final Forwarder forwarder;
+
+    public NewMessageHandler(TextProcessor processor, Forwarder forwarder) {
+        this.processor = processor;
+        this.forwarder = forwarder;
+    }
+
+    @Override
+    public void handleMessage(String key, ChangePack message) {
+        LOGGER.info("Handling message: " + message.toString());
+        stream(message)
+                .filter(c -> !Objects.isNull(c))
+                .filter(c -> !Objects.isNull(c.getNew()))
+                .filter(c -> !Objects.isNull(c.getNew().getPath()))
+                .filter(c -> !c.getNew().getPath().isEmpty())
+                .map(c -> processor.process(key, message))
+                .forEach(forwarder::send);
+    }
+
+    private Stream<ChangePack> stream(ChangePack changePack) {
+        return Stream.of(changePack);
+    }
+
+}
diff --git a/src/main/java/org/onap/avcnmanager/message/processing/NetconfTextProcessor.java b/src/main/java/org/onap/avcnmanager/message/processing/NetconfTextProcessor.java
new file mode 100644
index 0000000..2900c59
--- /dev/null
+++ b/src/main/java/org/onap/avcnmanager/message/processing/NetconfTextProcessor.java
@@ -0,0 +1,113 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Simulator
+ * ================================================================================
+ * Copyright (C) 2021 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.avcnmanager.message.processing;
+
+import org.onap.avcnmanager.message.data.ChangePack;
+import org.onap.avcnmanager.message.processing.targets.MapTarget;
+import org.onap.avcnmanager.message.processing.targets.StringBuilderTarget;
+import org.onap.avcnmanager.message.processing.targets.TargetContainer;
+import org.onap.avcnmanager.message.processing.tokens.ContainerToken;
+import org.onap.avcnmanager.message.processing.tokens.ListToken;
+import org.onap.avcnmanager.message.processing.tokens.Token;
+import org.onap.avcnmanager.message.processing.tokens.ValueToken;
+import org.springframework.stereotype.Component;
+
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+@Component
+public final class NetconfTextProcessor implements TextProcessor {
+    private static final Pattern PATTERN_PATH = Pattern.compile("/+");
+    private static final String DELIMITER = " , ";
+
+    @Override
+    public ParsingResult process(String key, ChangePack message) {
+        StringBuilderTarget dn = new StringBuilderTarget(new StringBuilder(), DELIMITER);
+        MapTarget attributesMap = new MapTarget(new HashMap<>());
+        LinkedList<Token> dequeOfTokens = new LinkedList<>();
+        String newPath = message.getNew().getPath();
+        String value = message.getNew().getValue();
+        if(!value.isEmpty()) {
+            newPath += " = " +value;
+        }
+
+        List<String> strings = splitToStrings(newPath, PATTERN_PATH);
+        convertToTypedTokens(strings, dequeOfTokens, 0);
+        removeConsecutiveDuplicatesFrom(dequeOfTokens);
+        dumpTokensIntoRespectiveContainers(dequeOfTokens, dn, attributesMap);
+
+        return emitResult(dn, attributesMap);
+    }
+
+    private static List<String> splitToStrings(String newPath, Pattern patternPath) {
+        return patternPath.splitAsStream(newPath)
+                    .filter(s -> !s.isEmpty())
+                    .collect(Collectors.toList());
+    }
+
+    private static void removeConsecutiveDuplicatesFrom(LinkedList<Token> dequeOfTokens) {
+        List<Token> toRemove = new LinkedList<>();
+        for (int i = 1; i < dequeOfTokens.size(); i++) {
+            if(ListToken.class.equals(dequeOfTokens.get(i-1).getClass())
+                && ListToken.class.equals(dequeOfTokens.get(i).getClass())) {
+                toRemove.add(dequeOfTokens.get(i-1));
+            }
+        }
+        dequeOfTokens.removeAll(toRemove);
+    }
+
+    private static void convertToTypedTokens(List<String> strings, LinkedList<Token> dequeOfTokens, int startIndex) {
+        if (strings.size() > startIndex) {
+            String str = strings.get(startIndex);
+            Token token = determineToken(str);
+            dequeOfTokens.add(token);
+
+            convertToTypedTokens(strings,dequeOfTokens, startIndex + 1);
+        }
+    }
+
+    private static Token determineToken(String str) {
+        if (str.endsWith("]")) {
+            return new ListToken(str);
+        } else if (str.contains("=")) {
+            return new ValueToken(str);
+        } else {
+            return new ContainerToken(str);
+        }
+    }
+
+    private static void dumpTokensIntoRespectiveContainers(LinkedList<Token> dequeOfTokens,
+                         TargetContainer<String> dnBuilder,
+                         TargetContainer<String> attributesBuilder) {
+        dequeOfTokens.forEach(t -> {
+            t.dump(dnBuilder);
+            t.dump(attributesBuilder);
+        });
+    }
+
+    private static ParsingResult emitResult(StringBuilderTarget dnSb, MapTarget attributesMap) {
+        dnSb.trimLastDelimiter();
+        return new ParsingResult(dnSb.stringValue(), attributesMap.mapValue());
+    }
+}
diff --git a/src/main/java/org/onap/avcnmanager/message/processing/ParsingResult.java b/src/main/java/org/onap/avcnmanager/message/processing/ParsingResult.java
new file mode 100644
index 0000000..d924a35
--- /dev/null
+++ b/src/main/java/org/onap/avcnmanager/message/processing/ParsingResult.java
@@ -0,0 +1,61 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Simulator
+ * ================================================================================
+ * Copyright (C) 2021 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.avcnmanager.message.processing;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Objects;
+
+public class ParsingResult {
+    private final String dn;
+    private final Map<String,String> attributesList;
+
+    @JsonCreator(mode = JsonCreator.Mode.PROPERTIES)
+    public ParsingResult(@JsonProperty("dn") String dn, @JsonProperty("attributesList") Map<String, String> attributesList) {
+        this.dn = dn;
+        this.attributesList = attributesList;
+    }
+
+    public String getDn() {
+        return dn;
+    }
+
+    public Map<String, String> getAttributesList() {
+        return Collections.unmodifiableMap(attributesList);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        ParsingResult that = (ParsingResult) o;
+        return Objects.equals(getDn(), that.getDn()) &&
+                Objects.equals(getAttributesList(), that.getAttributesList());
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(getDn(), getAttributesList());
+    }
+}
diff --git a/src/main/java/org/onap/avcnmanager/message/processing/TextProcessor.java b/src/main/java/org/onap/avcnmanager/message/processing/TextProcessor.java
new file mode 100644
index 0000000..d4ad6b9
--- /dev/null
+++ b/src/main/java/org/onap/avcnmanager/message/processing/TextProcessor.java
@@ -0,0 +1,27 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Simulator
+ * ================================================================================
+ * Copyright (C) 2021 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.avcnmanager.message.processing;
+
+import org.onap.avcnmanager.message.data.ChangePack;
+
+public interface TextProcessor {
+    ParsingResult process(String key, ChangePack message);
+}
diff --git a/src/main/java/org/onap/avcnmanager/message/processing/targets/MapTarget.java b/src/main/java/org/onap/avcnmanager/message/processing/targets/MapTarget.java
new file mode 100644
index 0000000..c9c1ae8
--- /dev/null
+++ b/src/main/java/org/onap/avcnmanager/message/processing/targets/MapTarget.java
@@ -0,0 +1,46 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Simulator
+ * ================================================================================
+ * Copyright (C) 2021 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.avcnmanager.message.processing.targets;
+
+import java.util.Collections;
+import java.util.Map;
+
+public class MapTarget implements TargetContainer<String> {
+    private final Map<String,String> map;
+
+    public MapTarget(Map<String, String> map) {
+        this.map = map;
+    }
+
+    @Override
+    public void acceptOne(String argument) {
+        //no op
+    }
+
+    @Override
+    public void acceptPair(String first, String second) {
+        map.put(first, second);
+    }
+
+    public Map<String,String> mapValue() {
+        return Collections.unmodifiableMap(map);
+    }
+}
diff --git a/src/main/java/org/onap/avcnmanager/message/processing/targets/StringBuilderTarget.java b/src/main/java/org/onap/avcnmanager/message/processing/targets/StringBuilderTarget.java
new file mode 100644
index 0000000..6b04da0
--- /dev/null
+++ b/src/main/java/org/onap/avcnmanager/message/processing/targets/StringBuilderTarget.java
@@ -0,0 +1,51 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Simulator
+ * ================================================================================
+ * Copyright (C) 2021 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.avcnmanager.message.processing.targets;
+
+public class StringBuilderTarget implements TargetContainer<String> {
+    private final StringBuilder stringBuilder;
+    private final String delimiter;
+
+    public StringBuilderTarget(StringBuilder stringBuilder, String delimiter) {
+        this.stringBuilder = stringBuilder;
+        this.delimiter = delimiter;
+    }
+
+    @Override
+    public void acceptOne(String argument) {
+        stringBuilder.append(argument).append(delimiter);
+    }
+
+    @Override
+    public void acceptPair(String first, String second) {
+        //no op
+    }
+
+    public String stringValue() {
+        return stringBuilder.toString();
+    }
+
+    public void trimLastDelimiter() {
+        if(this.stringBuilder.lastIndexOf(delimiter) == this.stringBuilder.length() - delimiter.length()) {
+            this.stringBuilder.setLength(this.stringBuilder.length() - delimiter.length());
+        }
+    }
+}
diff --git a/src/main/java/org/onap/avcnmanager/message/processing/targets/TargetContainer.java b/src/main/java/org/onap/avcnmanager/message/processing/targets/TargetContainer.java
new file mode 100644
index 0000000..8357ac2
--- /dev/null
+++ b/src/main/java/org/onap/avcnmanager/message/processing/targets/TargetContainer.java
@@ -0,0 +1,26 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Simulator
+ * ================================================================================
+ * Copyright (C) 2021 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.avcnmanager.message.processing.targets;
+
+public interface TargetContainer<T> {
+    void acceptOne(T argument);
+    void acceptPair(T first, T second);
+}
diff --git a/src/main/java/org/onap/avcnmanager/message/processing/tokens/BaseToken.java b/src/main/java/org/onap/avcnmanager/message/processing/tokens/BaseToken.java
new file mode 100644
index 0000000..11e79bf
--- /dev/null
+++ b/src/main/java/org/onap/avcnmanager/message/processing/tokens/BaseToken.java
@@ -0,0 +1,32 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Simulator
+ * ================================================================================
+ * Copyright (C) 2021 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.avcnmanager.message.processing.tokens;
+
+public abstract class BaseToken implements Token {
+    private final String value;
+    BaseToken(String value) {
+        this.value = value;
+    }
+
+    public String getValue() {
+        return value;
+    }
+}
diff --git a/src/main/java/org/onap/avcnmanager/message/processing/tokens/ContainerToken.java b/src/main/java/org/onap/avcnmanager/message/processing/tokens/ContainerToken.java
new file mode 100644
index 0000000..1d56889
--- /dev/null
+++ b/src/main/java/org/onap/avcnmanager/message/processing/tokens/ContainerToken.java
@@ -0,0 +1,49 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Simulator
+ * ================================================================================
+ * Copyright (C) 2021 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.avcnmanager.message.processing.tokens;
+
+import org.onap.avcnmanager.message.processing.targets.TargetContainer;
+
+import java.util.AbstractMap;
+import java.util.Map;
+
+public class ContainerToken extends BaseToken {
+    private static final Map.Entry<String,String> EMPTY_ENTRY = new AbstractMap.SimpleEntry<>("","");
+
+    public ContainerToken(String value) {
+        super(value);
+    }
+
+    @Override
+    public String stringValue() {
+        return getValue() + "= " + getValue();
+    }
+
+    @Override
+    public Map.Entry<String, String> pairValue() {
+        return EMPTY_ENTRY;
+    }
+
+    @Override
+    public void dump(TargetContainer<String> targetContainer) {
+        targetContainer.acceptOne(stringValue());
+    }
+}
diff --git a/src/main/java/org/onap/avcnmanager/message/processing/tokens/ListToken.java b/src/main/java/org/onap/avcnmanager/message/processing/tokens/ListToken.java
new file mode 100644
index 0000000..7046de7
--- /dev/null
+++ b/src/main/java/org/onap/avcnmanager/message/processing/tokens/ListToken.java
@@ -0,0 +1,59 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Simulator
+ * ================================================================================
+ * Copyright (C) 2021 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.avcnmanager.message.processing.tokens;
+
+import org.onap.avcnmanager.message.processing.targets.TargetContainer;
+
+import java.util.AbstractMap;
+import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+public class ListToken extends BaseToken {
+    private static final Map.Entry<String,String> EMPTY_ENTRY = new AbstractMap.SimpleEntry<>("","");
+    private static final Pattern PATTERN_LIST = Pattern.compile("(.*)?\\[(.*)?='(.*)?'\\]");
+
+    public ListToken(String value) {
+        super(value);
+    }
+
+    @Override
+    public String stringValue() {
+        StringBuilder sb = new StringBuilder();
+        Matcher m = PATTERN_LIST.matcher(getValue());
+        if (m.find()) {
+            String listName = m.group(1);
+            String value = m.group(3);
+            sb.append(listName).append("=").append(value);
+        }
+        return sb.toString();
+    }
+
+    @Override
+    public Map.Entry<String, String> pairValue() {
+        return EMPTY_ENTRY;
+    }
+
+    @Override
+    public void dump(TargetContainer<String> targetContainer) {
+        targetContainer.acceptOne(stringValue());
+    }
+}
diff --git a/src/main/java/org/onap/avcnmanager/message/processing/tokens/Token.java b/src/main/java/org/onap/avcnmanager/message/processing/tokens/Token.java
new file mode 100644
index 0000000..26720e5
--- /dev/null
+++ b/src/main/java/org/onap/avcnmanager/message/processing/tokens/Token.java
@@ -0,0 +1,31 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Simulator
+ * ================================================================================
+ * Copyright (C) 2021 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.avcnmanager.message.processing.tokens;
+
+import org.onap.avcnmanager.message.processing.targets.TargetContainer;
+
+import java.util.Map;
+
+public interface Token {
+    String stringValue();
+    Map.Entry<String,String> pairValue();
+    void dump(TargetContainer<String> targetContainer);
+}
diff --git a/src/main/java/org/onap/avcnmanager/message/processing/tokens/ValueToken.java b/src/main/java/org/onap/avcnmanager/message/processing/tokens/ValueToken.java
new file mode 100644
index 0000000..336de21
--- /dev/null
+++ b/src/main/java/org/onap/avcnmanager/message/processing/tokens/ValueToken.java
@@ -0,0 +1,60 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Simulator
+ * ================================================================================
+ * Copyright (C) 2021 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.avcnmanager.message.processing.tokens;
+
+import org.onap.avcnmanager.message.processing.targets.TargetContainer;
+
+import java.util.AbstractMap;
+import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+public class ValueToken extends BaseToken {
+    private static final Map.Entry<String,String> EMPTY_ENTRY = new AbstractMap.SimpleEntry<>("","");
+    private static final Pattern PATTERN_VALUE = Pattern.compile("(.*)=(.*)");
+
+    public ValueToken(String value) {
+        super(value);
+    }
+
+    @Override
+    public String stringValue() {
+        return "";
+    }
+
+    @Override
+    public Map.Entry<String, String> pairValue() {
+        Map.Entry<String,String> entry = EMPTY_ENTRY;
+        Matcher m = PATTERN_VALUE.matcher(getValue());
+        if (m.find()) {
+            String paramName = m.group(1).trim();
+            String paramValue = m.group(2).trim();
+            entry = new AbstractMap.SimpleEntry<>(paramName, paramValue);
+        }
+        return entry;
+    }
+
+    @Override
+    public void dump(TargetContainer<String> targetContainer) {
+        Map.Entry<String,String> entry = pairValue();
+        targetContainer.acceptPair(entry.getKey(), entry.getValue());
+    }
+}
diff --git a/src/main/java/org/onap/avcnmanager/message/serializers/ChangePackDeserializer.java b/src/main/java/org/onap/avcnmanager/message/serializers/ChangePackDeserializer.java
new file mode 100644
index 0000000..9b7feb6
--- /dev/null
+++ b/src/main/java/org/onap/avcnmanager/message/serializers/ChangePackDeserializer.java
@@ -0,0 +1,46 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Simulator
+ * ================================================================================
+ * Copyright (C) 2021 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.avcnmanager.message.serializers;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.kafka.common.serialization.Deserializer;
+import org.onap.avcnmanager.message.data.ChangePack;
+
+import java.io.IOException;
+import java.util.Map;
+
+public class ChangePackDeserializer implements Deserializer<ChangePack> {
+    private ObjectMapper objectMapper;
+
+    @Override
+    public void configure(Map<String, ?> configs, boolean isKey) {
+        objectMapper = new ObjectMapper();
+    }
+
+    @Override
+    public ChangePack deserialize(String s, byte[] bytes) {
+        try {
+            return objectMapper.readValue(bytes, ChangePack.class);
+        } catch (IOException e) {
+            throw new RuntimeException("Can not deserialize ChangePack from bytes: " + e.getMessage());
+        }
+    }
+}
diff --git a/src/main/java/org/onap/avcnmanager/message/serializers/ChangePackSerializer.java b/src/main/java/org/onap/avcnmanager/message/serializers/ChangePackSerializer.java
new file mode 100644
index 0000000..38988c4
--- /dev/null
+++ b/src/main/java/org/onap/avcnmanager/message/serializers/ChangePackSerializer.java
@@ -0,0 +1,45 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Simulator
+ * ================================================================================
+ * Copyright (C) 2021 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.avcnmanager.message.serializers;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.kafka.common.serialization.Serializer;
+import org.onap.avcnmanager.message.data.ChangePack;
+
+import java.util.Map;
+
+public class ChangePackSerializer implements Serializer<ChangePack> {
+    private ObjectMapper objectMapper;
+    @Override public void configure(Map<String, ?> configs, boolean isKey) {
+        objectMapper = new ObjectMapper();
+    }
+    @Override
+    public byte[] serialize(String s, ChangePack changePack) {
+        byte[] retVal;
+        try {
+            retVal = objectMapper.writeValueAsBytes(changePack);
+        } catch (JsonProcessingException e) {
+            throw new RuntimeException("Can not serialize ChangePack instance: " + changePack + "\nReason: " + e);
+        }
+        return retVal;
+    }
+}
diff --git a/src/main/java/org/onap/avcnmanager/utils/FileUtils.java b/src/main/java/org/onap/avcnmanager/utils/FileUtils.java
new file mode 100644
index 0000000..990fdf3
--- /dev/null
+++ b/src/main/java/org/onap/avcnmanager/utils/FileUtils.java
@@ -0,0 +1,39 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Simulator
+ * ================================================================================
+ * Copyright (C) 2021 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.avcnmanager.utils;
+
+import org.apache.commons.io.IOUtils;
+import org.springframework.stereotype.Component;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.Charset;
+
+@Component
+public class FileUtils {
+    public String readStringFromResourceFile(String pnfPayloadTemplate) {
+        try(InputStream inputStream = getClass().getClassLoader().getResourceAsStream(pnfPayloadTemplate)) {
+            return IOUtils.toString(inputStream, Charset.defaultCharset());
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+}
diff --git a/src/main/java/org/onap/avcnmanager/utils/JsonUtils.java b/src/main/java/org/onap/avcnmanager/utils/JsonUtils.java
new file mode 100644
index 0000000..7ee0dce
--- /dev/null
+++ b/src/main/java/org/onap/avcnmanager/utils/JsonUtils.java
@@ -0,0 +1,67 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Simulator
+ * ================================================================================
+ * Copyright (C) 2021 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.avcnmanager.utils;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.onap.avcnmanager.message.processing.ParsingResult;
+import org.springframework.stereotype.Component;
+
+import java.io.IOException;
+
+@Component
+public class JsonUtils {
+    private final ObjectMapper objectMapper;
+    private final FileUtils ioUtils;
+
+    public JsonUtils(FileUtils ioUtils) {
+        this.objectMapper = new ObjectMapper();
+        this.ioUtils = ioUtils;
+    }
+
+    public JsonNode convertToJsonNode(String json) {
+        try {
+            return objectMapper.readTree(json);
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    public String convertToJsonString(ParsingResult message) {
+        try {
+            return objectMapper.writeValueAsString(message);
+        } catch (JsonProcessingException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    public ObjectNode objectNodeFromJsonNode(JsonNode jsonNode) {
+        return (ObjectNode) jsonNode;
+    }
+
+    public JsonNode jsonNodeFromResourceFile(String fileName) {
+        String json = ioUtils.readStringFromResourceFile(fileName);
+        return convertToJsonNode(json);
+    }
+
+}
diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties
new file mode 100644
index 0000000..e7b3904
--- /dev/null
+++ b/src/main/resources/application.properties
@@ -0,0 +1,5 @@
+kafka.bootstrap.servers=kafka1:9092
+kafka.application.id=avcn-simulator
+kafka.source.topic=config
+rest.client.pnfsimulator.endpoint=http://pnf-simulator:5000/simulator/start
+rest.client.ves.endpoint=http://VES-HOST:8080/eventListener/v7
diff --git a/src/main/resources/logback.xml b/src/main/resources/logback.xml
new file mode 100644
index 0000000..3a6f4a5
--- /dev/null
+++ b/src/main/resources/logback.xml
@@ -0,0 +1,56 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ============LICENSE_START=======================================================
+  Simulator
+  ================================================================================
+  Copyright (C) 2021 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=========================================================
+  -->
+
+<Configuration complete="true" compact="true">
+
+    <Property name="outputFilename" value="avcnsimulator_output"/>
+    <Property name="log-path" value="/var/log/ONAP/avcnsimulator"/>
+    <Property name="archive" value="/var/log/ONAP/avcnsimulator/archive"/>
+    <property name="maxFileSize" value="50MB"/>
+    <property name="maxHistory" value="30"/>
+    <property name="totalSizeCap" value="10GB"/>
+    <include resource="org/springframework/boot/logging/logback/defaults.xml"/>
+
+    <appender name="Console" target="SYSTEM_OUT" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>${CONSOLE_LOG_PATTERN}</pattern>
+        </encoder>
+    </appender>
+
+    <appender name="ROLLING-FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
+        <encoder>
+            <pattern>${FILE_LOG_PATTERN}</pattern>
+        </encoder>
+        <File>${log-path}/${outputFilename}.log</File>
+        <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
+            <FileNamePattern>${archive}/${outputFilename}.%d{yyyy-MM-dd}.%i.log.zip</FileNamePattern>
+            <MaxFileSize>${maxFileSize}</MaxFileSize>
+            <MaxHistory>${maxHistory}</MaxHistory>
+            <TotalSizeCap>${totalSizeCap}</TotalSizeCap>
+        </rollingPolicy>
+    </appender>
+
+    <root level="info">
+        <appender-ref ref="Console"/>
+        <appender-ref ref="ROLLING-FILE"/>
+    </root>
+
+</Configuration>
diff --git a/src/main/resources/pnfsim_payload_notification.json b/src/main/resources/pnfsim_payload_notification.json
new file mode 100644
index 0000000..7172cb2
--- /dev/null
+++ b/src/main/resources/pnfsim_payload_notification.json
@@ -0,0 +1,13 @@
+{
+  "simulatorParams": {
+    "repeatCount": 1,
+    "repeatInterval": 1,
+    "vesServerUrl": ""
+  },
+  "templateName": "notification.json",
+  "patch": {},
+  "variables": {
+    "dn": "",
+    "attributesList": {}
+  }
+}
\ No newline at end of file
diff --git a/src/test/java/org/onap/avcnmanager/config/AVCNConfigTest.java b/src/test/java/org/onap/avcnmanager/config/AVCNConfigTest.java
new file mode 100644
index 0000000..5d2f43c
--- /dev/null
+++ b/src/test/java/org/onap/avcnmanager/config/AVCNConfigTest.java
@@ -0,0 +1,70 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Simulator
+ * ================================================================================
+ * Copyright (C) 2021 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.avcnmanager.config;
+
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.Topology;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.onap.avcnmanager.message.handlers.MessageHandler;
+
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class AVCNConfigTest {
+
+
+    private AVCNConfig kafkaStreamConfig;
+
+    @Mock
+    private MessageHandler messageHandler;
+
+    @BeforeEach
+    void setUp() {
+        MockitoAnnotations.initMocks(this);
+        kafkaStreamConfig = new AVCNConfig();
+    }
+
+    @Test
+    void validateKafkaStreamPropertiesAreCorrect() {
+        final String testServer = "http://kafka/test";
+        final String testApplication = "applicationTest";
+
+        Properties properties = kafkaStreamConfig.getKafkaStreamProperties(testServer, testApplication);
+
+        assertEquals(properties.getProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG), testServer);
+        assertEquals(properties.getProperty(StreamsConfig.APPLICATION_ID_CONFIG), testApplication);
+    }
+
+    @Test
+    void validateKafkaStreamTopologyIsCorrect() {
+        final String testTopic = "testTopic";
+
+        Topology topology = kafkaStreamConfig.getKafkaStreamTopology(testTopic, messageHandler);
+
+        assertTrue(topology.describe().toString().contains(testTopic));
+    }
+
+}
diff --git a/src/test/java/org/onap/avcnmanager/kafka/stream/AVCNKafkaStreamTest.java b/src/test/java/org/onap/avcnmanager/kafka/stream/AVCNKafkaStreamTest.java
new file mode 100644
index 0000000..2d30f92
--- /dev/null
+++ b/src/test/java/org/onap/avcnmanager/kafka/stream/AVCNKafkaStreamTest.java
@@ -0,0 +1,53 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Simulator
+ * ================================================================================
+ * Copyright (C) 2021 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.avcnmanager.kafka.stream;
+
+import org.apache.kafka.streams.KafkaStreams;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+class AVCNKafkaStreamTest {
+
+
+    @Mock
+    private KafkaStreams streams;
+
+    @BeforeEach
+    void setUp() {
+        MockitoAnnotations.initMocks(this);
+    }
+
+    @Test
+    void validateAVCNKafkaStream() {
+        AVCNKafkaStream stream = new AVCNKafkaStream(streams);
+        stream.startKafkaStream();
+
+        verify(streams, times(1) ).setUncaughtExceptionHandler(any(Thread.UncaughtExceptionHandler.class));
+        verify(streams, times(1) ).start();
+    }
+
+}
diff --git a/src/test/java/org/onap/avcnmanager/message/forwarders/RestForwarderTest.java b/src/test/java/org/onap/avcnmanager/message/forwarders/RestForwarderTest.java
new file mode 100644
index 0000000..6d224cb
--- /dev/null
+++ b/src/test/java/org/onap/avcnmanager/message/forwarders/RestForwarderTest.java
@@ -0,0 +1,74 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Simulator
+ * ================================================================================
+ * Copyright (C) 2021 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.avcnmanager.message.forwarders;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.onap.avcnmanager.message.processing.ParsingResult;
+import org.onap.avcnmanager.utils.FileUtils;
+import org.onap.avcnmanager.utils.JsonUtils;
+import org.springframework.http.HttpEntity;
+import org.springframework.web.client.RestTemplate;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+class RestForwarderTest {
+    private final JsonUtils jsonUtils;
+
+    RestForwarderTest() {
+        jsonUtils = new JsonUtils(new FileUtils());
+    }
+
+    @Test
+    @SuppressWarnings("unchecked")
+    void shouldSendResultsAsValidPnfSimRequest() {
+        RestTemplate restTemplateMock = mock(RestTemplate.class);
+
+        String pnfSimulatorEndpoint = "http://pnfsim_endpoint";
+        String vesEndpoint = "http://some_ves_endpoint";
+        String expectedPayload = new FileUtils().readStringFromResourceFile("sample_payload_notification.json");
+        JsonNode expectedJson = jsonUtils.convertToJsonNode(expectedPayload);
+        ((ObjectNode)expectedJson.get("simulatorParams")).put("vesServerUrl", vesEndpoint);
+        Map<String,String> sampleAttributes = new HashMap<>();
+        sampleAttributes.put("attribute1", "value1");
+        ParsingResult parsingResult = new ParsingResult("some/dn", sampleAttributes);
+
+        ArgumentCaptor<HttpEntity<String>> entityCaptor = ArgumentCaptor.forClass(HttpEntity.class);
+        ArgumentCaptor<String> endpointCaptor = ArgumentCaptor.forClass(String.class);
+
+        RestForwarder restForwarder = new RestForwarder(pnfSimulatorEndpoint, vesEndpoint, restTemplateMock, jsonUtils);
+        restForwarder.send(parsingResult);
+
+        verify(restTemplateMock).postForObject(endpointCaptor.capture(), entityCaptor.capture(), eq(String.class));
+        String stringBody = entityCaptor.getValue().getBody();
+        JsonNode node = jsonUtils.convertToJsonNode(stringBody);
+        assertEquals(expectedJson, node);
+        assertEquals(pnfSimulatorEndpoint, endpointCaptor.getValue());
+    }
+}
diff --git a/src/test/java/org/onap/avcnmanager/message/handlers/NewMessageHandlerTest.java b/src/test/java/org/onap/avcnmanager/message/handlers/NewMessageHandlerTest.java
new file mode 100644
index 0000000..96543f7
--- /dev/null
+++ b/src/test/java/org/onap/avcnmanager/message/handlers/NewMessageHandlerTest.java
@@ -0,0 +1,97 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Simulator
+ * ================================================================================
+ * Copyright (C) 2021 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.avcnmanager.message.handlers;
+
+import org.assertj.core.util.Maps;
+import org.junit.jupiter.api.Test;
+import org.onap.avcnmanager.message.data.Change;
+import org.onap.avcnmanager.message.data.ChangePack;
+import org.onap.avcnmanager.message.forwarders.Forwarder;
+import org.onap.avcnmanager.message.processing.ParsingResult;
+import org.onap.avcnmanager.message.processing.TextProcessor;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+class NewMessageHandlerTest {
+
+    @Test
+    void emptyChangeShouldBeSkippedFromProcessing() {
+        TextProcessor mockProcessor = mock(TextProcessor.class);
+        Forwarder mockForwarder = mock(Forwarder.class);
+        NewMessageHandler handler = new NewMessageHandler(mockProcessor, mockForwarder);
+
+        String key = "key";
+        ChangePack message = new ChangePack(emptyChange(), emptyChange(), "");
+        handler.handleMessage(key, message);
+
+        verify(mockProcessor, never()).process(key, message);
+        verify(mockProcessor, never()).process(anyString(), any(ChangePack.class));
+        verify(mockForwarder, never()).send(any(ParsingResult.class));
+    }
+
+    @Test
+    void validMessagesShouldBePassedToTextProcessor() {
+        TextProcessor mockProcessor = mock(TextProcessor.class);
+        Forwarder mockForwarder = mock(Forwarder.class);
+        NewMessageHandler handler = new NewMessageHandler(mockProcessor, mockForwarder);
+
+        String key = "key";
+        ChangePack message = new ChangePack(new Change("aaaa", "bbbb"), emptyChange(), "1");
+        handler.handleMessage(key, message);
+
+        verify(mockProcessor, times(1)).process(key, message);
+        verify(mockProcessor, times(1)).process(anyString(), any(ChangePack.class));
+    }
+
+    @Test
+    void processedChangeShouldBePassedToForwarder() {
+        List<ParsingResult> trapList = new ArrayList<>();
+        String key = "key";
+        ChangePack message = new ChangePack(new Change("aaaa", "bbbb"), emptyChange(), "1");
+
+        TextProcessor mockProcessor = this::simpleProcessing;
+        Forwarder mockForwarder = trapList::add;
+
+        NewMessageHandler handler = new NewMessageHandler(mockProcessor, mockForwarder);
+        handler.handleMessage(key, message);
+
+        assertEquals(1, trapList.size());
+        ParsingResult expected = simpleProcessing(key, message);
+        assertEquals(expected, trapList.get(0));
+    }
+
+    private ParsingResult simpleProcessing(String key, ChangePack message) {
+        return new ParsingResult(message.getNew().getPath(), Maps.newHashMap(message.getNew().getValue(), message.getType()));
+    }
+
+    private static Change emptyChange() {
+        return new Change("", "");
+    }
+}
diff --git a/src/test/java/org/onap/avcnmanager/message/processing/NetconfTextProcessorTest.java b/src/test/java/org/onap/avcnmanager/message/processing/NetconfTextProcessorTest.java
new file mode 100644
index 0000000..1afdadd
--- /dev/null
+++ b/src/test/java/org/onap/avcnmanager/message/processing/NetconfTextProcessorTest.java
@@ -0,0 +1,111 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Simulator
+ * ================================================================================
+ * Copyright (C) 2021 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.avcnmanager.message.processing;
+
+import org.junit.jupiter.api.Test;
+import org.onap.avcnmanager.message.data.Change;
+import org.onap.avcnmanager.message.data.ChangePack;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class NetconfTextProcessorTest {
+
+    @Test
+    void shouldReturnCorrentDnWithEmptyAttributeListForOneContainerOneListInstance() {
+        ChangePack changePack = new ChangePack(new Change("/example-sports:sports/person[name='name 1']",""), emptyChange(), "1");
+
+        String expectedDn = "example-sports:sports= example-sports:sports , person=name 1";
+        ParsingResult expected = new ParsingResult(expectedDn, Collections.emptyMap());
+        ParsingResult actual = new NetconfTextProcessor().process("aaa", changePack);
+
+        assertEquals(expected.getDn(), actual.getDn());
+        assertEquals(expected.getAttributesList(), actual.getAttributesList());
+    }
+
+    @Test
+    void shouldReturnCorrentDnWithFilledAttributeListForOneContainerOneListOneValue() {
+        ChangePack changePack = new ChangePack(new Change("/example-sports:sports/person[name='name 1']/name","name 2"), emptyChange(), "1");
+
+        String expectedDn = "example-sports:sports= example-sports:sports , person=name 1";
+        Map<String,String> expectedAttributesMap = new HashMap<>();
+        expectedAttributesMap.put("name", "name 2");
+
+        ParsingResult actual = new NetconfTextProcessor().process("aaa", changePack);
+
+        assertEquals(expectedDn, actual.getDn());
+        assertThat(actual.getAttributesList()).hasSameSizeAs(expectedAttributesMap);
+        assertThat(actual.getAttributesList()).containsAllEntriesOf(expectedAttributesMap);
+    }
+
+    @Test
+    void oneContainerTwoListsNestedNoValue() {
+        ChangePack changePack = new ChangePack(
+                new Change("/example-sports:sports/team[name='team 1']/player[name='player 1']",""), emptyChange(), "1");
+
+        String expectedDn = "example-sports:sports= example-sports:sports , player=player 1";
+        Map<String,String> expectedAttributesMap = Collections.emptyMap();
+
+        ParsingResult actual = new NetconfTextProcessor().process("aaa", changePack);
+
+        assertEquals(expectedDn, actual.getDn());
+        assertThat(actual.getAttributesList()).hasSameSizeAs(expectedAttributesMap);
+        assertThat(actual.getAttributesList()).containsAllEntriesOf(expectedAttributesMap);
+    }
+
+    @Test
+    void oneContainerTwoListsNestedOneValue() {
+        ChangePack changePack = new ChangePack(
+                new Change("/example-sports:sports/team[name='team 1']/player[name='player 1']/name","player 1"), emptyChange(), "1");
+
+        String expectedDn = "example-sports:sports= example-sports:sports , player=player 1";
+        Map<String,String> expectedAttributesMap = new HashMap<>();
+        expectedAttributesMap.put("name", "player 1");
+
+        ParsingResult actual = new NetconfTextProcessor().process("aaa", changePack);
+
+        assertEquals(expectedDn, actual.getDn());
+        assertThat(actual.getAttributesList()).hasSameSizeAs(expectedAttributesMap);
+        assertThat(actual.getAttributesList()).containsAllEntriesOf(expectedAttributesMap);
+    }
+
+    @Test
+    void emptyChangeShouldGenerateEmptyResponse() {
+        ChangePack changePack = new ChangePack(new Change("",""), emptyChange(), "1");
+
+        String expectedDn = "";
+        Map<String,String> expectedAttributesMap = new HashMap<>();
+
+        ParsingResult actual = new NetconfTextProcessor().process("aaa", changePack);
+
+        assertEquals(expectedDn, actual.getDn());
+        assertThat(actual.getAttributesList()).hasSameSizeAs(expectedAttributesMap);
+        assertThat(actual.getAttributesList()).containsAllEntriesOf(expectedAttributesMap);
+    }
+
+    private static Change emptyChange() {
+        return new Change("", "");
+    }
+}
diff --git a/src/test/java/org/onap/avcnmanager/message/serializers/ChangePackSerializersTest.java b/src/test/java/org/onap/avcnmanager/message/serializers/ChangePackSerializersTest.java
new file mode 100644
index 0000000..376be47
--- /dev/null
+++ b/src/test/java/org/onap/avcnmanager/message/serializers/ChangePackSerializersTest.java
@@ -0,0 +1,83 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Simulator
+ * ================================================================================
+ * Copyright (C) 2021 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.avcnmanager.message.serializers;
+
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.Serializer;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.onap.avcnmanager.message.data.Change;
+import org.onap.avcnmanager.message.data.ChangePack;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+class ChangePackSerializersTest {
+    private static Serializer<ChangePack> serializer;
+    private static Deserializer<ChangePack> deserializer;
+    private final List<ChangePack> testList;
+
+    ChangePackSerializersTest() {
+        testList = new ArrayList<>();
+        for (int i = 0; i < 1000; i++) {
+            Change aNew = new Change(Integer.toString(i),Integer.toString(i));
+            Change old = new Change(Integer.toString(i-1),Integer.toString(i-1));
+            ChangePack changePack = new ChangePack(aNew, old, "1");
+            testList.add(changePack);
+        }
+    }
+
+    String[] getTopics() {
+        return new String[] { "test" };
+    }
+
+    @BeforeAll
+    static void setup() {
+        serializer = new ChangePackSerializer();
+        deserializer = new ChangePackDeserializer();
+        serializer.configure(null, false);
+        deserializer.configure(null, false);
+    }
+
+    @Test
+    void shouldSerializeAndDeserializeCorrectly() {
+        for(String topic : getTopics()) {
+            testList.forEach(x -> {
+                ChangePack changePack = deserializer.deserialize(topic, serializer.serialize(topic, x));
+                assertThat(changePack).isNotSameAs(x);
+                assertThat(changePack).isEqualTo(x);
+            });
+        }
+    }
+
+    @Test
+    void shouldThrowWhenDeserializeGarbage() {
+        byte[] bytes = serializer.serialize("", testList.get(0));
+        new Random(System.currentTimeMillis()).nextBytes(bytes);
+
+        assertThrows(RuntimeException.class, () -> deserializer.deserialize("", bytes));
+    }
+
+}
diff --git a/src/test/resources/sample_payload_notification.json b/src/test/resources/sample_payload_notification.json
new file mode 100644
index 0000000..7528fe0
--- /dev/null
+++ b/src/test/resources/sample_payload_notification.json
@@ -0,0 +1,15 @@
+{
+  "simulatorParams": {
+    "repeatCount": 1,
+    "repeatInterval": 1,
+    "vesServerUrl": "http://ves"
+  },
+  "templateName": "notification.json",
+  "patch": {},
+  "variables": {
+    "dn": "some/dn",
+    "attributesList": {
+      "attribute1": "value1"
+    }
+  }
+}
\ No newline at end of file