Add ability to set custom initiating flows for NotaryService (#6832)

core/src/main/kotlin/net/corda/core/internal/notary/NotaryService.kt

@@ -14,6 +14,14 @@ abstract class NotaryService : SingletonSerializeAsToken() {
     abstract val services: ServiceHub
     abstract val notaryIdentityKey: PublicKey
 
+    /**
+     * Mapping between @InitiatingFlow classes and factory methods that produce responder flows.
+     * Can be overridden in case of advanced notary service that serves both custom and standard flows.
+     */
+    open val initiatingFlows = mapOf(
+            NotaryFlow.Client::class to ::createServiceFlow
+    )
+
     /**
      * Interfaces for the request and result formats of queries supported by notary services. To
      * implement a new query, you must:

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

@@ -189,7 +189,6 @@ import java.util.concurrent.TimeUnit.SECONDS
 import java.util.function.Consumer
 import javax.persistence.EntityManager
 import javax.sql.DataSource
-import kotlin.collections.ArrayList
 
 /**
  * A base node implementation that can be customised either for production (with real implementations that do real
@@ -1026,13 +1025,23 @@ abstract class AbstractNode<S>(val configuration: NodeConfiguration,
             service.run {
                 tokenize()
                 runOnStop += ::stop
-                flowManager.registerInitiatedCoreFlowFactory(NotaryFlow.Client::class, ::createServiceFlow)
+                registerInitiatingFlows()
                 start()
             }
             return service
         }
     }
 
+    private fun NotaryService.registerInitiatingFlows() {
+        if (configuration.notary?.enableOverridableFlows == true) {
+            initiatingFlows.forEach { (flow, factory) ->
+                flowManager.registerInitiatedCoreFlowFactory(flow, factory)
+            }
+        } else {
+            flowManager.registerInitiatedCoreFlowFactory(NotaryFlow.Client::class, ::createServiceFlow)
+        }
+    }
+
     protected open fun makeKeyManagementService(identityService: PersistentIdentityService): KeyManagementServiceInternal {
         // Place the long term identity key in the KMS. Eventually, this is likely going to be separated again because
         // the KMS is meant for derived temporary keys used in transactions, and we're not supposed to sign things with

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

@@ -166,7 +166,8 @@ data class NotaryConfig(
         /** Notary implementation-specific configuration parameters. */
         val extraConfig: Config? = null,
         val raft: RaftConfig? = null,
-        val bftSMaRt: BFTSmartConfig? = null
+        val bftSMaRt: BFTSmartConfig? = null,
+        val enableOverridableFlows: Boolean? = null
 )
 
 /**

node/src/main/kotlin/net/corda/node/services/config/schema/v1/ConfigSections.kt

@@ -208,10 +208,22 @@ internal object NotaryConfigSpec : Configuration.Specification<NotaryConfig>("No
     private val extraConfig by nestedObject().map(ConfigObject::toConfig).optional()
     private val raft by nested(RaftConfigSpec).optional()
     private val bftSMaRt by nested(BFTSmartConfigSpec).optional()
+    private val enableOverridableFlows by boolean().optional()
 
     override fun parseValid(configuration: Config, options: Configuration.Options): Valid<NotaryConfig> {
         val config = configuration.withOptions(options)
-        return valid(NotaryConfig(config[validating], config[serviceLegalName], config[className], config[etaMessageThresholdSeconds], config[extraConfig], config[raft], config[bftSMaRt]))
+        return valid(
+                NotaryConfig(
+                        config[validating],
+                        config[serviceLegalName],
+                        config[className],
+                        config[etaMessageThresholdSeconds],
+                        config[extraConfig],
+                        config[raft],
+                        config[bftSMaRt],
+                        config[enableOverridableFlows]
+                )
+        )
     }
 }