[CORDA-3304] Introduce max number of retries per invocation for reconnecting rpc

.idea/codeStyles/Project.xml

@@ -3,6 +3,10 @@
     <option name="LINE_SEPARATOR" value="&#10;" />
     <option name="RIGHT_MARGIN" value="140" />
     <option name="SOFT_MARGINS" value="140" />
+    <JavaCodeStyleSettings>
+      <option name="CLASS_COUNT_TO_USE_IMPORT_ON_DEMAND" value="999" />
+      <option name="NAMES_COUNT_TO_USE_IMPORT_ON_DEMAND" value="999" />
+    </JavaCodeStyleSettings>
     <JetCodeStyleSettings>
       <option name="PACKAGES_TO_USE_STAR_IMPORTS">
         <value>
@@ -11,6 +15,8 @@
           <package name="tornadofx" withSubpackages="false" static="false" />
         </value>
       </option>
+      <option name="NAME_COUNT_TO_USE_STAR_IMPORT" value="2147483647" />
+      <option name="NAME_COUNT_TO_USE_STAR_IMPORT_FOR_MEMBERS" value="2147483647" />
       <option name="CONTINUATION_INDENT_IN_PARAMETER_LISTS" value="true" />
       <option name="CONTINUATION_INDENT_IN_ARGUMENT_LISTS" value="true" />
       <option name="CONTINUATION_INDENT_FOR_EXPRESSION_BODIES" value="true" />

.idea/codeStyles/codeStyleConfig.xml

@@ -1,6 +1,5 @@
 <component name="ProjectCodeStyleConfiguration">
   <state>
     <option name="USE_PER_PROJECT_SETTINGS" value="true" />
-    <option name="PREFERRED_PROJECT_CODE_STYLE" value="Default" />
   </state>
 </component>

client/rpc/src/main/kotlin/net/corda/client/rpc/CordaRPCClient.kt

@@ -282,10 +282,12 @@ open class CordaRPCClientConfiguration @JvmOverloads constructor(
  *
  * @param onDisconnect implement this callback to perform logic when the RPC disconnects on connection disconnect
  * @param onReconnect implement this callback to perform logic when the RPC has reconnected after connection disconnect
+ * @param maxRetries the maximum number of retries per each individual RPC call. A negative number indicates infinite number of retries.
+ *                   The default value is 5.
  */
-class GracefulReconnect(val onDisconnect: () -> Unit = {}, val onReconnect: () -> Unit = {}) {
+class GracefulReconnect(val onDisconnect: () -> Unit = {}, val onReconnect: () -> Unit = {}, val maxRetries: Int = 5) {
-    constructor(onDisconnect: Runnable, onReconnect: Runnable ) :
+    constructor(onDisconnect: Runnable, onReconnect: Runnable, maxRetries: Int = 5) :
-        this(onDisconnect = { onDisconnect.run() }, onReconnect = { onReconnect.run() })
+        this(onDisconnect = { onDisconnect.run() }, onReconnect = { onReconnect.run() }, maxRetries = maxRetries)
 }
 
 /**

client/rpc/src/main/kotlin/net/corda/client/rpc/RPCException.kt

@@ -9,6 +9,13 @@ open class RPCException(message: String?, cause: Throwable?) : CordaRuntimeExcep
     constructor(msg: String) : this(msg, null)
 }
 
+/**
+ * Thrown to indicate an RPC operation has been retried for the [maxNumberOfRetries] unsuccessfully.
+ * @param maxNumberOfRetries the number of retries that had been performed.
+ * @param cause the cause of the last failed attempt.
+ */
+class MaxRpcRetryException(maxNumberOfRetries: Int, cause: Throwable?): RPCException("Max number of retries ($maxNumberOfRetries) was reached.", cause)
+
 /**
  * Signals that the underlying [RPCConnection] dropped.
  */

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

@@ -20,6 +20,7 @@ import net.corda.nodeapi.exceptions.RejectedCommandException
 import org.apache.activemq.artemis.api.core.ActiveMQConnectionTimedOutException
 import org.apache.activemq.artemis.api.core.ActiveMQSecurityException
 import org.apache.activemq.artemis.api.core.ActiveMQUnBlockedException
+import java.lang.RuntimeException
 import java.lang.reflect.InvocationHandler
 import java.lang.reflect.InvocationTargetException
 import java.lang.reflect.Method
@@ -55,7 +56,7 @@ class ReconnectingCordaRPCOps private constructor(
             username: String,
             password: String,
             rpcConfiguration: CordaRPCClientConfiguration,
-            gracefulReconnect: GracefulReconnect? = null,
+            gracefulReconnect: GracefulReconnect = GracefulReconnect(),
             sslConfiguration: ClientRpcSslOptions? = null,
             classLoader: ClassLoader? = null,
             observersPool: ExecutorService
@@ -117,7 +118,7 @@ class ReconnectingCordaRPCOps private constructor(
             val rpcConfiguration: CordaRPCClientConfiguration,
             val sslConfiguration: ClientRpcSslOptions? = null,
             val classLoader: ClassLoader?,
-            val gracefulReconnect: GracefulReconnect? = null,
+            val gracefulReconnect: GracefulReconnect = GracefulReconnect(),
             val observersPool: ExecutorService
     ) : RPCConnection<CordaRPCOps> {
         private var currentRPCConnection: CordaRPCConnection? = null
@@ -147,13 +148,13 @@ class ReconnectingCordaRPCOps private constructor(
             // First one to get here gets to do all the reconnect logic, including calling onDisconnect and onReconnect. This makes sure
             // that they're only called once per reconnect.
             currentState = DIED
-            gracefulReconnect?.onDisconnect?.invoke()
+            gracefulReconnect.onDisconnect.invoke()
             //TODO - handle error cases
             log.error("Reconnecting to ${this.nodeHostAndPorts} due to error: ${e.message}")
             log.debug("", e)
             connect()
             previousConnection?.forceClose()
-            gracefulReconnect?.onReconnect?.invoke()
+            gracefulReconnect.onReconnect.invoke()
         }
         /**
          * Called on external error.
@@ -249,9 +250,16 @@ class ReconnectingCordaRPCOps private constructor(
             }
         }
 
-        private fun doInvoke(method: Method, args: Array<out Any>?): Any? {
+        /**
-            // will stop looping when [method.invoke] succeeds
+         * This method retries the invoked operation in a loop by re-establishing the connection when there is a problem
-            while (true) {
+         * and checking if the [maxNumberOfRetries] has been exhausted.
+         *
+         * A negative number for [maxNumberOfRetries] means an unlimited number of retries will be performed.
+         */
+        private fun doInvoke(method: Method, args: Array<out Any>?, maxNumberOfRetries: Int): Any? {
+            var remainingRetries = maxNumberOfRetries
+            var lastException: Throwable? = null
+            while (remainingRetries != 0) {
                 try {
                     log.debug { "Invoking RPC $method..." }
                     return method.invoke(reconnectingRPCConnection.proxy, *(args ?: emptyArray())).also {
@@ -280,15 +288,19 @@ class ReconnectingCordaRPCOps private constructor(
                             checkIfIsStartFlow(method, e)
                         }
                     }
+                    lastException = e.targetException
+                    remainingRetries--
                 }
             }
+
+            throw MaxRpcRetryException(maxNumberOfRetries, lastException)
         }
 
         override fun invoke(proxy: Any, method: Method, args: Array<out Any>?): Any? {
             return when (method.returnType) {
                 DataFeed::class.java -> {
                     // Intercept the data feed methods and return a ReconnectingObservable instance
-                    val initialFeed: DataFeed<Any, Any?> = uncheckedCast(doInvoke(method, args))
+                    val initialFeed: DataFeed<Any, Any?> = uncheckedCast(doInvoke(method, args, reconnectingRPCConnection.gracefulReconnect.maxRetries))
                     val observable = ReconnectingObservable(reconnectingRPCConnection, initialFeed) {
                         // This handles reconnecting and creates new feeds.
                         uncheckedCast(this.invoke(reconnectingRPCConnection.proxy, method, args))
@@ -296,7 +308,7 @@ class ReconnectingCordaRPCOps private constructor(
                     initialFeed.copy(updates = observable)
                 }
                 // TODO - add handlers for Observable return types.
-                else -> doInvoke(method, args)
+                else -> doInvoke(method, args, reconnectingRPCConnection.gracefulReconnect.maxRetries)
             }
         }
     }