fixes #96: New procedure to receive data from the stream

Author: conker84

common/src/main/kotlin/streams/utils/StreamsUtils.kt

@@ -12,7 +12,10 @@ object StreamsUtils {
         return try {
             action()
         } catch (e: Throwable) {
-            when (e::class.java) {
+            if (toIgnore.isEmpty()) {
+                return null
+            }
+            return when (e::class.java) {
                 in toIgnore -> null
                 else -> throw e
             }

consumer/src/main/kotlin/streams/StreamsEventSink.kt

@@ -1,33 +1,51 @@
 package streams
 
-import org.neo4j.kernel.AvailabilityGuard
 import org.neo4j.kernel.configuration.Config
-import org.neo4j.kernel.internal.GraphDatabaseAPI
+import org.neo4j.logging.Log
 
 abstract class StreamsEventSink(private val config: Config,
-                                private val db: GraphDatabaseAPI): AvailabilityGuard.AvailabilityListener {
+                                private val queryExecution: StreamsEventSinkQueryExecution,
+                                private val streamsTopicService: StreamsTopicService,
+                                private val log: Log) {
 
-    abstract var streamsTopicService: StreamsTopicService?
+    abstract fun stop()
+
+    abstract fun start()
+
+    abstract fun getEventConsumerFactory(): StreamsEventConsumerFactory
+
+    abstract fun getEventSinkConfigMapper(): StreamsEventSinkConfigMapper
+
+}
+
+abstract class StreamsEventConsumer<T>(private val consumer: T, config: StreamsSinkConfiguration, private val log: Log) {
 
     abstract fun stop()
 
+    abstract fun withTopics(topics: Set<String>): StreamsEventConsumer<T>
+
     abstract fun start()
 
-    override fun unavailable() {
-        stop()
-    }
+    abstract fun read(): Map<String, List<Map<String, Any?>>>?
 
-    override fun available() {
-        start()
-    }
+}
 
+abstract class StreamsEventConsumerFactory {
+    abstract fun createStreamsEventConsumer(config: Map<String, String>, log: Log): StreamsEventConsumer<*>
 }
 
 object StreamsEventSinkFactory {
-    fun getStreamsEventSink(config: Config, db: GraphDatabaseAPI): StreamsEventSink {
+    fun getStreamsEventSink(config: Config, streamsQueryExecution: StreamsEventSinkQueryExecution,
+                            streamsTopicService: StreamsTopicService, log: Log): StreamsEventSink {
         return Class.forName(config.raw.getOrDefault("streams.sink", "streams.kafka.KafkaEventSink"))
                 .getConstructor(Config::class.java,
-                        GraphDatabaseAPI::class.java)
-                .newInstance(config, db) as StreamsEventSink
+                        StreamsEventSinkQueryExecution::class.java,
+                        StreamsTopicService::class.java,
+                        Log::class.java)
+                .newInstance(config, streamsQueryExecution, streamsTopicService, log) as StreamsEventSink
     }
+}
+
+abstract class StreamsEventSinkConfigMapper(private val baseConfiguration: Map<String, String>, private val mapping: Map<String, String>) {
+    abstract fun convert(config: Map<String, String>): Map<String, String>
 }

consumer/src/main/kotlin/streams/StreamsEventSinkExtensionFactory.kt

@@ -8,6 +8,7 @@ import org.neo4j.kernel.impl.spi.KernelContext
 import org.neo4j.kernel.internal.GraphDatabaseAPI
 import org.neo4j.kernel.lifecycle.Lifecycle
 import org.neo4j.kernel.lifecycle.LifecycleAdapter
+import streams.procedures.StreamsSinkProcedures
 import streams.utils.StreamsUtils
 
 class StreamsEventSinkExtensionFactory : KernelExtensionFactory<StreamsEventSinkExtensionFactory.Dependencies>("Streams.Consumer") {
@@ -33,8 +34,24 @@ class StreamsEventSinkExtensionFactory : KernelExtensionFactory<StreamsEventSink
 
         override fun start() {
             try {
-                eventSink = StreamsEventSinkFactory.getStreamsEventSink(configuration, db)
-                dependencies.availabilityGuard().addListener(eventSink)
+                dependencies.availabilityGuard().addListener(object: AvailabilityGuard.AvailabilityListener {
+                    override fun unavailable() {}
+
+                    override fun available() {
+                        val streamsSinkConfiguration = StreamsSinkConfiguration.from(configuration)
+                        val streamsTopicService = StreamsTopicService(db, streamsSinkConfiguration.topics)
+                        val streamsQueryExecution = StreamsEventSinkQueryExecution(streamsTopicService, db, log.getUserLog(StreamsEventSinkQueryExecution::class.java))
+                        StreamsSinkProcedures.registerStreamsSinkConfiguration(streamsSinkConfiguration)
+                        eventSink = StreamsEventSinkFactory.getStreamsEventSink(configuration,
+                                streamsQueryExecution,
+                                streamsTopicService,
+                                log.getUserLog(StreamsEventSinkFactory::class.java))
+                        eventSink.start()
+                        StreamsSinkProcedures.registerStreamsEventConsumerFactory(eventSink.getEventConsumerFactory())
+                        StreamsSinkProcedures.registerStreamsEventSinkConfigMapper(eventSink.getEventSinkConfigMapper())
+                    }
+
+                })
             } catch (e: Exception) {
                 e.printStackTrace()
                 streamsLog.error("Error initializing the streaming sink", e)

consumer/src/main/kotlin/streams/StreamsEventSinkQueryExecution.kt

@@ -1,21 +1,34 @@
 package streams
 
+import org.neo4j.graphdb.GraphDatabaseService
 import org.neo4j.kernel.internal.GraphDatabaseAPI
 import org.neo4j.logging.Log
+import streams.utils.Neo4jUtils
+import streams.utils.StreamsUtils
 
 class StreamsEventSinkQueryExecution(private val streamsTopicService: StreamsTopicService, private val db: GraphDatabaseAPI, val log: Log) {
 
-    private val UNWIND: String = "UNWIND {events} AS event"
-
     fun execute(topic: String, params: Collection<Any>) {
         val cypherQuery = streamsTopicService.get(topic)
         if (cypherQuery == null) {
             return
         }
+        val query = "${StreamsUtils.UNWIND} $cypherQuery"
         if(log.isDebugEnabled){
-            log.debug("Processing ${params.size} events from Kafka")
+            log.debug("Processing ${params.size} events with query: $query")
+        }
+        if (Neo4jUtils.isWriteableInstance(db)) {
+            try {
+                db.execute(query, mapOf("events" to params)).close()
+            } catch (e: Exception) {
+                log.error("Error while executing the query", e)
+            }
+        } else {
+            if(log.isDebugEnabled){
+                log.debug("Not writeable instance")
+            }
         }
-        db.execute("$UNWIND $cypherQuery", mapOf("events" to params)).close()
+
     }
 
     fun execute(map: Map<String, Collection<Any>>) {

consumer/src/main/kotlin/streams/StreamsSinkConfiguration.kt

@@ -8,22 +8,30 @@ private object StreamsSinkConfigurationConstants {
     const val STREAMS_CONFIG_PREFIX: String = "streams."
     const val STREAMS_SINK_TOPIC_CYPHER_PREFIX: String = "sink.topic.cypher."
     const val ENABLED = "sink.enabled"
+    const val PROCEDURES_ENABLED = "procedures.enabled"
 }
 
 data class StreamsSinkConfiguration(val enabled: Boolean = true,
+                                    val proceduresEnabled: Boolean = true,
                                     val sinkPollingInterval: Long = Long.MAX_VALUE,
                                     val topics: Map<String, String> = emptyMap()) {
 
     companion object {
         fun from(cfg: Config): StreamsSinkConfiguration {
+            return from(cfg.raw)
+        }
+
+        fun from(cfg: Map<String, String>): StreamsSinkConfiguration {
             val default = StreamsSinkConfiguration()
-            val config = cfg.raw
+            val config = cfg
                     .filterKeys { it.startsWith(StreamsSinkConfigurationConstants.STREAMS_CONFIG_PREFIX) }
                     .mapKeys { it.key.substring(StreamsSinkConfigurationConstants.STREAMS_CONFIG_PREFIX.length) }
             val topics = config
                     .filterKeys { it.startsWith(StreamsSinkConfigurationConstants.STREAMS_SINK_TOPIC_CYPHER_PREFIX) }
                     .mapKeys { it.key.replace(StreamsSinkConfigurationConstants.STREAMS_SINK_TOPIC_CYPHER_PREFIX, StringUtils.EMPTY) }
             return default.copy(enabled = config.getOrDefault(StreamsSinkConfigurationConstants.ENABLED, default.enabled).toString().toBoolean(),
+                    proceduresEnabled = config.getOrDefault(StreamsSinkConfigurationConstants.PROCEDURES_ENABLED, default.proceduresEnabled)
+                            .toString().toBoolean(),
                     sinkPollingInterval = config.getOrDefault("sink.polling.interval", default.sinkPollingInterval).toString().toLong(),
                     topics = topics)
         }

consumer/src/main/kotlin/streams/StreamsTopicService.kt

@@ -3,19 +3,45 @@ package streams
 import org.apache.commons.lang3.StringUtils
 import org.neo4j.kernel.impl.core.EmbeddedProxySPI
 import org.neo4j.kernel.impl.core.GraphProperties
-import org.neo4j.kernel.impl.logging.LogService
 import org.neo4j.kernel.internal.GraphDatabaseAPI
 import streams.utils.Neo4jUtils
 
 
 private const val STREAMS_TOPIC_KEY: String = "streams.sink.topic."
 
-class StreamsTopicService(private val db: GraphDatabaseAPI, private val streamsSinkConfiguration: StreamsSinkConfiguration) {
+class StreamsTopicService(private val db: GraphDatabaseAPI, private val topicMap: Map<String, String>) {
     private val properties: GraphProperties = db.dependencyResolver.resolveDependency(EmbeddedProxySPI::class.java).newGraphPropertiesProxy()
     private val log = Neo4jUtils.getLogService(db).getUserLog(StreamsTopicService::class.java)
 
     init {
-        setAll(streamsSinkConfiguration.topics)
+        clear()
+        setAll(topicMap)
+    }
+
+    fun clear() {
+        return db.beginTx().use {
+            val keys = properties.allProperties
+                    .filterKeys { it.startsWith(STREAMS_TOPIC_KEY) }
+                    .keys
+            keys.forEach {
+                properties.removeProperty(it)
+            }
+            it.success()
+        }
+    }
+
+    fun remove(topic: String) {
+        val key = "$STREAMS_TOPIC_KEY$topic"
+        return db.beginTx().use {
+            if (!properties.hasProperty(key)) {
+                if (log.isDebugEnabled) {
+                    log.debug("No query registered for topic $topic")
+                }
+                return
+            }
+            properties.removeProperty(key)
+            it.success()
+        }
     }
 
     fun set(topic: String, query: String) {