Redefine Routing

As all needed information to route messege is contained inside of
KafkaSink message, we can simply put this object as part of single Route.

Change-Id: I2e7df2e0193eb2af5283980d4d5c8df03ac94df9
Issue-ID: DCAEGEN2-1347
Signed-off-by: Filip Krzywka <filip.krzywka@nokia.com>

1 files changed, 18 insertions, 15 deletions

diff --git a/sources/hv-collector-main/src/test/kotlin/org/onap/dcae/collectors/veshv/main/MicrometerMetricsTest.kt b/sources/hv-collector-main/src/test/kotlin/org/onap/dcae/collectors/veshv/main/MicrometerMetricsTest.kt
index e452a5f4..f260f158 100644
--- a/sources/hv-collector-main/src/test/kotlin/org/onap/dcae/collectors/veshv/main/MicrometerMetricsTest.kt
+++ b/sources/hv-collector-main/src/test/kotlin/org/onap/dcae/collectors/veshv/main/MicrometerMetricsTest.kt
@@ -19,6 +19,7 @@
  */
 package org.onap.dcae.collectors.veshv.main
 
+import arrow.core.Option
 import arrow.core.Try
 import io.micrometer.core.instrument.Counter
 import io.micrometer.core.instrument.Gauge
@@ -44,6 +45,7 @@ import org.onap.dcae.collectors.veshv.domain.VesMessage
 import org.onap.dcae.collectors.veshv.tests.utils.emptyWireProtocolFrame
 import org.onap.dcae.collectors.veshv.tests.utils.vesEvent
 import org.onap.dcae.collectors.veshv.tests.utils.wireProtocolFrame
+import org.onap.ves.VesEventOuterClass
 import java.time.Instant
 import java.time.temporal.Temporal
 import java.util.concurrent.TimeUnit
@@ -383,23 +385,24 @@ object MicrometerMetricsTest : Spek({
 })
 
 fun routedMessage(topic: String, partition: Int = 0) =
-        vesEvent().let { evt ->
-            RoutedMessage(topic, partition,
-                    VesMessage(evt.commonEventHeader, wireProtocolFrame(evt)))
-        }
+        vesEvent().run { toRoutedMessage(topic, partition) }
 
 fun routedMessageReceivedAt(topic: String, receivedAt: Temporal, partition: Int = 0) =
-        vesEvent().let { evt ->
-            RoutedMessage(topic, partition,
-                    VesMessage(evt.commonEventHeader, wireProtocolFrame(evt).copy(receivedAt = receivedAt)))
-        }
+        vesEvent().run { toRoutedMessage(topic, partition, receivedAt) }
 
 fun routedMessageSentAt(topic: String, sentAt: Instant, partition: Int = 0) =
-        vesEvent().let { evt ->
-            val builder = evt.toBuilder()
+        vesEvent().run {
+            val builder = toBuilder()
             builder.commonEventHeaderBuilder.lastEpochMicrosec = sentAt.epochSecond * 1000000 + sentAt.nano / 1000
-            builder.build()
-        }.let { evt ->
-            RoutedMessage(topic, partition,
-                    VesMessage(evt.commonEventHeader, wireProtocolFrame(evt)))
-        }
\ No newline at end of file
+            builder.build().toRoutedMessage(topic, partition)
+        }
+
+private fun VesEventOuterClass.VesEvent.toRoutedMessage(topic: String,
+                                                        partition: Int,
+                                                        receivedAt: Temporal = Instant.now()) =
+        RoutedMessage(
+                VesMessage(this.commonEventHeader, wireProtocolFrame(this).copy(receivedAt = receivedAt)),
+                topic,
+                Option.just(partition)
+        )
+