CORE-18320 Support for a flow specific timeout on Flow Message API

application/src/main/java/net/corda/v5/application/messaging/FlowMessaging.java

@@ -9,6 +9,7 @@
 import net.corda.v5.base.types.MemberX500Name;
 import org.jetbrains.annotations.NotNull;
 
+import java.time.Duration;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -106,6 +107,29 @@ public interface FlowMessaging {
     @NotNull
     FlowSession initiateFlow(@NotNull MemberX500Name x500Name, boolean requireClose);
 
+    /**
+     * Creates a communication session with a counterparty's {@link ResponderFlow}. Subsequently, you may send/receive using
+     * this session object. Note that this function does not communicate in itself. The counter-flow will be kicked off
+     * by the first send/receive.
+     * <p>
+     * Initiated flows are initiated with context based on the context of the initiating flow at the point in time this
+     * method is called. The context of the initiating flow is snapshotted by the returned session. Altering the flow
+     * context has no effect on the context of the session after this point, and therefore it has no effect on the
+     * context of the initiated flow either.
+     *
+     * @param x500Name The X500 name of the member to communicate with.
+     * @param requireClose When set to true, the initiated party will send a close message after calling FlowSession.close()
+     *                     and the initiating party will suspend and wait to receive the message when they call FlowSession.close().
+     *                     When set to false the session is marked as terminated immediately when close() is called.
+     * @param sessionTimeout The duration that Corda waits when no message has been received from a counterparty before
+     *                       causing the session to error. If set to `null`, value set in Corda Configuration will be used.
+     *
+     * @return The session.
+     */
+    @Suspendable
+    @NotNull
+    FlowSession initiateFlow(@NotNull MemberX500Name x500Name, boolean requireClose, Duration sessionTimeout);
+
     /**
      * Creates a communication session with another member. Subsequently, you may send/receive using this session object.
      * Note that this function does not communicate in itself. The counter-flow will be kicked off by the first
@@ -189,6 +213,57 @@ public interface FlowMessaging {
     @NotNull
     FlowSession initiateFlow(@NotNull MemberX500Name x500Name, boolean requireClose, @NotNull FlowContextPropertiesBuilder flowContextPropertiesBuilder);
 
+    /**
+     * Creates a communication session with another member. Subsequently, you may send/receive using this session object.
+     * Note that this function does not communicate in itself. The counter-flow will be kicked off by the first
+     * send/receive.
+     * <p>
+     * This overload takes a builder of context properties. Any properties set or modified against the context passed to
+     * this builder will be propagated to initiated flows and all that flow's initiated flows and sub flows down the
+     * stack. The properties passed to the builder are pre-populated with the current flow context properties, see
+     * {@link FlowContextProperties}. Altering the current flow context has no effect on the context of the session after the
+     * builder is applied and the session returned by this method, and therefore it has no effect on the context of the
+     * initiated flow either.
+     * <p>
+     * Example of use in Kotlin.
+     * ```Kotlin
+     * val flowSession = flowMessaging.initiateFlow(virtualNodeName) { flowContextProperties ->
+     *      flowContextProperties["key"] = "value"
+     * }
+     * ```
+     * Example of use in Java.
+     * ```Java
+     * FlowSession flowSession = flowMessaging.initiateFlow(virtualNodeName, (flowContextProperties) -> {
+     *      flowContextProperties.put("key", "value");
+     * });
+     * ```
+     *
+     * @param x500Name The X500 name of the member to communicate with.
+     * @param requireClose When set to true, the initiated party will send a close message after calling FlowSession.close()
+     *                     and the initiating party will suspend and wait to receive the message when they call FlowSession.close().
+     *                     When set to false the session is marked as terminated immediately when close() is called.
+     * @param sessionTimeout The duration that Corda waits when no message has been received from a counterparty before
+     *                       causing the session to error. If set to `null`, value set in Corda Configuration will
+     *                       be used.
+     * @param flowContextPropertiesBuilder A builder of context properties.
+     *
+     * @return The session.
+     *
+     * @throws IllegalArgumentException if the builder tries to set a property for which a platform property already
+     * exists or if the key is prefixed by {@link FlowContextProperties#CORDA_RESERVED_PREFIX}. See also
+     * {@link FlowContextProperties}. Any other
+     * exception thrown by the builder will also be thrown through here and should be avoided in the provided
+     * implementation, see {@link FlowContextPropertiesBuilder}.
+     */
+    @Suspendable
+    @NotNull
+    FlowSession initiateFlow(
+            @NotNull MemberX500Name x500Name,
+            boolean requireClose,
+            Duration sessionTimeout,
+            @NotNull FlowContextPropertiesBuilder flowContextPropertiesBuilder
+    );
+
     /**
      * Suspends until a message has been received for each session in the specified {@code sessions}.
      * <p>

application/src/test/java/net/corda/v5/application/messaging/FlowMessagingJavaApiTest.java

@@ -39,7 +39,7 @@ public void initiateFlowPartyWithBuilder() {
     @Test
     public void initiateFlowPartyWithBuilderRequireCloseTrue() {
         final MemberX500Name counterparty = new MemberX500Name("Alice Corp", "LDN", "GB");
-        when(flowMessaging.initiateFlow(eq(counterparty), eq(true), any())).thenReturn(flowSession);
+        when(flowMessaging.initiateFlow(eq(counterparty), eq(true), any(FlowContextPropertiesBuilder.class))).thenReturn(flowSession);
 
         FlowSession result = flowMessaging.initiateFlow(counterparty, true, (contextProperties) -> contextProperties.put("key", "value"));
 
@@ -50,7 +50,7 @@ public void initiateFlowPartyWithBuilderRequireCloseTrue() {
     @Test
     public void initiateFlowPartyWithBuilderRequireCloseFalse() {
         final MemberX500Name counterparty = new MemberX500Name("Alice Corp", "LDN", "GB");
-        when(flowMessaging.initiateFlow(eq(counterparty), eq(false), any())).thenReturn(flowSession);
+        when(flowMessaging.initiateFlow(eq(counterparty), eq(false), any(FlowContextPropertiesBuilder.class))).thenReturn(flowSession);
 
         FlowSession result = flowMessaging.initiateFlow(counterparty, false, (contextProperties) -> contextProperties.put("key", "value"));
 

data/avro-schema/src/main/resources/avro/net/corda/data/flow/state/session/SessionState.avsc

@@ -34,6 +34,15 @@
       "type": "boolean",
       "doc": "True if the user has set requireClose to be true when calling initiate flow. False otherwise."
     },
+    {
+      "name": "sessionTimeout",
+      "type": [
+        "null",
+        "int"
+      ],
+      "default": null,
+      "doc": "The length of time in milliseconds that Corda waits when no message has been received from a counterparty before causing the session to error. If not set, value from configuration is used."
+    },
     {
       "name": "receivedEventsState",
       "type": "net.corda.data.flow.state.session.SessionProcessState",

gradle.properties

@@ -5,7 +5,7 @@ cordaProductVersion = 5.2.0
 # NOTE: update this each time this module contains a breaking change
 ## NOTE: currently this is a top level revision, so all API versions will line up, but this could be moved to
 ##   a per module property in which case module versions can change independently.
-cordaApiRevision = 12
+cordaApiRevision = 13
 
 # Main
 kotlinVersion = 1.8.21