fixes #191: The latest version of plugin fails to start polling messages

Author: conker84

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

@@ -41,12 +41,10 @@ object Neo4jUtils {
                 .resolveDependency(LogService::class.java)
     }
 
-    fun isEnterpriseEdition(db: GraphDatabaseAPI): Boolean {
+    fun isCluster(db: GraphDatabaseAPI): Boolean {
         try {
-            return db.execute("""
-                CALL dbms.components() YIELD edition
-                RETURN edition = "enterprise" AS isEnterprise
-            """.trimIndent()).columnAs<Boolean>("isEnterprise").next()
+            db.execute("CALL dbms.cluster.role()").columnAs<String>("role").next()
+            return true
         } catch (e: QueryExecutionException) {
             if (e.statusCode.equals("Neo.ClientError.Procedure.ProcedureNotFound", ignoreCase = true)) {
                 return false

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

@@ -1,8 +1,6 @@
 package streams.utils
 
-import org.junit.AfterClass
-import org.junit.BeforeClass
-import org.junit.Test
+import org.junit.*
 import org.neo4j.kernel.internal.GraphDatabaseAPI
 import org.neo4j.test.TestGraphDatabaseFactory
 import kotlin.test.assertFalse
@@ -34,8 +32,8 @@ class Neo4jUtilsTest {
     }
 
     @Test
-    fun shouldCheckIfIsEnterpriseEdition() {
-        val isEnterprise = Neo4jUtils.isEnterpriseEdition(db)
+    fun shouldCheckIfIsACluster() {
+        val isEnterprise = Neo4jUtils.isCluster(db)
         assertFalse { isEnterprise }
     }
 

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

@@ -30,9 +30,9 @@ class StreamsEventSinkExtensionFactory : KernelExtensionFactory<StreamsEventSink
 
     class StreamsEventLifecycle(private val dependencies: StreamsEventSinkExtensionFactory.Dependencies): LifecycleAdapter() {
         private val db = dependencies.graphdatabaseAPI()
-        private val log = dependencies.log()
+        private val logService = dependencies.log()
         private val configuration = dependencies.config()
-        private var streamsLog = log.getUserLog(StreamsEventLifecycle::class.java)
+        private var streamsLog = logService.getUserLog(StreamsEventLifecycle::class.java)
 
         private lateinit var eventSink: StreamsEventSink
 
@@ -42,42 +42,36 @@ class StreamsEventSinkExtensionFactory : KernelExtensionFactory<StreamsEventSink
                     override fun unavailable() {}
 
                     override fun available() {
+                        streamsLog.info("Initialising the Streams Sink module")
                         val streamsSinkConfiguration = StreamsSinkConfiguration.from(configuration)
-
                         val streamsTopicService = StreamsTopicService(db)
                         streamsTopicService.clearAll()
                         streamsTopicService.setAll(streamsSinkConfiguration.topics)
                         val strategyMap = TopicUtils.toStrategyMap(streamsSinkConfiguration.topics,
                                 streamsSinkConfiguration.sourceIdStrategyConfig)
                         val streamsQueryExecution = StreamsEventSinkQueryExecution(streamsTopicService, db,
-                                log.getUserLog(StreamsEventSinkQueryExecution::class.java),
+                                logService.getUserLog(StreamsEventSinkQueryExecution::class.java),
                                 strategyMap)
 
                         // Create the Sink
-                        val sinkLog = log.getUserLog(StreamsEventSinkFactory::class.java)
+                        val log = logService.getUserLog(StreamsEventSinkFactory::class.java)
                         eventSink = StreamsEventSinkFactory
                                 .getStreamsEventSink(configuration,
                                         streamsQueryExecution,
                                         streamsTopicService,
-
-                                        sinkLog,
+                                        log,
                                         db)
                         // start the Sink
-                        if (Neo4jUtils.isEnterpriseEdition(db)) {
-                            sinkLog.info("The Sink module is running in an enterprise edition, checking for the ${Neo4jUtils.LEADER}")
-                            Neo4jUtils.executeInLeader(db, sinkLog) { initSinkModule() }
+                        if (Neo4jUtils.isCluster(db)) {
+                            log.info("The Sink module is running in a cluster, checking for the ${Neo4jUtils.LEADER}")
+                            Neo4jUtils.executeInLeader(db, log) { initSinkModule() }
                         } else {
                             // check if is writeable instance
                             Neo4jUtils.executeInWriteableInstance(db) { initSinkModule() }
                         }
 
                         // Register required services for the Procedures
                         StreamsSinkProcedures.registerStreamsSinkConfiguration(streamsSinkConfiguration)
-                        eventSink = StreamsEventSinkFactory.getStreamsEventSink(configuration,
-                                streamsQueryExecution,
-                                streamsTopicService,
-                                log.getUserLog(StreamsEventSinkFactory::class.java), db)
-                        eventSink.start()
                         StreamsSinkProcedures.registerStreamsEventConsumerFactory(eventSink.getEventConsumerFactory())
                         StreamsSinkProcedures.registerStreamsEventSinkConfigMapper(eventSink.getEventSinkConfigMapper())
                     }