Convert long lived services supporting protocol listeners to plugins

core/src/main/kotlin/com/r3corda/core/node/CordaPluginRegistry.kt

@@ -27,4 +27,12 @@ interface CordaPluginRegistry {
      * This is used to extend the white listed protocols that can be initiated from the ServiceHub invokeProtocolAsync method.
      */
     val requiredProtocols: Map<String, Set<String>>
+
+    /**
+     * List of additional long lived services to be hosted within the node.
+     * They are expected to have a single parameter constructor that takes a ServiceHubInternal as input.
+     * The ServiceHubInternal will be fully constructed before the plugin service is created and will
+     * allow access to the protocol factory and protocol initiation entry points there.
+     */
+    val servicePlugins: List<Class<*>>
 }

node/src/main/kotlin/com/r3corda/node/internal/AbstractNode.kt

@@ -17,6 +17,7 @@ import com.r3corda.core.protocols.ProtocolLogic
 import com.r3corda.core.protocols.ProtocolLogicRefFactory
 import com.r3corda.core.random63BitValue
 import com.r3corda.core.seconds
+import com.r3corda.core.serialization.SingletonSerializeAsToken
 import com.r3corda.core.serialization.deserialize
 import com.r3corda.core.serialization.serialize
 import com.r3corda.node.api.APIServer
@@ -60,7 +61,7 @@ import java.util.*
 // In theory the NodeInfo for the node should be passed in, instead, however currently this is constructed by the
 // AbstractNode. It should be possible to generate the NodeInfo outside of AbstractNode, so it can be passed in.
 abstract class AbstractNode(val dir: Path, val configuration: NodeConfiguration, val networkMapService: NodeInfo?,
-                            val advertisedServices: Set<ServiceType>, val platformClock: Clock) {
+                            val advertisedServices: Set<ServiceType>, val platformClock: Clock): SingletonSerializeAsToken() {
     companion object {
         val PRIVATE_KEY_FILE_NAME = "identity-private-key"
         val PUBLIC_IDENTITY_FILE_NAME = "identity-public"
@@ -124,6 +125,11 @@ abstract class AbstractNode(val dir: Path, val configuration: NodeConfiguration,
     lateinit var api: APIServer
     lateinit var scheduler: SchedulerService
     lateinit var protocolLogicFactory: ProtocolLogicRefFactory
+    var customServices: List<Any> = emptyList()
+    inline fun <reified T: Any>getCustomService() : T {
+        return customServices.single{ x-> x is T } as T
+    }
+
     var isPreviousCheckpointsPresent = false
         private set
 
@@ -151,7 +157,7 @@ abstract class AbstractNode(val dir: Path, val configuration: NodeConfiguration,
         net = makeMessagingService()
         netMapCache = InMemoryNetworkMapCache(net)
         wallet = NodeWalletService(services)
-        makeInterestRatesOracleService()
+
         identity = makeIdentityService()
         // 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
@@ -159,17 +165,18 @@ abstract class AbstractNode(val dir: Path, val configuration: NodeConfiguration,
         keyManagement = E2ETestKeyManagementService(setOf(storage.myLegalIdentityKey))
         api = APIServerImpl(this)
         scheduler = NodeSchedulerService(services)
-        smm = StateMachineManager(services,
-                listOf(storage, net, wallet, keyManagement, identity, platformClock, scheduler, interestRatesService),
-                checkpointStorage,
-                serverThread)
 
         protocolLogicFactory = initialiseProtocolLogicFactory()
 
-        // This object doesn't need to be referenced from this class because it registers handlers on the network
+        val tokenizableServices = mutableListOf(storage, net, wallet, keyManagement, identity, platformClock, scheduler)
-        // service and so that keeps it from being collected.
+
-        DataVendingService(net, services)
+        buildPluginServices(tokenizableServices)
-        NotaryChangeService(net, smm, services.networkMapCache)
+
+
+        smm = StateMachineManager(services,
+                listOf(tokenizableServices),
+                checkpointStorage,
+                serverThread)
 
         buildAdvertisedServices()
 
@@ -199,6 +206,20 @@ abstract class AbstractNode(val dir: Path, val configuration: NodeConfiguration,
         return ProtocolLogicRefFactory(protocolWhitelist)
     }
 
+    private fun buildPluginServices(tokenizableServices: MutableList<Any>) {
+        val pluginServices = pluginRegistries.flatMap { x -> x.servicePlugins }
+        val serviceList = mutableListOf<Any>()
+        for (serviceClass in pluginServices) {
+            val service = serviceClass.getConstructor(ServiceHubInternal::class.java).newInstance(services)
+            serviceList.add(service)
+            tokenizableServices.add(service)
+            if(service is AcceptsFileUpload) {
+                _servicesThatAcceptUploads += service
+            }
+        }
+        customServices = serviceList
+    }
+
 
     /**
      * Run any tasks that are needed to ensure the node is in a correct state before running start().
@@ -280,14 +301,6 @@ abstract class AbstractNode(val dir: Path, val configuration: NodeConfiguration,
         }
     }
 
-    lateinit var interestRatesService: NodeInterestRates.Service
-
-    open protected fun makeInterestRatesOracleService() {
-        // TODO: Once the service has data, automatically register with the network map service (once built).
-        interestRatesService = NodeInterestRates.Service(this)
-        _servicesThatAcceptUploads += interestRatesService
-    }
-
     protected open fun makeIdentityService(): IdentityService {
         val service = InMemoryIdentityService()
         if (networkMapService != null)

node/src/main/kotlin/com/r3corda/node/internal/testing/IRSSimulation.kt

@@ -15,7 +15,6 @@ import com.r3corda.core.node.services.linearHeadsOfType
 import com.r3corda.core.node.services.testing.MockIdentityService
 import com.r3corda.core.random63BitValue
 import com.r3corda.core.success
-import com.r3corda.node.services.FixingSessionInitiationHandler
 import com.r3corda.node.services.network.InMemoryMessagingNetwork
 import com.r3corda.node.utilities.JsonSupport
 import com.r3corda.protocols.TwoPartyDealProtocol
@@ -40,11 +39,6 @@ class IRSSimulation(networkSendManuallyPumped: Boolean, runAsync: Boolean, laten
     private val executeOnNextIteration = Collections.synchronizedList(LinkedList<() -> Unit>())
 
     override fun startMainSimulation(): ListenableFuture<Unit> {
-
-        // TODO: until we have general session initiation
-        FixingSessionInitiationHandler.register(banks[0])
-        FixingSessionInitiationHandler.register(banks[1])
-
         val future = SettableFuture.create<Unit>()
 
         nodeAKey = banks[0].keyManagement.freshKey()

node/src/main/kotlin/com/r3corda/node/internal/testing/Simulation.kt

@@ -115,9 +115,10 @@ abstract class Simulation(val networkSendManuallyPumped: Boolean,
             }
 
             return object : SimulatedNode(dir, cfg, network, networkMapAddr, advertisedServices, id, keyPair) {
-                override fun makeInterestRatesOracleService() {
+                override fun start(): MockNetwork.MockNode {
-                    super.makeInterestRatesOracleService()
+                    super.start()
-                    interestRatesService.upload(javaClass.getResourceAsStream("example.rates.txt"))
+                    getCustomService<NodeInterestRates.Service>().upload(javaClass.getResourceAsStream("example.rates.txt"))
+                    return this
                 }
             }
         }

node/src/main/kotlin/com/r3corda/node/services/FixingSessionInitiationHandler.kt

@@ -1,22 +0,0 @@
-package com.r3corda.node.services
-
-import com.r3corda.core.serialization.deserialize
-import com.r3corda.node.internal.AbstractNode
-import com.r3corda.protocols.TwoPartyDealProtocol
-
-/**
- * This is a temporary handler required for establishing random sessionIDs for the [Fixer] and [Floater] as part of
- * running scheduled fixings for the [InterestRateSwap] contract.
- *
- * TODO: This will be replaced with the automatic sessionID / session setup work.
- */
-object FixingSessionInitiationHandler {
-
-    fun register(node: AbstractNode) {
-        node.net.addMessageHandler("${TwoPartyDealProtocol.FIX_INITIATE_TOPIC}.0") { msg, registration ->
-            val initiation = msg.data.deserialize<TwoPartyDealProtocol.FixingSessionInitiation>()
-            val protocol = TwoPartyDealProtocol.Fixer(initiation)
-            node.smm.add("fixings", protocol)
-        }
-    }
-}

node/src/main/kotlin/com/r3corda/node/services/NotaryChangeService.kt

@@ -1,30 +1,38 @@
 package com.r3corda.node.services
 
 import com.r3corda.core.messaging.Ack
-import com.r3corda.core.messaging.MessagingService
+import com.r3corda.core.node.CordaPluginRegistry
-import com.r3corda.core.node.services.NetworkMapCache
 import com.r3corda.node.services.api.AbstractNodeService
-import com.r3corda.node.services.statemachine.StateMachineManager
+import com.r3corda.node.services.api.ServiceHubInternal
 import com.r3corda.protocols.AbstractStateReplacementProtocol
 import com.r3corda.protocols.NotaryChangeProtocol
 
-/**
+
- * A service that monitors the network for requests for changing the notary of a state,
+object NotaryChange {
- * and immediately runs the [NotaryChangeProtocol] if the auto-accept criteria are met.
+    class Plugin : CordaPluginRegistry {
- */
+        override val webApis: List<Class<*>> = emptyList()
-class NotaryChangeService(net: MessagingService, val smm: StateMachineManager, networkMapCache: NetworkMapCache) : AbstractNodeService(net, networkMapCache) {
+        override val requiredProtocols: Map<String, Set<String>> = emptyMap()
-    init {
+        override val servicePlugins: List<Class<*>> = listOf(Service::class.java)
-        addMessageHandler(NotaryChangeProtocol.TOPIC,
-                { req: AbstractStateReplacementProtocol.Handshake -> handleChangeNotaryRequest(req) }
-        )
     }
 
-    private fun handleChangeNotaryRequest(req: AbstractStateReplacementProtocol.Handshake): Ack {
+    /**
-        val protocol = NotaryChangeProtocol.Acceptor(
+     * A service that monitors the network for requests for changing the notary of a state,
-                req.replyToParty,
+     * and immediately runs the [NotaryChangeProtocol] if the auto-accept criteria are met.
-                req.sessionID,
+     */
-                req.sessionIdForSend)
+    class Service(val services: ServiceHubInternal) : AbstractNodeService(services.networkService, services.networkMapCache) {
-        smm.add(NotaryChangeProtocol.TOPIC, protocol)
+        init {
-        return Ack
+            addMessageHandler(NotaryChangeProtocol.TOPIC,
+                    { req: AbstractStateReplacementProtocol.Handshake -> handleChangeNotaryRequest(req) }
+            )
+        }
+
+        private fun handleChangeNotaryRequest(req: AbstractStateReplacementProtocol.Handshake): Ack {
+            val protocol = NotaryChangeProtocol.Acceptor(
+                    req.replyToParty,
+                    req.sessionID,
+                    req.sessionIdForSend)
+            services.startProtocol(NotaryChangeProtocol.TOPIC, protocol)
+            return Ack
+        }
     }
 }

node/src/main/kotlin/com/r3corda/node/services/clientapi/FixingSessionInitiation.kt

@@ -0,0 +1,31 @@
+package com.r3corda.node.services.clientapi
+
+import com.r3corda.core.node.CordaPluginRegistry
+import com.r3corda.core.serialization.deserialize
+import com.r3corda.node.internal.AbstractNode
+import com.r3corda.node.services.api.ServiceHubInternal
+import com.r3corda.protocols.TwoPartyDealProtocol
+
+/**
+ * This is a temporary handler required for establishing random sessionIDs for the [Fixer] and [Floater] as part of
+ * running scheduled fixings for the [InterestRateSwap] contract.
+ *
+ * TODO: This will be replaced with the automatic sessionID / session setup work.
+ */
+object FixingSessionInitiation {
+    class Plugin: CordaPluginRegistry {
+        override val webApis: List<Class<*>> = emptyList()
+        override val requiredProtocols: Map<String, Set<String>> = emptyMap()
+        override val servicePlugins: List<Class<*>> = listOf(Service::class.java)
+    }
+
+    class Service(services: ServiceHubInternal) {
+        init {
+            services.networkService.addMessageHandler("${TwoPartyDealProtocol.FIX_INITIATE_TOPIC}.0") { msg, registration ->
+                val initiation = msg.data.deserialize<TwoPartyDealProtocol.FixingSessionInitiation>()
+                val protocol = TwoPartyDealProtocol.Fixer(initiation)
+                services.startProtocol("fixings", protocol)
+            }
+        }
+    }
+}

node/src/main/kotlin/com/r3corda/node/services/clientapi/NodeInterestRates.kt

@@ -13,9 +13,9 @@ import com.r3corda.core.node.CordaPluginRegistry
 import com.r3corda.core.node.services.ServiceType
 import com.r3corda.core.protocols.ProtocolLogic
 import com.r3corda.core.utilities.ProgressTracker
-import com.r3corda.node.internal.AbstractNode
 import com.r3corda.node.services.api.AbstractNodeService
 import com.r3corda.node.services.api.AcceptsFileUpload
+import com.r3corda.node.services.api.ServiceHubInternal
 import com.r3corda.node.utilities.FiberBox
 import com.r3corda.protocols.RatesFixProtocol
 import com.r3corda.protocols.ServiceRequestMessage
@@ -42,12 +42,23 @@ import javax.annotation.concurrent.ThreadSafe
  */
 object NodeInterestRates {
     object Type : ServiceType("corda.interest_rates")
+
+    /**
+     * Register the protocol that is used with the Fixing integration tests.
+     */
+    class Plugin : CordaPluginRegistry {
+        override val webApis: List<Class<*>> = emptyList()
+        override val requiredProtocols: Map<String, Set<String>> = mapOf(Pair(TwoPartyDealProtocol.FixingRoleDecider::class.java.name, setOf(Duration::class.java.name, StateRef::class.java.name)))
+        override val servicePlugins: List<Class<*>> = listOf(NodeInterestRates.Service::class.java)
+        override val staticServeDirs: Map<String, String> = emptyMap()
+    }
+
     /**
      * The Service that wraps [Oracle] and handles messages/network interaction/request scrubbing.
      */
-    class Service(node: AbstractNode) : AcceptsFileUpload, AbstractNodeService(node.services.networkService, node.services.networkMapCache) {
+    class Service(services: ServiceHubInternal) : AcceptsFileUpload, AbstractNodeService(services.networkService, services.networkMapCache) {
-        val ss = node.services.storageService
+        val ss = services.storageService
-        val oracle = Oracle(ss.myLegalIdentity, ss.myLegalIdentityKey, node.services.clock)
+        val oracle = Oracle(ss.myLegalIdentity, ss.myLegalIdentityKey, services.clock)
 
         private val logger = LoggerFactory.getLogger(Service::class.java)
 
@@ -65,7 +76,7 @@ object NodeInterestRates {
                              * Interest rates become available when they are uploaded via the web as per [DataUploadServlet],
                              * if they haven't already been uploaded that way.
                              */
-                            node.smm.add("fixing", FixQueryHandler(this, req as RatesFixProtocol.QueryRequest))
+                            services.startProtocol("fixing", FixQueryHandler(this, req as RatesFixProtocol.QueryRequest))
                             Unit
                         }
                     },
@@ -96,15 +107,6 @@ object NodeInterestRates {
             }
         }
 
-        /**
-         * Register the protocol that is used with the Fixing integration tests.
-         */
-        class FixingServicePlugin : CordaPluginRegistry {
-            override val webApis: List<Class<*>> = emptyList()
-            override val requiredProtocols: Map<String, Set<String>> = mapOf(Pair(TwoPartyDealProtocol.FixingRoleDecider::class.java.name, setOf(Duration::class.java.name, StateRef::class.java.name)))
-            override val staticServeDirs: Map<String, String> = emptyMap()
-        }
-
         // File upload support
         override val dataTypePrefix = "interest-rates"
         override val acceptableFileExtensions = listOf(".rates", ".txt")

node/src/main/kotlin/com/r3corda/node/services/persistence/DataVendingService.kt

@@ -4,104 +4,110 @@ import com.r3corda.core.contracts.SignedTransaction
 import com.r3corda.core.crypto.Party
 import com.r3corda.core.failure
 import com.r3corda.core.messaging.MessagingService
+import com.r3corda.core.node.CordaPluginRegistry
 import com.r3corda.core.serialization.serialize
 import com.r3corda.core.success
 import com.r3corda.core.utilities.loggerFor
 import com.r3corda.node.services.api.AbstractNodeService
 import com.r3corda.node.services.api.ServiceHubInternal
-import com.r3corda.protocols.FetchAttachmentsProtocol
+import com.r3corda.protocols.*
-import com.r3corda.protocols.FetchDataProtocol
-import com.r3corda.protocols.FetchTransactionsProtocol
-import com.r3corda.protocols.PartyRequestMessage
-import com.r3corda.protocols.ResolveTransactionsProtocol
 import java.io.InputStream
 import javax.annotation.concurrent.ThreadSafe
 
-/**
+object DataVending {
- * This class sets up network message handlers for requests from peers for data keyed by hash. It is a piece of simple
+
- * glue that sits between the network layer and the database layer.
+    class Plugin : CordaPluginRegistry {
- *
+        override val webApis: List<Class<*>> = emptyList()
- * Note that in our data model, to be able to name a thing by hash automatically gives the power to request it. There
+        override val requiredProtocols: Map<String, Set<String>> = emptyMap()
- * are no access control lists. If you want to keep some data private, then you must be careful who you give its name
+        override val servicePlugins: List<Class<*>> = listOf(Service::class.java)
- * to, and trust that they will not pass the name onwards. If someone suspects some data might exist but does not have
+    }
- * its name, then the 256-bit search space they'd have to cover makes it physically impossible to enumerate, and as
+
- * such the hash of a piece of data can be seen as a type of password allowing access to it.
+    /**
- *
+     * This class sets up network message handlers for requests from peers for data keyed by hash. It is a piece of simple
- * Additionally, because nodes do not store invalid transactions, requesting such a transaction will always yield null.
+     * glue that sits between the network layer and the database layer.
- */
+     *
-@ThreadSafe
+     * Note that in our data model, to be able to name a thing by hash automatically gives the power to request it. There
-// TODO:  I don't like that this needs ServiceHubInternal, but passing in a state machine breaks MockServices because
+     * are no access control lists. If you want to keep some data private, then you must be careful who you give its name
+     * to, and trust that they will not pass the name onwards. If someone suspects some data might exist but does not have
+     * its name, then the 256-bit search space they'd have to cover makes it physically impossible to enumerate, and as
+     * such the hash of a piece of data can be seen as a type of password allowing access to it.
+     *
+     * Additionally, because nodes do not store invalid transactions, requesting such a transaction will always yield null.
+     */
+    @ThreadSafe
+    // TODO:  I don't like that this needs ServiceHubInternal, but passing in a state machine breaks MockServices because
 //        the state machine isn't set when this is constructed. [NodeSchedulerService] has the same problem, and both
 //        should be fixed at the same time.
-class DataVendingService(net: MessagingService, private val services: ServiceHubInternal) : AbstractNodeService(net, services.networkMapCache) {
+    class Service(net: MessagingService, private val services: ServiceHubInternal) : AbstractNodeService(net, services.networkMapCache) {
-    companion object {
+        companion object {
-        val logger = loggerFor<DataVendingService>()
+            val logger = loggerFor<DataVending.Service>()
 
-        /** Topic for messages notifying a node of a new transaction */
+            /** Topic for messages notifying a node of a new transaction */
-        val NOTIFY_TX_PROTOCOL_TOPIC = "platform.wallet.notify_tx"
+            val NOTIFY_TX_PROTOCOL_TOPIC = "platform.wallet.notify_tx"
-    }
-
-    val storage = services.storageService
-
-    data class NotifyTxRequestMessage(val tx: SignedTransaction, override val replyToParty: Party, override val sessionID: Long) : PartyRequestMessage
-    data class NotifyTxResponseMessage(val accepted: Boolean)
-
-    init {
-        addMessageHandler(FetchTransactionsProtocol.TOPIC,
-                { req: FetchDataProtocol.Request -> handleTXRequest(req) },
-                { message, e -> logger.error("Failure processing data vending request.", e) }
-        )
-        addMessageHandler(FetchAttachmentsProtocol.TOPIC,
-                { req: FetchDataProtocol.Request -> handleAttachmentRequest(req) },
-                { message, e -> logger.error("Failure processing data vending request.", e) }
-        )
-        addMessageHandler(NOTIFY_TX_PROTOCOL_TOPIC,
-                { req: NotifyTxRequestMessage -> handleTXNotification(req) },
-                { message, e -> logger.error("Failure processing data vending request.", e) }
-        )
-    }
-
-    private fun handleTXNotification(req: NotifyTxRequestMessage): Unit {
-        // TODO: We should have a whitelist of contracts we're willing to accept at all, and reject if the transaction
-        //       includes us in any outside that list. Potentially just if it includes any outside that list at all.
-
-        // TODO: Do we want to be able to reject specific transactions on more complex rules, for example reject incoming
-        //       cash without from unknown parties?
-
-        services.startProtocol(NOTIFY_TX_PROTOCOL_TOPIC, ResolveTransactionsProtocol(req.tx, req.replyToParty))
-                .success {
-                    services.recordTransactions(req.tx)
-                    val resp = NotifyTxResponseMessage(true)
-                    val msg = net.createMessage(NOTIFY_TX_PROTOCOL_TOPIC + "." + req.sessionID, resp.serialize().bits)
-                    net.send(msg, req.getReplyTo(services.networkMapCache))
-                }.failure {
-                    val resp = NotifyTxResponseMessage(false)
-                    val msg = net.createMessage(NOTIFY_TX_PROTOCOL_TOPIC + "." + req.sessionID, resp.serialize().bits)
-                    net.send(msg, req.getReplyTo(services.networkMapCache))
-                }
-    }
-
-    private fun handleTXRequest(req: FetchDataProtocol.Request): List<SignedTransaction?> {
-        require(req.hashes.isNotEmpty())
-        return req.hashes.map {
-            val tx = storage.validatedTransactions.getTransaction(it)
-            if (tx == null)
-                logger.info("Got request for unknown tx $it")
-            tx
         }
-    }
 
-    private fun handleAttachmentRequest(req: FetchDataProtocol.Request): List<ByteArray?> {
+        val storage = services.storageService
-        // TODO: Use Artemis message streaming support here, called "large messages". This avoids the need to buffer.
+
-        require(req.hashes.isNotEmpty())
+        data class NotifyTxRequestMessage(val tx: SignedTransaction, override val replyToParty: Party, override val sessionID: Long) : PartyRequestMessage
-        return req.hashes.map {
+        data class NotifyTxResponseMessage(val accepted: Boolean)
-            val jar: InputStream? = storage.attachments.openAttachment(it)?.open()
+
-            if (jar == null) {
+        init {
-                logger.info("Got request for unknown attachment $it")
+            addMessageHandler(FetchTransactionsProtocol.TOPIC,
-                null
+                    { req: FetchDataProtocol.Request -> handleTXRequest(req) },
-            } else {
+                    { message, e -> logger.error("Failure processing data vending request.", e) }
-                jar.readBytes()
+            )
+            addMessageHandler(FetchAttachmentsProtocol.TOPIC,
+                    { req: FetchDataProtocol.Request -> handleAttachmentRequest(req) },
+                    { message, e -> logger.error("Failure processing data vending request.", e) }
+            )
+            addMessageHandler(NOTIFY_TX_PROTOCOL_TOPIC,
+                    { req: NotifyTxRequestMessage -> handleTXNotification(req) },
+                    { message, e -> logger.error("Failure processing data vending request.", e) }
+            )
+        }
+
+        private fun handleTXNotification(req: NotifyTxRequestMessage): Unit {
+            // TODO: We should have a whitelist of contracts we're willing to accept at all, and reject if the transaction
+            //       includes us in any outside that list. Potentially just if it includes any outside that list at all.
+
+            // TODO: Do we want to be able to reject specific transactions on more complex rules, for example reject incoming
+            //       cash without from unknown parties?
+
+            services.startProtocol(NOTIFY_TX_PROTOCOL_TOPIC, ResolveTransactionsProtocol(req.tx, req.replyToParty))
+                    .success {
+                        services.recordTransactions(req.tx)
+                        val resp = NotifyTxResponseMessage(true)
+                        val msg = net.createMessage(NOTIFY_TX_PROTOCOL_TOPIC + "." + req.sessionID, resp.serialize().bits)
+                        net.send(msg, req.getReplyTo(services.networkMapCache))
+                    }.failure {
+                val resp = NotifyTxResponseMessage(false)
+                val msg = net.createMessage(NOTIFY_TX_PROTOCOL_TOPIC + "." + req.sessionID, resp.serialize().bits)
+                net.send(msg, req.getReplyTo(services.networkMapCache))
+            }
+        }
+
+        private fun handleTXRequest(req: FetchDataProtocol.Request): List<SignedTransaction?> {
+            require(req.hashes.isNotEmpty())
+            return req.hashes.map {
+                val tx = storage.validatedTransactions.getTransaction(it)
+                if (tx == null)
+                    logger.info("Got request for unknown tx $it")
+                tx
+            }
+        }
+
+        private fun handleAttachmentRequest(req: FetchDataProtocol.Request): List<ByteArray?> {
+            // TODO: Use Artemis message streaming support here, called "large messages". This avoids the need to buffer.
+            require(req.hashes.isNotEmpty())
+            return req.hashes.map {
+                val jar: InputStream? = storage.attachments.openAttachment(it)?.open()
+                if (jar == null) {
+                    logger.info("Got request for unknown attachment $it")
+                    null
+                } else {
+                    jar.readBytes()
+                }
             }
         }
     }
-}
+}

node/src/main/resources/META-INF/services/com.r3corda.core.node.CordaPluginRegistry

@@ -1,2 +1,5 @@
 # Register a ServiceLoader service extending from com.r3corda.node.CordaPluginRegistry
-com.r3corda.node.services.clientapi.NodeInterestRates$Service$FixingServicePlugin
+com.r3corda.node.services.clientapi.FixingSessionInitiation$Plugin
+com.r3corda.node.services.clientapi.NodeInterestRates$Plugin
+com.r3corda.node.services.NotaryChange$Plugin
+com.r3corda.node.services.persistence.DataVending$Plugin

node/src/test/kotlin/com/r3corda/node/services/MockServices.kt

@@ -14,7 +14,7 @@ import com.r3corda.node.services.api.MonitoringService
 import com.r3corda.node.services.api.ServiceHubInternal
 import com.r3corda.node.services.network.MockNetworkMapCache
 import com.r3corda.node.services.network.NetworkMapService
-import com.r3corda.node.services.persistence.DataVendingService
+import com.r3corda.node.services.persistence.DataVending
 import com.r3corda.node.services.statemachine.StateMachineManager
 import com.r3corda.node.services.wallet.NodeWalletService
 import java.time.Clock
@@ -68,7 +68,7 @@ open class MockServices(
         if (net != null && storage != null) {
             // Creating this class is sufficient, we don't have to store it anywhere, because it registers a listener
             // on the networking service, so that will keep it from being collected.
-            DataVendingService(net, this)
+            DataVending.Service(this)
         }
     }
 }

node/src/test/kotlin/com/r3corda/node/services/NodeInterestRatesTest.kt

@@ -105,7 +105,7 @@ class NodeInterestRatesTest {
     fun network() {
         val net = MockNetwork()
         val (n1, n2) = net.createTwoNodes()
-        n2.interestRatesService.oracle.knownFixes = TEST_DATA
+        n2.getCustomService<NodeInterestRates.Service>().oracle.knownFixes = TEST_DATA
 
         val tx = TransactionType.General.Builder()
         val fixOf = NodeInterestRates.parseFixOf("LIBOR 2016-03-16 1M")

src/main/kotlin/com/r3corda/demos/IRSDemo.kt

@@ -17,7 +17,6 @@ import com.r3corda.demos.protocols.UpdateBusinessDayProtocol
 import com.r3corda.node.internal.AbstractNode
 import com.r3corda.node.internal.Node
 import com.r3corda.node.internal.testing.MockNetwork
-import com.r3corda.node.services.FixingSessionInitiationHandler
 import com.r3corda.node.services.clientapi.NodeInterestRates
 import com.r3corda.node.services.config.NodeConfiguration
 import com.r3corda.node.services.config.NodeConfigurationFromConfig
@@ -265,6 +264,7 @@ class IRSDemoPluginRegistry : CordaPluginRegistry {
             Pair(AutoOfferProtocol.Requester::class.java.name, setOf(InterestRateSwap.State::class.java.name)),
             Pair(UpdateBusinessDayProtocol.Broadcast::class.java.name, setOf(java.time.LocalDate::class.java.name)),
             Pair(ExitServerProtocol.Broadcast::class.java.name, setOf(kotlin.Int::class.java.name)))
+    override val servicePlugins: List<Class<*>> = emptyList()
 }
 
 private class NotSetupException: Throwable {
@@ -332,11 +332,6 @@ private fun runNode(cliParams: CliParams.RunNode): Int {
         val networkMap = createRecipient(cliParams.mapAddress)
 
         val node = startNode(cliParams, networkMap)
-        // Register handlers for the demo
-        AutoOfferProtocol.Handler.register(node)
-        UpdateBusinessDayProtocol.Handler.register(node)
-        ExitServerProtocol.Handler.register(node)
-        FixingSessionInitiationHandler.register(node)
 
         if (cliParams.uploadRates) {
             runUploadRates(cliParams.apiAddress)

src/main/kotlin/com/r3corda/demos/protocols/AutoOfferProtocol.kt

@@ -6,11 +6,12 @@ import com.google.common.util.concurrent.Futures
 import com.r3corda.core.contracts.DealState
 import com.r3corda.core.contracts.SignedTransaction
 import com.r3corda.core.crypto.Party
+import com.r3corda.core.node.CordaPluginRegistry
 import com.r3corda.core.protocols.ProtocolLogic
 import com.r3corda.core.random63BitValue
 import com.r3corda.core.serialization.deserialize
 import com.r3corda.core.utilities.ProgressTracker
-import com.r3corda.node.internal.Node
+import com.r3corda.node.services.api.ServiceHubInternal
 import com.r3corda.protocols.TwoPartyDealProtocol
 
 /**
@@ -27,7 +28,14 @@ object AutoOfferProtocol {
                                 val notary: Party,
                                 val otherSessionID: Long, val dealBeingOffered: DealState)
 
-    object Handler {
+    class Plugin: CordaPluginRegistry {
+        override val webApis: List<Class<*>> = emptyList()
+        override val requiredProtocols: Map<String, Set<String>> = emptyMap()
+        override val servicePlugins: List<Class<*>> = listOf(Service::class.java)
+    }
+
+
+    class Service(services: ServiceHubInternal) {
 
         object RECEIVED : ProgressTracker.Step("Received offer")
         object DEALING : ProgressTracker.Step("Starting the deal protocol") {
@@ -46,16 +54,16 @@ object AutoOfferProtocol {
             }
         }
 
-        fun register(node: Node) {
+        init {
-            node.net.addMessageHandler("$TOPIC.0") { msg, registration ->
+            services.networkService.addMessageHandler("$TOPIC.0") { msg, registration ->
                 val progressTracker = tracker()
                 progressTracker.currentStep = RECEIVED
                 val autoOfferMessage = msg.data.deserialize<AutoOfferMessage>()
                 // Put the deal onto the ledger
                 progressTracker.currentStep = DEALING
                 val seller = TwoPartyDealProtocol.Instigator(autoOfferMessage.otherSide, autoOfferMessage.notary,
-                        autoOfferMessage.dealBeingOffered, node.services.keyManagementService.freshKey(), autoOfferMessage.otherSessionID, progressTracker.getChildProgressTracker(DEALING)!!)
+                        autoOfferMessage.dealBeingOffered, services.keyManagementService.freshKey(), autoOfferMessage.otherSessionID, progressTracker.getChildProgressTracker(DEALING)!!)
-                val future = node.smm.add("${TwoPartyDealProtocol.DEAL_TOPIC}.seller", seller)
+                val future = services.startProtocol("${TwoPartyDealProtocol.DEAL_TOPIC}.seller", seller)
                 // This is required because we are doing child progress outside of a subprotocol.  In future, we should just wrap things like this in a protocol to avoid it
                 Futures.addCallback(future, Callback() {
                     seller.progressTracker.currentStep = ProgressTracker.DONE

src/main/kotlin/com/r3corda/demos/protocols/ExitServerProtocol.kt

@@ -2,10 +2,11 @@ package com.r3corda.demos.protocols
 
 import co.paralleluniverse.fibers.Suspendable
 import co.paralleluniverse.strands.Strand
+import com.r3corda.core.node.CordaPluginRegistry
 import com.r3corda.core.node.NodeInfo
 import com.r3corda.core.protocols.ProtocolLogic
 import com.r3corda.core.serialization.deserialize
-import com.r3corda.node.internal.Node
+import com.r3corda.node.services.api.ServiceHubInternal
 import com.r3corda.node.services.network.MockNetworkMapCache
 import java.util.concurrent.TimeUnit
 
@@ -19,10 +20,16 @@ object ExitServerProtocol {
 
     data class ExitMessage(val exitCode: Int)
 
-    object Handler {
+    class Plugin: CordaPluginRegistry {
+        override val webApis: List<Class<*>> = emptyList()
+        override val requiredProtocols: Map<String, Set<String>> = emptyMap()
+        override val servicePlugins: List<Class<*>> = listOf(Service::class.java)
+    }
 
-        fun register(node: Node) {
+    class Service(services: ServiceHubInternal) {
-            node.net.addMessageHandler("$TOPIC.0") { msg, registration ->
+
+        init {
+            services.networkService.addMessageHandler("$TOPIC.0") { msg, registration ->
                 // Just to validate we got the message
                 if (enabled) {
                     val message = msg.data.deserialize<ExitMessage>()

src/main/kotlin/com/r3corda/demos/protocols/UpdateBusinessDayProtocol.kt

@@ -1,12 +1,14 @@
 package com.r3corda.demos.protocols
 
 import co.paralleluniverse.fibers.Suspendable
+import com.r3corda.core.node.CordaPluginRegistry
 import com.r3corda.core.node.NodeInfo
 import com.r3corda.core.protocols.ProtocolLogic
 import com.r3corda.core.serialization.deserialize
 import com.r3corda.core.utilities.ProgressTracker
 import com.r3corda.demos.DemoClock
 import com.r3corda.node.internal.Node
+import com.r3corda.node.services.api.ServiceHubInternal
 import com.r3corda.node.services.network.MockNetworkMapCache
 import java.time.LocalDate
 
@@ -19,12 +21,18 @@ object UpdateBusinessDayProtocol {
 
     data class UpdateBusinessDayMessage(val date: LocalDate)
 
-    object Handler {
+    class Plugin: CordaPluginRegistry {
+        override val webApis: List<Class<*>> = emptyList()
+        override val requiredProtocols: Map<String, Set<String>> = emptyMap()
+        override val servicePlugins: List<Class<*>> = listOf(Service::class.java)
+    }
 
-        fun register(node: Node) {
+    class Service(services: ServiceHubInternal) {
-            node.net.addMessageHandler("${TOPIC}.0") { msg, registration ->
+
+        init {
+            services.networkService.addMessageHandler("${TOPIC}.0") { msg, registration ->
                 val updateBusinessDayMessage = msg.data.deserialize<UpdateBusinessDayMessage>()
-                (node.services.clock as DemoClock).updateDate(updateBusinessDayMessage.date)
+                (services.clock as DemoClock).updateDate(updateBusinessDayMessage.date)
             }
         }
     }

src/main/resources/META-INF/services/com.r3corda.core.node.CordaPluginRegistry

@@ -1,2 +1,5 @@
 # Register a ServiceLoader service extending from com.r3corda.node.CordaPluginRegistry
-com.r3corda.demos.IRSDemoPluginRegistry
+com.r3corda.demos.IRSDemoPluginRegistry
+com.r3corda.demos.protocols.AutoOfferProtocol$Plugin
+com.r3corda.demos.protocols.ExitServerProtocol$Plugin
+com.r3corda.demos.protocols.UpdateBusinessDayProtocol$Plugin