fixes #472: Fix race condition on startup 4.2.x (#473)

Author: conker84

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

@@ -9,8 +9,8 @@ import org.neo4j.kernel.internal.GraphDatabaseAPI
 import org.neo4j.kernel.lifecycle.Lifecycle
 import org.neo4j.kernel.lifecycle.LifecycleAdapter
 import org.neo4j.logging.internal.LogService
-import streams.config.StreamsConfig
 import streams.extensions.isSystemDb
+import java.util.concurrent.atomic.AtomicReference
 
 class StreamsEventSinkExtensionFactory : ExtensionFactory<StreamsEventSinkExtensionFactory.Dependencies>(ExtensionType.DATABASE,"Streams.Consumer") {
 
@@ -29,21 +29,22 @@ class StreamsEventSinkExtensionFactory : ExtensionFactory<StreamsEventSinkExtens
         private val db = dependencies.graphdatabaseAPI()
         private val logService = dependencies.log()
         private val streamsLog = logService.getUserLog(StreamsEventLifecycle::class.java)
-        private val availabilityListener: StreamsEventSinkAvailabilityListener? =  if (db.isSystemDb()) {
+        private val availabilityListener: AtomicReference<StreamsEventSinkAvailabilityListener> = AtomicReference(null)
+
+        private fun createStreamsEventSinkAvailabilityListener() = if (db.isSystemDb()) {
             null
         } else {
             StreamsEventSinkAvailabilityListener(dependencies)
         }
 
         override fun start() {
-            availabilityListener?.let {
-                dependencies.availabilityGuard().addListener(availabilityListener)
-            }
+            availabilityListener.updateAndGet { it ?: createStreamsEventSinkAvailabilityListener() }
+                    ?.let { dependencies.availabilityGuard().addListener(it) }
         }
 
         override fun stop() {
             try {
-                availabilityListener?.let {
+                availabilityListener.getAndSet(null)?.let {
                     it.shutdown()
                     dependencies.availabilityGuard().removeListener(it)
                 }

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

@@ -9,6 +9,7 @@ import org.neo4j.kernel.lifecycle.Lifecycle
 import org.neo4j.kernel.lifecycle.LifecycleAdapter
 import org.neo4j.logging.internal.LogService
 import streams.extensions.isSystemDb
+import java.util.concurrent.atomic.AtomicReference
 
 class StreamsExtensionFactory : ExtensionFactory<StreamsExtensionFactory.Dependencies>(ExtensionType.DATABASE,"Streams.Producer") {
     override fun newInstance(context: ExtensionContext, dependencies: Dependencies): Lifecycle {
@@ -26,26 +27,27 @@ class StreamsExtensionFactory : ExtensionFactory<StreamsExtensionFactory.Depende
 }
 
 class StreamsEventRouterLifecycle(private val availabilityGuard: AvailabilityGuard,
-                                  db: GraphDatabaseAPI,
-                                  log: LogService): LifecycleAdapter() {
+                                  private val db: GraphDatabaseAPI,
+                                  private val log: LogService): LifecycleAdapter() {
 
     private val streamsLog = log.getUserLog(StreamsEventRouterLifecycle::class.java)
 
-    private val streamsEventRouterAvailabilityListener: StreamsEventRouterAvailabilityListener? = if (db.isSystemDb()) {
+    private val availabilityListener: AtomicReference<StreamsEventRouterAvailabilityListener> = AtomicReference(null)
+
+    private fun createStreamsEventRouterAvailabilityListener() = if (db.isSystemDb()) {
         null
     } else {
         StreamsEventRouterAvailabilityListener(db, log)
     }
 
     override fun start() {
-        streamsEventRouterAvailabilityListener?.also {
-            availabilityGuard.addListener(it)
-        }
+        availabilityListener.updateAndGet { it ?: createStreamsEventRouterAvailabilityListener() }
+                ?.let { availabilityGuard.addListener(it) }
     }
 
     override fun stop() {
         try {
-            streamsEventRouterAvailabilityListener?.also {
+            availabilityListener.getAndSet(null)?.let {
                 it.shutdown()
                 availabilityGuard.removeListener(it)
             }