Implement blocking consul calls

[dcaegen2/collectors/hv-ves.git] / hv-collector-core / src / main / kotlin / org / onap / dcae / collectors / veshv / impl / adapters / AdapterFactory.kt

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
index cdfcfd3..2a8a396 100644 (file)
--- 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
@@ -2,7 +2,7 @@
  * ============LICENSE_START=======================================================
  * dcaegen2-collectors-veshv
  * ================================================================================
- * Copyright (C) 2018 NOKIA Intellectual Property. All rights reserved.
+ * Copyright (C) 2018 NOKIA
  * ================================================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -19,21 +19,11 @@
  */
 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 org.onap.dcae.collectors.veshv.impl.adapters.kafka.KafkaSinkProvider
 import reactor.ipc.netty.http.client.HttpClient
-import reactor.kafka.sender.KafkaSender
-import reactor.kafka.sender.SenderOptions
-import java.nio.ByteBuffer
+import java.time.Duration
 
 /**
  * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
@@ -41,25 +31,11 @@ import java.nio.ByteBuffer
  */
 object AdapterFactory {
     fun kafkaSink(): SinkProvider = KafkaSinkProvider()
+    fun loggingSink(): SinkProvider = LoggingSinkProvider()
 
-    fun staticConfigurationProvider(config: CollectorConfiguration) =
-            object : ConfigurationProvider {
-                override fun invoke() = Flux.just(config)
-            }
+    fun consulConfigurationProvider(url: String, firstRequestDelay: Duration): ConfigurationProvider =
+            ConsulConfigurationProvider(url, httpAdapter(), firstRequestDelay)
 
-    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())
 }