Merge pull request #546 from corda/merges/march-13-14-55

Merges: March 13 14:55

client/rpc/src/integration-test/kotlin/net/corda/client/rpc/RPCStabilityTests.kt

@@ -326,6 +326,46 @@ class RPCStabilityTests {
         }
     }
 
+    interface NoOps : RPCOps {
+        fun subscribe(): Observable<Nothing>
+    }
+
+    @Test
+    fun `observables error when connection breaks`() {
+        rpcDriver {
+            val ops = object : NoOps {
+                override val protocolVersion = 0
+                override fun subscribe(): Observable<Nothing> {
+                    return PublishSubject.create<Nothing>()
+                }
+            }
+            val serverFollower = shutdownManager.follower()
+            val serverPort = startRpcServer<NoOps>(ops = ops).getOrThrow().broker.hostAndPort!!
+            serverFollower.unfollow()
+
+            val clientConfiguration = RPCClientConfiguration.default.copy(connectionRetryInterval = 500.millis, maxReconnectAttempts = 1)
+            val clientFollower = shutdownManager.follower()
+            val client = startRpcClient<NoOps>(serverPort, configuration = clientConfiguration).getOrThrow()
+            clientFollower.unfollow()
+
+            var terminateHandlerCalled = false
+            var errorHandlerCalled = false
+            val subscription = client.subscribe()
+                     .doOnTerminate{ terminateHandlerCalled = true }
+                     .doOnError { errorHandlerCalled = true }
+                     .subscribe()
+
+            serverFollower.shutdown()
+            Thread.sleep(100)
+
+            assertTrue(terminateHandlerCalled)
+            assertTrue(errorHandlerCalled)
+            assertTrue(subscription.isUnsubscribed)
+
+            clientFollower.shutdown() // Driver would do this after the new server, causing hang.
+        }
+    }
+
     interface ThreadOps : RPCOps {
         fun sendMessage(id: Int, msgNo: Int): String
     }

client/rpc/src/main/kotlin/net/corda/client/rpc/internal/RPCClientProxyHandler.kt

@@ -418,6 +418,14 @@ class RPCClientProxyHandler(
         when (event) {
             FailoverEventType.FAILURE_DETECTED -> {
                 log.warn("RPC server unavailable. RPC calls are being buffered.")
+                log.warn("Terminating observables.")
+                val m = observableContext.observableMap.asMap()
+                m.keys.forEach { k ->
+                    observationExecutorPool.run(k) {
+                        m[k]?.onError(RPCException("Connection failure detected."))
+                    }
+                }
+                observableContext.observableMap.invalidateAll()
             }
 
             FailoverEventType.FAILOVER_COMPLETED -> {
@@ -432,8 +440,6 @@ class RPCClientProxyHandler(
                         "will throw an RPCException.")
                 rpcReplyMap.forEach { id, replyFuture ->
                     replyFuture.setException(RPCException("Could not re-connect to RPC server. Failover failed."))
-                    val observable = observableContext.observableMap.getIfPresent(id)
-                    observable?.onError(RPCException("Could not re-connect to RPC server. Failover failed."))
                 }
                 outgoingRequestBuffer.clear()
                 rpcReplyMap.clear()
@@ -517,4 +523,4 @@ object RpcClientObservableSerializer : Serializer<Observable<*>>() {
         val rpcRequestOrObservableId = kryo.context[RPCApi.RpcRequestOrObservableIdKey] as InvocationId
         return observableContext.callSiteMap?.get(rpcRequestOrObservableId)
     }
-}
+}

core/build.gradle

@@ -118,8 +118,6 @@ dependencies {
     // This may be temporary until we experiment with other ways to do on-the-fly contract specialisation via an API.
     compile "org.apache.commons:commons-jexl3:3.0"
     compile 'commons-lang:commons-lang:2.6'
-    // For JSON
-    compile "com.fasterxml.jackson.core:jackson-databind:${jackson_version}"
 
     // Java ed25519 implementation. See https://github.com/str4d/ed25519-java/
     compile "net.i2p.crypto:eddsa:$eddsa_version"

docs/source/api-flows.rst

@@ -683,6 +683,12 @@ We then update the progress tracker's current step as we progress through the fl
         :end-before: DOCEND 18
         :dedent: 12
 
+HTTP and database calls
+-----------------------
+HTTP, database and other calls to external resources are allowed in flows. However, their support is currently limited:
+
+* The call must be executed in a BLOCKING way. Flows don't currently support suspending to await the response to a call to an external resource
+* The call must be idempotent. If the flow fails and has to restart from a checkpoint, the call will also be replayed
 
 Concurrency, Locking and Waiting
 --------------------------------

docs/source/key-concepts-transactions.rst

@@ -105,7 +105,7 @@ will not be valid, and will not be accepted as inputs to subsequent transactions
 
 Other transaction components
 ----------------------------
-As well as input states and output states, transactions may contain:
+As well as input states and output states, transactions contain:
 
 * Commands
 * Attachments
@@ -178,4 +178,4 @@ In some cases, we want a transaction proposed to only be approved during a certa
 * A bond may only be redeemed before its expiry date
 
 In such cases, we can add a *time-window* to the transaction. Time-windows specify the time window during which the
-transaction can be committed. We discuss time-windows in the section on :doc:`key-concepts-time-windows`.
+transaction can be committed. We discuss time-windows in the section on :doc:`key-concepts-time-windows`.

docs/source/running-a-node.rst

@@ -8,29 +8,15 @@ Running nodes locally
 
 There are several ways to run a Corda node locally for testing purposes.
 
-Starting all nodes at once
---------------------------
-
-.. note:: ``runnodes`` is a shell script (or batch file on Windows) that is generated by ``deployNodes`` to allow you
-   to quickly start up all nodes and their webservers. ``runnodes`` should only be used for testing purposes.
-
-Start the nodes with ``runnodes`` by running the following command from the root of the project:
-
-* Linux/macOS: ``build/nodes/runnodes``
-* Windows: ``call build\nodes\runnodes.bat``
-
-.. warning:: On macOS, do not click/change focus until all the node terminal windows have opened, or some processes may
-   fail to start.
-
-If you receive an ``OutOfMemoryError`` exception when interacting with the nodes, you need to increase the amount of
-Java heap memory available to them, which you can do when running them individually. See
-:ref:`starting-an-individual-corda-node`.
+Starting a Corda node using DemoBench
+-------------------------------------
+See the instructions in :doc:`demobench`.
 
 .. _starting-an-individual-corda-node:
 
-Starting an individual Corda node
----------------------------------
-Run the node by opening a terminal window in the node's folder and running:
+Starting a Corda node from the command line
+-------------------------------------------
+Run a node by opening a terminal window in the node's folder and running:
 
 .. code-block:: shell
 
@@ -57,8 +43,8 @@ Optionally run the node's webserver as well by opening a terminal window in the
 
 .. warning:: The node webserver is for testing purposes only and will be removed soon.
 
-Starting a node with remote debugging enabled
----------------------------------------------
+Enabling remote debugging
+~~~~~~~~~~~~~~~~~~~~~~~~~
 To enable remote debugging of the node, run the following from the terminal window:
 
 ``java -Dcapsule.jvm.args="-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=5005" -jar corda.jar``
@@ -73,4 +59,18 @@ To enable export of JMX metrics over HTTP via `Jolokia <https://jolokia.org/>`_,
 
 This command line will start the node with JMX metrics accessible via HTTP on port 7005.
 
-See :ref:`Monitoring your node <jolokia_ref>` for further details.
+See :ref:`Monitoring your node <jolokia_ref>` for further details.
+
+Starting all nodes at once from the command line
+------------------------------------------------
+If you created your nodes using ``deployNodes``, a ``runnodes`` shell script (or batch file on Windows) will have been
+generated to allow you to quickly start up all nodes and their webservers. ``runnodes`` should only be used for testing
+purposes.
+
+Start the nodes with ``runnodes`` by running the following command from the root of the project:
+
+* Linux/macOS: ``build/nodes/runnodes``
+* Windows: ``call build\nodes\runnodes.bat``
+
+.. warning:: On macOS, do not click/change focus until all the node terminal windows have opened, or some processes may
+fail to start.

finance/build.gradle

@@ -42,6 +42,9 @@ dependencies {
     // TODO Remove this once we have app configs
     compile "com.typesafe:config:$typesafe_config_version"
 
+    // For JSON
+    compile "com.fasterxml.jackson.core:jackson-databind:${jackson_version}"
+
     testCompile project(':test-utils')
     testCompile project(path: ':core', configuration: 'testArtifacts')
     testCompile "junit:junit:$junit_version"
@@ -72,6 +75,12 @@ artifacts {
 
 jar {
     baseName 'corda-finance'
+    exclude "META-INF/*.DSA"
+    exclude "META-INF/*.RSA"
+    exclude "META-INF/*.SF"
+    exclude "META-INF/*.MF"
+    exclude "META-INF/LICENSE"
+    exclude "META-INF/NOTICE"
 }
 
 publish {

node-api/build.gradle

@@ -61,6 +61,8 @@ dependencies {
     // Pure-Java Snappy compression
     compile 'org.iq80.snappy:snappy:0.4'
 
+    compile "com.fasterxml.jackson.core:jackson-databind:$jackson_version"
+
     // Unit testing helpers.
     testCompile "junit:junit:$junit_version"
     testCompile "org.assertj:assertj-core:$assertj_version"

tools/shell/src/integration-test/kotlin/net/corda/tools/shell/InteractiveShellIntegrationTest.kt

@@ -31,6 +31,7 @@ import org.apache.activemq.artemis.api.core.ActiveMQSecurityException
 import org.assertj.core.api.Assertions.assertThat
 import org.assertj.core.api.Assertions.assertThatThrownBy
 import org.bouncycastle.util.io.Streams
+import org.junit.Ignore
 import org.junit.Test
 import kotlin.test.assertTrue
 
@@ -183,6 +184,7 @@ class InteractiveShellIntegrationTest {
         }
     }
 
+    @Ignore
     @Test
     fun `ssh run flows via standalone shell over ssl to node`() {
         val user = User("mark", "dadada", setOf(Permissions.startFlow<SSHServerTest.FlowICanRun>(),