fixes #167 Added the AvailabilityGuard in the StreamsEventSinkExtensionFactory (#168)

Author: conker84

producer/src/main/kotlin/streams/StreamsConstraintsService.kt

@@ -12,16 +12,19 @@ import java.util.*
 import java.util.concurrent.ConcurrentHashMap
 
 class StreamsConstraintsService(private val db: GraphDatabaseService, private val poolInterval: Long): Closeable {
-    override fun close() = runBlocking {
-        job.cancelAndJoin()
-    }
 
     private val nodeConstraints = ConcurrentHashMap<String, Set<Constraint>>()
     private val relConstraints = ConcurrentHashMap<String, Set<Constraint>>()
 
-    private val job: Job
+    private lateinit var job: Job
+
+    override fun close() = runBlocking {
+        if (::job.isInitialized) {
+            job.cancelAndJoin()
+        }
+    }
 
-    init {
+    fun start() {
         job = GlobalScope.launch(Dispatchers.IO) {
             while (isActive) {
                 StreamsUtils.ignoreExceptions({

producer/src/main/kotlin/streams/StreamsExtensionFactory.kt

@@ -1,5 +1,6 @@
 package streams
 
+import org.neo4j.kernel.AvailabilityGuard
 import org.neo4j.kernel.configuration.Config
 import org.neo4j.kernel.extension.ExtensionType
 import org.neo4j.kernel.extension.KernelExtensionFactory
@@ -17,18 +18,20 @@ class StreamsExtensionFactory : KernelExtensionFactory<StreamsExtensionFactory.D
         val configuration = dependencies.config()
         val streamHandler = StreamsEventRouterFactory.getStreamsEventRouter(log, configuration)
         val streamsEventRouterConfiguration = StreamsEventRouterConfiguration.from(configuration.raw)
-        return StreamsEventRouterLifecycle(db, streamHandler, streamsEventRouterConfiguration, log)
+        return StreamsEventRouterLifecycle(db, streamHandler, streamsEventRouterConfiguration, dependencies.availabilityGuard(), log)
     }
 
     interface Dependencies {
         fun graphdatabaseAPI(): GraphDatabaseAPI
         fun log(): LogService
         fun config(): Config
+        fun availabilityGuard(): AvailabilityGuard
     }
 }
 
 class StreamsEventRouterLifecycle(val db: GraphDatabaseAPI, val streamHandler: StreamsEventRouter,
                                   val streamsEventRouterConfiguration: StreamsEventRouterConfiguration,
+                                  private val availabilityGuard: AvailabilityGuard,
                                   private val log: LogService): LifecycleAdapter() {
     private val streamsLog = log.getUserLog(StreamsEventRouterLifecycle::class.java)
     private lateinit var txHandler: StreamsTransactionEventHandler
@@ -53,6 +56,13 @@ class StreamsEventRouterLifecycle(val db: GraphDatabaseAPI, val streamHandler: S
             streamsConstraintsService = StreamsConstraintsService(db, streamsEventRouterConfiguration.schemaPollingInterval)
             txHandler = StreamsTransactionEventHandler(streamHandler, streamsConstraintsService, streamsEventRouterConfiguration)
             db.registerTransactionEventHandler(txHandler)
+            availabilityGuard.addListener(object: AvailabilityGuard.AvailabilityListener {
+                override fun unavailable() {}
+
+                override fun available() {
+                    streamsConstraintsService.start()
+                }
+            })
         }
     }
 

producer/src/test/kotlin/streams/StreamsTransactionEventHandlerRelTest.kt

@@ -40,6 +40,7 @@ class StreamsTransactionEventHandlerRelTest {
         Mockito.`when`(schemaMock.constraints).thenReturn(emptyList())
         Mockito.`when`(dbMock.schema()).thenReturn(schemaMock)
         streamsConstraintsService = StreamsConstraintsService(dbMock, 0)
+        streamsConstraintsService.start()
         handler = StreamsTransactionEventHandler(MockStreamsEventRouter(),
                 streamsConstraintsService, StreamsEventRouterConfiguration())
         MockStreamsEventRouter.reset()

producer/src/test/kotlin/streams/StreamsTransactionEventHandlerTest.kt

@@ -38,6 +38,7 @@ class StreamsTransactionEventHandlerTest {
         Mockito.`when`(schemaMock.getConstraints(Mockito.any(RelationshipType::class.java))).thenReturn(emptyList())
         Mockito.`when`(dbMock.schema()).thenReturn(schemaMock)
         streamsConstraintsService = StreamsConstraintsService(dbMock, 0)
+        streamsConstraintsService.start()
         handler = StreamsTransactionEventHandler(MockStreamsEventRouter(),
                 streamsConstraintsService, StreamsEventRouterConfiguration())
         MockStreamsEventRouter.reset()