HV VES Collector seed code

Contains squashed commits up to 11fe6b63 (2018-05-30). The whole
contains a basic project structure. We are trying to put rest of the
commits one by one so we do not loose the history.

Bellow there are messages of the single commits in this squashed bulk:

Basic project setup
Create base maven project with Gitlab CI configuration.
Piotr Jaszczyk <piotr.jaszczyk@nokia.com>

Merging guildeline
Piotr Jaszczyk <piotr.jaszczyk@nokia.com>

Add remote branch delete command
Piotr Jaszczyk <piotr.jaszczyk@nokia.com>

Sample runtime in Kotlin - PoC
Piotr Jaszczyk <piotr.jaszczyk@nokia.com>

Setup project internal architecture
Piotr Jaszczyk <piotr.jaszczyk@nokia.com>

Message routing
Determine target topic and partition by VES Common Header.
Piotr Jaszczyk <piotr.jaszczyk@nokia.com>

Parse GPB message header
fkrzywka <filip.krzywka@nokia.com>

Set listen port based on command line args
Use Apache Commons CLI to parse cmd line args.
Piotr Jaszczyk <piotr.jaszczyk@nokia.com>

Drop invalid GPB messages
Instead of propagating error and closing stream just drop the message
and proceed. Final handling logic may include closing the connection or
sending some message depending on the specification.
Piotr Jaszczyk <piotr.jaszczyk@nokia.com>

Add Apache license file
Piotr Jaszczyk <piotr.jaszczyk@nokia.com>

Convert to maven multi-module project
fkrzywka <filip.krzywka@nokia.com>

Component tests with current GPB schema
* Using v5 draft protobuf definition
* Code reorganized to so component boundaries are more visible
Piotr Jaszczyk <piotr.jaszczyk@nokia.com>

Thin logging facade over slf4j
Piotr Jaszczyk <piotr.jaszczyk@nokia.com>

Introduce code analysis tools
Piotr Jaszczyk <piotr.jaszczyk@nokia.com>

Implemented reading configuration from consul

Ves Common Header validation added (required parameters existance check)

Micro benchmark for direct vs on-heap NIO buffers
Piotr Jaszczyk <piotr.jaszczyk@nokia.com>

Decode wire protocol and fix (most?) memory leaks
Proposed wire protocol is just a suggestion and will (should) change
in the future.
Netty's ByteBuf is a reference-counted wrapper over a memory chunk. It
is crucial to free unused buffers by means of release() method.
The general rule regarding memory management was suggested. Let's put
all memory-cleanup logic in main VesHvCollector class so other classes
could focus on their job.
Piotr Jaszczyk <piotr.jaszczyk@nokia.com>

Minor cleanup
Piotr Jaszczyk <piotr.jaszczyk@nokia.com>

Add license info in files
Piotr Jaszczyk <piotr.jaszczyk@nokia.com>

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

58 files changed, 4641 insertions, 0 deletions

diff --git a/.gitignore b/.gitignore
new file mode 100644
index 00000000..11ddc442
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,52 @@
+##################
+## MAVEN RULES  ##
+##################
+
+target/
+pom.xml.tag
+pom.xml.releaseBackup
+pom.xml.versionsBackup
+pom.xml.next
+release.properties
+dependency-reduced-pom.xml
+buildNumber.properties
+.mvn/timing.properties
+
+# Avoid ignoring Maven wrapper jar file (.jar files are usually ignored)
+!/.mvn/wrapper/maven-wrapper.jar
+
+
+##################
+## JAVA RULES   ##
+##################
+
+# Compiled class file
+*.class
+
+# Log file
+*.log
+
+# BlueJ files
+*.ctxt
+
+# Mobile Tools for Java (J2ME)
+.mtj.tmp/
+
+# Package Files #
+*.jar
+*.war
+*.nar
+*.ear
+*.zip
+*.tar.gz
+*.rar
+
+# virtual machine crash logs, see http://www.java.com/en/download/help/error_hotspot.xml
+hs_err_pid*
+
+
+##################
+## OTHER RULES  ##
+##################
+.idea/
+*.iml
diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml
new file mode 100644
index 00000000..796bc236
--- /dev/null
+++ b/.gitlab-ci.yml
@@ -0,0 +1,16 @@
+image: archive.docker-registry.eecloud.nsn-net.net/imp/matryoshka:latest
+
+stages:
+  - build
+
+build:
+  stage: build
+  script:
+    - mvn -e -T2 -Panalysis clean install
+  artifacts:
+    paths:
+    - hv-collector-coverage/target/site/jacoco-aggregate
+    - hv-collector-core/target/reports
+    - hv-collector-main/target/reports
+    - hv-collector-utils/target/reports
+    
\ No newline at end of file
diff --git a/.gitreview b/.gitreview
new file mode 100644
index 00000000..fa0d2b2d
--- /dev/null
+++ b/.gitreview
@@ -0,0 +1,5 @@
+[gerrit]
+host=gerrit.onap.org
+port=29418
+project=dcaegen2/collectors/hv-ves.git
+
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
new file mode 100644
index 00000000..6fc7e0ae
--- /dev/null
+++ b/CONTRIBUTING.md
@@ -0,0 +1,74 @@
+# Contribution guidelines
+Project will probably be moved to ONAP Gerrit repository. Thus, please follow ONAP contribution guidelines.
+
+## Merging
+
+First of all let's avoid merge commits. To achieve it we should:
+
+* rebase master changes to task branches,
+* manually merge task branches to master with `--ff-only option`.
+
+The last part needs some comment. In Gerrit master history is linear and easy to follow. That is probably the greatest (only? :wink:) advantage of Gerrit over GitLab. In GitLab EE it is possible to configure MRs so merge commits are avoided. However in GitLab CE it is not possible. To mitigate this, we should manually merge branches into master.
+
+Some commands which you might find useful:
+
+```shell
+# 1. Update master
+git checkout master
+git fetch
+git merge -ff-only
+
+# 2. Rebase branch on fresh master
+git checkout $BRANCH
+git rebase -i master
+git push --force
+
+# 3. Merge branch
+git checkout master
+git merge --ff-only $BRANCH
+git branch -d $BRANCH
+git push origin --delete $BRANCH
+git push
+
+```
+
+## Sign-off rules
+Each commit has to be signed-off. By signing the commit you certify that you did follow rules defined in the document https://gerrit.onap.org/r/static/signoffrules.txt:
+
+```
+Sign your work
+
+To improve tracking of who did what, especially with contributions that
+can percolate to their final resting place in the code base through
+several layers of maintainers, we've introduced a "sign-off" procedure
+on all contributions submitted.
+
+The sign-off is a simple line at the end of the explanation for the
+contribution, which certifies that you wrote it or otherwise have the
+right to pass it on as an open-source contribution. When you sign-off
+your contribution you are stating the following:
+
+    By making a contribution to this project, I certify that:
+
+    (a) The contribution was created in whole or in part by me and I
+        have the right to submit it under the open source license
+        indicated in the file; or
+
+    (b) The contribution is based upon previous work that, to the best
+        of my knowledge, is covered under an appropriate open source
+        license and I have the right under that license to submit that
+        work with modifications, whether created in whole or in part
+        by me, under the same open source license (unless I am
+        permitted to submit under a different license), as indicated
+        in the file; or
+
+    (c) The contribution was provided directly to me by some other
+        person who certified (a), (b) or (c) and I have not modified
+        it.
+
+    (d) I understand and agree that this project and the contribution
+        are public and that a record of the contribution (including all
+        personal information I submit with it, including my sign-off) is
+        maintained indefinitely and may be redistributed consistent with
+        this project or the open source license(s) involved.
+```
diff --git a/LICENSE.txt b/LICENSE.txt
new file mode 100644
index 00000000..d6456956
--- /dev/null
+++ b/LICENSE.txt
@@ -0,0 +1,202 @@
+
+                                 Apache License
+                           Version 2.0, January 2004
+                        http://www.apache.org/licenses/
+
+   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+   1. Definitions.
+
+      "License" shall mean the terms and conditions for use, reproduction,
+      and distribution as defined by Sections 1 through 9 of this document.
+
+      "Licensor" shall mean the copyright owner or entity authorized by
+      the copyright owner that is granting the License.
+
+      "Legal Entity" shall mean the union of the acting entity and all
+      other entities that control, are controlled by, or are under common
+      control with that entity. For the purposes of this definition,
+      "control" means (i) the power, direct or indirect, to cause the
+      direction or management of such entity, whether by contract or
+      otherwise, or (ii) ownership of fifty percent (50%) or more of the
+      outstanding shares, or (iii) beneficial ownership of such entity.
+
+      "You" (or "Your") shall mean an individual or Legal Entity
+      exercising permissions granted by this License.
+
+      "Source" form shall mean the preferred form for making modifications,
+      including but not limited to software source code, documentation
+      source, and configuration files.
+
+      "Object" form shall mean any form resulting from mechanical
+      transformation or translation of a Source form, including but
+      not limited to compiled object code, generated documentation,
+      and conversions to other media types.
+
+      "Work" shall mean the work of authorship, whether in Source or
+      Object form, made available under the License, as indicated by a
+      copyright notice that is included in or attached to the work
+      (an example is provided in the Appendix below).
+
+      "Derivative Works" shall mean any work, whether in Source or Object
+      form, that is based on (or derived from) the Work and for which the
+      editorial revisions, annotations, elaborations, or other modifications
+      represent, as a whole, an original work of authorship. For the purposes
+      of this License, Derivative Works shall not include works that remain
+      separable from, or merely link (or bind by name) to the interfaces of,
+      the Work and Derivative Works thereof.
+
+      "Contribution" shall mean any work of authorship, including
+      the original version of the Work and any modifications or additions
+      to that Work or Derivative Works thereof, that is intentionally
+      submitted to Licensor for inclusion in the Work by the copyright owner
+      or by an individual or Legal Entity authorized to submit on behalf of
+      the copyright owner. For the purposes of this definition, "submitted"
+      means any form of electronic, verbal, or written communication sent
+      to the Licensor or its representatives, including but not limited to
+      communication on electronic mailing lists, source code control systems,
+      and issue tracking systems that are managed by, or on behalf of, the
+      Licensor for the purpose of discussing and improving the Work, but
+      excluding communication that is conspicuously marked or otherwise
+      designated in writing by the copyright owner as "Not a Contribution."
+
+      "Contributor" shall mean Licensor and any individual or Legal Entity
+      on behalf of whom a Contribution has been received by Licensor and
+      subsequently incorporated within the Work.
+
+   2. Grant of Copyright License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      copyright license to reproduce, prepare Derivative Works of,
+      publicly display, publicly perform, sublicense, and distribute the
+      Work and such Derivative Works in Source or Object form.
+
+   3. Grant of Patent License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      (except as stated in this section) patent license to make, have made,
+      use, offer to sell, sell, import, and otherwise transfer the Work,
+      where such license applies only to those patent claims licensable
+      by such Contributor that are necessarily infringed by their
+      Contribution(s) alone or by combination of their Contribution(s)
+      with the Work to which such Contribution(s) was submitted. If You
+      institute patent litigation against any entity (including a
+      cross-claim or counterclaim in a lawsuit) alleging that the Work
+      or a Contribution incorporated within the Work constitutes direct
+      or contributory patent infringement, then any patent licenses
+      granted to You under this License for that Work shall terminate
+      as of the date such litigation is filed.
+
+   4. Redistribution. You may reproduce and distribute copies of the
+      Work or Derivative Works thereof in any medium, with or without
+      modifications, and in Source or Object form, provided that You
+      meet the following conditions:
+
+      (a) You must give any other recipients of the Work or
+          Derivative Works a copy of this License; and
+
+      (b) You must cause any modified files to carry prominent notices
+          stating that You changed the files; and
+
+      (c) You must retain, in the Source form of any Derivative Works
+          that You distribute, all copyright, patent, trademark, and
+          attribution notices from the Source form of the Work,
+          excluding those notices that do not pertain to any part of
+          the Derivative Works; and
+
+      (d) If the Work includes a "NOTICE" text file as part of its
+          distribution, then any Derivative Works that You distribute must
+          include a readable copy of the attribution notices contained
+          within such NOTICE file, excluding those notices that do not
+          pertain to any part of the Derivative Works, in at least one
+          of the following places: within a NOTICE text file distributed
+          as part of the Derivative Works; within the Source form or
+          documentation, if provided along with the Derivative Works; or,
+          within a display generated by the Derivative Works, if and
+          wherever such third-party notices normally appear. The contents
+          of the NOTICE file are for informational purposes only and
+          do not modify the License. You may add Your own attribution
+          notices within Derivative Works that You distribute, alongside
+          or as an addendum to the NOTICE text from the Work, provided
+          that such additional attribution notices cannot be construed
+          as modifying the License.
+
+      You may add Your own copyright statement to Your modifications and
+      may provide additional or different license terms and conditions
+      for use, reproduction, or distribution of Your modifications, or
+      for any such Derivative Works as a whole, provided Your use,
+      reproduction, and distribution of the Work otherwise complies with
+      the conditions stated in this License.
+
+   5. Submission of Contributions. Unless You explicitly state otherwise,
+      any Contribution intentionally submitted for inclusion in the Work
+      by You to the Licensor shall be under the terms and conditions of
+      this License, without any additional terms or conditions.
+      Notwithstanding the above, nothing herein shall supersede or modify
+      the terms of any separate license agreement you may have executed
+      with Licensor regarding such Contributions.
+
+   6. Trademarks. This License does not grant permission to use the trade
+      names, trademarks, service marks, or product names of the Licensor,
+      except as required for reasonable and customary use in describing the
+      origin of the Work and reproducing the content of the NOTICE file.
+
+   7. Disclaimer of Warranty. Unless required by applicable law or
+      agreed to in writing, Licensor provides the Work (and each
+      Contributor provides its Contributions) on an "AS IS" BASIS,
+      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+      implied, including, without limitation, any warranties or conditions
+      of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+      PARTICULAR PURPOSE. You are solely responsible for determining the
+      appropriateness of using or redistributing the Work and assume any
+      risks associated with Your exercise of permissions under this License.
+
+   8. Limitation of Liability. In no event and under no legal theory,
+      whether in tort (including negligence), contract, or otherwise,
+      unless required by applicable law (such as deliberate and grossly
+      negligent acts) or agreed to in writing, shall any Contributor be
+      liable to You for damages, including any direct, indirect, special,
+      incidental, or consequential damages of any character arising as a
+      result of this License or out of the use or inability to use the
+      Work (including but not limited to damages for loss of goodwill,
+      work stoppage, computer failure or malfunction, or any and all
+      other commercial damages or losses), even if such Contributor
+      has been advised of the possibility of such damages.
+
+   9. Accepting Warranty or Additional Liability. While redistributing
+      the Work or Derivative Works thereof, You may choose to offer,
+      and charge a fee for, acceptance of support, warranty, indemnity,
+      or other liability obligations and/or rights consistent with this
+      License. However, in accepting such obligations, You may act only
+      on Your own behalf and on Your sole responsibility, not on behalf
+      of any other Contributor, and only if You agree to indemnify,
+      defend, and hold each Contributor harmless for any liability
+      incurred by, or claims asserted against, such Contributor by reason
+      of your accepting any such warranty or additional liability.
+
+   END OF TERMS AND CONDITIONS
+
+   APPENDIX: How to apply the Apache License to your work.
+
+      To apply the Apache License to your work, attach the following
+      boilerplate notice, with the fields enclosed by brackets "[]"
+      replaced with your own identifying information. (Don't include
+      the brackets!)  The text should be enclosed in the appropriate
+      comment syntax for the file format. We also recommend that a
+      file or class name and description of purpose be included on the
+      same "printed page" as the copyright notice for easier
+      identification within third-party archives.
+
+   Copyright [yyyy] [name of copyright owner]
+
+   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.
diff --git a/README.md b/README.md
new file mode 100644
index 00000000..4690eaf7
--- /dev/null
+++ b/README.md
@@ -0,0 +1,4 @@
+# VES High Volume Collector
+
+ONAP component for collecting high volume data, eg, RTPM (Real Time Performance Management).
+
diff --git a/hv-collector-analysis/pom.xml b/hv-collector-analysis/pom.xml
new file mode 100644
index 00000000..a977c5ae
--- /dev/null
+++ b/hv-collector-analysis/pom.xml
@@ -0,0 +1,39 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ ============LICENSE_START=======================================================
+  ~ dcaegen2-collectors-veshv
+  ~ ================================================================================
+  ~ Copyright (C) 2018 NOKIA Intellectual Property. 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>
+
+  <licenses>
+    <license>
+      <name>The Apache Software License, Version 2.0</name>
+      <url>http://www.apache.org/licenses/LICENSE-2.0.txt</url>
+    </license>
+  </licenses>
+
+  <groupId>org.onap.dcaegen2.collectors.veshv</groupId>
+  <artifactId>hv-collector-analysis</artifactId>
+  <version>1.0.0-SNAPSHOT</version>
+  <description>VES HighVolume Collector :: Code analysis configuration</description>
+
+
+</project>
\ No newline at end of file
diff --git a/hv-collector-analysis/src/main/resources/onap-detekt-config.yml b/hv-collector-analysis/src/main/resources/onap-detekt-config.yml
new file mode 100644
index 00000000..318d48ac
--- /dev/null
+++ b/hv-collector-analysis/src/main/resources/onap-detekt-config.yml
@@ -0,0 +1,475 @@
+autoCorrect: true
+failFast: false
+
+test-pattern: # Configure exclusions for test sources
+  active: true
+  patterns: # Test file regexes
+    - '.*/test/.*'
+    - '.*Test.kt'
+    - '.*Spec.kt'
+  exclude-rule-sets:
+    - 'comments'
+  exclude-rules:
+    - 'NamingRules'
+    - 'WildcardImport'
+    - 'MagicNumber'
+    - 'MaxLineLength'
+    - 'LateinitUsage'
+    - 'StringLiteralDuplication'
+    - 'SpreadOperator'
+    - 'TooManyFunctions'
+    - 'ForEachOnRange'
+
+build:
+  maxIssues: 3
+  weights:
+    # complexity: 2
+    # LongParameterList: 1
+    # style: 1
+    # comments: 1
+
+processors:
+  active: true
+  exclude:
+  # - 'FunctionCountProcessor'
+  # - 'PropertyCountProcessor'
+  # - 'ClassCountProcessor'
+  # - 'PackageCountProcessor'
+  # - 'KtFileCountProcessor'
+
+console-reports:
+  active: true
+  exclude:
+  #  - 'ProjectStatisticsReport'
+  #  - 'ComplexityReport'
+  #  - 'NotificationReport'
+  #  - 'FindingsReport'
+  #  - 'BuildFailureReport'
+
+output-reports:
+  active: true
+  exclude:
+  #  - 'HtmlOutputReport'
+  #  - 'PlainOutputReport'
+  #  - 'XmlOutputReport'
+
+comments:
+  active: true
+  CommentOverPrivateFunction:
+    active: false
+  CommentOverPrivateProperty:
+    active: false
+  EndOfSentenceFormat:
+    active: false
+    endOfSentenceFormat: ([.?!][ \t\n\r\f<])|([.?!]$)
+  UndocumentedPublicClass:
+    active: false
+    searchInNestedClass: true
+    searchInInnerClass: true
+    searchInInnerObject: true
+    searchInInnerInterface: true
+  UndocumentedPublicFunction:
+    active: false
+
+complexity:
+  active: true
+  ComplexCondition:
+    active: true
+    threshold: 4
+  ComplexInterface:
+    active: false
+    threshold: 10
+    includeStaticDeclarations: false
+  ComplexMethod:
+    active: true
+    threshold: 10
+    ignoreSingleWhenExpression: false
+  LabeledExpression:
+    active: false
+  LargeClass:
+    active: true
+    threshold: 150
+  LongMethod:
+    active: true
+    threshold: 20
+  LongParameterList:
+    active: true
+    threshold: 6
+    ignoreDefaultParameters: false
+  MethodOverloading:
+    active: false
+    threshold: 6
+  NestedBlockDepth:
+    active: true
+    threshold: 4
+  StringLiteralDuplication:
+    active: false
+    threshold: 3
+    ignoreAnnotation: true
+    excludeStringsWithLessThan5Characters: true
+    ignoreStringsRegex: '$^'
+  TooManyFunctions:
+    active: true
+    thresholdInFiles: 11
+    thresholdInClasses: 11
+    thresholdInInterfaces: 11
+    thresholdInObjects: 11
+    thresholdInEnums: 11
+    ignoreDeprecated: false
+
+empty-blocks:
+  active: true
+  EmptyCatchBlock:
+    active: true
+    allowedExceptionNameRegex: "^(_|(ignore|expected).*)"
+  EmptyClassBlock:
+    active: true
+  EmptyDefaultConstructor:
+    active: true
+  EmptyDoWhileBlock:
+    active: true
+  EmptyElseBlock:
+    active: true
+  EmptyFinallyBlock:
+    active: true
+  EmptyForBlock:
+    active: true
+  EmptyFunctionBlock:
+    active: true
+    ignoreOverriddenFunctions: false
+  EmptyIfBlock:
+    active: true
+  EmptyInitBlock:
+    active: true
+  EmptyKtFile:
+    active: true
+  EmptySecondaryConstructor:
+    active: true
+  EmptyWhenBlock:
+    active: true
+  EmptyWhileBlock:
+    active: true
+
+exceptions:
+  active: true
+  ExceptionRaisedInUnexpectedLocation:
+    active: false
+    methodNames: 'toString,hashCode,equals,finalize'
+  InstanceOfCheckForException:
+    active: false
+  NotImplementedDeclaration:
+    active: false
+  PrintStackTrace:
+    active: false
+  RethrowCaughtException:
+    active: false
+  ReturnFromFinally:
+    active: false
+  SwallowedException:
+    active: false
+  ThrowingExceptionFromFinally:
+    active: false
+  ThrowingExceptionInMain:
+    active: false
+  ThrowingExceptionsWithoutMessageOrCause:
+    active: false
+    exceptions: 'IllegalArgumentException,IllegalStateException,IOException'
+  ThrowingNewInstanceOfSameException:
+    active: false
+  TooGenericExceptionCaught:
+    active: true
+    exceptionNames:
+     - ArrayIndexOutOfBoundsException
+     - Error
+     - Exception
+     - IllegalMonitorStateException
+     - NullPointerException
+     - IndexOutOfBoundsException
+     - RuntimeException
+     - Throwable
+  TooGenericExceptionThrown:
+    active: true
+    exceptionNames:
+     - Error
+     - Exception
+     - Throwable
+     - RuntimeException
+
+formatting:
+  active: true
+  android: false
+  autoCorrect: true
+  ChainWrapping:
+    active: true
+    autoCorrect: true
+  CommentSpacing:
+    active: true
+    autoCorrect: true
+  Filename:
+    active: true
+  FinalNewline:
+    active: true
+    autoCorrect: true
+  ImportOrdering:
+    active: true
+    autoCorrect: true
+  Indentation:
+    active: true
+    autoCorrect: true
+    indentSize: 4
+    continuationIndentSize: 4
+  MaximumLineLength:
+    active: true
+    maxLineLength: 120
+  ModifierOrdering:
+    active: true
+    autoCorrect: true
+  NoBlankLineBeforeRbrace:
+    active: true
+    autoCorrect: true
+  NoConsecutiveBlankLines:
+    active: true
+    autoCorrect: true
+  NoEmptyClassBody:
+    active: true
+    autoCorrect: true
+  NoItParamInMultilineLambda:
+    active: true
+  NoLineBreakAfterElse:
+    active: true
+    autoCorrect: true
+  NoLineBreakBeforeAssignment:
+    active: true
+    autoCorrect: true
+  NoMultipleSpaces:
+    active: true
+    autoCorrect: true
+  NoSemicolons:
+    active: true
+    autoCorrect: true
+  NoTrailingSpaces:
+    active: true
+    autoCorrect: true
+  NoUnitReturn:
+    active: true
+    autoCorrect: true
+  NoUnusedImports:
+    active: true
+    autoCorrect: true
+  NoWildcardImports:
+    active: true
+    autoCorrect: true
+  ParameterListWrapping:
+    active: true
+    autoCorrect: true
+    indentSize: 4
+  SpacingAroundColon:
+    active: true
+    autoCorrect: true
+  SpacingAroundComma:
+    active: true
+    autoCorrect: true
+  SpacingAroundCurly:
+    active: true
+    autoCorrect: true
+  SpacingAroundKeyword:
+    active: true
+    autoCorrect: true
+  SpacingAroundOperators:
+    active: true
+    autoCorrect: true
+  SpacingAroundRangeOperator:
+    active: true
+    autoCorrect: true
+  StringTemplate:
+    active: true
+    autoCorrect: true
+
+naming:
+  active: true
+  ClassNaming:
+    active: true
+    classPattern: '[A-Z$][a-zA-Z0-9$]*'
+  EnumNaming:
+    active: true
+    enumEntryPattern: '^[A-Z][_a-zA-Z0-9]*'
+  ForbiddenClassName:
+    active: false
+    forbiddenName: ''
+  FunctionMaxLength:
+    active: false
+    maximumFunctionNameLength: 30
+  FunctionMinLength:
+    active: false
+    minimumFunctionNameLength: 3
+  FunctionNaming:
+    active: true
+    functionPattern: '^([a-z$][a-zA-Z$0-9]*)|(`.*`)$'
+    excludeClassPattern: '$^'
+  MatchingDeclarationName:
+    active: true
+  MemberNameEqualsClassName:
+    active: false
+    ignoreOverriddenFunction: true
+  ObjectPropertyNaming:
+    active: true
+    propertyPattern: '[A-Za-z][_A-Za-z0-9]*'
+    constantPattern: '[A-Za-z][_A-Za-z0-9]*'
+  PackageNaming:
+    active: true
+    packagePattern: '^[a-z]+(\.[a-z][a-z0-9]*)*$'
+  TopLevelPropertyNaming:
+    active: true
+    constantPattern: '[A-Z][_A-Z0-9]*'
+    propertyPattern: '[A-Za-z][_A-Za-z0-9]*'
+    privatePropertyPattern: '(_)?[A-Za-z][A-Za-z0-9]*'
+  VariableMaxLength:
+    active: false
+    maximumVariableNameLength: 64
+  VariableMinLength:
+    active: false
+    minimumVariableNameLength: 1
+  VariableNaming:
+    active: true
+    variablePattern: '[a-z][A-Za-z0-9]*'
+    privateVariablePattern: '(_)?[a-z][A-Za-z0-9]*'
+    excludeClassPattern: '$^'
+
+performance:
+  active: true
+  ForEachOnRange:
+    active: true
+  SpreadOperator:
+    active: true
+  UnnecessaryTemporaryInstantiation:
+    active: true
+
+potential-bugs:
+  active: true
+  DuplicateCaseInWhenExpression:
+    active: true
+  EqualsAlwaysReturnsTrueOrFalse:
+    active: false
+  EqualsWithHashCodeExist:
+    active: true
+  ExplicitGarbageCollectionCall:
+    active: true
+  InvalidRange:
+    active: false
+  IteratorHasNextCallsNextMethod:
+    active: false
+  IteratorNotThrowingNoSuchElementException:
+    active: false
+  LateinitUsage:
+    active: false
+    excludeAnnotatedProperties: ""
+    ignoreOnClassesPattern: ""
+  UnconditionalJumpStatementInLoop:
+    active: false
+  UnreachableCode:
+    active: true
+  UnsafeCallOnNullableType:
+    active: false
+  UnsafeCast:
+    active: false
+  UselessPostfixExpression:
+    active: false
+  WrongEqualsTypeParameter:
+    active: false
+
+style:
+  active: true
+  CollapsibleIfStatements:
+    active: false
+  DataClassContainsFunctions:
+    active: false
+    conversionFunctionPrefix: 'to'
+  EqualsNullCall:
+    active: false
+  ExpressionBodySyntax:
+    active: false
+  ForbiddenComment:
+    active: true
+    values: 'TODO:,FIXME:,STOPSHIP:'
+  ForbiddenImport:
+    active: false
+    imports: ''
+  FunctionOnlyReturningConstant:
+    active: false
+    ignoreOverridableFunction: true
+    excludedFunctions: 'describeContents'
+  LoopWithTooManyJumpStatements:
+    active: false
+    maxJumpCount: 1
+  MagicNumber:
+    active: true
+    ignoreNumbers: '-1,0,1,2'
+    ignoreHashCodeFunction: false
+    ignorePropertyDeclaration: false
+    ignoreConstantDeclaration: true
+    ignoreCompanionObjectPropertyDeclaration: true
+    ignoreAnnotation: false
+    ignoreNamedArgument: true
+    ignoreEnums: false
+  MaxLineLength:
+    active: true
+    maxLineLength: 120
+    excludePackageStatements: false
+    excludeImportStatements: false
+    excludeCommentStatements: false
+  MayBeConst:
+    active: false
+  ModifierOrder:
+    active: true
+  NestedClassesVisibility:
+    active: false
+  NewLineAtEndOfFile:
+    active: true
+  NoTabs:
+    active: false
+  OptionalAbstractKeyword:
+    active: true
+  OptionalUnit:
+    active: false
+  OptionalWhenBraces:
+    active: false
+  ProtectedMemberInFinalClass:
+    active: false
+  RedundantVisibilityModifierRule:
+    active: false
+  ReturnCount:
+    active: true
+    max: 2
+    excludedFunctions: "equals"
+  SafeCast:
+    active: true
+  SerialVersionUIDInSerializableClass:
+    active: false
+  SpacingBetweenPackageAndImports:
+    active: false
+  ThrowsCount:
+    active: true
+    max: 2
+  TrailingWhitespace:
+    active: false
+  UnnecessaryAbstractClass:
+    active: false
+  UnnecessaryInheritance:
+    active: false
+  UnnecessaryParentheses:
+    active: false
+  UntilInsteadOfRangeTo:
+    active: false
+  UnusedImports:
+    active: false
+  UnusedPrivateMember:
+    active: false
+    allowedNames: "(_|ignored|expected)"
+  UseDataClass:
+    active: false
+    excludeAnnotatedClasses: ""
+  UtilityClassWithPublicConstructor:
+    active: false
+  WildcardImport:
+    active: true
+    excludeImports: 'java.util.*,kotlinx.android.synthetic.*'
diff --git a/hv-collector-core/pom.xml b/hv-collector-core/pom.xml
new file mode 100644
index 00000000..ed501a44
--- /dev/null
+++ b/hv-collector-core/pom.xml
@@ -0,0 +1,142 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ ============LICENSE_START=======================================================
+  ~ dcaegen2-collectors-veshv
+  ~ ================================================================================
+  ~ Copyright (C) 2018 NOKIA Intellectual Property. 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>
+
+  <licenses>
+    <license>
+      <name>The Apache Software License, Version 2.0</name>
+      <url>http://www.apache.org/licenses/LICENSE-2.0.txt</url>
+    </license>
+  </licenses>
+
+  <parent>
+    <groupId>org.onap.dcaegen2.collectors.veshv</groupId>
+    <artifactId>ves-hv-collector</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+    <relativePath>..</relativePath>
+  </parent>
+
+  <artifactId>hv-collector-core</artifactId>
+  <description>VES HighVolume Collector :: Core</description>
+
+  <properties>
+    <skipAnalysis>false</skipAnalysis>
+  </properties>
+
+  <build>
+    <plugins>
+      <plugin>
+        <artifactId>kotlin-maven-plugin</artifactId>
+        <groupId>org.jetbrains.kotlin</groupId>
+      </plugin>
+      <plugin>
+        <artifactId>maven-surefire-plugin</artifactId>
+        <groupId>org.apache.maven.plugins</groupId>
+      </plugin>
+      <plugin>
+        <groupId>org.jacoco</groupId>
+        <artifactId>jacoco-maven-plugin</artifactId>
+      </plugin>
+    </plugins>
+  </build>
+
+  <dependencies>
+    <dependency>
+      <groupId>${project.parent.groupId}</groupId>
+      <artifactId>protobuf</artifactId>
+      <version>${project.parent.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>${project.parent.groupId}</groupId>
+      <artifactId>hv-collector-utils</artifactId>
+      <version>${project.parent.version}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>org.jetbrains.kotlin</groupId>
+      <artifactId>kotlin-reflect</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>io.projectreactor</groupId>
+      <artifactId>reactor-core</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>io.projectreactor.addons</groupId>
+      <artifactId>reactor-extra</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>io.projectreactor.ipc</groupId>
+      <artifactId>reactor-netty</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>io.projectreactor.kafka</groupId>
+      <artifactId>reactor-kafka</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>javax.json</groupId>
+      <artifactId>javax.json-api</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.glassfish</groupId>
+      <artifactId>javax.json</artifactId>
+      <scope>runtime</scope>
+    </dependency>
+
+
+    <dependency>
+      <groupId>com.nhaarman</groupId>
+      <artifactId>mockito-kotlin</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-core</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.assertj</groupId>
+      <artifactId>assertj-core</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.jetbrains.kotlin</groupId>
+      <artifactId>kotlin-test</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.jetbrains.spek</groupId>
+      <artifactId>spek-api</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.jetbrains.spek</groupId>
+      <artifactId>spek-junit-platform-engine</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>io.projectreactor</groupId>
+      <artifactId>reactor-test</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>ch.qos.logback</groupId>
+      <artifactId>logback-classic</artifactId>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+
+</project>
diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/boundary/adapters.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/boundary/adapters.kt
new file mode 100644
index 00000000..d4de1b5b
--- /dev/null
+++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/boundary/adapters.kt
@@ -0,0 +1,53 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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.dcae.collectors.veshv.boundary
+
+import org.onap.dcae.collectors.veshv.domain.CollectorConfiguration
+import org.onap.dcae.collectors.veshv.domain.RoutedMessage
+import org.onap.dcae.collectors.veshv.domain.VesMessage
+import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader
+import reactor.core.publisher.Flux
+
+interface Sink {
+    fun send(messages: Flux<RoutedMessage>): Flux<VesMessage>
+}
+
+@FunctionalInterface
+interface SinkProvider {
+    operator fun invoke(config: CollectorConfiguration): Sink
+
+    companion object {
+        fun just(sink: Sink): SinkProvider =
+                object : SinkProvider {
+                    override fun invoke(config: CollectorConfiguration): Sink = sink
+                }
+    }
+}
+
+interface ConfigurationProvider {
+    operator fun invoke(): Flux<CollectorConfiguration>
+
+    companion object {
+        fun from(function: () -> Flux<CollectorConfiguration>): ConfigurationProvider =
+                object : ConfigurationProvider {
+                    override fun invoke(): Flux<CollectorConfiguration> = function()
+                }
+    }
+}
diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/boundary/api.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/boundary/api.kt
new file mode 100644
index 00000000..809ba32f
--- /dev/null
+++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/boundary/api.kt
@@ -0,0 +1,39 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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.dcae.collectors.veshv.boundary
+
+import io.netty.buffer.ByteBuf
+import org.onap.dcae.collectors.veshv.domain.ServerConfiguration
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+
+interface Collector {
+    fun handleConnection(dataStream: Flux<ByteBuf>): Mono<Void>
+}
+
+typealias CollectorProvider = () -> Collector
+
+interface Server {
+    fun start(): Mono<Void>
+}
+
+interface ServerFactory {
+    fun createServer(serverConfig: ServerConfiguration, collector: CollectorProvider): Server
+}
diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/CollectorConfiguration.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/CollectorConfiguration.kt
new file mode 100644
index 00000000..f5e525e7
--- /dev/null
+++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/CollectorConfiguration.kt
@@ -0,0 +1,26 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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.dcae.collectors.veshv.domain
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since May 2018
+ */
+data class CollectorConfiguration(val kafkaBootstrapServers: String, val routing: Routing)
diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/RoutedMessage.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/RoutedMessage.kt
new file mode 100644
index 00000000..2cf2b082
--- /dev/null
+++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/RoutedMessage.kt
@@ -0,0 +1,22 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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.dcae.collectors.veshv.domain
+
+data class RoutedMessage(val topic: String, val partition: Int, val message: VesMessage)
diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/ServerConfiguration.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/ServerConfiguration.kt
new file mode 100644
index 00000000..ea78706e
--- /dev/null
+++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/ServerConfiguration.kt
@@ -0,0 +1,26 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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.dcae.collectors.veshv.domain
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since May 2018
+ */
+data class ServerConfiguration(val port: Int, val configurationUrl: String)
diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/VesMessage.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/VesMessage.kt
new file mode 100644
index 00000000..79b48804
--- /dev/null
+++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/VesMessage.kt
@@ -0,0 +1,29 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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.dcae.collectors.veshv.domain
+
+import io.netty.buffer.ByteBuf
+import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since May 2018
+ */
+data class VesMessage(val header: CommonEventHeader, val rawMessage: ByteBuf)
diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/WireFrame.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/WireFrame.kt
new file mode 100644
index 00000000..306b7762
--- /dev/null
+++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/WireFrame.kt
@@ -0,0 +1,93 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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.dcae.collectors.veshv.domain
+
+import io.netty.buffer.ByteBuf
+import io.netty.buffer.ByteBufAllocator
+
+/**
+ * Wire frame structure is presented bellow. All fields are in network byte order (big-endian).
+ *
+ * ```
+ *     ┌─────┬───────────────────────┬───────────────────────┬───────────────────────┐
+ *     │octet│           0           │           1           │           2           │
+ *     ├─────┼──┬──┬──┬──┬──┬──┬──┬──┼──┬──┬──┬──┬──┬──┬──┬──┼──┬──┬──┬──┬──┬──┬──┬──┤
+ *     │ bit │ 0│  │  │  │  │  │  │  │ 8│  │  │  │  │  │  │  │16│  │  │  │  │  │  │  │ ...
+ *     ├─────┼──┴──┴──┴──┴──┴──┴──┴──┼──┴──┴──┴──┴──┴──┴──┴──┼──┴──┴──┴──┴──┴──┴──┴──┤
+ *     │field│          0xFF         │     major version     │     minor version     │
+ *     └─────┴───────────────────────┴───────────────────────┴───────────────────────┘
+ *     ┌─────┬───────────────────────┬───────────────────────┬───────────────────────┬───────────────────────┐
+ *     │octet│           3           │           4           │           5           │           6           │
+ *     ├─────┼──┬──┬──┬──┬──┬──┬──┬──┼──┬──┬──┬──┬──┬──┬──┬──┼──┬──┬──┬──┬──┬──┬──┬──┼──┬──┬──┬──┬──┬──┬──┬──┤
+ * ... │ bit │24│  │  │  │  │  │  │  │  │  │  │  │  │  │  │  │  │  │  │  │  │  │  │  │  │  │  │  │  │  │  │  │ ...
+ *     ├─────┼──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┤
+ *     │field│                                         payload size                                          │
+ *     └─────┴───────────────────────────────────────────────────────────────────────────────────────────────┘
+ *     ┌─────┬───────────────────────
+ *     │octet│           7         ...
+ *     ├─────┼──┬──┬──┬──┬──┬──┬──┬──
+ * ... │ bit │56│  │  │  │  │  │  │...
+ *     ├─────┼──┴──┴──┴──┴──┴──┴──┴──
+ *     │field│   protobuf payload
+ *     └─────┴───────────────────────
+ * ```
+ *
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since May 2018
+ */
+data class WireFrame(val payload: ByteBuf,
+                     val mark: Short,
+                     val majorVersion: Short,
+                     val minorVersion: Short,
+                     val payloadSize: Int) {
+    fun isValid(): Boolean {
+        return mark == FF_BYTE
+                && majorVersion == SUPPORTED_MAJOR_VERSION
+                && payload.readableBytes() == payloadSize
+    }
+
+    fun encode(allocator: ByteBufAllocator): ByteBuf {
+        val bb = allocator.buffer(HEADER_SIZE + payload.readableBytes())
+
+        bb.writeByte(mark.toInt())
+        bb.writeByte(majorVersion.toInt())
+        bb.writeByte(minorVersion.toInt())
+        bb.writeInt(payloadSize)
+        bb.writeBytes(payload)
+
+        return bb
+    }
+
+    companion object {
+        fun decode(byteBuf: ByteBuf): WireFrame {
+            val mark = byteBuf.readUnsignedByte()
+            val majorVersion = byteBuf.readUnsignedByte()
+            val minorVersion = byteBuf.readUnsignedByte()
+            val payloadSize = byteBuf.readInt()
+            val payload = byteBuf.slice()
+
+            return WireFrame(payload, mark, majorVersion, minorVersion, payloadSize)
+        }
+
+        private const val HEADER_SIZE = 3 + java.lang.Integer.BYTES
+        private const val FF_BYTE: Short = 0xFF
+        private const val SUPPORTED_MAJOR_VERSION: Short = 1
+    }
+}
diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/routing.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/routing.kt
new file mode 100644
index 00000000..3f826f7c
--- /dev/null
+++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/routing.kt
@@ -0,0 +1,82 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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.dcae.collectors.veshv.domain
+
+import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader
+import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader.Domain
+
+data class Routing(val routes: List<Route>) {
+
+    fun routeFor(commonHeader: CommonEventHeader): Route? = routes.find { it.applies(commonHeader) }
+}
+
+data class Route(val domain: Domain, val targetTopic: String, val partitioning: (CommonEventHeader) -> Int) {
+
+    fun applies(commonHeader: CommonEventHeader) = commonHeader.domain == domain
+
+    operator fun invoke(message: VesMessage): RoutedMessage =
+            RoutedMessage(targetTopic, partitioning(message.header), message)
+}
+
+
+/*
+Configuration DSL
+ */
+
+fun routing(init: RoutingBuilder.() -> Unit): RoutingBuilder {
+    val conf = RoutingBuilder()
+    conf.init()
+    return conf
+}
+
+class RoutingBuilder {
+    private val routes: MutableList<RouteBuilder> = mutableListOf()
+
+    fun defineRoute(init: RouteBuilder.() -> Unit): RouteBuilder {
+        val rule = RouteBuilder()
+        rule.init()
+        routes.add(rule)
+        return rule
+    }
+
+    fun build() = Routing(routes.map { it.build() }.toList())
+}
+
+class RouteBuilder {
+
+    private lateinit var domain: Domain
+    private lateinit var targetTopic: String
+    private lateinit var partitioning: (CommonEventHeader) -> Int
+
+    fun fromDomain(domain: Domain) {
+        this.domain = domain
+    }
+
+    fun toTopic(targetTopic: String) {
+        this.targetTopic = targetTopic
+    }
+
+    fun withFixedPartitioning(num: Int = 1) {
+        partitioning = { _ -> num }
+    }
+
+    fun build() = Route(domain, targetTopic, partitioning)
+
+}
diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/factory/CollectorFactory.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/factory/CollectorFactory.kt
new file mode 100644
index 00000000..cb018cca
--- /dev/null
+++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/factory/CollectorFactory.kt
@@ -0,0 +1,59 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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.dcae.collectors.veshv.factory
+
+import org.onap.dcae.collectors.veshv.boundary.Collector
+import org.onap.dcae.collectors.veshv.boundary.CollectorProvider
+import org.onap.dcae.collectors.veshv.boundary.ConfigurationProvider
+import org.onap.dcae.collectors.veshv.boundary.SinkProvider
+import org.onap.dcae.collectors.veshv.domain.CollectorConfiguration
+import org.onap.dcae.collectors.veshv.impl.MessageValidator
+import org.onap.dcae.collectors.veshv.impl.Router
+import org.onap.dcae.collectors.veshv.impl.VesDecoder
+import org.onap.dcae.collectors.veshv.impl.VesHvCollector
+import org.onap.dcae.collectors.veshv.impl.WireDecoder
+import reactor.core.publisher.Flux
+import java.util.concurrent.atomic.AtomicReference
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since May 2018
+ */
+class CollectorFactory(val configuration: ConfigurationProvider, val sinkProvider: SinkProvider) {
+    fun createVesHvCollectorProvider(): CollectorProvider {
+        val collector: AtomicReference<Collector> = AtomicReference()
+        createVesHvCollector().subscribe(collector::set)
+        return collector::get
+    }
+
+    private fun createVesHvCollector(): Flux<Collector> =
+            configuration().map(this::createVesHvCollector)
+
+    private fun createVesHvCollector(config: CollectorConfiguration): Collector {
+        return VesHvCollector(
+                WireDecoder(),
+                VesDecoder(),
+                MessageValidator(),
+                Router(config.routing),
+                sinkProvider(config))
+    }
+
+}
+
diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/factory/ServerFactory.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/factory/ServerFactory.kt
new file mode 100644
index 00000000..5e60fa56
--- /dev/null
+++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/factory/ServerFactory.kt
@@ -0,0 +1,33 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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.dcae.collectors.veshv.factory
+
+import org.onap.dcae.collectors.veshv.boundary.CollectorProvider
+import org.onap.dcae.collectors.veshv.boundary.Server
+import org.onap.dcae.collectors.veshv.domain.ServerConfiguration
+import org.onap.dcae.collectors.veshv.impl.NettyTcpServer
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since May 2018
+ */
+object ServerFactory {
+    val createNettyTcpServer: (ServerConfiguration, CollectorProvider) -> Server = ::NettyTcpServer
+}
diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/MessageValidator.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/MessageValidator.kt
new file mode 100644
index 00000000..55e65cf7
--- /dev/null
+++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/MessageValidator.kt
@@ -0,0 +1,48 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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.dcae.collectors.veshv.impl
+
+import org.onap.dcae.collectors.veshv.domain.VesMessage
+import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader
+
+internal class MessageValidator {
+
+    val requiredFieldDescriptors = listOf(
+            "version",
+            "eventName",
+            "domain",
+            "eventId",
+            "sourceName",
+            "reportingEntityName",
+            "priority",
+            "startEpochMicrosec",
+            "lastEpochMicrosec",
+            "sequence")
+    .map { fieldName -> CommonEventHeader.getDescriptor().findFieldByName(fieldName)}
+
+    fun isValid(message: VesMessage): Boolean {
+        val header = message.header
+        return allMandatoryFieldsArePresent(header) && header.domain == CommonEventHeader.Domain.HVRANMEAS
+    }
+
+    private fun allMandatoryFieldsArePresent(header: CommonEventHeader) =
+            requiredFieldDescriptors
+                    .all { fieldDescriptor -> header.hasField(fieldDescriptor) }
+}
diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/NettyTcpServer.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/NettyTcpServer.kt
new file mode 100644
index 00000000..ca77df2a
--- /dev/null
+++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/NettyTcpServer.kt
@@ -0,0 +1,66 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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.dcae.collectors.veshv.impl
+
+import org.onap.dcae.collectors.veshv.boundary.CollectorProvider
+import org.onap.dcae.collectors.veshv.boundary.Server
+import org.onap.dcae.collectors.veshv.domain.ServerConfiguration
+import org.onap.dcae.collectors.veshv.utils.logging.Logger
+import org.reactivestreams.Publisher
+import reactor.core.publisher.Mono
+import reactor.ipc.netty.NettyInbound
+import reactor.ipc.netty.NettyOutbound
+import reactor.ipc.netty.tcp.TcpServer
+import java.util.function.BiFunction
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since May 2018
+ */
+internal class NettyTcpServer(val serverConfig: ServerConfiguration,
+                              val collectorProvider: CollectorProvider) : Server {
+
+    override fun start(): Mono<Void> {
+        logger.info { "Listening on port ${serverConfig.port}" }
+        return Mono.defer {
+            val nettyContext = TcpServer.create(serverConfig.port)
+                    .start(BiFunction<NettyInbound, NettyOutbound, Publisher<Void>> { t, u ->
+                        handleConnection(t, u)
+                    })
+            Mono.never<Void>().doFinally { _ -> nettyContext.shutdown() }
+        }
+    }
+
+    private fun handleConnection(nettyInbound: NettyInbound, nettyOutbound: NettyOutbound): Mono<Void> {
+        logger.debug("Got connection")
+        val pipe = collectorProvider().handleConnection(nettyInbound.receive())
+
+        val hello = nettyOutbound
+                .options { it.flushOnEach() }
+                .sendString(Mono.just("ONAP_VES_HV/0.1\n"))
+                .then()
+
+        return hello.then(pipe)
+    }
+
+    companion object {
+        private val logger = Logger(NettyTcpServer::class)
+    }
+}
diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/Router.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/Router.kt
new file mode 100644
index 00000000..8c005d42
--- /dev/null
+++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/Router.kt
@@ -0,0 +1,9 @@
+package org.onap.dcae.collectors.veshv.impl
+
+import org.onap.dcae.collectors.veshv.domain.RoutedMessage
+import org.onap.dcae.collectors.veshv.domain.Routing
+import org.onap.dcae.collectors.veshv.domain.VesMessage
+
+class Router(private val routing: Routing) {
+    fun findDestination(message: VesMessage): RoutedMessage? = routing.routeFor(message.header)?.invoke(message)
+}
diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/VesDecoder.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/VesDecoder.kt
new file mode 100644
index 00000000..6ace06e4
--- /dev/null
+++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/VesDecoder.kt
@@ -0,0 +1,50 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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.dcae.collectors.veshv.impl
+
+import com.google.protobuf.InvalidProtocolBufferException
+import io.netty.buffer.ByteBuf
+import org.onap.dcae.collectors.veshv.domain.VesMessage
+import org.onap.dcae.collectors.veshv.utils.logging.Logger
+import org.onap.ves.VesEventV5.VesEvent
+import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader
+import reactor.core.publisher.Mono
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since May 2018
+ */
+internal class VesDecoder {
+
+    fun decode(bb: ByteBuf): VesMessage? =
+            try {
+                val decodedHeader = VesEvent.parseFrom(bb.nioBuffer()).commonEventHeader
+                VesMessage(decodedHeader, bb)
+            } catch (ex: InvalidProtocolBufferException) {
+                logger.warn { "Dropping incoming message. Invalid protocol buffer: ${ex.message}" }
+                logger.debug("Cause", ex)
+                null
+            }
+
+
+    companion object {
+        private val logger = Logger(VesDecoder::class)
+    }
+}
diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/VesHvCollector.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/VesHvCollector.kt
new file mode 100644
index 00000000..af9d0b0a
--- /dev/null
+++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/VesHvCollector.kt
@@ -0,0 +1,83 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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.dcae.collectors.veshv.impl
+
+import io.netty.buffer.ByteBuf
+import org.onap.dcae.collectors.veshv.boundary.Collector
+import org.onap.dcae.collectors.veshv.boundary.Sink
+import org.onap.dcae.collectors.veshv.domain.RoutedMessage
+import org.onap.dcae.collectors.veshv.domain.VesMessage
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since May 2018
+ */
+internal class VesHvCollector(
+        val wireDecoder: WireDecoder,
+        val protobufDecoder: VesDecoder,
+        val validator: MessageValidator,
+        val router: Router,
+        val sink: Sink) : Collector {
+    override fun handleConnection(dataStream: Flux<ByteBuf>): Mono<Void> =
+            dataStream
+                    .flatMap(this::decodeWire)
+                    .flatMap(this::decodeProtobuf)
+                    .filter(this::validate)
+                    .flatMap(this::findRoute)
+                    .compose(sink::send)
+                    .doOnNext(this::releaseMemory)
+                    .then()
+
+    private fun decodeWire(wire: ByteBuf) = releaseWhenNull(wire, wireDecoder::decode)
+
+    private fun decodeProtobuf(protobuf: ByteBuf) = releaseWhenNull(protobuf, protobufDecoder::decode)
+
+    private fun validate(msg: VesMessage): Boolean {
+        val valid = validator.isValid(msg)
+        if (!valid) {
+            msg.rawMessage.release()
+        }
+        return valid
+    }
+
+    private fun findRoute(msg: VesMessage): Mono<RoutedMessage> {
+        val routedMessage = router.findDestination(msg)
+        return if (routedMessage == null)
+            Mono.empty()
+        else
+            Mono.just(routedMessage)
+    }
+
+    private fun releaseMemory(msg: VesMessage) {
+        msg.rawMessage.release()
+    }
+
+    private fun <T>releaseWhenNull(input: ByteBuf, mapper: (ByteBuf) -> T?): Mono<T> {
+        val result = mapper(input)
+        return if (result == null) {
+            input.release()
+            Mono.empty()
+        } else {
+            Mono.just(result)
+        }
+    }
+}
diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/WireDecoder.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/WireDecoder.kt
new file mode 100644
index 00000000..66ced2d7
--- /dev/null
+++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/WireDecoder.kt
@@ -0,0 +1,44 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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.dcae.collectors.veshv.impl
+
+import io.netty.buffer.ByteBuf
+import org.onap.dcae.collectors.veshv.domain.WireFrame
+import org.onap.dcae.collectors.veshv.utils.logging.Logger
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since May 2018
+ */
+internal class WireDecoder {
+    fun decode(byteBuf: ByteBuf): ByteBuf? =
+            try {
+                WireFrame.decode(byteBuf)
+                        .takeIf { it.isValid() }
+                        .let { it?.payload }
+            } catch (ex: IndexOutOfBoundsException) {
+                logger.debug { "Wire protocol frame could not be decoded - input is too small" }
+                null
+            }
+
+    companion object {
+        private val logger = Logger(WireDecoder::class)
+    }
+}
diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/AdapterFactory.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/AdapterFactory.kt
new file mode 100644
index 00000000..cdfcfd30
--- /dev/null
+++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/AdapterFactory.kt
@@ -0,0 +1,65 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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.dcae.collectors.veshv.impl.adapters
+
+import org.apache.kafka.clients.producer.ProducerConfig.BOOTSTRAP_SERVERS_CONFIG
+import org.apache.kafka.clients.producer.ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG
+import org.apache.kafka.clients.producer.ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG
+import org.apache.kafka.common.serialization.ByteBufferSerializer
+import org.apache.kafka.common.serialization.StringSerializer
+import org.onap.dcae.collectors.veshv.boundary.ConfigurationProvider
+import org.onap.dcae.collectors.veshv.boundary.Sink
+import org.onap.dcae.collectors.veshv.boundary.SinkProvider
+import org.onap.dcae.collectors.veshv.domain.CollectorConfiguration
+import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader
+import reactor.core.publisher.Flux
+import reactor.ipc.netty.http.client.HttpClient
+import reactor.kafka.sender.KafkaSender
+import reactor.kafka.sender.SenderOptions
+import java.nio.ByteBuffer
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since May 2018
+ */
+object AdapterFactory {
+    fun kafkaSink(): SinkProvider = KafkaSinkProvider()
+
+    fun staticConfigurationProvider(config: CollectorConfiguration) =
+            object : ConfigurationProvider {
+                override fun invoke() = Flux.just(config)
+            }
+
+    private class KafkaSinkProvider : SinkProvider {
+        override fun invoke(config: CollectorConfiguration): Sink {
+            val sender = KafkaSender.create(
+                    SenderOptions.create<CommonEventHeader, ByteBuffer>()
+                            .producerProperty(BOOTSTRAP_SERVERS_CONFIG, config.kafkaBootstrapServers)
+                            .producerProperty(KEY_SERIALIZER_CLASS_CONFIG, StringSerializer::class.java)
+                            .producerProperty(VALUE_SERIALIZER_CLASS_CONFIG, ByteBufferSerializer::class.java))
+            return KafkaSink(sender)
+        }
+    }
+
+    fun consulConfigurationProvider(url: String): ConfigurationProvider =
+            ConsulConfigurationProvider(url, httpAdapter())
+    fun httpAdapter(): HttpAdapter = HttpAdapter(HttpClient.create())
+}
+
diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/ConsulConfigurationProvider.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/ConsulConfigurationProvider.kt
new file mode 100644
index 00000000..ef6c2f76
--- /dev/null
+++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/ConsulConfigurationProvider.kt
@@ -0,0 +1,63 @@
+package org.onap.dcae.collectors.veshv.impl.adapters
+
+import org.onap.dcae.collectors.veshv.boundary.ConfigurationProvider
+import org.onap.dcae.collectors.veshv.domain.CollectorConfiguration
+import org.onap.ves.VesEventV5
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Flux
+import java.io.StringReader
+import java.time.Duration
+import java.util.*
+import java.util.concurrent.atomic.AtomicReference
+import javax.json.Json
+import javax.json.JsonObject
+
+
+/**
+ * @author Jakub Dudycz <jakub.dudycz@nokia.com>
+ * @since May 2018
+ */
+internal class ConsulConfigurationProvider(private val url: String, private val http: HttpAdapter)
+    : ConfigurationProvider {
+
+
+    private val logger = LoggerFactory.getLogger(ConsulConfigurationProvider::class.java)
+    private var lastConfigurationHash: AtomicReference<Int> = AtomicReference()
+
+    override fun invoke(): Flux<CollectorConfiguration> =
+            Flux.interval(Duration.ZERO, REFRESH_INTERVAL)
+                    .flatMap { http.getResponse(url) }
+                    .filter { body -> body.hashCode() != lastConfigurationHash.get() }
+                    .doOnNext { body -> lastConfigurationHash.set(body.hashCode()) }
+                    .map { str -> getConfigurationJson(str) }
+                    .map { json -> createCollectorConfiguration(json) }
+
+    private fun getConfigurationJson(str: String): JsonObject {
+        val response = Json.createReader(StringReader(str)).readArray().getJsonObject(0)
+        val decodedValue = String(
+                Base64.getDecoder().decode(response.getString("Value")))
+        logger.info("Obtained new configuration from consul:\n$decodedValue")
+        return Json.createReader(StringReader(decodedValue)).readObject()
+    }
+
+    private fun createCollectorConfiguration(configuration: JsonObject): CollectorConfiguration {
+
+        val routing = configuration.getJsonObject("routing")
+
+        return CollectorConfiguration(
+                kafkaBootstrapServers = configuration.getString("kafkaBootstrapServers"),
+                routing = org.onap.dcae.collectors.veshv.domain.routing {
+                    defineRoute {
+                        fromDomain(VesEventV5.VesEvent.CommonEventHeader.Domain.forNumber(routing.getInt("fromDomain")))
+                        toTopic(routing.getString("toTopic"))
+                        withFixedPartitioning()
+                    }
+                }.build()
+        )
+    }
+
+    companion object {
+        private const val REFRESH_INTERVAL_MINUTES: Long = 5
+        private val REFRESH_INTERVAL = Duration.ofMinutes(REFRESH_INTERVAL_MINUTES)
+    }
+}
diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/HttpAdapter.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/HttpAdapter.kt
new file mode 100644
index 00000000..c09af015
--- /dev/null
+++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/HttpAdapter.kt
@@ -0,0 +1,29 @@
+package org.onap.dcae.collectors.veshv.impl.adapters
+
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Mono
+import reactor.core.publisher.toMono
+import reactor.ipc.netty.http.client.HttpClient
+import reactor.ipc.netty.http.client.HttpClientResponse
+import java.nio.charset.Charset
+
+/**
+ * @author Jakub Dudycz <jakub.dudycz@nokia.com>
+ * @since May 2018
+ */
+open class HttpAdapter(private val httpClient: HttpClient) {
+
+    private val logger = LoggerFactory.getLogger(HttpAdapter::class.java)
+
+    open fun getResponse(url: String): Mono<String> =
+            httpClient.get(url)
+                    .onErrorResume { e -> unableToGetResource(e, url) }
+                    .flatMap { res -> res.receiveContent().toMono() }
+                    .map { content -> content.content().toString(Charset.defaultCharset()) }
+
+
+    private fun unableToGetResource(e: Throwable, url: String): Mono<HttpClientResponse> {
+        logger.info("Failed to get resource on path: $url\n${e.localizedMessage}")
+        return Mono.empty()
+    }
+}
diff --git a/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/KafkaSink.kt b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/KafkaSink.kt
new file mode 100644
index 00000000..8b558a85
--- /dev/null
+++ b/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/KafkaSink.kt
@@ -0,0 +1,68 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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.dcae.collectors.veshv.impl.adapters
+
+import org.onap.dcae.collectors.veshv.boundary.Sink
+import org.onap.dcae.collectors.veshv.domain.RoutedMessage
+import org.onap.dcae.collectors.veshv.domain.VesMessage
+import org.onap.dcae.collectors.veshv.utils.logging.Logger
+import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader
+import reactor.core.publisher.Flux
+import reactor.kafka.sender.KafkaSender
+import reactor.kafka.sender.SenderRecord
+import reactor.kafka.sender.SenderResult
+import java.nio.ByteBuffer
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since May 2018
+ */
+internal class KafkaSink(private val sender: KafkaSender<CommonEventHeader, ByteBuffer>) : Sink {
+
+    override fun send(messages: Flux<RoutedMessage>): Flux<VesMessage> {
+        val records = messages.map(this::vesToKafkaRecord)
+        return sender.send(records)
+                .doOnNext(::logException)
+                .filter(::isSuccessful)
+                .map { it.correlationMetadata() }
+    }
+
+    private fun vesToKafkaRecord(msg: RoutedMessage): SenderRecord<CommonEventHeader, ByteBuffer, VesMessage> {
+        return SenderRecord.create(
+                msg.topic,
+                msg.partition,
+                System.currentTimeMillis(),
+                msg.message.header,
+                msg.message.rawMessage.nioBuffer(),
+                msg.message)
+    }
+
+    private fun logException(senderResult: SenderResult<out Any>) {
+        if (senderResult.exception() != null) {
+            logger.warn(senderResult.exception()) { "Failed to send message to Kafka" }
+        }
+    }
+
+    private fun isSuccessful(senderResult: SenderResult<out Any>)  = senderResult.exception() == null
+
+    companion object {
+        val logger = Logger(KafkaSink::class)
+    }
+}
diff --git a/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/MessageValidatorTest.kt b/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/MessageValidatorTest.kt
new file mode 100644
index 00000000..8f6b76fb
--- /dev/null
+++ b/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/MessageValidatorTest.kt
@@ -0,0 +1,108 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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.dcae.collectors.veshv.impl
+
+import com.google.protobuf.ByteString
+import io.netty.buffer.ByteBuf
+import io.netty.buffer.Unpooled
+import io.netty.buffer.Unpooled.wrappedBuffer
+import org.jetbrains.spek.api.Spek
+import org.jetbrains.spek.api.dsl.given
+import org.jetbrains.spek.api.dsl.it
+import org.jetbrains.spek.api.dsl.on
+import org.onap.dcae.collectors.veshv.domain.VesMessage
+import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader
+import org.onap.ves.VesEventV5.VesEvent
+import org.assertj.core.api.Assertions.assertThat
+import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader.*
+
+internal object MessageValidatorTest : Spek({
+
+    fun vesMessageBytes(commonHeader: CommonEventHeader): ByteBuf {
+        val msg = VesEvent.newBuilder()
+                .setCommonEventHeader(commonHeader)
+                .setHvRanMeasFields(ByteString.copyFromUtf8("high volume data"))
+                .build()
+        return wrappedBuffer(msg.toByteArray())
+    }
+
+    given("Message validator") {
+        val cut = MessageValidator()
+
+        on("ves hv message including header with fully initialized fields") {
+            val commonHeader = newBuilder()
+                    .setVersion("1.9")
+                    .setEventName("Sample event name")
+                    .setDomain(Domain.HVRANMEAS)
+                    .setEventId("Sample event Id")
+                    .setSourceName("Sample Source")
+                    .setReportingEntityName(ByteString.copyFromUtf8("Sample byte String"))
+                    .setPriority(Priority.MEDIUM)
+                    .setStartEpochMicrosec(120034455)
+                    .setLastEpochMicrosec(120034459)
+                    .setSequence(2)
+                    .build()
+
+            it("should accept message with fully initialized message header") {
+                val vesMessage = VesMessage(commonHeader, vesMessageBytes(commonHeader))
+                assertThat(cut.isValid(vesMessage)).describedAs("message validation result").isTrue()
+            }
+
+            it("should reject message with domain other than HVRANMEAS") {
+                Domain.values()
+                        .filter { it != Domain.HVRANMEAS && it != Domain.UNRECOGNIZED }
+                        .forEach { domain ->
+                            val header = newBuilder(commonHeader).setDomain(domain).build()
+                            val vesMessage = VesMessage(header, vesMessageBytes(header))
+                            assertThat(cut.isValid(vesMessage))
+                                    .describedAs("message with $domain domain")
+                                    .isFalse()
+                        }
+            }
+        }
+
+        on("ves hv message bytes") {
+            val vesMessage = VesMessage(getDefaultInstance(), Unpooled.EMPTY_BUFFER)
+            it("should not accept message with default header") {
+                assertThat(cut.isValid(vesMessage)).describedAs("message validation result").isFalse()
+            }
+        }
+
+
+        on("ves hv message including header with not initialized fields") {
+            val commonHeader = newBuilder()
+                    .setVersion("1.9")
+                    .setEventName("Sample event name")
+                    .setEventId("Sample event Id")
+                    .setSourceName("Sample Source")
+                    .build()
+            val msg = VesEvent.newBuilder()
+                    .setCommonEventHeader(commonHeader)
+                    .setHvRanMeasFields(ByteString.copyFromUtf8("high volume data !!!"))
+                    .build()
+            val rawMessageBytes = wrappedBuffer(msg.toByteArray())
+
+            it("should not accept not fully initialized message header ") {
+                val vesMessage = VesMessage(commonHeader, rawMessageBytes)
+                assertThat(cut.isValid(vesMessage)).describedAs("message validation result").isFalse()
+            }
+        }
+    }
+})
\ No newline at end of file
diff --git a/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/RouterTest.kt b/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/RouterTest.kt
new file mode 100644
index 00000000..ac91aaeb
--- /dev/null
+++ b/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/RouterTest.kt
@@ -0,0 +1,92 @@
+package org.onap.dcae.collectors.veshv.impl
+
+import io.netty.buffer.Unpooled
+import org.assertj.core.api.Assertions.assertThat
+import org.jetbrains.spek.api.Spek
+import org.jetbrains.spek.api.dsl.given
+import org.jetbrains.spek.api.dsl.it
+import org.jetbrains.spek.api.dsl.on
+import org.onap.dcae.collectors.veshv.domain.VesMessage
+import org.onap.dcae.collectors.veshv.domain.routing
+import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader
+import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader.Domain
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since May 2018
+ */
+object RouterTest : Spek({
+    given("sample configuration") {
+        val config = routing {
+
+            defineRoute {
+                fromDomain(Domain.HVRANMEAS)
+                toTopic("ves_rtpm")
+                withFixedPartitioning(2)
+            }
+
+            defineRoute {
+                fromDomain(Domain.SYSLOG)
+                toTopic("ves_trace")
+                withFixedPartitioning()
+            }
+        }.build()
+        val cut = Router(config)
+
+        on("message with existing route (rtpm)") {
+            val message = VesMessage(vesCommonHeaderWithDomain(Domain.HVRANMEAS), Unpooled.EMPTY_BUFFER)
+            val result = cut.findDestination(message)
+
+            it("should have route available") {
+                assertThat(result).isNotNull()
+            }
+
+            it("should be routed to proper partition") {
+                assertThat(result?.partition).isEqualTo(2)
+            }
+
+            it("should be routed to proper topic") {
+                assertThat(result?.topic).isEqualTo("ves_rtpm")
+            }
+
+            it("should be routed with a given message") {
+                assertThat(result?.message).isSameAs(message)
+            }
+        }
+
+        on("message with existing route (trace)") {
+            val message = VesMessage(vesCommonHeaderWithDomain(Domain.SYSLOG), Unpooled.EMPTY_BUFFER)
+            val result = cut.findDestination(message)
+
+            it("should have route available") {
+                assertThat(result).isNotNull()
+            }
+
+            it("should be routed to proper partition") {
+                assertThat(result?.partition).isEqualTo(1)
+            }
+
+            it("should be routed to proper topic") {
+                assertThat(result?.topic).isEqualTo("ves_trace")
+            }
+
+            it("should be routed with a given message") {
+                assertThat(result?.message).isSameAs(message)
+            }
+        }
+
+        on("message with unknown route") {
+            val message = VesMessage(vesCommonHeaderWithDomain(Domain.HEARTBEAT), Unpooled.EMPTY_BUFFER)
+            val result = cut.findDestination(message)
+
+            it("should not have route available") {
+                assertThat(result).isNull()
+            }
+        }
+    }
+})
+
+private fun vesCommonHeaderWithDomain(domain: Domain) =
+        CommonEventHeader.getDefaultInstance().toBuilder()
+                .setDomain(domain)
+                .build()
\ No newline at end of file
diff --git a/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/VesDecoderTest.kt b/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/VesDecoderTest.kt
new file mode 100644
index 00000000..4d9c9239
--- /dev/null
+++ b/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/VesDecoderTest.kt
@@ -0,0 +1,69 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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.dcae.collectors.veshv.impl
+
+import com.google.protobuf.ByteString
+import io.netty.buffer.Unpooled.wrappedBuffer
+import org.assertj.core.api.Assertions.assertThat
+import org.jetbrains.spek.api.Spek
+import org.jetbrains.spek.api.dsl.given
+import org.jetbrains.spek.api.dsl.it
+import org.jetbrains.spek.api.dsl.on
+import org.onap.dcae.collectors.veshv.domain.VesMessage
+import org.onap.ves.VesEventV5.VesEvent
+import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader
+import reactor.test.StepVerifier
+import java.nio.charset.Charset
+
+
+internal object VesDecoderTest : Spek({
+
+    given("ves message decoder") {
+        val cut = VesDecoder()
+
+        on("ves hv message bytes") {
+            val commonHeader = CommonEventHeader.getDefaultInstance()
+            val msg = VesEvent.newBuilder()
+                    .setCommonEventHeader(commonHeader)
+                    .setHvRanMeasFields(ByteString.copyFromUtf8("highvolume measurements"))
+                    .build()
+            val rawMessageBytes = wrappedBuffer(msg.toByteArray())
+
+
+            it("should decode only header and pass it on along with raw message") {
+                val expectedMessage = VesMessage(
+                        commonHeader,
+                        rawMessageBytes
+                )
+
+                assertThat(cut.decode(rawMessageBytes)).isEqualTo(expectedMessage)
+
+            }
+        }
+
+        on("invalid ves hv message bytes") {
+            val rawMessageBytes = wrappedBuffer("ala ma kota".toByteArray(Charset.defaultCharset()))
+
+            it("should return empty result") {
+                assertThat(cut.decode(rawMessageBytes)).isNull()
+            }
+        }
+    }
+})
diff --git a/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/WireDecoderTest.kt b/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/WireDecoderTest.kt
new file mode 100644
index 00000000..3563bf6d
--- /dev/null
+++ b/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/WireDecoderTest.kt
@@ -0,0 +1,104 @@
+package org.onap.dcae.collectors.veshv.impl
+
+import io.netty.buffer.Unpooled
+import io.netty.buffer.UnpooledByteBufAllocator
+import org.assertj.core.api.Assertions.assertThat
+import org.jetbrains.spek.api.Spek
+import org.jetbrains.spek.api.dsl.describe
+import org.jetbrains.spek.api.dsl.given
+import org.jetbrains.spek.api.dsl.it
+import org.onap.dcae.collectors.veshv.domain.WireFrame
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk></piotr.jaszczyk>@nokia.com>
+ * @since May 2018
+ */
+internal object WireDecoderTest : Spek({
+    describe("decoding wire protocol") {
+        val cut = WireDecoder()
+
+        fun decode(frame: WireFrame) =
+                cut.decode(
+                        frame.encode(UnpooledByteBufAllocator.DEFAULT))
+
+        given("empty input") {
+            val input = Unpooled.EMPTY_BUFFER
+
+            it("should yield empty result") {
+                assertThat(cut.decode(input)).isNull()
+            }
+        }
+
+        given("input without 0xFF first byte") {
+            val input = WireFrame(
+                    payload = Unpooled.EMPTY_BUFFER,
+                    mark = 0x10,
+                    majorVersion = 1,
+                    minorVersion = 2,
+                    payloadSize = 0)
+
+            it("should yield empty result") {
+                assertThat(decode(input)).isNull()
+            }
+        }
+
+        given("input with unsupported major version") {
+            val input = WireFrame(
+                    payload = Unpooled.EMPTY_BUFFER,
+                    mark = 0xFF,
+                    majorVersion = 100,
+                    minorVersion = 2,
+                    payloadSize = 0)
+
+            it("should yield empty result") {
+                assertThat(decode(input)).isNull()
+            }
+        }
+
+        given("input with too small payload size") {
+            val input = WireFrame(
+                    payload = Unpooled.wrappedBuffer(byteArrayOf(1, 2 ,3)),
+                    mark = 0xFF,
+                    majorVersion = 1,
+                    minorVersion = 0,
+                    payloadSize = 1)
+
+            it("should yield empty result") {
+                assertThat(decode(input)).isNull()
+            }
+        }
+
+        given("input with too big payload size") {
+            val input = WireFrame(
+                    payload = Unpooled.wrappedBuffer(byteArrayOf(1, 2 ,3)),
+                    mark = 0xFF,
+                    majorVersion = 1,
+                    minorVersion = 0,
+                    payloadSize = 8)
+
+            it("should yield empty result") {
+                assertThat(decode(input)).isNull()
+            }
+        }
+
+        given("valid input") {
+            val payload = byteArrayOf(6, 9, 8, 6)
+            val input = WireFrame(
+                    payload = Unpooled.wrappedBuffer(payload),
+                    mark = 0xFF,
+                    majorVersion = 1,
+                    minorVersion = 0,
+                    payloadSize = payload.size)
+
+
+            it("should yield Google Protocol Buffers payload") {
+                val result = decode(input)!!
+
+                val actualPayload = ByteArray(result.readableBytes())
+                result.readBytes(actualPayload)
+
+                assertThat(actualPayload).containsExactly(*payload)
+            }
+        }
+    }
+})
diff --git a/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/ConsulConfigurationProviderTest.kt b/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/ConsulConfigurationProviderTest.kt
new file mode 100644
index 00000000..bf65c859
--- /dev/null
+++ b/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/ConsulConfigurationProviderTest.kt
@@ -0,0 +1,60 @@
+package org.onap.dcae.collectors.veshv.impl.adapters
+
+import com.nhaarman.mockito_kotlin.mock
+import com.nhaarman.mockito_kotlin.whenever
+import org.jetbrains.spek.api.Spek
+import org.jetbrains.spek.api.dsl.given
+import org.jetbrains.spek.api.dsl.it
+import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader.Domain
+import reactor.core.publisher.Mono
+import java.util.*
+import kotlin.test.assertEquals
+
+/**
+ * @author Jakub Dudycz <jakub.dudycz@nokia.com>
+ * @since May 2018
+ */
+internal object ConsulConfigurationProviderTest : Spek({
+
+    given("valid resource url") {
+        val testUrl = "http://valid-url/"
+        val httpAdapterMock: HttpAdapter = mock()
+        val consulConfigProvider = ConsulConfigurationProvider(testUrl, httpAdapterMock)
+
+        whenever(httpAdapterMock.getResponse(testUrl)).thenReturn(Mono.just(constructConsulResponse()))
+
+
+        it("should create valid collector configuration") {
+            val response = consulConfigProvider().blockFirst()
+            assertEquals("val1", response.kafkaBootstrapServers)
+            val route = response.routing.routes[0]
+            assertEquals(Domain.MEASUREMENTS_FOR_VF_SCALING, route.domain)
+            assertEquals("val3", route.targetTopic)
+        }
+    }
+})
+
+fun constructConsulResponse(): String {
+
+    val config = """{
+        "kafkaBootstrapServers": "val1",
+        "routing": {
+            "fromDomain": 2,
+            "toTopic": "val3"
+        }
+    }"""
+
+    val encodedValue = String(Base64.getEncoder().encode(config.toByteArray()))
+
+    return """[
+        {
+            "CreateIndex": 100,
+            "ModifyIndex": 200,
+            "LockIndex": 200,
+            "Key": "zip",
+            "Flags": 0,
+            "Value": "$encodedValue",
+            "Session": "adf4238a-882b-9ddc-4a9d-5b6758e4159e"
+        }
+    ]"""
+}
diff --git a/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/HttpAdapterTest.kt b/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/HttpAdapterTest.kt
new file mode 100644
index 00000000..10e8b6bf
--- /dev/null
+++ b/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/HttpAdapterTest.kt
@@ -0,0 +1,60 @@
+package org.onap.dcae.collectors.veshv.impl.adapters
+
+import com.nhaarman.mockito_kotlin.mock
+import com.nhaarman.mockito_kotlin.whenever
+import io.netty.buffer.Unpooled
+import io.netty.handler.codec.http.HttpContent
+import org.jetbrains.spek.api.Spek
+import org.jetbrains.spek.api.dsl.given
+import org.jetbrains.spek.api.dsl.it
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+import reactor.ipc.netty.http.client.HttpClient
+import reactor.ipc.netty.http.client.HttpClientResponse
+import java.nio.charset.Charset
+import kotlin.test.assertEquals
+
+/**
+ * @author Jakub Dudycz <jakub.dudycz@nokia.com>
+ * @since May 2018
+ */
+internal object HttpAdapterTest : Spek({
+
+    given("valid resource url") {
+
+        val httpClientMock: HttpClient = mock()
+        val httpAdapter = HttpAdapter(httpClientMock)
+        val testUrl = "http://valid-url/"
+        val responseContent = """{"key1": "value1", "key2": "value2"}"""
+        val httpResponse = createHttpResponseMock(responseContent)
+        whenever(httpClientMock.get(testUrl)).thenReturn(Mono.just(httpResponse))
+
+        it("should return response string") {
+            assertEquals(responseContent, httpAdapter.getResponse(testUrl).block())
+        }
+    }
+
+    given("invalid resource url") {
+
+        val httpClientMock: HttpClient = mock()
+        val httpAdapter = HttpAdapter(httpClientMock)
+        val testUrl = "http://invalid-url/"
+        whenever(httpClientMock.get(testUrl)).thenReturn(Mono.error(Exception("Test exception")))
+
+
+        it("should return null response") {
+            assertEquals(null, httpAdapter.getResponse(testUrl).block())
+        }
+    }
+})
+
+fun createHttpResponseMock(content: String): HttpClientResponse {
+    val responseMock: HttpClientResponse = mock()
+    val contentMock: HttpContent = mock()
+    val contentByteBuff = Unpooled.copiedBuffer(content, Charset.defaultCharset())
+
+    whenever(responseMock.receiveContent()).thenReturn(Flux.just(contentMock))
+    whenever(contentMock.content()).thenReturn(contentByteBuff)
+
+    return responseMock
+}
diff --git a/hv-collector-core/src/test/resources/logback.xml b/hv-collector-core/src/test/resources/logback.xml
new file mode 100644
index 00000000..809f62d4
--- /dev/null
+++ b/hv-collector-core/src/test/resources/logback.xml
@@ -0,0 +1,35 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration>
+  <property name="LOG_FILE"
+    value="${LOG_FILE:-${LOG_PATH:-${LOG_TEMP:-${java.io.tmpdir:-/tmp}}/}ves-hv.log}"/>
+  <property name="FILE_LOG_PATTERN" value="%d{yyyy-MM-dd'T'HH:mm:ss.SSSXXX,UTC} %-5level [%-40.40logger{10}] - %msg%n"/>
+
+  <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
+    <encoder>
+      <pattern>
+        %d{yyyy-MM-dd'T'HH:mm:ss.SSSXXX,UTC} %highlight(%-5level) [%-40.40logger{10}] - %msg%n
+      </pattern>
+    </encoder>
+  </appender>
+
+    <appender name="ROLLING-FILE"
+      class="ch.qos.logback.core.rolling.RollingFileAppender">
+      <encoder>
+        <pattern>${FILE_LOG_PATTERN}</pattern>
+      </encoder>
+      <file>${LOG_FILE}</file>
+      <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
+        <fileNamePattern>${LOG_FILE}.%d{yyyy-MM-dd}.log</fileNamePattern>
+        <maxFileSize>50MB</maxFileSize>
+        <maxHistory>30</maxHistory>
+        <totalSizeCap>10GB</totalSizeCap>
+      </rollingPolicy>
+    </appender>
+
+  <logger name="org.onap.dcae.collectors.veshv" level="DEBUG"/>
+
+  <root level="INFO">
+      <appender-ref ref="CONSOLE"/>
+      <appender-ref ref="ROLLING-FILE"/>
+    </root>
+</configuration>
\ No newline at end of file
diff --git a/hv-collector-coverage/pom.xml b/hv-collector-coverage/pom.xml
new file mode 100644
index 00000000..5132b1f5
--- /dev/null
+++ b/hv-collector-coverage/pom.xml
@@ -0,0 +1,97 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ ============LICENSE_START=======================================================
+  ~ dcaegen2-collectors-veshv
+  ~ ================================================================================
+  ~ Copyright (C) 2018 NOKIA Intellectual Property. 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>
+
+  <licenses>
+    <license>
+      <name>The Apache Software License, Version 2.0</name>
+      <url>http://www.apache.org/licenses/LICENSE-2.0.txt</url>
+    </license>
+  </licenses>
+
+  <parent>
+    <groupId>org.onap.dcaegen2.collectors.veshv</groupId>
+    <artifactId>ves-hv-collector</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+    <relativePath>..</relativePath>
+  </parent>
+
+  <artifactId>hv-collector-coverage</artifactId>
+  <description>VES HighVolume Collector :: Code coverage</description>
+  <packaging>pom</packaging>
+
+  <properties>
+    <failIfMissingUnitTests>false</failIfMissingUnitTests>
+    <failIfMissingComponentTests>false</failIfMissingComponentTests>
+  </properties>
+
+  <profiles>
+    <profile>
+      <id>analysis</id>
+      <build>
+        <plugins>
+          <plugin>
+            <groupId>org.jacoco</groupId>
+            <artifactId>jacoco-maven-plugin</artifactId>
+            <executions>
+              <execution>
+                <id>default-report</id>
+                <phase>package</phase>
+                <goals>
+                  <goal>report-aggregate</goal>
+                </goals>
+              </execution>
+            </executions>
+          </plugin>
+        </plugins>
+      </build>
+    </profile>
+  </profiles>
+
+  <dependencies>
+    <dependency>
+      <groupId>${project.parent.groupId}</groupId>
+      <artifactId>hv-collector-core</artifactId>
+      <version>${project.parent.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>${project.parent.groupId}</groupId>
+      <artifactId>hv-collector-ct</artifactId>
+      <version>${project.parent.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>${project.parent.groupId}</groupId>
+      <artifactId>hv-collector-main</artifactId>
+      <version>${project.parent.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>${project.parent.groupId}</groupId>
+      <artifactId>hv-collector-utils</artifactId>
+      <version>${project.parent.version}</version>
+    </dependency>
+
+  </dependencies>
+
+
+</project>
\ No newline at end of file
diff --git a/hv-collector-ct/pom.xml b/hv-collector-ct/pom.xml
new file mode 100644
index 00000000..6ad304f7
--- /dev/null
+++ b/hv-collector-ct/pom.xml
@@ -0,0 +1,100 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ ============LICENSE_START=======================================================
+  ~ dcaegen2-collectors-veshv
+  ~ ================================================================================
+  ~ Copyright (C) 2018 NOKIA Intellectual Property. 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>
+
+    <licenses>
+        <license>
+            <name>The Apache Software License, Version 2.0</name>
+            <url>http://www.apache.org/licenses/LICENSE-2.0.txt</url>
+        </license>
+    </licenses>
+
+    <parent>
+        <groupId>org.onap.dcaegen2.collectors.veshv</groupId>
+        <artifactId>ves-hv-collector</artifactId>
+        <version>1.0.0-SNAPSHOT</version>
+        <relativePath>..</relativePath>
+    </parent>
+
+    <artifactId>hv-collector-ct</artifactId>
+    <description>VES HighVolume Collector :: Component tests</description>
+
+    <properties>
+        <failIfMissingUnitTests>false</failIfMissingUnitTests>
+        <failIfMissingComponentTests>true</failIfMissingComponentTests>
+    </properties>
+
+    <build>
+        <plugins>
+            <plugin>
+                <artifactId>kotlin-maven-plugin</artifactId>
+                <groupId>org.jetbrains.kotlin</groupId>
+            </plugin>
+            <plugin>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <groupId>org.apache.maven.plugins</groupId>
+            </plugin>
+        </plugins>
+    </build>
+
+    <dependencies>
+        <dependency>
+            <groupId>${project.parent.groupId}</groupId>
+            <artifactId>hv-collector-core</artifactId>
+            <version>${project.parent.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>ch.qos.logback</groupId>
+            <artifactId>logback-classic</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>commons-cli</groupId>
+            <artifactId>commons-cli</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.jetbrains.kotlin</groupId>
+            <artifactId>kotlin-test</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.jetbrains.spek</groupId>
+            <artifactId>spek-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.jetbrains.spek</groupId>
+            <artifactId>spek-junit-platform-engine</artifactId>
+        </dependency>
+    </dependencies>
+
+
+</project>
\ No newline at end of file
diff --git a/hv-collector-ct/src/test/kotlin/org/onap/dcae/collectors/veshv/tests/component/Sut.kt b/hv-collector-ct/src/test/kotlin/org/onap/dcae/collectors/veshv/tests/component/Sut.kt
new file mode 100644
index 00000000..f8eefa5a
--- /dev/null
+++ b/hv-collector-ct/src/test/kotlin/org/onap/dcae/collectors/veshv/tests/component/Sut.kt
@@ -0,0 +1,50 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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.dcae.collectors.veshv.tests.component
+
+import io.netty.buffer.ByteBuf
+import org.onap.dcae.collectors.veshv.boundary.Collector
+import org.onap.dcae.collectors.veshv.boundary.SinkProvider
+import org.onap.dcae.collectors.veshv.domain.RoutedMessage
+import org.onap.dcae.collectors.veshv.factory.CollectorFactory
+import org.onap.dcae.collectors.veshv.tests.fakes.FakeConfigurationProvider
+import org.onap.dcae.collectors.veshv.tests.fakes.FakeSink
+import reactor.core.publisher.Flux
+import java.time.Duration
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since May 2018
+ */
+internal class Sut {
+    val configurationProvider = FakeConfigurationProvider()
+    val sink = FakeSink()
+    private val collectorFactory = CollectorFactory(configurationProvider, SinkProvider.just(sink))
+    val collectorProvider = collectorFactory.createVesHvCollectorProvider()
+
+    val collector: Collector
+        get() = collectorProvider()
+
+    fun handleConnection(vararg packets: ByteBuf): List<RoutedMessage> {
+        collector.handleConnection(Flux.fromArray(packets)).block(Duration.ofSeconds(10))
+
+        return sink.sentMessages
+    }
+}
diff --git a/hv-collector-ct/src/test/kotlin/org/onap/dcae/collectors/veshv/tests/component/VesHvSpecification.kt b/hv-collector-ct/src/test/kotlin/org/onap/dcae/collectors/veshv/tests/component/VesHvSpecification.kt
new file mode 100644
index 00000000..2cfb785e
--- /dev/null
+++ b/hv-collector-ct/src/test/kotlin/org/onap/dcae/collectors/veshv/tests/component/VesHvSpecification.kt
@@ -0,0 +1,94 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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.dcae.collectors.veshv.tests.component
+
+import org.assertj.core.api.Assertions.assertThat
+import org.jetbrains.spek.api.Spek
+import org.jetbrains.spek.api.dsl.describe
+import org.onap.dcae.collectors.veshv.tests.fakes.HVRANMEAS_TOPIC
+import org.onap.dcae.collectors.veshv.tests.fakes.basicConfiguration
+import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader.Domain
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since May 2018
+ */
+object VesHvSpecification : Spek({
+    describe("VES High Volume Collector") {
+        system("should handle multiple HV RAN events") { sut ->
+            sut.configurationProvider.updateConfiguration(basicConfiguration)
+            val messages = sut.handleConnection(vesMessage(Domain.HVRANMEAS), vesMessage(Domain.HVRANMEAS))
+
+            assertThat(messages)
+                    .describedAs("should send all events")
+                    .hasSize(2)
+        }
+
+        system("should release memory for each incoming message") { sut ->
+            sut.configurationProvider.updateConfiguration(basicConfiguration)
+            val msgWithInvalidDomain = vesMessage(Domain.OTHER)
+            val msgWithInvalidPayload = invalidVesMessage()
+            val msgWithInvalidFrame = invalidWireFrame()
+            val validMessage = vesMessage(Domain.HVRANMEAS)
+
+            sut.handleConnection(msgWithInvalidDomain, msgWithInvalidPayload, msgWithInvalidFrame, validMessage)
+
+            assertThat(msgWithInvalidDomain.refCnt())
+                    .describedAs("message with invalid domain should be released")
+                    .isEqualTo(0)
+            assertThat(msgWithInvalidPayload.refCnt())
+                    .describedAs("message with invalid payload should be released")
+                    .isEqualTo(0)
+            assertThat(msgWithInvalidFrame.refCnt())
+                    .describedAs("message with invalid frame should be released")
+                    .isEqualTo(0)
+            assertThat(validMessage.refCnt())
+                    .describedAs("handled message should be released")
+                    .isEqualTo(0)
+        }
+    }
+
+    describe("message routing") {
+        system("should direct message to a topic by means of routing configuration") { sut ->
+            sut.configurationProvider.updateConfiguration(basicConfiguration)
+
+            val messages = sut.handleConnection(vesMessage(Domain.HVRANMEAS))
+            assertThat(messages).describedAs("number of routed messages").hasSize(1)
+
+            val msg = messages[0]
+            assertThat(msg.topic).describedAs("routed message topic").isEqualTo(HVRANMEAS_TOPIC)
+            assertThat(msg.partition).describedAs("routed message partition").isEqualTo(1)
+        }
+
+        system("should drop message if route was not found") { sut ->
+            sut.configurationProvider.updateConfiguration(basicConfiguration)
+            val messages = sut.handleConnection(
+                    vesMessage(Domain.OTHER, "first"),
+                    vesMessage(Domain.HVRANMEAS, "second"),
+                    vesMessage(Domain.HEARTBEAT, "third"))
+
+            assertThat(messages).describedAs("number of routed messages").hasSize(1)
+
+            val msg = messages[0]
+            assertThat(msg.topic).describedAs("routed message topic").isEqualTo(HVRANMEAS_TOPIC)
+            assertThat(msg.message.header.eventId).describedAs("routed message eventId").isEqualTo("second")
+        }
+    }
+})
diff --git a/hv-collector-ct/src/test/kotlin/org/onap/dcae/collectors/veshv/tests/component/spek_extensions.kt b/hv-collector-ct/src/test/kotlin/org/onap/dcae/collectors/veshv/tests/component/spek_extensions.kt
new file mode 100644
index 00000000..c30ba7e1
--- /dev/null
+++ b/hv-collector-ct/src/test/kotlin/org/onap/dcae/collectors/veshv/tests/component/spek_extensions.kt
@@ -0,0 +1,31 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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.dcae.collectors.veshv.tests.component
+
+import org.jetbrains.spek.api.dsl.Pending
+import org.jetbrains.spek.api.dsl.TestContainer
+
+internal fun TestContainer.system(description: String, body: (Sut) -> Unit) {
+    test("system $description", body = { body(Sut()) })
+}
+
+internal fun TestContainer.xsystem(description: String, reason: String? = null, body: (Sut) -> Unit = {}) {
+    test("system $description", Pending.Yes(reason), body = { body(Sut()) })
+}
diff --git a/hv-collector-ct/src/test/kotlin/org/onap/dcae/collectors/veshv/tests/component/utils.kt b/hv-collector-ct/src/test/kotlin/org/onap/dcae/collectors/veshv/tests/component/utils.kt
new file mode 100644
index 00000000..e4958fdb
--- /dev/null
+++ b/hv-collector-ct/src/test/kotlin/org/onap/dcae/collectors/veshv/tests/component/utils.kt
@@ -0,0 +1,81 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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.dcae.collectors.veshv.tests.component
+
+import com.google.protobuf.ByteString
+import io.netty.buffer.ByteBuf
+import io.netty.buffer.ByteBufAllocator
+import io.netty.buffer.PooledByteBufAllocator
+import io.netty.buffer.Unpooled
+import org.onap.ves.VesEventV5.VesEvent
+import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader
+import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader.Domain
+import java.nio.charset.Charset
+import java.util.*
+
+val alocator: ByteBufAllocator = PooledByteBufAllocator.DEFAULT
+
+fun vesMessage(domain: Domain = Domain.OTHER, id: String = UUID.randomUUID().toString()) = alocator.buffer().run {
+    writeByte(0xFF) // always 0xFF
+    writeByte(1)   // major version
+    writeByte(0)   // minor version
+
+    val gpb = vesEvent(domain, id).toByteString().asReadOnlyByteBuffer()
+    writeInt(gpb.limit())  // ves event size in bytes
+    writeBytes(gpb)  // ves event as GPB bytes
+}
+
+
+fun invalidVesMessage() = alocator.buffer().run {
+    writeByte(0xFF) // always 0xFF
+    writeByte(1)   // major version
+    writeByte(0)   // minor version
+
+    val invalidGpb = "some random data".toByteArray(Charsets.UTF_8)
+    writeInt(invalidGpb.size)  // ves event size in bytes
+    writeBytes(invalidGpb)
+
+}
+
+fun invalidWireFrame() = alocator.buffer().run {
+    writeByte(0xFF)
+    writeByte(1)
+    writeByte(0)
+}
+
+fun vesEvent(domain: Domain = Domain.OTHER, id: String = UUID.randomUUID().toString()) =
+        VesEvent.newBuilder()
+                .setCommonEventHeader(
+                        CommonEventHeader.getDefaultInstance().toBuilder()
+                                .setVersion("1.0")
+                                .setEventId(id)
+                                .setDomain(domain)
+                                .setEventName("Sample event name")
+                                .setSourceName("Sample Source")
+                                .setReportingEntityName(ByteString.copyFromUtf8("Sample byte String"))
+                                .setPriority(CommonEventHeader.Priority.MEDIUM)
+                                .setStartEpochMicrosec(120034455)
+                                .setLastEpochMicrosec(120034459)
+                                .setSequence(1))
+                .setHvRanMeasFields(ByteString.EMPTY)
+                .build()
+
+
+
diff --git a/hv-collector-ct/src/test/kotlin/org/onap/dcae/collectors/veshv/tests/fakes/configuration.kt b/hv-collector-ct/src/test/kotlin/org/onap/dcae/collectors/veshv/tests/fakes/configuration.kt
new file mode 100644
index 00000000..728e0c5c
--- /dev/null
+++ b/hv-collector-ct/src/test/kotlin/org/onap/dcae/collectors/veshv/tests/fakes/configuration.kt
@@ -0,0 +1,52 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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.dcae.collectors.veshv.tests.fakes
+
+import org.onap.dcae.collectors.veshv.boundary.ConfigurationProvider
+import org.onap.dcae.collectors.veshv.domain.CollectorConfiguration
+import org.onap.dcae.collectors.veshv.domain.routing
+import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader.Domain
+import reactor.core.publisher.FluxProcessor
+import reactor.core.publisher.UnicastProcessor
+
+
+const val HVRANMEAS_TOPIC = "ves_hvRanMeas"
+
+val basicConfiguration: CollectorConfiguration = CollectorConfiguration(
+        kafkaBootstrapServers = "localhost:9969",
+        routing = routing {
+            defineRoute {
+                fromDomain(Domain.HVRANMEAS)
+                toTopic(HVRANMEAS_TOPIC)
+                withFixedPartitioning()
+            }
+        }.build()
+)
+
+
+class FakeConfigurationProvider : ConfigurationProvider {
+    private val configStream: FluxProcessor<CollectorConfiguration, CollectorConfiguration> = UnicastProcessor.create()
+
+    fun updateConfiguration(collectorConfiguration: CollectorConfiguration) {
+        configStream.onNext(collectorConfiguration)
+    }
+
+    override fun invoke() = configStream
+}
\ No newline at end of file
diff --git a/hv-collector-ct/src/test/kotlin/org/onap/dcae/collectors/veshv/tests/fakes/sink.kt b/hv-collector-ct/src/test/kotlin/org/onap/dcae/collectors/veshv/tests/fakes/sink.kt
new file mode 100644
index 00000000..7da72151
--- /dev/null
+++ b/hv-collector-ct/src/test/kotlin/org/onap/dcae/collectors/veshv/tests/fakes/sink.kt
@@ -0,0 +1,43 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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.dcae.collectors.veshv.tests.fakes
+
+import org.onap.dcae.collectors.veshv.boundary.Sink
+import org.onap.dcae.collectors.veshv.domain.RoutedMessage
+import org.onap.dcae.collectors.veshv.domain.VesMessage
+import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader
+import reactor.core.publisher.Flux
+import java.util.*
+import java.util.concurrent.ConcurrentLinkedDeque
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since May 2018
+ */
+class FakeSink : Sink {
+    private val sent: Deque<RoutedMessage> = ConcurrentLinkedDeque()
+
+    val sentMessages: List<RoutedMessage>
+        get() = sent.toList()
+
+    override fun send(messages: Flux<RoutedMessage>): Flux<VesMessage> {
+        return messages.doOnNext(sent::addLast).map { it.message }
+    }
+}
diff --git a/hv-collector-ct/src/test/resources/logback-test.xml b/hv-collector-ct/src/test/resources/logback-test.xml
new file mode 100644
index 00000000..809f62d4
--- /dev/null
+++ b/hv-collector-ct/src/test/resources/logback-test.xml
@@ -0,0 +1,35 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration>
+  <property name="LOG_FILE"
+    value="${LOG_FILE:-${LOG_PATH:-${LOG_TEMP:-${java.io.tmpdir:-/tmp}}/}ves-hv.log}"/>
+  <property name="FILE_LOG_PATTERN" value="%d{yyyy-MM-dd'T'HH:mm:ss.SSSXXX,UTC} %-5level [%-40.40logger{10}] - %msg%n"/>
+
+  <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
+    <encoder>
+      <pattern>
+        %d{yyyy-MM-dd'T'HH:mm:ss.SSSXXX,UTC} %highlight(%-5level) [%-40.40logger{10}] - %msg%n
+      </pattern>
+    </encoder>
+  </appender>
+
+    <appender name="ROLLING-FILE"
+      class="ch.qos.logback.core.rolling.RollingFileAppender">
+      <encoder>
+        <pattern>${FILE_LOG_PATTERN}</pattern>
+      </encoder>
+      <file>${LOG_FILE}</file>
+      <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
+        <fileNamePattern>${LOG_FILE}.%d{yyyy-MM-dd}.log</fileNamePattern>
+        <maxFileSize>50MB</maxFileSize>
+        <maxHistory>30</maxHistory>
+        <totalSizeCap>10GB</totalSizeCap>
+      </rollingPolicy>
+    </appender>
+
+  <logger name="org.onap.dcae.collectors.veshv" level="DEBUG"/>
+
+  <root level="INFO">
+      <appender-ref ref="CONSOLE"/>
+      <appender-ref ref="ROLLING-FILE"/>
+    </root>
+</configuration>
\ No newline at end of file
diff --git a/hv-collector-main/pom.xml b/hv-collector-main/pom.xml
new file mode 100644
index 00000000..9f02ed58
--- /dev/null
+++ b/hv-collector-main/pom.xml
@@ -0,0 +1,99 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ ============LICENSE_START=======================================================
+  ~ dcaegen2-collectors-veshv
+  ~ ================================================================================
+  ~ Copyright (C) 2018 NOKIA Intellectual Property. 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>
+
+    <licenses>
+        <license>
+            <name>The Apache Software License, Version 2.0</name>
+            <url>http://www.apache.org/licenses/LICENSE-2.0.txt</url>
+        </license>
+    </licenses>
+
+    <parent>
+        <groupId>org.onap.dcaegen2.collectors.veshv</groupId>
+        <artifactId>ves-hv-collector</artifactId>
+        <version>1.0.0-SNAPSHOT</version>
+        <relativePath>..</relativePath>
+    </parent>
+
+    <artifactId>hv-collector-main</artifactId>
+    <description>VES HighVolume Collector :: Main</description>
+
+    <properties>
+        <skipAnalysis>false</skipAnalysis>
+    </properties>
+
+    <build>
+        <plugins>
+            <plugin>
+                <artifactId>kotlin-maven-plugin</artifactId>
+                <groupId>org.jetbrains.kotlin</groupId>
+            </plugin>
+            <plugin>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <groupId>org.apache.maven.plugins</groupId>
+            </plugin>
+        </plugins>
+    </build>
+
+    <dependencies>
+        <dependency>
+            <groupId>${project.parent.groupId}</groupId>
+            <artifactId>hv-collector-core</artifactId>
+            <version>${project.parent.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>ch.qos.logback</groupId>
+            <artifactId>logback-classic</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>commons-cli</groupId>
+            <artifactId>commons-cli</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.jetbrains.kotlin</groupId>
+            <artifactId>kotlin-test</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.jetbrains.spek</groupId>
+            <artifactId>spek-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.jetbrains.spek</groupId>
+            <artifactId>spek-junit-platform-engine</artifactId>
+        </dependency>
+    </dependencies>
+
+
+</project>
diff --git a/hv-collector-main/src/main/kotlin/org/onap/dcae/collectors/veshv/main/ArgBasedServerConfiguration.kt b/hv-collector-main/src/main/kotlin/org/onap/dcae/collectors/veshv/main/ArgBasedServerConfiguration.kt
new file mode 100644
index 00000000..6311b6c0
--- /dev/null
+++ b/hv-collector-main/src/main/kotlin/org/onap/dcae/collectors/veshv/main/ArgBasedServerConfiguration.kt
@@ -0,0 +1,82 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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.dcae.collectors.veshv.main
+
+import org.apache.commons.cli.*
+import org.onap.dcae.collectors.veshv.domain.ServerConfiguration
+
+internal object DefaultValues {
+    const val PORT = 8600
+    const val CONFIG_URL = ""
+}
+
+internal object ArgBasedServerConfiguration {
+    private val OPT_PORT = Option.builder("p")
+            .longOpt("listen-port")
+            .hasArg()
+            .desc("Listen port")
+            .build()
+
+    private val OPT_CONFIG_URL = Option.builder("c")
+            .longOpt("config-url")
+            .optionalArg(true)
+            .hasArg()
+            .desc("Url of ves configuration on consul")
+            .build()
+
+    private val options by lazy {
+        val options = Options()
+        options.addOption(OPT_PORT)
+        options.addOption(OPT_CONFIG_URL)
+        options
+    }
+
+    fun parse(args: Array<out String>): ServerConfiguration {
+        val parser = DefaultParser()
+
+        try {
+            parser.parse(options, args).run {
+                return ServerConfiguration(
+                        intValue(OPT_PORT, DefaultValues.PORT),
+                        stringValue(OPT_CONFIG_URL, DefaultValues.CONFIG_URL))
+            }
+        } catch (ex: Exception) {
+            throw WrongArgumentException(ex)
+        }
+    }
+
+    private fun CommandLine.intValue(option: Option, default: Int) =
+            getOptionValue(option.opt)?.toInt() ?: default
+
+    private fun CommandLine.stringValue(option: Option, default: String) =
+            getOptionValue(option.opt) ?: default
+
+
+    class WrongArgumentException(parent: Exception) : Exception(parent.message, parent) {
+        fun printMessage() {
+            println(message)
+        }
+
+        fun printHelp(programName: String) {
+            val formatter = HelpFormatter()
+            formatter.printHelp(programName, options)
+        }
+    }
+}
diff --git a/hv-collector-main/src/main/kotlin/org/onap/dcae/collectors/veshv/main/main.kt b/hv-collector-main/src/main/kotlin/org/onap/dcae/collectors/veshv/main/main.kt
new file mode 100644
index 00000000..d81a063d
--- /dev/null
+++ b/hv-collector-main/src/main/kotlin/org/onap/dcae/collectors/veshv/main/main.kt
@@ -0,0 +1,72 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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.dcae.collectors.veshv.main
+
+import org.onap.dcae.collectors.veshv.boundary.ConfigurationProvider
+import org.onap.dcae.collectors.veshv.domain.CollectorConfiguration
+import org.onap.dcae.collectors.veshv.domain.ServerConfiguration
+import org.onap.dcae.collectors.veshv.domain.routing
+import org.onap.dcae.collectors.veshv.factory.CollectorFactory
+import org.onap.dcae.collectors.veshv.factory.ServerFactory
+import org.onap.dcae.collectors.veshv.impl.adapters.AdapterFactory
+import org.onap.dcae.collectors.veshv.main.ArgBasedServerConfiguration.WrongArgumentException
+import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader.Domain
+import org.slf4j.LoggerFactory
+import kotlin.system.exitProcess
+
+private val logger = LoggerFactory.getLogger("main")
+
+fun main(args: Array<String>) {
+    try {
+        val serverConfiguration = ArgBasedServerConfiguration.parse(args)
+
+        val collectorProvider = CollectorFactory(
+                resolveConfigurationProvider(serverConfiguration),
+                AdapterFactory.kafkaSink()
+        ).createVesHvCollectorProvider()
+        ServerFactory.createNettyTcpServer(serverConfiguration, collectorProvider).start().block()
+    } catch (ex: WrongArgumentException) {
+        ex.printMessage()
+        ex.printHelp("java org.onap.dcae.collectors.veshv.main.MainKt")
+        exitProcess(1)
+    }
+}
+
+
+private fun resolveConfigurationProvider(serverConfiguration: ServerConfiguration): ConfigurationProvider {
+
+    if (serverConfiguration.configurationUrl.isEmpty()) {
+        logger.info("Configuration url not specified - using default config")
+        val sampleConfig = CollectorConfiguration(
+                kafkaBootstrapServers = "dmaap.cluster.local:9969",
+                routing = routing {
+                    defineRoute {
+                        fromDomain(Domain.HVRANMEAS)
+                        toTopic("ves_hvRanMeas")
+                        withFixedPartitioning()
+                    }
+                }.build()
+        )
+        return AdapterFactory.staticConfigurationProvider(sampleConfig)
+    }
+
+    logger.info("Using configuration url: ${serverConfiguration.configurationUrl}")
+    return AdapterFactory.consulConfigurationProvider(serverConfiguration.configurationUrl)
+}
diff --git a/hv-collector-main/src/main/resources/logback.xml b/hv-collector-main/src/main/resources/logback.xml
new file mode 100644
index 00000000..809f62d4
--- /dev/null
+++ b/hv-collector-main/src/main/resources/logback.xml
@@ -0,0 +1,35 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration>
+  <property name="LOG_FILE"
+    value="${LOG_FILE:-${LOG_PATH:-${LOG_TEMP:-${java.io.tmpdir:-/tmp}}/}ves-hv.log}"/>
+  <property name="FILE_LOG_PATTERN" value="%d{yyyy-MM-dd'T'HH:mm:ss.SSSXXX,UTC} %-5level [%-40.40logger{10}] - %msg%n"/>
+
+  <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
+    <encoder>
+      <pattern>
+        %d{yyyy-MM-dd'T'HH:mm:ss.SSSXXX,UTC} %highlight(%-5level) [%-40.40logger{10}] - %msg%n
+      </pattern>
+    </encoder>
+  </appender>
+
+    <appender name="ROLLING-FILE"
+      class="ch.qos.logback.core.rolling.RollingFileAppender">
+      <encoder>
+        <pattern>${FILE_LOG_PATTERN}</pattern>
+      </encoder>
+      <file>${LOG_FILE}</file>
+      <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
+        <fileNamePattern>${LOG_FILE}.%d{yyyy-MM-dd}.log</fileNamePattern>
+        <maxFileSize>50MB</maxFileSize>
+        <maxHistory>30</maxHistory>
+        <totalSizeCap>10GB</totalSizeCap>
+      </rollingPolicy>
+    </appender>
+
+  <logger name="org.onap.dcae.collectors.veshv" level="DEBUG"/>
+
+  <root level="INFO">
+      <appender-ref ref="CONSOLE"/>
+      <appender-ref ref="ROLLING-FILE"/>
+    </root>
+</configuration>
\ No newline at end of file
diff --git a/hv-collector-main/src/test/kotlin/org/onap/dcae/collectors/veshv/main/ArgBasedServerConfigurationTest.kt b/hv-collector-main/src/test/kotlin/org/onap/dcae/collectors/veshv/main/ArgBasedServerConfigurationTest.kt
new file mode 100644
index 00000000..0d2188ca
--- /dev/null
+++ b/hv-collector-main/src/test/kotlin/org/onap/dcae/collectors/veshv/main/ArgBasedServerConfigurationTest.kt
@@ -0,0 +1,72 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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.dcae.collectors.veshv.main
+
+import org.assertj.core.api.Assertions.assertThat
+import org.jetbrains.spek.api.Spek
+import org.jetbrains.spek.api.dsl.given
+import org.jetbrains.spek.api.dsl.it
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since May 2018
+ */
+object ArgBasedServerConfigurationTest : Spek({
+    val cut = ArgBasedServerConfiguration
+    val configurationUrl = "http://test-address/test"
+
+    fun parse(vararg cmdLine: String) = cut.parse(cmdLine)
+
+    given("all parameters are present in the long form") {
+        val result = parse("--listen-port", "6969", "--config-url", configurationUrl)
+
+        it("should set proper port") {
+            assertThat(result.port).isEqualTo(6969)
+        }
+
+        it("should set proper config url") {
+            assertThat(result.configurationUrl).isEqualTo(configurationUrl)
+        }
+    }
+
+    given("all parameters are present in the short form") {
+        val result = parse("-p", "666", "-c", configurationUrl)
+
+        it("should set proper port") {
+            assertThat(result.port).isEqualTo(666)
+        }
+
+        it("should set proper config url") {
+            assertThat(result.configurationUrl).isEqualTo(configurationUrl)
+        }
+    }
+
+    given("all optional parameters are absent") {
+        val result = parse()
+
+        it("should set default port") {
+            assertThat(result.port).isEqualTo(DefaultValues.PORT)
+        }
+
+        it("should set default config url") {
+            assertThat(result.configurationUrl).isEqualTo(DefaultValues.CONFIG_URL)
+        }
+    }
+})
\ No newline at end of file
diff --git a/hv-collector-main/src/test/kotlin/org/onap/dcae/collectors/veshv/main/NioBuffersTest.kt b/hv-collector-main/src/test/kotlin/org/onap/dcae/collectors/veshv/main/NioBuffersTest.kt
new file mode 100644
index 00000000..b46d5a28
--- /dev/null
+++ b/hv-collector-main/src/test/kotlin/org/onap/dcae/collectors/veshv/main/NioBuffersTest.kt
@@ -0,0 +1,87 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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.dcae.collectors.veshv.main
+
+import org.jetbrains.spek.api.Spek
+import org.jetbrains.spek.api.dsl.describe
+import org.jetbrains.spek.api.dsl.it
+import java.nio.ByteBuffer
+
+fun Int.toKibibytes(): Int = this * 1024
+fun Int.toMebibytes(): Int = this * 1024 * 1024
+
+
+object NioBuffersTest : Spek({
+    val BUFFER_SIZES = listOf(128.toKibibytes(), 512.toKibibytes(), 1.toMebibytes(), 2.toMebibytes())
+    val NUMBER_OF_ITERATIONS = 100
+
+    fun measureCopyTimeInNanos(bb1: ByteBuffer, bb2: ByteBuffer): Double {
+        bb1.clear()
+        bb2.clear()
+        val start = System.nanoTime()
+        while (bb2.remaining() > 0)
+            bb2.putInt(bb1.getInt())
+        val time = System.nanoTime() - start
+        val operations = bb1.capacity() / Integer.BYTES
+        return time.toDouble() / operations
+    }
+
+    fun measureAverageCopyTimeInNanos(bb1: ByteBuffer, bb2: ByteBuffer): Double =
+            (0..NUMBER_OF_ITERATIONS).map { measureCopyTimeInNanos(bb1, bb2) }.average()
+
+    fun measureAndPrintAverageCopyTime(message: String, bb1: ByteBuffer, bb2: ByteBuffer) {
+        val avg = measureAverageCopyTimeInNanos(bb1, bb2)
+        System.out.printf("Each putInt+getInt for %s took an average of %.1f ns%n", message, avg)
+    }
+
+    for (singleBufferSize in BUFFER_SIZES) {
+
+        describe("$singleBufferSize bytes buffers") {
+            describe("direct buffers") {
+
+                val bb1 = ByteBuffer.allocateDirect(singleBufferSize)
+                val bb2 = ByteBuffer.allocateDirect(singleBufferSize)
+
+                it("should be heated up") {
+                    measureAverageCopyTimeInNanos(bb1, bb2)
+                }
+
+                it("should work fast") {
+                    measureAndPrintAverageCopyTime("direct buffers of $singleBufferSize bytes", bb1, bb2)
+                }
+            }
+
+            describe("on-heap buffers") {
+
+                val bb1 = ByteBuffer.allocate(singleBufferSize)
+                val bb2 = ByteBuffer.allocate(singleBufferSize)
+
+                it("should be heated up") {
+                    measureAverageCopyTimeInNanos(bb1, bb2)
+                }
+
+                it("should work fast") {
+                    measureAndPrintAverageCopyTime("onheap buffers of $singleBufferSize bytes", bb1, bb2)
+                }
+            }
+        }
+    }
+
+})
\ No newline at end of file
diff --git a/hv-collector-utils/pom.xml b/hv-collector-utils/pom.xml
new file mode 100644
index 00000000..ddd0a50c
--- /dev/null
+++ b/hv-collector-utils/pom.xml
@@ -0,0 +1,106 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ ============LICENSE_START=======================================================
+  ~ dcaegen2-collectors-veshv
+  ~ ================================================================================
+  ~ Copyright (C) 2018 NOKIA Intellectual Property. 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>
+
+  <licenses>
+    <license>
+      <name>The Apache Software License, Version 2.0</name>
+      <url>http://www.apache.org/licenses/LICENSE-2.0.txt</url>
+    </license>
+  </licenses>
+
+  <properties>
+    <skipAnalysis>false</skipAnalysis>
+  </properties>
+
+  <parent>
+    <groupId>org.onap.dcaegen2.collectors.veshv</groupId>
+    <artifactId>ves-hv-collector</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+    <relativePath>..</relativePath>
+  </parent>
+
+  <artifactId>hv-collector-utils</artifactId>
+  <description>VES HighVolume Collector :: Utilities</description>
+
+
+  <build>
+    <plugins>
+      <plugin>
+        <artifactId>kotlin-maven-plugin</artifactId>
+        <groupId>org.jetbrains.kotlin</groupId>
+      </plugin>
+      <plugin>
+        <artifactId>maven-surefire-plugin</artifactId>
+        <groupId>org.apache.maven.plugins</groupId>
+      </plugin>
+    </plugins>
+  </build>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.jetbrains.kotlin</groupId>
+      <artifactId>kotlin-reflect</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.slf4j</groupId>
+      <artifactId>slf4j-api</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>com.nhaarman</groupId>
+      <artifactId>mockito-kotlin</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-core</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.assertj</groupId>
+      <artifactId>assertj-core</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.jetbrains.kotlin</groupId>
+      <artifactId>kotlin-test</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.jetbrains.spek</groupId>
+      <artifactId>spek-api</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.jetbrains.spek</groupId>
+      <artifactId>spek-junit-platform-engine</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>io.projectreactor</groupId>
+      <artifactId>reactor-test</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>ch.qos.logback</groupId>
+      <artifactId>logback-classic</artifactId>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+
+
+</project>
\ No newline at end of file
diff --git a/hv-collector-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/utils/logging/Logger.kt b/hv-collector-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/utils/logging/Logger.kt
new file mode 100644
index 00000000..ab9fc2e6
--- /dev/null
+++ b/hv-collector-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/utils/logging/Logger.kt
@@ -0,0 +1,99 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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.dcae.collectors.veshv.utils.logging
+
+import org.slf4j.LoggerFactory
+import kotlin.reflect.KClass
+
+class Logger(val logger: org.slf4j.Logger) {
+    constructor(clazz: KClass<out Any>) : this(LoggerFactory.getLogger(clazz.java))
+
+    //
+    // DEBUG
+    //
+
+    fun debug(message: String) {
+        logger.debug(message)
+    }
+
+    fun debug(message: String, ex: Exception) {
+        logger.debug(message, ex)
+    }
+
+    fun debug(messageProvider: () -> String) {
+        if (logger.isDebugEnabled) {
+            logger.debug(messageProvider())
+        }
+    }
+
+    fun debug(ex: Exception, messageProvider: () -> String) {
+        if (logger.isDebugEnabled) {
+            logger.debug(messageProvider(), ex)
+        }
+    }
+
+    //
+    // INFO
+    //
+    fun info(message: String) {
+        logger.info(message)
+    }
+
+    fun info(message: String, ex: Exception) {
+        logger.info(message, ex)
+    }
+
+    fun info(messageProvider: () -> String) {
+        if (logger.isInfoEnabled) {
+            logger.info(messageProvider())
+        }
+    }
+
+    fun info(ex: Exception, messageProvider: () -> String) {
+        if (logger.isInfoEnabled) {
+            logger.info(messageProvider(), ex)
+        }
+    }
+
+
+    //
+    // WARN
+    //
+
+    fun warn(message: String) {
+        logger.warn(message)
+    }
+
+    fun warn(message: String, ex: Exception) {
+        logger.warn(message, ex)
+    }
+
+    fun warn(messageProvider: () -> String) {
+        if (logger.isWarnEnabled) {
+            logger.warn(messageProvider())
+        }
+    }
+
+    fun warn(ex: Exception, messageProvider: () -> String) {
+        if (logger.isWarnEnabled) {
+            logger.warn(messageProvider(), ex)
+        }
+    }
+}
\ No newline at end of file
diff --git a/hv-collector-utils/src/test/kotlin/org/onap/dcae/collectors/veshv/utils/logging/LoggerTest.kt b/hv-collector-utils/src/test/kotlin/org/onap/dcae/collectors/veshv/utils/logging/LoggerTest.kt
new file mode 100644
index 00000000..9e9af1fa
--- /dev/null
+++ b/hv-collector-utils/src/test/kotlin/org/onap/dcae/collectors/veshv/utils/logging/LoggerTest.kt
@@ -0,0 +1,186 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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.dcae.collectors.veshv.utils.logging
+
+import com.nhaarman.mockito_kotlin.mock
+import com.nhaarman.mockito_kotlin.verify
+import com.nhaarman.mockito_kotlin.verifyNoMoreInteractions
+import com.nhaarman.mockito_kotlin.whenever
+import org.jetbrains.spek.api.Spek
+import org.jetbrains.spek.api.dsl.describe
+import org.jetbrains.spek.api.dsl.it
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since May 2018
+ */
+object LoggerTest : Spek({
+
+    lateinit var slf4jLogger: org.slf4j.Logger
+    lateinit var cut: Logger
+
+    beforeEachTest {
+        slf4jLogger = mock()
+        cut = Logger(slf4jLogger)
+    }
+
+    afterEachTest {
+        verifyNoMoreInteractions(slf4jLogger)
+    }
+
+    describe("Thin Kotlin logging facade for Slf4j") {
+        val message = "sample message"
+        val exception = Exception("fail")
+
+        describe("debug levels") {
+            it("should log message") {
+                cut.debug(message)
+                verify(slf4jLogger).debug(message)
+            }
+
+            it("should log message with exception") {
+                cut.debug(message, exception)
+                verify(slf4jLogger).debug(message, exception)
+            }
+
+            describe("lazy logging message") {
+
+                it("should log when debug is ON") {
+                    whenever(slf4jLogger.isDebugEnabled).thenReturn(true)
+                    cut.debug { message }
+                    verify(slf4jLogger).isDebugEnabled
+                    verify(slf4jLogger).debug(message)
+                }
+
+                it("should not log when debug is OFF") {
+                    whenever(slf4jLogger.isDebugEnabled).thenReturn(false)
+                    cut.debug { message }
+                    verify(slf4jLogger).isDebugEnabled
+                }
+            }
+
+            describe("lazy logging message with exception") {
+
+                it("should log when debug is ON") {
+                    whenever(slf4jLogger.isDebugEnabled).thenReturn(true)
+                    cut.debug(exception) { message }
+                    verify(slf4jLogger).isDebugEnabled
+                    verify(slf4jLogger).debug(message, exception)
+                }
+
+                it("should not log when debug is OFF") {
+                    whenever(slf4jLogger.isDebugEnabled).thenReturn(false)
+                    cut.debug(exception) { message }
+                    verify(slf4jLogger).isDebugEnabled
+                }
+            }
+        }
+
+        describe("info levels") {
+            it("should log message") {
+                cut.info(message)
+                verify(slf4jLogger).info(message)
+            }
+
+            it("should log message with exception") {
+                cut.info(message, exception)
+                verify(slf4jLogger).info(message, exception)
+            }
+
+            describe("lazy logging message") {
+
+                it("should log when debug is ON") {
+                    whenever(slf4jLogger.isInfoEnabled).thenReturn(true)
+                    cut.info { message }
+                    verify(slf4jLogger).isInfoEnabled
+                    verify(slf4jLogger).info(message)
+                }
+
+                it("should not log when debug is OFF") {
+                    whenever(slf4jLogger.isInfoEnabled).thenReturn(false)
+                    cut.info { message }
+                    verify(slf4jLogger).isInfoEnabled
+                }
+            }
+
+            describe("lazy logging message with exception") {
+
+                it("should log when debug is ON") {
+                    whenever(slf4jLogger.isInfoEnabled).thenReturn(true)
+                    cut.info(exception) { message }
+                    verify(slf4jLogger).isInfoEnabled
+                    verify(slf4jLogger).info(message, exception)
+                }
+
+                it("should not log when debug is OFF") {
+                    whenever(slf4jLogger.isInfoEnabled).thenReturn(false)
+                    cut.info(exception) { message }
+                    verify(slf4jLogger).isInfoEnabled
+                }
+            }
+        }
+
+        describe("warning levels") {
+            it("should log message") {
+                cut.warn(message)
+                verify(slf4jLogger).warn(message)
+            }
+
+            it("should log message with exception") {
+                cut.warn(message, exception)
+                verify(slf4jLogger).warn(message, exception)
+            }
+
+            describe("lazy logging message") {
+
+                it("should log when debug is ON") {
+                    whenever(slf4jLogger.isWarnEnabled).thenReturn(true)
+                    cut.warn { message }
+                    verify(slf4jLogger).isWarnEnabled
+                    verify(slf4jLogger).warn(message)
+                }
+
+                it("should not log when debug is OFF") {
+                    whenever(slf4jLogger.isWarnEnabled).thenReturn(false)
+                    cut.warn { message }
+                    verify(slf4jLogger).isWarnEnabled
+                }
+            }
+
+            describe("lazy logging message with exception") {
+
+                it("should log when debug is ON") {
+                    whenever(slf4jLogger.isWarnEnabled).thenReturn(true)
+                    cut.warn(exception) { message }
+                    verify(slf4jLogger).isWarnEnabled
+                    verify(slf4jLogger).warn(message, exception)
+                }
+
+                it("should not log when debug is OFF") {
+                    whenever(slf4jLogger.isWarnEnabled).thenReturn(false)
+                    cut.warn(exception) { message }
+                    verify(slf4jLogger).isWarnEnabled
+                }
+            }
+        }
+
+
+    }
+})
diff --git a/pom.xml b/pom.xml
new file mode 100644
index 00000000..db46a4f9
--- /dev/null
+++ b/pom.xml
@@ -0,0 +1,460 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ ============LICENSE_START=======================================================
+  ~ dcaegen2-collectors-veshv
+  ~ ================================================================================
+  ~ Copyright (C) 2018 NOKIA Intellectual Property. 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>
+
+  <licenses>
+    <license>
+      <name>The Apache Software License, Version 2.0</name>
+      <url>http://www.apache.org/licenses/LICENSE-2.0.txt</url>
+    </license>
+  </licenses>
+
+  <groupId>org.onap.dcaegen2.collectors.veshv</groupId>
+  <artifactId>ves-hv-collector</artifactId>
+  <version>1.0.0-SNAPSHOT</version>
+  <name>dcaegen2-collectors-veshv</name>
+  <description>VES HighVolume Collector</description>
+  <packaging>pom</packaging>
+
+  <modules>
+    <module>hv-collector-core</module>
+    <module>hv-collector-main</module>
+    <module>hv-collector-ct</module>
+    <module>protobuf</module>
+    <module>hv-collector-utils</module>
+    <module>hv-collector-coverage</module>
+    <module>hv-collector-analysis</module>
+  </modules>
+
+  <properties>
+    <kotlin.version>1.2.41</kotlin.version>
+    <protobuf.version>3.5.1</protobuf.version>
+    <protoc-jar-maven-plugin.version>3.5.1.1</protoc-jar-maven-plugin.version>
+    <protobuf-generated-files.directory>${project.build.directory}/generated-sources/proto/main/java/
+    </protobuf-generated-files.directory>
+    <maven-compiler-plugin.version>3.7.0</maven-compiler-plugin.version>
+    <build-helper-maven-plugin.version>1.7</build-helper-maven-plugin.version>
+
+    <junit-platform.version>1.2.0-RC1</junit-platform.version>
+    <junit-jupiter.version>5.2.0-RC1</junit-jupiter.version>
+    <spek.version>1.1.5</spek.version>
+    <maven-surefire-plugin.version>2.21.0</maven-surefire-plugin.version>
+
+    <failIfMissingUnitTests>true</failIfMissingUnitTests>
+    <failIfMissingComponentTests>false</failIfMissingComponentTests>
+    <skipAnalysis>true</skipAnalysis>
+  </properties>
+
+
+  <build>
+    <sourceDirectory>${project.basedir}/src/main/kotlin</sourceDirectory>
+    <testSourceDirectory>${project.basedir}/src/test/kotlin</testSourceDirectory>
+    <resources>
+      <resource>
+        <directory>${project.basedir}/src/main/resources</directory>
+      </resource>
+    </resources>
+    <pluginManagement>
+      <plugins>
+        <plugin>
+          <groupId>org.apache.maven.plugins</groupId>
+          <artifactId>maven-resources-plugin</artifactId>
+          <version>3.1.0</version>
+          <configuration>
+            <encoding>UTF-8</encoding>
+          </configuration>
+        </plugin>
+        <plugin>
+          <artifactId>kotlin-maven-plugin</artifactId>
+          <groupId>org.jetbrains.kotlin</groupId>
+          <version>${kotlin.version}</version>
+          <configuration>
+            <jvmTarget>1.8</jvmTarget>
+          </configuration>
+          <executions>
+            <execution>
+              <id>compile</id>
+              <goals>
+                <goal>compile</goal>
+              </goals>
+              <configuration>
+                <sourceDirs>
+                  <source>${project.build.sourceDirectory}</source>
+                  <source>${project.build.directory}/generated-sources/annotations</source>
+                </sourceDirs>
+              </configuration>
+            </execution>
+            <execution>
+              <id>test-compile</id>
+              <goals>
+                <goal>test-compile</goal>
+              </goals>
+              <configuration>
+                <sourceDirs>
+                  <source>${project.build.testSourceDirectory}</source>
+                </sourceDirs>
+              </configuration>
+            </execution>
+          </executions>
+        </plugin>
+        <plugin>
+          <groupId>com.github.os72</groupId>
+          <artifactId>protoc-jar-maven-plugin</artifactId>
+          <version>${protoc-jar-maven-plugin.version}</version>
+        </plugin>
+        <plugin>
+          <groupId>org.codehaus.mojo</groupId>
+          <artifactId>build-helper-maven-plugin</artifactId>
+          <version>${build-helper-maven-plugin.version}</version>
+          <executions>
+            <execution>
+              <id>add-source</id>
+              <phase>generate-sources</phase>
+              <goals>
+                <goal>add-source</goal>
+              </goals>
+              <configuration>
+                <sources>
+                  <source>${protobuf-generated-files.directory}</source>
+                </sources>
+              </configuration>
+            </execution>
+          </executions>
+        </plugin>
+        <plugin>
+          <groupId>org.apache.maven.plugins</groupId>
+          <artifactId>maven-compiler-plugin</artifactId>
+          <version>${maven-compiler-plugin.version}</version>
+          <configuration>
+            <source>8</source>
+            <target>8</target>
+            <encoding>UTF-8</encoding>
+            <showWarnings>true</showWarnings>
+            <showDeprecation>true</showDeprecation>
+            <failOnWarning>false</failOnWarning>
+          </configuration>
+          <dependencies>
+            <dependency>
+              <groupId>org.ow2.asm</groupId>
+              <artifactId>asm</artifactId>
+              <version>6.1.1</version> <!-- Use newer version of ASM -->
+            </dependency>
+          </dependencies>
+        </plugin>
+        <!--
+         Due to a memory leak in Surefire 2.20 and issues running on Java 9, the junit-platform-surefire-provider
+         currently only works with Surefire 2.19.1.
+         For updates see https://junit.org/junit5/docs/current/user-guide/#running-tests-build-maven
+         -->
+        <plugin>
+          <artifactId>maven-surefire-plugin</artifactId>
+          <groupId>org.apache.maven.plugins</groupId>
+          <version>${maven-surefire-plugin.version}</version>
+          <executions>
+            <execution>
+              <id>default-test</id>
+              <goals>
+                <goal>test</goal>
+              </goals>
+              <configuration>
+                <failIfNoTests>${failIfMissingUnitTests}</failIfNoTests>
+                <forkCount>1</forkCount>
+                <includes>
+                  <include>**/*Test.*</include>
+                </includes>
+                <!--<argLine>&#45;&#45;add-modules org.junit.jupiter.api,reactor.test,assertj.core</argLine>-->
+              </configuration>
+            </execution>
+            <execution>
+              <id>component-tests</id>
+              <phase>verify</phase>
+              <goals>
+                <goal>test</goal>
+              </goals>
+              <configuration>
+                <failIfNoTests>${failIfMissingComponentTests}</failIfNoTests>
+                <forkCount>1</forkCount>
+                <includes>
+                  <include>**/*Specification.*</include>
+                </includes>
+              </configuration>
+            </execution>
+          </executions>
+          <dependencies>
+            <dependency>
+              <groupId>org.apache.commons</groupId>
+              <artifactId>commons-lang3</artifactId>
+              <version>3.7</version>
+            </dependency>
+            <dependency>
+              <groupId>org.junit.platform</groupId>
+              <artifactId>junit-platform-surefire-provider</artifactId>
+              <version>${junit-platform.version}</version>
+              <scope>runtime</scope>
+            </dependency>
+            <dependency>
+              <groupId>org.jetbrains.spek</groupId>
+              <artifactId>spek-junit-platform-engine</artifactId>
+              <version>${spek.version}</version>
+              <scope>runtime</scope>
+            </dependency>
+          </dependencies>
+        </plugin>
+      </plugins>
+    </pluginManagement>
+  </build>
+
+  <profiles>
+    <profile>
+      <id>analysis</id>
+      <activation>
+        <activeByDefault>true</activeByDefault>
+      </activation>
+      <build>
+        <plugins>
+          <plugin>
+            <groupId>org.jacoco</groupId>
+            <artifactId>jacoco-maven-plugin</artifactId>
+            <version>0.8.1</version>
+            <executions>
+              <execution>
+                <id>default-prepare-agent</id>
+                <goals>
+                  <goal>prepare-agent</goal>
+                </goals>
+              </execution>
+              <execution>
+                <id>default-prepare-agent-integration</id>
+                <goals>
+                  <goal>prepare-agent-integration</goal>
+                </goals>
+              </execution>
+            </executions>
+          </plugin>
+
+          <plugin>
+            <groupId>org.apache.maven.plugins</groupId>
+            <artifactId>maven-antrun-plugin</artifactId>
+            <version>1.8</version>
+            <executions>
+              <execution>
+                <!-- This can be run separately with mvn antrun:run@detekt -->
+                <id>detekt</id>
+                <phase>verify</phase>
+                <configuration>
+                  <target name="detekt" unless="${skipAnalysis}">
+                    <java taskname="detekt" dir="${basedir}"
+                      fork="true"
+                      failonerror="true"
+                      classname="io.gitlab.arturbosch.detekt.cli.Main"
+                      classpathref="maven.plugin.classpath">
+                      <arg value="--input"/>
+                      <arg value="${basedir}/src/main/kotlin"/>
+                      <arg value="--config-resource"/>
+                      <arg value="onap-detekt-config.yml"/>
+                      <arg value="--filters"/>
+                      <arg value=".*/target/.*,.*/resources/.*"/>
+                      <arg value="--output"/>
+                      <arg value="${basedir}/target/reports"/>
+                      <arg value="--output-name"/>
+                      <arg value="detekt-report"/>
+                      <arg value="--baseline"/>
+                      <arg value="${basedir}/target/reports/baseline.xml"/>
+                    </java>
+                  </target>
+                </configuration>
+                <goals>
+                  <goal>run</goal>
+                </goals>
+              </execution>
+            </executions>
+            <dependencies>
+              <dependency>
+                <groupId>io.gitlab.arturbosch.detekt</groupId>
+                <artifactId>detekt-cli</artifactId>
+                <version>1.0.0.RC7</version>
+              </dependency>
+              <dependency>
+                <groupId>${project.groupId}</groupId>
+                <artifactId>hv-collector-analysis</artifactId>
+                <version>${project.version}</version>
+              </dependency>
+            </dependencies>
+          </plugin>
+        </plugins>
+      </build>
+    </profile>
+  </profiles>
+
+  <reporting>
+    <plugins>
+      <plugin>
+        <groupId>org.jacoco</groupId>
+        <artifactId>jacoco-maven-plugin</artifactId>
+        <reportSets>
+          <reportSet>
+            <reports>
+              <report>report</report>
+            </reports>
+          </reportSet>
+        </reportSets>
+      </plugin>
+    </plugins>
+  </reporting>
+
+  <pluginRepositories>
+    <pluginRepository>
+      <id>arturbosch-code-analysis</id>
+      <name>arturbosch-code-analysis (for detekt)</name>
+      <url>https://dl.bintray.com/arturbosch/code-analysis/</url>
+      <layout>default</layout>
+      <releases>
+        <enabled>true</enabled>
+        <updatePolicy>never</updatePolicy>
+      </releases>
+      <snapshots>
+        <enabled>false</enabled>
+        <updatePolicy>never</updatePolicy>
+      </snapshots>
+    </pluginRepository>
+  </pluginRepositories>
+
+  <dependencyManagement>
+    <dependencies>
+      <dependency>
+        <groupId>com.google.guava</groupId>
+        <artifactId>guava</artifactId>
+        <version>25.0-jre</version>
+        <exclusions>
+          <exclusion>
+            <groupId>com.google.code.findbugs</groupId>
+            <artifactId>jsr305</artifactId>
+          </exclusion>
+        </exclusions>
+      </dependency>
+      <dependency>
+        <groupId>org.jetbrains.kotlin</groupId>
+        <artifactId>kotlin-stdlib-jdk8</artifactId>
+        <version>${kotlin.version}</version>
+      </dependency>
+      <dependency>
+        <groupId>org.jetbrains.kotlin</groupId>
+        <artifactId>kotlin-reflect</artifactId>
+        <version>${kotlin.version}</version>
+      </dependency>
+      <dependency>
+        <groupId>ch.qos.logback</groupId>
+        <artifactId>logback-classic</artifactId>
+        <version>1.3.0-alpha4</version>
+        <scope>runtime</scope>
+      </dependency>
+      <dependency>
+        <groupId>org.slf4j</groupId>
+        <artifactId>slf4j-api</artifactId>
+        <version>1.8.0-beta1</version>
+      </dependency>
+      <dependency>
+        <groupId>io.projectreactor</groupId>
+        <artifactId>reactor-bom</artifactId>
+        <version>Bismuth-SR8</version>
+        <type>pom</type>
+        <scope>import</scope>
+      </dependency>
+      <dependency>
+        <groupId>com.google.protobuf</groupId>
+        <artifactId>protobuf-java</artifactId>
+        <version>${protobuf.version}</version>
+      </dependency>
+      <dependency>
+        <groupId>commons-cli</groupId>
+        <artifactId>commons-cli</artifactId>
+        <version>1.4</version>
+      </dependency>
+      <dependency>
+        <groupId>javax.json</groupId>
+        <artifactId>javax.json-api</artifactId>
+        <version>1.1.2</version>
+      </dependency>
+      <dependency>
+        <groupId>org.glassfish</groupId>
+        <artifactId>javax.json</artifactId>
+        <version>1.1.2</version>
+      </dependency>
+
+      <!-- Test dependencies -->
+
+      <dependency>
+        <groupId>org.jetbrains.spek</groupId>
+        <artifactId>spek-api</artifactId>
+        <version>${spek.version}</version>
+        <scope>test</scope>
+      </dependency>
+      <dependency>
+        <groupId>org.jetbrains.spek</groupId>
+        <artifactId>spek-junit-platform-engine</artifactId>
+        <version>${spek.version}</version>
+        <scope>test</scope>
+      </dependency>
+
+      <dependency>
+        <groupId>org.assertj</groupId>
+        <artifactId>assertj-core</artifactId>
+        <version>3.9.1</version>
+        <scope>test</scope>
+      </dependency>
+      <dependency>
+        <groupId>com.nhaarman</groupId>
+        <artifactId>mockito-kotlin</artifactId>
+        <version>1.5.0</version>
+        <scope>test</scope>
+        <exclusions>
+          <exclusion>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+          </exclusion>
+        </exclusions>
+      </dependency>
+      <dependency>
+        <groupId>org.mockito</groupId>
+        <artifactId>mockito-core</artifactId>
+        <version>2.18.3</version>
+        <scope>test</scope>
+      </dependency>
+      <dependency>
+        <groupId>org.jetbrains.kotlin</groupId>
+        <artifactId>kotlin-test</artifactId>
+        <version>${kotlin.version}</version>
+        <scope>test</scope>
+      </dependency>
+      <dependency>
+        <groupId>io.projectreactor</groupId>
+        <artifactId>reactor-test</artifactId>
+        <version>3.1.7.RELEASE</version>
+        <scope>test</scope>
+      </dependency>
+    </dependencies>
+  </dependencyManagement>
+</project>
+
+
diff --git a/protobuf/pom.xml b/protobuf/pom.xml
new file mode 100644
index 00000000..08f06330
--- /dev/null
+++ b/protobuf/pom.xml
@@ -0,0 +1,96 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ ============LICENSE_START=======================================================
+  ~ dcaegen2-collectors-veshv
+  ~ ================================================================================
+  ~ Copyright (C) 2018 NOKIA Intellectual Property. 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>
+
+    <licenses>
+        <license>
+            <name>The Apache Software License, Version 2.0</name>
+            <url>http://www.apache.org/licenses/LICENSE-2.0.txt</url>
+        </license>
+    </licenses>
+
+    <parent>
+        <groupId>org.onap.dcaegen2.collectors.veshv</groupId>
+        <artifactId>ves-hv-collector</artifactId>
+        <version>1.0.0-SNAPSHOT</version>
+        <relativePath>..</relativePath>
+    </parent>
+
+    <artifactId>protobuf</artifactId>
+    <description>Google Protocol Buffers schemas for VES-HV-Collector</description>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.codehaus.mojo</groupId>
+                <artifactId>build-helper-maven-plugin</artifactId>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+            </plugin>
+            <plugin>
+                <groupId>com.github.os72</groupId>
+                <artifactId>protoc-jar-maven-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <phase>generate-sources</phase>
+                        <goals>
+                            <goal>run</goal>
+                        </goals>
+                        <configuration>
+                            <protocArtifact>com.google.protobuf:protoc:${protobuf.version}</protocArtifact>
+                            <inputDirectories>
+                                <include>${project.basedir}/src/main/proto</include>
+                            </inputDirectories>
+                            <outputTargets>
+                                <outputTarget>
+                                    <type>java</type>
+                                    <addSources>none</addSources>
+                                    <outputDirectory>${protobuf-generated-files.directory}</outputDirectory>
+                                </outputTarget>
+                            </outputTargets>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
+            <plugin>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <groupId>org.apache.maven.plugins</groupId>
+                <configuration>
+                    <skipTests>true</skipTests>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
+
+    <dependencies>
+        <dependency>
+            <groupId>com.google.protobuf</groupId>
+            <artifactId>protobuf-java</artifactId>
+        </dependency>
+    </dependencies>
+
+</project>
\ No newline at end of file
diff --git a/protobuf/src/main/proto/HVRanMeasFields-v5.proto b/protobuf/src/main/proto/HVRanMeasFields-v5.proto
new file mode 100644
index 00000000..cc375211
--- /dev/null
+++ b/protobuf/src/main/proto/HVRanMeasFields-v5.proto
@@ -0,0 +1,54 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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=========================================================
+ */
+syntax = "proto3";
+package org.onap.ves;
+
+// Definition for RTPM
+
+message HVRanMeasFields {
+  message HVRanMeasPayload {
+    message PMObject {
+      message HVRanMeas {
+        uint32 measurement_id = 1;
+        repeated uint32 counter_subid = 2;
+        repeated sint64 counter_value = 3;
+        repeated uint32 missing_counter_subid = 4;
+        bool suspectFlagIncomplete = 5; // (some is data missing due to internal error)
+        bool suspectFlagOutOfSync = 6; // (source time not aligned)
+      }
+
+      string uri = 1; // monitored object URI
+      repeated HVRanMeas hvRanMeas = 2; // performance counters grouped by measurement types
+    }
+    repeated PMObject pmObject = 1;
+  }
+
+  message AdditionalField {
+    string name = 1;
+    string value = 2;
+  }
+
+  string hvRanMeasFieldsVersion = 1; // version of HVRanMeasFields message
+  uint32 period_ms = 2; // period configured for reporting the data in milliseconds
+  string timezone = 3; // timezone of Network Function sending the data
+  string pmDictionaryVsn = 4; // vendor name + schema version E.g. NOKIA_LN7.0, uniquely identify the relevant PM dictionary
+  HVRanMeasPayload hvRanMeasPayload = 5; // objects being monitored
+  repeated AdditionalField additionalFields = 6; // array of name-value pairs if needed
+}
\ No newline at end of file
diff --git a/protobuf/src/main/proto/VesEvent-v5.proto b/protobuf/src/main/proto/VesEvent-v5.proto
new file mode 100644
index 00000000..4ba6f6a5
--- /dev/null
+++ b/protobuf/src/main/proto/VesEvent-v5.proto
@@ -0,0 +1,86 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property. 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=========================================================
+ */
+syntax = "proto3";
+package org.onap.ves;
+
+message VesEvent {
+
+  // VES CommonEventHeader adapted to GPB (Google Protocol Buffers)
+  // Source: https://git.opnfv.org/ves/tree/tests/docs/ves_data_model.json
+  //         2017-05-13 Align with VES 5.0 schema.
+  //         blob: ca948ff67e8a2de4e2a47cffc4d4d2893170ab76
+
+  message CommonEventHeader {
+    string version = 1; // required, "version of the event header"
+    enum Domain {
+      FAULT = 0;
+      HEARTBEAT = 1;
+      MEASUREMENTS_FOR_VF_SCALING = 2;
+      MOBILE_FLOW = 3;
+      SIP_SIGNALING = 4;
+      STATE_CHANGE = 5;
+      SYSLOG = 6;
+      THRESHOLD_CROSSING_ALERT = 7;
+      VOICE_QUALITY = 8;
+      OTHER = 9;
+      HVRANMEAS = 10;
+    }
+    Domain domain = 2; // required, "the eventing domain associated with the event" [map to string]
+
+    uint32 sequence = 3; // required, "ordering of events communicated by an event source instance or 0 if not needed"
+
+    enum Priority {
+      HIGH = 0;
+      MEDIUM = 1;
+      NORMAL = 2;
+      LOW = 3;
+    }
+    Priority priority = 4; // required, "processing priority"
+
+    string eventId = 5; // required, "event key that is unique to the event source"
+    string eventName = 6; // required, "unique event name"
+    string eventType = 7; // "for example - applicationVnf, guestOS, hostOS, platform"
+
+    uint64 lastEpochMicrosec = 8; // required, "the latest unix time aka epoch time associated with the event from any component--as microseconds elapsed since 1 Jan 1970 not including leap seconds"
+    uint64 startEpochMicrosec = 9; // required, "the earliest unix time aka epoch time associated with the event from any component--as microseconds elapsed since 1 Jan 1970 not including leap seconds"
+
+    string nfNamingCode = 10; // "4 character network function type, aligned with vnf naming standards"
+    string nfcNamingCode = 11; // "3 character network function component type, aligned with vfc naming standards"
+
+    string reportingEntityId = 12; // "UUID identifying the entity reporting the event, for example an OAM VM; must be populated by the ATT enrichment process"
+    bytes reportingEntityName = 13; // required, "name of the entity reporting the event, for example, an EMS name; may be the same as sourceName"
+    bytes sourceId = 14; // "UUID identifying the entity experiencing the event issue; must be populated by the ATT enrichment process"
+    string sourceName = 15; // required, "name of the entity experiencing the event issue"
+
+    reserved "InternalHeaderFields"; // "enrichment fields for internal VES Event Listener service use only, not supplied by event sources"
+    reserved 100;
+  }
+
+  CommonEventHeader commonEventHeader = 1;
+
+  oneof eventFields // required, payload, each high-volume domain has its specific GPB schema
+  {
+    bytes hvRanMeasFields = 2; // if domain==HVRANMEAS, GPB schema: HVRanMeasFields.proto
+  }
+}
+
+message VesEventList {
+  repeated VesEvent vesEvent = 1;
+}