add the sink lifecycle procedures (#320)

* add the sink lifecycle procedures

* remove call to dbms.procedures

* added apoc check

Author: conker84

common/src/main/kotlin/streams/events/StreamsPluginStatus.kt

@@ -0,0 +1,3 @@
+package streams.events
+
+enum class StreamsPluginStatus { RUNNING, STOPPED, UNKNOWN }

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

@@ -14,7 +14,7 @@ import kotlin.streams.toList
 
 object Neo4jUtils {
     @JvmStatic val LEADER = "LEADER"
-    fun isWriteableInstance(db: GraphDatabaseAPI): Boolean {
+    fun isWriteableInstance(db: GraphDatabaseAPI, isCluster: Boolean = false): Boolean {
         try {
             val isSlave = StreamsUtils.ignoreExceptions(
                     {
@@ -26,8 +26,12 @@ object Neo4jUtils {
                 return false
             }
 
-            val role = db.execute("CALL dbms.cluster.role()").columnAs<String>("role").next()
-            return role.equals(LEADER, ignoreCase = true)
+            return if (isCluster) {
+                val role = db.execute("CALL dbms.cluster.role() YIELD role RETURN role").columnAs<String>("role").next()
+                return role.equals(LEADER, ignoreCase = true)
+            } else {
+                true
+            }
         } catch (e: QueryExecutionException) {
             if (e.statusCode.equals("Neo.ClientError.Procedure.ProcedureNotFound", ignoreCase = true)) {
                 return true
@@ -53,6 +57,19 @@ object Neo4jUtils {
         }
     }
 
+    fun hasApoc(db: GraphDatabaseAPI): Boolean {
+        try {
+            db.execute("RETURN apoc.version() AS version").columnAs<String>("version").next()
+            return true
+        } catch (e: QueryExecutionException) {
+            if (e.statusCode.equals("Neo.ClientError.Statement.SyntaxError", ignoreCase = true)
+                    && e.message!!.contains("Unknown function", ignoreCase = true)) {
+                return false
+            }
+            throw e
+        }
+    }
+
     fun clusterHasLeader(db: GraphDatabaseAPI): Boolean {
         try {
             return db.execute("""

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

@@ -1,5 +1,8 @@
 package streams.utils
 
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.delay
+
 object StreamsUtils {
 
     const val UNWIND: String = "UNWIND {events} AS event"
@@ -22,4 +25,14 @@ object StreamsUtils {
         }
     }
 
+    fun blockUntilTrueOrTimeout(timeout: Long, delay: Long = 1000, action: () -> Boolean): Boolean = runBlocking {
+        val start = System.currentTimeMillis()
+        var success = action()
+        while (System.currentTimeMillis() - start < timeout && !success) {
+            delay(delay)
+            success = action()
+        }
+        success
+    }
+
 }

common/src/test/kotlin/streams/utils/Neo4jUtilsTest.kt

@@ -37,4 +37,9 @@ class Neo4jUtilsTest {
         assertFalse { isEnterprise }
     }
 
+    @Test
+    fun `should not have APOC`() {
+        assertFalse { Neo4jUtils.hasApoc(db) }
+    }
+
 }

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

@@ -3,6 +3,7 @@ package streams
 import org.neo4j.kernel.configuration.Config
 import org.neo4j.kernel.internal.GraphDatabaseAPI
 import org.neo4j.logging.Log
+import streams.events.StreamsPluginStatus
 
 abstract class StreamsEventSink(private val config: Config,
                                 private val queryExecution: StreamsEventSinkQueryExecution,
@@ -23,6 +24,8 @@ abstract class StreamsEventSink(private val config: Config,
 
     open fun printInvalidTopics() {}
 
+    abstract fun status(): StreamsPluginStatus
+
 }
 
 object StreamsEventSinkFactory {

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

@@ -1,5 +1,8 @@
 package streams
 
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.launch
 import org.neo4j.kernel.availability.AvailabilityGuard
 import org.neo4j.kernel.availability.AvailabilityListener
 import org.neo4j.kernel.configuration.Config
@@ -72,6 +75,7 @@ class StreamsEventSinkExtensionFactory : KernelExtensionFactory<StreamsEventSink
                             StreamsSinkProcedures.registerStreamsSinkConfiguration(streamsSinkConfiguration)
                             StreamsSinkProcedures.registerStreamsEventConsumerFactory(eventSink.getEventConsumerFactory())
                             StreamsSinkProcedures.registerStreamsEventSinkConfigMapper(eventSink.getEventSinkConfigMapper())
+                            StreamsSinkProcedures.registerStreamsEventSink(eventSink)
                         } catch (e: Exception) {
                             streamsLog.error("Error initializing the streaming sink:", e)
                         }
@@ -80,6 +84,28 @@ class StreamsEventSinkExtensionFactory : KernelExtensionFactory<StreamsEventSink
         }
 
         private fun initSinkModule(streamsTopicService: StreamsTopicService, streamsSinkConfiguration: StreamsSinkConfiguration) {
+            if (streamsSinkConfiguration.checkApocTimeout > -1) {
+                GlobalScope.launch(Dispatchers.IO) {
+                    val success = StreamsUtils.blockUntilTrueOrTimeout(streamsSinkConfiguration.checkApocTimeout, streamsSinkConfiguration.checkApocInterval) {
+                        val hasApoc = Neo4jUtils.hasApoc(db)
+                        if (!hasApoc && streamsLog.isDebugEnabled) {
+                            streamsLog.debug("APOC not loaded yet, next check in ${streamsSinkConfiguration.checkApocInterval} ms")
+                        }
+                        hasApoc
+                    }
+                    if (success) {
+                        initSink(streamsTopicService, streamsSinkConfiguration)
+                    } else {
+                        streamsLog.info("Streams Sink plugin not loaded as APOC are not installed")
+                    }
+                }
+            } else {
+                initSink(streamsTopicService, streamsSinkConfiguration)
+            }
+
+        }
+
+        private fun initSink(streamsTopicService: StreamsTopicService, streamsSinkConfiguration: StreamsSinkConfiguration) {
             streamsTopicService.clearAll()
             streamsTopicService.setAll(streamsSinkConfiguration.topics)
             eventSink.start()

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

@@ -1,22 +1,28 @@
 package streams
 
 import org.neo4j.kernel.configuration.Config
+import streams.extensions.toPointCase
+import streams.serialization.JSONUtils
 import streams.service.TopicUtils
 import streams.service.sink.strategy.SourceIdIngestionStrategyConfig
 import streams.service.Topics
+import java.util.Properties
 
 
 object StreamsSinkConfigurationConstants {
-    const val STREAMS_CONFIG_PREFIX: String = "streams."
+    const val CHECK_APOC_TIMEOUT = "check.apoc.timeout"
+    const val CHECK_APOC_INTERVAL = "check.apoc.interval"
+    const val STREAMS_CONFIG_PREFIX = "streams."
     const val ENABLED = "sink.enabled"
     const val PROCEDURES_ENABLED = "procedures.enabled"
 }
 
 data class StreamsSinkConfiguration(val enabled: Boolean = false,
                                     val proceduresEnabled: Boolean = true,
-                                    val sinkPollingInterval: Long = 10000,
                                     val topics: Topics = Topics(),
                                     val errorConfig: Map<String,Any?> = emptyMap(),
+                                    val checkApocTimeout: Long = -1,
+                                    val checkApocInterval: Long = 1000,
                                     val sourceIdStrategyConfig: SourceIdIngestionStrategyConfig = SourceIdIngestionStrategyConfig()) {
 
     companion object {
@@ -47,9 +53,12 @@ data class StreamsSinkConfiguration(val enabled: Boolean = false,
             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,
                     errorConfig = errorHandler,
+                    checkApocTimeout = config.getOrDefault(StreamsSinkConfigurationConstants.CHECK_APOC_TIMEOUT, default.checkApocTimeout)
+                            .toString().toLong(),
+                    checkApocInterval = config.getOrDefault(StreamsSinkConfigurationConstants.CHECK_APOC_INTERVAL, default.checkApocInterval)
+                            .toString().toLong(),
                     sourceIdStrategyConfig = sourceIdStrategyConfig)
         }
 

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

@@ -1,123 +1,79 @@
 package streams
 
-import org.neo4j.kernel.impl.core.EmbeddedProxySPI
-import org.neo4j.kernel.impl.core.GraphProperties
 import org.neo4j.kernel.internal.GraphDatabaseAPI
-import streams.serialization.JSONUtils
-import streams.service.STREAMS_TOPIC_KEY
 import streams.service.TopicType
 import streams.service.Topics
 import streams.utils.Neo4jUtils
+import java.util.Collections
+import java.util.concurrent.ConcurrentHashMap
 
-class StreamsTopicService(private val db: GraphDatabaseAPI) {
-    private val properties: GraphProperties = db.dependencyResolver.resolveDependency(EmbeddedProxySPI::class.java).newGraphPropertiesProxy()
+class StreamsTopicService(db: GraphDatabaseAPI) {
+    private val log = Neo4jUtils.getLogService(db).getUserLog(StreamsTopicService::class.java)
 
-    fun clearAll() { // TODO move to Neo4jUtils#executeInWriteableInstance
-        if (!Neo4jUtils.isWriteableInstance(db)) {
-            return
-        }
-        return db.beginTx().use {
-            val keys = properties.allProperties
-                    .filterKeys { it.startsWith(STREAMS_TOPIC_KEY) }
-                    .keys
-            keys.forEach {
-                properties.removeProperty(it)
-            }
-            it.success()
-        }
+    private val storage = ConcurrentHashMap<TopicType, Any>()
+
+    fun clearAll() {
+        storage.clear()
     }
 
-    fun set(topicType: TopicType, data: Any) = Neo4jUtils.executeInWriteableInstance(db) {
-        db.beginTx().use {
-            if (properties.hasProperty(topicType.key)) {
-                val topicData = JSONUtils.readValue<Any>(properties.getProperty(topicType.key))
-                val newData = when (topicData) {
-                    is Map<*, *> -> topicData + (data as Map<String, Any?>)
-                    is Collection<*> -> topicData + (data as Collection<String>)
-                    else -> throw RuntimeException("Unsupported data $data for topic type $topicType")
-                }
-                properties.setProperty(topicType.key, JSONUtils.writeValueAsString(newData))
-            } else {
-                properties.setProperty(topicType.key, JSONUtils.writeValueAsString(data))
-            }
-            it.success()
+    fun set(topicType: TopicType, data: Any) {
+        val runtimeException = RuntimeException("Unsupported data $data for topic type $topicType")
+        var oldData = storage[topicType]
+        oldData = oldData ?: when (data) {
+            is Map<*, *> -> emptyMap<String, Any?>()
+            is Collection<*> -> emptyList<String>()
+            else -> throw runtimeException
+        }
+        val newData = when (oldData) {
+            is Map<*, *> -> oldData + (data as Map<String, Any?>)
+            is Collection<*> -> oldData + (data as Collection<String>)
+            else -> throw runtimeException
         }
+        storage[topicType] = newData
     }
 
-    fun remove(topicType: TopicType, topic: String) = Neo4jUtils.executeInWriteableInstance(db) {
-        db.beginTx().use {
-            if (properties.hasProperty(topicType.key)) {
-                val topicData = JSONUtils.readValue<Any>(properties.getProperty(topicType.key))
-                val newData = when (topicData) {
-                    is Map<*, *> -> topicData.filterKeys { it.toString() != topic }
-                    is Collection<*> -> topicData.filter { it.toString() != topic }
-                    else -> throw RuntimeException("Unsupported data $topicData for topic type $topicType")
-                }
-                val isEmpty = when (newData) {
-                    is Map<*, *> -> newData.isEmpty()
-                    is Collection<*> -> newData.isEmpty()
-                    else -> throw RuntimeException("Unsupported data $topicData for topic type $topicType")
-                }
-                if (isEmpty) {
-                    properties.removeProperty(topicType.key)
-                } else {
-                    properties.setProperty(topicType.key, JSONUtils.writeValueAsString(newData))
-                }
-            }
-            it.success()
+    fun remove(topicType: TopicType, topic: String) {
+        val topicData = storage[topicType] ?: return
+
+        val runtimeException = RuntimeException("Unsupported data $topicData for topic type $topicType")
+        val filteredData = when (topicData) {
+            is Map<*, *> -> topicData.filterKeys { it.toString() != topic }
+            is Collection<*> -> topicData.filter { it.toString() != topic }
+            else -> throw runtimeException
         }
+
+        storage[topicType] = filteredData
     }
 
-    fun getTopicType(topic: String) = Neo4jUtils.executeInWriteableInstance(db) {
-        db.beginTx().use {
-            TopicType.values().find {
-                if (!properties.hasProperty(it.key)) {
-                    false
-                } else {
-                    val data = JSONUtils.readValue<Any>(properties.getProperty(it.key))
-                    when (data) {
-                        is Map<*, *> -> data.containsKey(topic)
-                        is Collection<*> -> data.contains(topic)
-                        else -> false
-                    }
+    fun getTopicType(topic: String) = TopicType.values()
+            .find {
+                val topicData = storage[it]
+                when (topicData) {
+                    is Map<*, *> -> topicData.containsKey(topic)
+                    is Collection<*> -> topicData.contains(topic)
+                    else -> false
                 }
             }
-        }
-    }
 
-    fun getTopics() = db.beginTx().use {
-        TopicType.values()
-                .filter { properties.hasProperty(it.key) }
-                .flatMap {
-                    val data = JSONUtils.readValue<Any>(properties.getProperty(it.key))
-                    when (data) {
-                        is Map<*, *> -> data.keys
-                        is Collection<*> -> data.toSet()
-                        else -> emptySet()
-                    }
-                }.toSet() as Set<String>
-    }
+    fun getTopics() = TopicType.values()
+            .flatMap {
+                val data = storage[it]
+                when (data) {
+                    is Map<*, *> -> data.keys
+                    is Collection<*> -> data.toSet()
+                    else -> emptySet<String>()
+                }
+            }.toSet() as Set<String>
 
     fun setAll(topics: Topics) {
         topics.asMap().forEach { topicType, data ->
             set(topicType, data)
         }
     }
 
-    fun getCypherTemplate(topic: String) = db.beginTx().use {
-        if (properties.hasProperty(TopicType.CYPHER.key)) {
-            val data = JSONUtils.readValue<Map<String, String>>(properties.getProperty(TopicType.CYPHER.key))
-            data[topic]
-        } else {
-            null
-        }
-    }
+    fun getCypherTemplate(topic: String) = (storage.getOrDefault(TopicType.CYPHER, emptyMap<String, String>()) as Map<String, String>)
+            .let { it[topic] }
 
-    fun getAll() = db.beginTx().use {
-        TopicType.values()
-                .filter { properties.hasProperty(it.key) }
-                .map { it to JSONUtils.readValue<Any>(properties.getProperty(it.key)) }
-                .toMap()
-    }
+    fun getAll(): Map<TopicType, Any> = Collections.unmodifiableMap(storage)
 
 }