fixes #200: Move the streamsTopicService into the initSinkModule method

Author: conker84

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

@@ -89,4 +89,16 @@ object Neo4jUtils {
         }
     }
 
+    fun waitForTheLeader(db: GraphDatabaseAPI, log: Log, timeout: Long = 120000, action: () -> Unit) {
+        GlobalScope.launch(Dispatchers.IO) {
+            val start = System.currentTimeMillis()
+            val delay: Long = 2000
+            while (!clusterHasLeader(db) && System.currentTimeMillis() - start < timeout) {
+                log.info("$LEADER not found, new check comes in $delay milliseconds...")
+                delay(delay)
+            }
+            action()
+        }
+    }
+
 }

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

@@ -43,8 +43,6 @@ class StreamsEventSinkExtensionFactory : KernelExtensionFactory<StreamsEventSink
                         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,
@@ -62,10 +60,10 @@ class StreamsEventSinkExtensionFactory : KernelExtensionFactory<StreamsEventSink
                         // start the Sink
                         if (Neo4jUtils.isCluster(db)) {
                             log.info("The Sink module is running in a cluster, checking for the ${Neo4jUtils.LEADER}")
-                            Neo4jUtils.executeInLeader(db, log) { initSinkModule() }
+                            Neo4jUtils.waitForTheLeader(db, log) { initSinkModule(streamsTopicService, streamsSinkConfiguration) }
                         } else {
                             // check if is writeable instance
-                            Neo4jUtils.executeInWriteableInstance(db) { initSinkModule() }
+                            Neo4jUtils.executeInWriteableInstance(db) { initSinkModule(streamsTopicService, streamsSinkConfiguration) }
                         }
 
                         // Register required services for the Procedures
@@ -80,7 +78,9 @@ class StreamsEventSinkExtensionFactory : KernelExtensionFactory<StreamsEventSink
             }
         }
 
-        private fun initSinkModule() {
+        private fun initSinkModule(streamsTopicService: StreamsTopicService, streamsSinkConfiguration: StreamsSinkConfiguration) {
+            streamsTopicService.clearAll()
+            streamsTopicService.setAll(streamsSinkConfiguration.topics)
             eventSink.start()
             streamsLog.info("Streams Sink module initialised")
         }