Add a configuration option to set how often should a node check for new nodeinfos on disk (#1851)

* Add a configuration option to set how often should a node check for new NodeInfo files in additional-node-infos

node/src/integration-test/kotlin/net/corda/node/services/network/NodeInfoWatcherTest.kt

@@ -50,7 +50,7 @@ class NodeInfoWatcherTest : NodeBasedTest() {
     fun start() {
         val identityService = InMemoryIdentityService(trustRoot = DEV_TRUST_ROOT)
         keyManagementService = MockKeyManagementService(identityService, ALICE_KEY)
-        nodeInfoWatcher = NodeInfoWatcher(folder.root.toPath(), scheduler)
+        nodeInfoWatcher = NodeInfoWatcher(folder.root.toPath(), scheduler = scheduler)
         nodeInfoPath = folder.root.toPath() / CordformNode.NODE_INFO_DIRECTORY
     }
 

node/src/main/kotlin/net/corda/node/services/config/NodeConfiguration.kt

@@ -2,6 +2,7 @@ package net.corda.node.services.config
 
 import net.corda.core.identity.CordaX500Name
 import net.corda.core.utilities.NetworkHostAndPort
+import net.corda.core.utilities.seconds
 import net.corda.node.internal.NetworkMapInfo
 import net.corda.node.services.messaging.CertificateChainCheckPolicy
 import net.corda.nodeapi.User
@@ -33,6 +34,7 @@ interface NodeConfiguration : NodeSSLConfiguration {
     val messageRedeliveryDelaySeconds: Int
     val notary: NotaryConfig?
     val activeMQServer: ActiveMqServerConfiguration
+    val additionalNodeInfoPollingFrequencyMsec: Long
 }
 
 data class NotaryConfig(val validating: Boolean, val raft: RaftConfig? = null, val bftSMaRt: BFTSMaRtConfiguration? = null) {
@@ -86,7 +88,8 @@ data class FullNodeConfiguration(
         override val devMode: Boolean = false,
         val useTestClock: Boolean = false,
         val detectPublicIp: Boolean = true,
-        override val activeMQServer: ActiveMqServerConfiguration
+        override val activeMQServer: ActiveMqServerConfiguration,
+        override val additionalNodeInfoPollingFrequencyMsec: Long = 5.seconds.toMillis()
 ) : NodeConfiguration {
     override val exportJMXto: String get() = "http"
 

node/src/main/kotlin/net/corda/node/services/network/NodeInfoWatcher.kt

@@ -8,10 +8,10 @@ import net.corda.core.node.services.KeyManagementService
 import net.corda.core.serialization.deserialize
 import net.corda.core.serialization.serialize
 import net.corda.core.utilities.loggerFor
+import net.corda.core.utilities.seconds
 import rx.Observable
 import rx.Scheduler
 import rx.schedulers.Schedulers
-import java.nio.file.Files
 import java.nio.file.Path
 import java.util.concurrent.TimeUnit
 import kotlin.streams.toList
@@ -22,13 +22,17 @@ import kotlin.streams.toList
  * - Poll a directory for new serialized [NodeInfo]
  *
  * @param path the base path of a node.
+ * @param pollFrequencyMsec how often to poll the filesystem in milliseconds. Any value smaller than 5 seconds will
+ *        be treated as 5 seconds.
  * @param scheduler a [Scheduler] for the rx [Observable] returned by [nodeInfoUpdates], this is mainly useful for
  *        testing. It defaults to the io scheduler which is the appropriate value for production uses.
  */
 class NodeInfoWatcher(private val nodePath: Path,
+                      pollFrequencyMsec: Long = 5.seconds.toMillis(),
                       private val scheduler: Scheduler = Schedulers.io()) {
 
     private val nodeInfoDirectory = nodePath / CordformNode.NODE_INFO_DIRECTORY
+    private val pollFrequencyMsec: Long
 
     companion object {
         private val logger = loggerFor<NodeInfoWatcher>()
@@ -56,6 +60,10 @@ class NodeInfoWatcher(private val nodePath: Path,
         }
     }
 
+    init {
+        this.pollFrequencyMsec = maxOf(pollFrequencyMsec, 5.seconds.toMillis())
+    }
+
     /**
      * Read all the files contained in [nodePath] / [CordformNode.NODE_INFO_DIRECTORY] and keep watching
      * the folder for further updates.
@@ -67,7 +75,7 @@ class NodeInfoWatcher(private val nodePath: Path,
      *      than once.
      */
     fun nodeInfoUpdates(): Observable<NodeInfo> {
-        return Observable.interval(5, TimeUnit.SECONDS, scheduler)
+        return Observable.interval(pollFrequencyMsec, TimeUnit.MILLISECONDS, scheduler)
                 .flatMapIterable { loadFromDirectory() }
     }
 

node/src/main/kotlin/net/corda/node/services/network/PersistentNetworkMapCache.kt

@@ -39,6 +39,7 @@ import rx.Observable
 import rx.subjects.PublishSubject
 import java.security.PublicKey
 import java.security.SignatureException
+import java.time.Duration
 import java.util.*
 import javax.annotation.concurrent.ThreadSafe
 import kotlin.collections.HashMap
@@ -87,7 +88,8 @@ open class PersistentNetworkMapCache(private val serviceHub: ServiceHubInternal)
                     .sortedBy { it.name.toString() }
         }
 
-    private val nodeInfoSerializer = NodeInfoWatcher(serviceHub.configuration.baseDirectory)
+    private val nodeInfoSerializer = NodeInfoWatcher(serviceHub.configuration.baseDirectory,
+            serviceHub.configuration.additionalNodeInfoPollingFrequencyMsec)
 
     init {
         loadFromFiles()

node/src/test/kotlin/net/corda/node/services/config/FullNodeConfigurationTest.kt

@@ -1,6 +1,7 @@
 package net.corda.node.services.config
 
 import net.corda.core.utilities.NetworkHostAndPort
+import net.corda.core.utilities.seconds
 import net.corda.nodeapi.User
 import net.corda.testing.ALICE
 import net.corda.testing.node.MockServices.Companion.makeTestDataSourceProperties
@@ -33,7 +34,8 @@ class FullNodeConfigurationTest {
                 notary = null,
                 certificateChainCheckPolicies = emptyList(),
                 devMode = true,
-                activeMQServer = ActiveMqServerConfiguration(BridgeConfiguration(0, 0, 0.0)))
+                activeMQServer = ActiveMqServerConfiguration(BridgeConfiguration(0, 0, 0.0)),
+                additionalNodeInfoPollingFrequencyMsec = 5.seconds.toMillis())
 
         fun configWithRPCUsername(username: String) {
             testConfiguration.copy(rpcUsers = listOf(User(username, "pass", emptySet())))