CORDA-716 Rename one TestClock to DemoClock, and unduplicate code (co…

…rda#1988)

node/src/main/kotlin/net/corda/node/internal/CordaClock.kt

@@ -0,0 +1,61 @@
+package net.corda.node.internal
+
+import net.corda.core.serialization.SerializeAsToken
+import net.corda.core.serialization.SerializeAsTokenContext
+import net.corda.core.serialization.SingletonSerializationToken
+import rx.Observable
+import rx.Subscriber
+import rx.subscriptions.Subscriptions
+import java.time.Clock
+import java.time.Instant
+import java.time.ZoneId
+import java.util.concurrent.CopyOnWriteArraySet
+import java.util.concurrent.atomic.AtomicLong
+import javax.annotation.concurrent.ThreadSafe
+
+/** A [Clock] that tokenizes itself when serialized, and delegates to an underlying [Clock] implementation. */
+abstract class CordaClock : Clock(), SerializeAsToken {
+    protected abstract val delegateClock: Clock
+    private val token = SingletonSerializationToken.singletonSerializationToken(javaClass)
+    override fun toToken(context: SerializeAsTokenContext) = token.registerWithContext(context, this)
+    override fun instant(): Instant = delegateClock.instant()
+    override fun getZone(): ZoneId = delegateClock.zone
+    @Deprecated("Do not use this. Instead seek to use ZonedDateTime methods.", level = DeprecationLevel.ERROR)
+    override fun withZone(zone: ZoneId) = throw UnsupportedOperationException("Tokenized clock does not support withZone()")
+}
+
+@ThreadSafe
+class SimpleClock(override val delegateClock: Clock) : CordaClock()
+
+/**
+ * An abstract class with helper methods for a type of Clock that might have it's concept of "now" adjusted externally.
+ * e.g. for testing (so unit tests do not have to wait for timeouts in realtime) or for demos and simulations.
+ */
+abstract class MutableClock(private var _delegateClock: Clock) : CordaClock() {
+    override var delegateClock
+        @Synchronized get() = _delegateClock
+        @Synchronized set(clock) {
+            _delegateClock = clock
+        }
+    private val _version = AtomicLong(0L)
+    /** This is an observer on the mutation count of this [Clock], which reflects the occurence of mutations. */
+    val mutations: Observable<Long> by lazy {
+        Observable.create { subscriber: Subscriber<in Long> ->
+            if (!subscriber.isUnsubscribed) {
+                mutationObservers.add(subscriber)
+                // This is not very intuitive, but subscribing to a subscriber observes unsubscribes.
+                subscriber.add(Subscriptions.create { mutationObservers.remove(subscriber) })
+            }
+        }
+    }
+    private val mutationObservers = CopyOnWriteArraySet<Subscriber<in Long>>()
+    /** Must be called by subclasses when they mutate (but not just with the passage of time as per the "wall clock"). */
+    protected fun notifyMutationObservers() {
+        val version = _version.incrementAndGet()
+        for (observer in mutationObservers) {
+            if (!observer.isUnsubscribed) {
+                observer.onNext(version)
+            }
+        }
+    }
+}

node/src/main/kotlin/net/corda/node/internal/Node.kt

@@ -15,7 +15,6 @@ import net.corda.core.utilities.loggerFor
 import net.corda.node.VersionInfo
 import net.corda.node.internal.cordapp.CordappLoader
 import net.corda.node.serialization.KryoServerSerializationScheme
-import net.corda.node.serialization.NodeClock
 import net.corda.node.services.RPCUserService
 import net.corda.node.services.RPCUserServiceImpl
 import net.corda.node.services.api.SchemaService
@@ -25,7 +24,7 @@ import net.corda.node.services.messaging.MessagingService
 import net.corda.node.services.messaging.NodeMessagingClient
 import net.corda.node.utilities.AddressUtils
 import net.corda.node.utilities.AffinityExecutor
-import net.corda.node.utilities.TestClock
+import net.corda.node.utilities.DemoClock
 import net.corda.nodeapi.ArtemisMessagingComponent
 import net.corda.nodeapi.internal.ShutdownHook
 import net.corda.nodeapi.internal.addShutdownHook
@@ -67,7 +66,7 @@ open class Node(configuration: NodeConfiguration,
         }
 
         private fun createClock(configuration: NodeConfiguration): Clock {
-            return if (configuration.useTestClock) TestClock() else NodeClock()
+            return (if (configuration.useTestClock) ::DemoClock else ::SimpleClock)(Clock.systemUTC())
         }
 
         private val sameVmNodeCounter = AtomicInteger()

node/src/main/kotlin/net/corda/node/utilities/DemoClock.kt

@@ -0,0 +1,23 @@
+package net.corda.node.utilities
+
+import net.corda.core.internal.until
+import net.corda.node.internal.MutableClock
+import java.time.Clock
+import java.time.LocalDate
+import javax.annotation.concurrent.ThreadSafe
+
+/** A [Clock] that can have the date advanced for use in demos. */
+@ThreadSafe
+class DemoClock(delegateClock: Clock) : MutableClock(delegateClock) {
+    @Synchronized
+    fun updateDate(date: LocalDate): Boolean {
+        val currentDate = LocalDate.now(this)
+        if (currentDate.isBefore(date)) {
+            // It's ok to increment
+            delegateClock = Clock.offset(delegateClock, currentDate.atStartOfDay() until date.atStartOfDay())
+            notifyMutationObservers()
+            return true
+        }
+        return false
+    }
+}

samples/irs-demo/cordapp/src/test/kotlin/net/corda/irs/flows/UpdateBusinessDayFlow.kt

@@ -7,11 +7,10 @@ import net.corda.core.flows.InitiatingFlow
 import net.corda.core.flows.StartableByRPC
 import net.corda.core.identity.Party
 import net.corda.core.flows.*
-import net.corda.core.node.NodeInfo
 import net.corda.core.serialization.CordaSerializable
 import net.corda.core.utilities.ProgressTracker
 import net.corda.core.utilities.unwrap
-import net.corda.node.utilities.TestClock
+import net.corda.node.utilities.DemoClock
 import java.time.LocalDate
 
 /**
@@ -27,7 +26,7 @@ object UpdateBusinessDayFlow {
         @Suspendable
         override fun call() {
             val message = otherPartySession.receive<UpdateBusinessDayMessage>().unwrap { it }
-            (serviceHub.clock as TestClock).updateDate(message.date)
+            (serviceHub.clock as DemoClock).updateDate(message.date)
         }
     }
 

samples/network-visualiser/src/main/kotlin/net/corda/netmap/simulation/Simulation.kt

@@ -110,7 +110,8 @@ abstract class Simulation(val networkSendManuallyPumped: Boolean,
     init {
         // Advance node clocks when current time is changed
         dateChanges.subscribe {
-            clocks.setTo(currentDateAndTime.toInstant(ZoneOffset.UTC))
+            val instant = currentDateAndTime.toInstant(ZoneOffset.UTC)
+            clocks.forEach { it.setTo(instant) }
         }
     }
 

testing/node-driver/src/main/kotlin/net/corda/testing/node/MockNode.kt

@@ -51,6 +51,7 @@ import java.math.BigInteger
 import java.nio.file.Path
 import java.security.KeyPair
 import java.security.PublicKey
+import java.time.Clock
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.AtomicInteger
 
@@ -217,7 +218,7 @@ class MockNetwork(defaultParameters: MockNetworkParameters = MockNetworkParamete
 
     open class MockNode(args: MockNodeArgs) : AbstractNode(
             args.config,
-            TestClock(),
+            TestClock(Clock.systemUTC()),
             MOCK_VERSION_INFO,
             CordappLoader.createDefaultWithTestPackages(args.config, args.network.cordappPackages),
             args.network.busyLatch

testing/node-driver/src/main/kotlin/net/corda/testing/node/TestClock.kt

@@ -1,30 +1,16 @@
 package net.corda.testing.node
 
 import net.corda.core.internal.until
-import net.corda.core.serialization.SerializeAsToken
-import net.corda.core.serialization.SerializeAsTokenContext
-import net.corda.core.serialization.SingletonSerializationToken.Companion.singletonSerializationToken
 import net.corda.node.internal.MutableClock
 import java.time.Clock
 import java.time.Duration
 import java.time.Instant
-import java.time.ZoneId
 import javax.annotation.concurrent.ThreadSafe
 
-
-/**
- * A [Clock] that can have the time advanced for use in testing.
- */
+/** A [Clock] that can have the time advanced for use in testing. */
 @ThreadSafe
-class TestClock(private var delegateClock: Clock = Clock.systemUTC()) : MutableClock(), SerializeAsToken {
-
-    private val token = singletonSerializationToken(javaClass)
-
-    override fun toToken(context: SerializeAsTokenContext) = token.registerWithContext(context, this)
-
-    /**
-     * Advance this [Clock] by the specified [Duration] for testing purposes.
-     */
+class TestClock(delegateClock: Clock) : MutableClock(delegateClock) {
+    /** Advance this [Clock] by the specified [Duration] for testing purposes. */
     @Synchronized
     fun advanceBy(duration: Duration) {
         delegateClock = offset(delegateClock, duration)
@@ -33,28 +19,8 @@ class TestClock(private var delegateClock: Clock = Clock.systemUTC()) : MutableC
 
     /**
      * Move this [Clock] to the specified [Instant] for testing purposes.
-     *
      * This will only be approximate due to the time ticking away, but will be some time shortly after the requested [Instant].
      */
     @Synchronized
     fun setTo(newInstant: Instant) = advanceBy(instant() until newInstant)
-
-    @Synchronized override fun instant(): Instant {
-        return delegateClock.instant()
-    }
-
-    @Deprecated("Do not use this. Instead seek to use ZonedDateTime methods.", level = DeprecationLevel.ERROR)
-    override fun withZone(zone: ZoneId): Clock {
-        throw UnsupportedOperationException("Tokenized clock does not support withZone()")
-    }
-
-    @Synchronized override fun getZone(): ZoneId {
-        return delegateClock.zone
-    }
 }
-
-/**
- * A helper method to set several [TestClock]s to approximately the same time.  The clocks may drift by the time it
- * takes for each [TestClock] to have it's time set and any observers to execute.
- */
-fun Iterable<TestClock>.setTo(instant: Instant) = this.forEach { it.setTo(instant) }