Moved topic parameter from send/receive methods to a ProtocolLogic property

contracts/src/main/kotlin/com/r3corda/protocols/TwoPartyTradeProtocol.kt

@@ -42,7 +42,8 @@ import java.util.*
  * To see an example of how to use this class, look at the unit tests.
  */
 object TwoPartyTradeProtocol {
-    val TRADE_TOPIC = "platform.trade"
+
+    val TOPIC = "platform.trade"
 
     class UnacceptablePriceException(val givenPrice: Amount<Issued<Currency>>) : Exception()
     class AssetMismatchException(val expectedTypeName: String, val typeName: String) : Exception() {
@@ -82,6 +83,8 @@ object TwoPartyTradeProtocol {
             fun tracker() = ProgressTracker(AWAITING_PROPOSAL, VERIFYING, SIGNING, NOTARY, SENDING_SIGS)
         }
 
+        override val topic: String get() = TOPIC
+
         @Suspendable
         override fun call(): SignedTransaction {
             val partialTX: SignedTransaction = receiveAndCheckProposedTransaction()
@@ -108,7 +111,7 @@ object TwoPartyTradeProtocol {
             // Make the first message we'll send to kick off the protocol.
             val hello = SellerTradeInfo(assetToSell, price, myKeyPair.public, sessionID)
 
-            val maybeSTX = sendAndReceive<SignedTransaction>(TRADE_TOPIC, otherSide, buyerSessionID, sessionID, hello)
+            val maybeSTX = sendAndReceive<SignedTransaction>(otherSide, buyerSessionID, sessionID, hello)
 
             progressTracker.currentStep = VERIFYING
 
@@ -166,7 +169,7 @@ object TwoPartyTradeProtocol {
 
             logger.trace { "Built finished transaction, sending back to secondary!" }
 
-            send(TRADE_TOPIC, otherSide, buyerSessionID, SignaturesFromSeller(ourSignature, notarySignature))
+            send(otherSide, buyerSessionID, SignaturesFromSeller(ourSignature, notarySignature))
             return fullySigned
         }
     }
@@ -185,6 +188,7 @@ object TwoPartyTradeProtocol {
 
         object SWAPPING_SIGNATURES : ProgressTracker.Step("Swapping signatures with the seller")
 
+        override val topic: String get() = TOPIC
         override val progressTracker = ProgressTracker(RECEIVING, VERIFYING, SIGNING, SWAPPING_SIGNATURES)
 
         @Suspendable
@@ -195,8 +199,6 @@ object TwoPartyTradeProtocol {
             val (ptx, cashSigningPubKeys) = assembleSharedTX(tradeRequest)
             val stx = signWithOurKeys(cashSigningPubKeys, ptx)
 
-            // exitProcess(0)
-
             val signatures = swapSignaturesWithSeller(stx, tradeRequest.sessionID)
 
             logger.trace { "Got signatures from seller, verifying ... " }
@@ -215,7 +217,7 @@ object TwoPartyTradeProtocol {
         private fun receiveAndValidateTradeRequest(): SellerTradeInfo {
             progressTracker.currentStep = RECEIVING
             // Wait for a trade request to come in on our pre-provided session ID.
-            val maybeTradeRequest = receive<SellerTradeInfo>(TRADE_TOPIC, sessionID)
+            val maybeTradeRequest = receive<SellerTradeInfo>(sessionID)
 
             progressTracker.currentStep = VERIFYING
             maybeTradeRequest.validate {
@@ -246,7 +248,7 @@ object TwoPartyTradeProtocol {
 
             // TODO: Protect against the seller terminating here and leaving us in the lurch without the final tx.
 
-            return sendAndReceive<SignaturesFromSeller>(TRADE_TOPIC, otherSide, theirSessionID, sessionID, stx).validate { it }
+            return sendAndReceive<SignaturesFromSeller>(otherSide, theirSessionID, sessionID, stx).validate { it }
         }
 
         private fun signWithOurKeys(cashSigningPubKeys: List<PublicKey>, ptx: TransactionBuilder): SignedTransaction {

core/src/main/kotlin/com/r3corda/core/protocols/ProtocolLogic.kt

@@ -25,6 +25,7 @@ import org.slf4j.Logger
  * it to the [subProtocol] method. It will return the result of that protocol when it completes.
  */
 abstract class ProtocolLogic<T> {
+
     /** Reference to the [Fiber] instance that is the top level controller for the entire flow. */
     lateinit var psm: ProtocolStateMachine<*>
 
@@ -38,24 +39,29 @@ abstract class ProtocolLogic<T> {
      */
     val serviceHub: ServiceHub get() = psm.serviceHub
 
+    /**
+     * The topic to use when communicating with other parties. If more than one topic is required then use sub-protocols.
+     * Note that this is temporary until protocol sessions are properly implemented.
+     */
+    protected abstract val topic: String
+
     // Kotlin helpers that allow the use of generic types.
-    inline fun <reified T : Any> sendAndReceive(topic: String,
+    inline fun <reified T : Any> sendAndReceive(destination: Party,
-                                                destination: Party,
                                                 sessionIDForSend: Long,
                                                 sessionIDForReceive: Long,
                                                 payload: Any): UntrustworthyData<T> {
         return psm.sendAndReceive(topic, destination, sessionIDForSend, sessionIDForReceive, payload, T::class.java)
     }
 
-    inline fun <reified T : Any> receive(topic: String, sessionIDForReceive: Long): UntrustworthyData<T> {
+    inline fun <reified T : Any> receive(sessionIDForReceive: Long): UntrustworthyData<T> {
-        return receive(topic, sessionIDForReceive, T::class.java)
+        return receive(sessionIDForReceive, T::class.java)
     }
 
-    @Suspendable fun <T : Any> receive(topic: String, sessionIDForReceive: Long, clazz: Class<T>): UntrustworthyData<T> {
+    @Suspendable fun <T : Any> receive(sessionIDForReceive: Long, receiveType: Class<T>): UntrustworthyData<T> {
-        return psm.receive(topic, sessionIDForReceive, clazz)
+        return psm.receive(topic, sessionIDForReceive, receiveType)
     }
 
-    @Suspendable fun send(topic: String, destination: Party, sessionID: Long, payload: Any) {
+    @Suspendable fun send(destination: Party, sessionID: Long, payload: Any) {
         psm.send(topic, destination, sessionID, payload)
     }
 
@@ -99,4 +105,5 @@ abstract class ProtocolLogic<T> {
     /** This is where you fill out your business logic. */
     @Suspendable
     abstract fun call(): T
+
 }

core/src/main/kotlin/com/r3corda/protocols/FetchAttachmentsProtocol.kt

@@ -13,13 +13,14 @@ import java.io.InputStream
  */
 class FetchAttachmentsProtocol(requests: Set<SecureHash>,
                                otherSide: Party) : FetchDataProtocol<Attachment, ByteArray>(requests, otherSide) {
+
     companion object {
         const val TOPIC = "platform.fetch.attachment"
     }
 
-    override fun load(txid: SecureHash): Attachment? = serviceHub.storageService.attachments.openAttachment(txid)
+    override val topic: String get() = TOPIC
 
-    override val queryTopic: String = TOPIC
+    override fun load(txid: SecureHash): Attachment? = serviceHub.storageService.attachments.openAttachment(txid)
 
     override fun convert(wire: ByteArray): Attachment {
         return object : Attachment {

core/src/main/kotlin/com/r3corda/protocols/FetchDataProtocol.kt

@@ -33,11 +33,9 @@ abstract class FetchDataProtocol<T : NamedByHash, W : Any>(
     class HashNotFound(val requested: SecureHash) : BadAnswer()
     class DownloadedVsRequestedDataMismatch(val requested: SecureHash, val got: SecureHash) : BadAnswer()
 
-    class Request(val hashes: List<SecureHash>, replyTo: Party, override val sessionID: Long) : AbstractRequestMessage(replyTo)
+    data class Request(val hashes: List<SecureHash>, override val replyToParty: Party, override val sessionID: Long) : PartyRequestMessage
     data class Result<T : NamedByHash>(val fromDisk: List<T>, val downloaded: List<T>)
 
-    protected abstract val queryTopic: String
-
     @Suspendable
     override fun call(): Result<T> {
         // Load the items we have from disk and figure out which we're missing.
@@ -51,7 +49,7 @@ abstract class FetchDataProtocol<T : NamedByHash, W : Any>(
             val sid = random63BitValue()
             val fetchReq = Request(toFetch, serviceHub.storageService.myLegalIdentity, sid)
             // TODO: Support "large message" response streaming so response sizes are not limited by RAM.
-            val maybeItems = sendAndReceive<ArrayList<W?>>(queryTopic, otherSide, 0, sid, fetchReq)
+            val maybeItems = sendAndReceive<ArrayList<W?>>(otherSide, 0, sid, fetchReq)
             // Check for a buggy/malicious peer answering with something that we didn't ask for.
             val downloaded = validateFetchResponse(maybeItems, toFetch)
             maybeWriteToDisk(downloaded)

core/src/main/kotlin/com/r3corda/protocols/FetchTransactionsProtocol.kt

@@ -14,10 +14,12 @@ import com.r3corda.core.crypto.SecureHash
  */
 class FetchTransactionsProtocol(requests: Set<SecureHash>, otherSide: Party) :
         FetchDataProtocol<SignedTransaction, SignedTransaction>(requests, otherSide) {
+
     companion object {
         const val TOPIC = "platform.fetch.tx"
     }
 
+    override val topic: String get() = TOPIC
+
     override fun load(txid: SecureHash): SignedTransaction? = serviceHub.storageService.validatedTransactions.getTransaction(txid)
-    override val queryTopic: String = TOPIC
 }

core/src/main/kotlin/com/r3corda/protocols/NotaryProtocol.kt

@@ -22,8 +22,8 @@ import com.r3corda.core.utilities.UntrustworthyData
 import java.security.PublicKey
 
 object NotaryProtocol {
-    val TOPIC = "platform.notary.request"
+
-    val TOPIC_INITIATE = "platform.notary.initiate"
+    val TOPIC = "platform.notary"
 
     /**
      * A protocol to be used for obtaining a signature from a [NotaryService] ascertaining the transaction
@@ -34,6 +34,7 @@ object NotaryProtocol {
      */
     class Client(private val stx: SignedTransaction,
                  override val progressTracker: ProgressTracker = Client.tracker()) : ProtocolLogic<DigitalSignature.LegallyIdentifiable>() {
+
         companion object {
 
             object REQUESTING : ProgressTracker.Step("Requesting signature by Notary service")
@@ -43,6 +44,8 @@ object NotaryProtocol {
             fun tracker() = ProgressTracker(REQUESTING, VALIDATING)
         }
 
+        override val topic: String get() = TOPIC
+
         lateinit var notaryParty: Party
 
         @Suspendable
@@ -54,10 +57,10 @@ object NotaryProtocol {
             val receiveSessionID = random63BitValue()
 
             val handshake = Handshake(serviceHub.storageService.myLegalIdentity, sendSessionID, receiveSessionID)
-            sendAndReceive<Ack>(TOPIC_INITIATE, notaryParty, 0, receiveSessionID, handshake)
+            sendAndReceive<Ack>(notaryParty, 0, receiveSessionID, handshake)
 
             val request = SignRequest(stx, serviceHub.storageService.myLegalIdentity)
-            val response = sendAndReceive<Result>(TOPIC, notaryParty, sendSessionID, receiveSessionID, request)
+            val response = sendAndReceive<Result>(notaryParty, sendSessionID, receiveSessionID, request)
 
             val notaryResult = validateResponse(response)
             return notaryResult.sig ?: throw NotaryException(notaryResult.error!!)
@@ -113,27 +116,26 @@ object NotaryProtocol {
                        val receiveSessionID: Long,
                        val timestampChecker: TimestampChecker,
                        val uniquenessProvider: UniquenessProvider) : ProtocolLogic<Unit>() {
+
+        override val topic: String get() = TOPIC
+
         @Suspendable
         override fun call() {
-            val request = receive<SignRequest>(TOPIC, receiveSessionID).validate { it }
+            val (stx, reqIdentity) = receive<SignRequest>(receiveSessionID).validate { it }
-            val stx = request.tx
             val wtx = stx.tx
-            val reqIdentity = request.callerIdentity
 
-            val result: Result
+            val result = try {
-            try {
                 validateTimestamp(wtx)
                 beforeCommit(stx, reqIdentity)
                 commitInputStates(wtx, reqIdentity)
 
                 val sig = sign(stx.txBits)
-                result = Result.noError(sig)
+                Result.noError(sig)
-
             } catch(e: NotaryException) {
-                result = Result.withError(e.error)
+                Result.withError(e.error)
             }
 
-            send(TOPIC, otherSide, sendSessionID, result)
+            send(otherSide, sendSessionID, result)
         }
 
         private fun validateTimestamp(tx: WireTransaction) {
@@ -178,14 +180,13 @@ object NotaryProtocol {
         }
     }
 
-    class Handshake(
+    data class Handshake(
-            replyTo: Party,
+            override  val replyToParty: Party,
             val sendSessionID: Long,
-            override val sessionID: Long) : AbstractRequestMessage(replyTo)
+            override val sessionID: Long) : PartyRequestMessage
 
     /** TODO: The caller must authenticate instead of just specifying its identity */
-    class SignRequest(val tx: SignedTransaction,
+    data class SignRequest(val tx: SignedTransaction, val callerIdentity: Party)
-                      val callerIdentity: Party)
 
     data class Result private constructor(val sig: DigitalSignature.LegallyIdentifiable?, val error: NotaryError?) {
         companion object {

core/src/main/kotlin/com/r3corda/protocols/RatesFixProtocol.kt

@@ -33,10 +33,9 @@ open class RatesFixProtocol(protected val tx: TransactionBuilder,
                             private val rateTolerance: BigDecimal,
                             private val timeOut: Duration,
                             override val progressTracker: ProgressTracker = RatesFixProtocol.tracker(fixOf.name)) : ProtocolLogic<Unit>() {
+
     companion object {
         val TOPIC = "platform.rates.interest.fix"
-        val TOPIC_SIGN = TOPIC + ".sign"
-        val TOPIC_QUERY = TOPIC + ".query"
 
         class QUERYING(val name: String) : ProgressTracker.Step("Querying oracle for $name interest rate")
         object WORKING : ProgressTracker.Step("Working with data returned by oracle")
@@ -45,10 +44,12 @@ open class RatesFixProtocol(protected val tx: TransactionBuilder,
         fun tracker(fixName: String) = ProgressTracker(QUERYING(fixName), WORKING, SIGNING)
     }
 
+    override val topic: String get() = TOPIC
+
     class FixOutOfRange(val byAmount: BigDecimal) : Exception()
 
-    class QueryRequest(val queries: List<FixOf>, replyTo: Party, override val sessionID: Long, val deadline: Instant) : AbstractRequestMessage(replyTo)
+    data class QueryRequest(val queries: List<FixOf>, override val replyToParty: Party, override val sessionID: Long, val deadline: Instant) : PartyRequestMessage
-    class SignRequest(val tx: WireTransaction, replyTo: Party, override val sessionID: Long) : AbstractRequestMessage(replyTo)
+    data class SignRequest(val tx: WireTransaction, override val replyToParty: Party, override val sessionID: Long) : PartyRequestMessage
 
     @Suspendable
     override fun call() {
@@ -79,11 +80,11 @@ open class RatesFixProtocol(protected val tx: TransactionBuilder,
     }
 
     @Suspendable
-    fun sign(): DigitalSignature.LegallyIdentifiable {
+    private fun sign(): DigitalSignature.LegallyIdentifiable {
         val sessionID = random63BitValue()
         val wtx = tx.toWireTransaction()
         val req = SignRequest(wtx, serviceHub.storageService.myLegalIdentity, sessionID)
-        val resp = sendAndReceive<DigitalSignature.LegallyIdentifiable>(TOPIC_SIGN, oracle, 0, sessionID, req)
+        val resp = sendAndReceive<DigitalSignature.LegallyIdentifiable>(oracle, 0, sessionID, req)
 
         return resp.validate { sig ->
             check(sig.signer == oracle)
@@ -93,12 +94,12 @@ open class RatesFixProtocol(protected val tx: TransactionBuilder,
     }
 
     @Suspendable
-    fun query(): Fix {
+    private fun query(): Fix {
         val sessionID = random63BitValue()
         val deadline = suggestInterestRateAnnouncementTimeWindow(fixOf.name, oracle.name, fixOf.forDay).end
         val req = QueryRequest(listOf(fixOf), serviceHub.storageService.myLegalIdentity, sessionID, deadline)
         // TODO: add deadline to receive
-        val resp = sendAndReceive<ArrayList<Fix>>(TOPIC_QUERY, oracle, 0, sessionID, req)
+        val resp = sendAndReceive<ArrayList<Fix>>(oracle, 0, sessionID, req)
 
         return resp.validate {
             val fix = it.first()

core/src/main/kotlin/com/r3corda/protocols/ResolveTransactionsProtocol.kt

@@ -70,6 +70,8 @@ class ResolveTransactionsProtocol(private val txHashes: Set<SecureHash>,
         serviceHub.recordTransactions(downloadedSignedTxns)
     }
 
+    override val topic: String get() = throw UnsupportedOperationException()
+
     @Suspendable
     private fun fetchDependenciesAndCheckSignatures(depsToCheck: Set<SecureHash>,
                                                     toVerify: HashSet<LedgerTransaction>,

core/src/main/kotlin/com/r3corda/protocols/ServiceRequestMessage.kt

@@ -6,14 +6,17 @@ import com.r3corda.core.node.services.NetworkMapCache
 
 /**
  * Abstract superclass for request messages sent to services, which includes common
- * fields such as replyTo and replyToTopic.
+ * fields such as replyTo and sessionID.
  */
 interface ServiceRequestMessage {
     val sessionID: Long
     fun getReplyTo(networkMapCache: NetworkMapCache): MessageRecipients
 }
 
-abstract class AbstractRequestMessage(val replyToParty: Party): ServiceRequestMessage {
+interface PartyRequestMessage : ServiceRequestMessage {
+
+    val replyToParty: Party
+
     override fun getReplyTo(networkMapCache: NetworkMapCache): MessageRecipients {
         return networkMapCache.partyNodes.single { it.identity == replyToParty }.address
     }

core/src/main/kotlin/com/r3corda/protocols/TwoPartyDealProtocol.kt

@@ -30,7 +30,10 @@ import java.time.Duration
  *
  */
 object TwoPartyDealProtocol {
+
     val DEAL_TOPIC = "platform.deal"
+    /** This topic exists purely for [FixingSessionInitiation] to be sent from [FixingRoleDecider] to [FixingSessionInitiationHandler] */
+    val FIX_INITIATE_TOPIC = "platform.fix.initiate"
 
     class DealMismatchException(val expectedDeal: ContractState, val actualDeal: ContractState) : Exception() {
         override fun toString() = "The submitted deal didn't match the expected: $expectedDeal vs $actualDeal"
@@ -69,6 +72,8 @@ object TwoPartyDealProtocol {
             fun tracker() = ProgressTracker(AWAITING_PROPOSAL, VERIFYING, SIGNING, NOTARY, SENDING_SIGS, RECORDING, COPYING_TO_REGULATOR)
         }
 
+        override val topic: String get() = DEAL_TOPIC
+
         abstract val payload: U
         abstract val notaryNode: NodeInfo
         abstract val otherSide: Party
@@ -83,7 +88,7 @@ object TwoPartyDealProtocol {
 
             // Make the first message we'll send to kick off the protocol.
             val hello = Handshake(payload, myKeyPair.public, sessionID)
-            val maybeSTX = sendAndReceive<SignedTransaction>(DEAL_TOPIC, otherSide, otherSessionID, sessionID, hello)
+            val maybeSTX = sendAndReceive<SignedTransaction>(otherSide, otherSessionID, sessionID, hello)
 
             return maybeSTX
         }
@@ -152,7 +157,7 @@ object TwoPartyDealProtocol {
                 // Copy the transaction to every regulator in the network. This is obviously completely bogus, it's
                 // just for demo purposes.
                 for (regulator in regulators) {
-                    send("regulator.all.seeing.eye", regulator.identity, 0, fullySigned)
+                    send(regulator.identity, 0, fullySigned)
                 }
             }
 
@@ -178,7 +183,7 @@ object TwoPartyDealProtocol {
 
             logger.trace { "Built finished transaction, sending back to other party!" }
 
-            send(DEAL_TOPIC, otherSide, otherSessionID, SignaturesFromPrimary(ourSignature, notarySignature))
+            send(otherSide, otherSessionID, SignaturesFromPrimary(ourSignature, notarySignature))
             return fullySigned
         }
     }
@@ -202,6 +207,8 @@ object TwoPartyDealProtocol {
             fun tracker() = ProgressTracker(RECEIVING, VERIFYING, SIGNING, SWAPPING_SIGNATURES, RECORDING)
         }
 
+        override val topic: String get() = DEAL_TOPIC
+
         abstract val otherSide: Party
         abstract val sessionID: Long
 
@@ -234,7 +241,7 @@ object TwoPartyDealProtocol {
         private fun receiveAndValidateHandshake(): Handshake<U> {
             progressTracker.currentStep = RECEIVING
             // Wait for a trade request to come in on our pre-provided session ID.
-            val handshake = receive<Handshake<U>>(DEAL_TOPIC, sessionID)
+            val handshake = receive<Handshake<U>>(sessionID)
 
             progressTracker.currentStep = VERIFYING
             handshake.validate {
@@ -249,7 +256,7 @@ object TwoPartyDealProtocol {
 
             // TODO: Protect against the seller terminating here and leaving us in the lurch without the final tx.
 
-            return sendAndReceive<SignaturesFromPrimary>(DEAL_TOPIC, otherSide, theirSessionID, sessionID, stx).validate { it }
+            return sendAndReceive<SignaturesFromPrimary>(otherSide, theirSessionID, sessionID, stx).validate { it }
         }
 
         private fun signWithOurKeys(signingPubKeys: List<PublicKey>, ptx: TransactionBuilder): SignedTransaction {
@@ -424,8 +431,6 @@ object TwoPartyDealProtocol {
             serviceHub.networkMapCache.notaryNodes.filter { it.identity == dealToFix.state.notary }.single()
     }
 
-    /** This topic exists purely for [FixingSessionInitiation] to be sent from [FixingRoleDecider] to [FixingSessionInitiationHandler] */
-    val FIX_INITIATE_TOPIC = "platform.fix.initiate"
 
     /** Used to set up the session between [Floater] and [Fixer] */
     data class FixingSessionInitiation(val sessionID: Long, val party: Party, val sender: Party, val timeout: Duration)
@@ -439,13 +444,18 @@ object TwoPartyDealProtocol {
      *
      * TODO: Replace [FixingSessionInitiation] and [FixingSessionInitiationHandler] with generic session initiation logic once it exists.
      */
-    class FixingRoleDecider(val ref: StateRef, val timeout: Duration, override val progressTracker: ProgressTracker = tracker(ref.toString())) : ProtocolLogic<Unit>() {
+    class FixingRoleDecider(val ref: StateRef,
+                            val timeout: Duration,
+                            override val progressTracker: ProgressTracker = tracker(ref.toString())) : ProtocolLogic<Unit>() {
+
         companion object {
             class LOADING(ref: String) : ProgressTracker.Step("Loading state $ref to decide fixing role")
 
             fun tracker(ref: String) = ProgressTracker(LOADING(ref))
         }
 
+        override val topic: String get() = FIX_INITIATE_TOPIC
+
         @Suspendable
         override fun call(): Unit {
             progressTracker.nextStep()
@@ -458,7 +468,7 @@ object TwoPartyDealProtocol {
                 val initation = FixingSessionInitiation(sessionID, sortedParties[0], serviceHub.storageService.myLegalIdentity, timeout)
 
                 // Send initiation to other side to launch one side of the fixing protocol (the Fixer).
-                send(FIX_INITIATE_TOPIC, sortedParties[1], 0, initation)
+                send(sortedParties[1], 0, initation)
 
                 // Then start the other side of the fixing protocol.
                 val protocol = Floater(ref, sessionID)
@@ -466,4 +476,5 @@ object TwoPartyDealProtocol {
             }
         }
     }
+
 }

core/src/main/kotlin/protocols/AbstractStateReplacementProtocol.kt

@@ -10,8 +10,8 @@ import com.r3corda.core.node.NodeInfo
 import com.r3corda.core.protocols.ProtocolLogic
 import com.r3corda.core.random63BitValue
 import com.r3corda.core.utilities.ProgressTracker
-import com.r3corda.protocols.AbstractRequestMessage
 import com.r3corda.protocols.NotaryProtocol
+import com.r3corda.protocols.PartyRequestMessage
 import com.r3corda.protocols.ResolveTransactionsProtocol
 import java.security.PublicKey
 
@@ -32,9 +32,9 @@ abstract class AbstractStateReplacementProtocol<T> {
         val stx: SignedTransaction
     }
 
-    class Handshake(val sessionIdForSend: Long,
+    data class Handshake(val sessionIdForSend: Long,
-                    replyTo: Party,
+                         override val replyToParty: Party,
-                    override val sessionID: Long) : AbstractRequestMessage(replyTo)
+                         override val sessionID: Long) : PartyRequestMessage
 
     abstract class Instigator<S : ContractState, T>(val originalState: StateAndRef<S>,
                                                     val modification: T,
@@ -48,9 +48,6 @@ abstract class AbstractStateReplacementProtocol<T> {
             fun tracker() = ProgressTracker(SIGNING, NOTARY)
         }
 
-        abstract val TOPIC_CHANGE: String
-        abstract val TOPIC_INITIATE: String
-
         @Suspendable
         override fun call(): StateAndRef<S> {
             val (stx, participants) = assembleTx()
@@ -88,7 +85,7 @@ abstract class AbstractStateReplacementProtocol<T> {
             }
 
             val allSignatures = participantSignatures + getNotarySignature(stx)
-            sessions.forEach { send(TOPIC_CHANGE, it.key.identity, it.value, allSignatures) }
+            sessions.forEach { send(it.key.identity, it.value, allSignatures) }
 
             return allSignatures
         }
@@ -99,9 +96,9 @@ abstract class AbstractStateReplacementProtocol<T> {
             val proposal = assembleProposal(originalState.ref, modification, stx)
 
             val handshake = Handshake(sessionIdForSend, serviceHub.storageService.myLegalIdentity, sessionIdForReceive)
-            sendAndReceive<Ack>(TOPIC_INITIATE, node.identity, 0, sessionIdForReceive, handshake)
+            sendAndReceive<Ack>(node.identity, 0, sessionIdForReceive, handshake)
 
-            val response = sendAndReceive<Result>(TOPIC_CHANGE, node.identity, sessionIdForSend, sessionIdForReceive, proposal)
+            val response = sendAndReceive<Result>(node.identity, sessionIdForSend, sessionIdForReceive, proposal)
             val participantSignature = response.validate {
                 if (it.sig == null) throw StateReplacementException(it.error!!)
                 else {
@@ -136,13 +133,10 @@ abstract class AbstractStateReplacementProtocol<T> {
             fun tracker() = ProgressTracker(VERIFYING, APPROVING, REJECTING)
         }
 
-        abstract val TOPIC_CHANGE: String
-        abstract val TOPIC_INITIATE: String
-
         @Suspendable
         override fun call() {
             progressTracker.currentStep = VERIFYING
-            val proposal = receive<Proposal<T>>(TOPIC_CHANGE, sessionIdForReceive).validate { it }
+            val proposal = receive<Proposal<T>>(sessionIdForReceive).validate { it }
 
             try {
                 verifyProposal(proposal)
@@ -168,7 +162,7 @@ abstract class AbstractStateReplacementProtocol<T> {
 
             val mySignature = sign(stx)
             val response = Result.noError(mySignature)
-            val swapSignatures = sendAndReceive<List<DigitalSignature.WithKey>>(TOPIC_CHANGE, otherSide, sessionIdForSend, sessionIdForReceive, response)
+            val swapSignatures = sendAndReceive<List<DigitalSignature.WithKey>>(otherSide, sessionIdForSend, sessionIdForReceive, response)
 
             val allSignatures = swapSignatures.validate { signatures ->
                 signatures.forEach { it.verifyWithECDSA(stx.txBits) }
@@ -184,7 +178,7 @@ abstract class AbstractStateReplacementProtocol<T> {
         private fun reject(e: StateReplacementRefused) {
             progressTracker.currentStep = REJECTING
             val response = Result.withError(e)
-            send(TOPIC_CHANGE, otherSide, sessionIdForSend, response)
+            send(otherSide, sessionIdForSend, response)
         }
 
         /**

core/src/main/kotlin/protocols/NotaryChangeProtocol.kt

@@ -16,8 +16,8 @@ import java.security.PublicKey
  * use the new updated state for future transactions.
  */
 object NotaryChangeProtocol: AbstractStateReplacementProtocol<Party>() {
-    val TOPIC_INITIATE = "platform.notary.change.initiate"
+
-    val TOPIC_CHANGE = "platform.notary.change.execute"
+    val TOPIC = "platform.notary.change"
 
     data class Proposal(override val stateRef: StateRef,
                         override val modification: Party,
@@ -28,10 +28,7 @@ object NotaryChangeProtocol: AbstractStateReplacementProtocol<Party>() {
                                         progressTracker: ProgressTracker = tracker())
         : AbstractStateReplacementProtocol.Instigator<T, Party>(originalState, newNotary, progressTracker) {
 
-        override val TOPIC_CHANGE: String
+        override val topic: String get() = TOPIC
-            get() = NotaryChangeProtocol.TOPIC_CHANGE
-        override val TOPIC_INITIATE: String
-            get() = NotaryChangeProtocol.TOPIC_INITIATE
 
         override fun assembleProposal(stateRef: StateRef, modification: Party, stx: SignedTransaction): AbstractStateReplacementProtocol.Proposal<Party>
             = NotaryChangeProtocol.Proposal(stateRef, modification, stx)
@@ -53,10 +50,8 @@ object NotaryChangeProtocol: AbstractStateReplacementProtocol<Party>() {
                    sessionIdForReceive: Long,
                    override val progressTracker: ProgressTracker = tracker())
     : AbstractStateReplacementProtocol.Acceptor<Party>(otherSide, sessionIdForSend, sessionIdForReceive) {
-        override val TOPIC_CHANGE: String
+
-            get() = NotaryChangeProtocol.TOPIC_CHANGE
+        override val topic: String get() = TOPIC
-        override val TOPIC_INITIATE: String
-            get() = NotaryChangeProtocol.TOPIC_INITIATE
 
         /**
          * Check the notary change proposal.

core/src/test/java/com/r3corda/core/protocols/ProtocolLogicRefFromJavaTest.java

@@ -2,6 +2,7 @@ package com.r3corda.core.protocols;
 
 
 import com.google.common.collect.Sets;
+import org.jetbrains.annotations.NotNull;
 import org.junit.Test;
 
 public class ProtocolLogicRefFromJavaTest {
@@ -15,6 +16,12 @@ public class ProtocolLogicRefFromJavaTest {
         public Void call() {
             return null;
         }
+
+        @NotNull
+        @Override
+        protected String getTopic() {
+            throw new UnsupportedOperationException();
+        }
     }
 
     public static class JavaNoArgProtocolLogic extends ProtocolLogic<Void> {
@@ -26,6 +33,12 @@ public class ProtocolLogicRefFromJavaTest {
         public Void call() {
             return null;
         }
+
+        @NotNull
+        @Override
+        protected String getTopic() {
+            throw new UnsupportedOperationException();
+        }
     }
 
     @Test

core/src/test/kotlin/com/r3corda/core/protocols/ProtocolLogicRefTest.kt

@@ -18,17 +18,21 @@ class ProtocolLogicRefTest {
 
         override fun call(): Unit {
         }
+
+        override val topic: String get() = throw UnsupportedOperationException()
     }
 
     class KotlinNoArgProtocolLogic : ProtocolLogic<Unit>() {
         override fun call(): Unit {
         }
+        override val topic: String get() = throw UnsupportedOperationException()
     }
 
     @Suppress("UNUSED_PARAMETER") // We will never use A or b
     class NotWhiteListedKotlinProtocolLogic(A: Int, b: String) : ProtocolLogic<Unit>() {
         override fun call(): Unit {
         }
+        override val topic: String get() = throw UnsupportedOperationException()
     }
 
     lateinit var factory: ProtocolLogicRefFactory

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

@@ -61,8 +61,8 @@ class TradeSimulation(runAsync: Boolean, latencyInjector: InMemoryMessagingNetwo
         showConsensusFor(listOf(buyer, seller, notary))
         showProgressFor(listOf(buyer, seller))
 
-        val buyerFuture = buyer.smm.add("bank.$buyerBankIndex.${TwoPartyTradeProtocol.TRADE_TOPIC}.buyer", buyerProtocol)
+        val buyerFuture = buyer.smm.add("bank.$buyerBankIndex.${TwoPartyTradeProtocol.TOPIC}.buyer", buyerProtocol)
-        val sellerFuture = seller.smm.add("bank.$sellerBankIndex.${TwoPartyTradeProtocol.TRADE_TOPIC}.seller", sellerProtocol)
+        val sellerFuture = seller.smm.add("bank.$sellerBankIndex.${TwoPartyTradeProtocol.TOPIC}.seller", sellerProtocol)
 
         return Futures.successfulAsList(buyerFuture, sellerFuture)
     }

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

@@ -14,7 +14,7 @@ import protocols.NotaryChangeProtocol
  */
 class NotaryChangeService(net: MessagingService, val smm: StateMachineManager, networkMapCache: NetworkMapCache) : AbstractNodeService(net, networkMapCache) {
     init {
-        addMessageHandler(NotaryChangeProtocol.TOPIC_INITIATE,
+        addMessageHandler(NotaryChangeProtocol.TOPIC,
                 { req: AbstractStateReplacementProtocol.Handshake -> handleChangeNotaryRequest(req) }
         )
     }
@@ -24,7 +24,7 @@ class NotaryChangeService(net: MessagingService, val smm: StateMachineManager, n
                 req.replyToParty,
                 req.sessionID,
                 req.sessionIdForSend)
-        smm.add(NotaryChangeProtocol.TOPIC_CHANGE, protocol)
+        smm.add(NotaryChangeProtocol.TOPIC, protocol)
         return Ack
     }
 }

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

@@ -17,6 +17,7 @@ import com.r3corda.node.services.api.AbstractNodeService
 import com.r3corda.node.services.api.AcceptsFileUpload
 import com.r3corda.node.utilities.FiberBox
 import com.r3corda.protocols.RatesFixProtocol
+import com.r3corda.protocols.ServiceRequestMessage
 import org.slf4j.LoggerFactory
 import java.io.InputStream
 import java.math.BigDecimal
@@ -48,32 +49,36 @@ object NodeInterestRates {
         private val logger = LoggerFactory.getLogger(Service::class.java)
 
         init {
-            addMessageHandler(RatesFixProtocol.TOPIC_SIGN,
+            addMessageHandler(RatesFixProtocol.TOPIC,
-                    { req: RatesFixProtocol.SignRequest -> oracle.sign(req.tx) },
+                    { req: ServiceRequestMessage ->
-                    { message, e -> logger.error("Exception during interest rate oracle request processing", e) }
+                        if (req is RatesFixProtocol.SignRequest) {
-            )
+                            oracle.sign(req.tx)
-            addMessageHandler(RatesFixProtocol.TOPIC_QUERY,
+                        }
-                    { req: RatesFixProtocol.QueryRequest ->
+                        else {
-                        /**
+                            /**
-                         * We put this into a protocol so that if it blocks waiting for the interest rate to become
+                             * We put this into a protocol so that if it blocks waiting for the interest rate to become
-                         * available, we a) don't block this thread and b) allow the fact we are waiting
+                             * available, we a) don't block this thread and b) allow the fact we are waiting
-                         * to be persisted/checkpointed.
+                             * to be persisted/checkpointed.
-                         * Interest rates become available when they are uploaded via the web as per [DataUploadServlet],
+                             * Interest rates become available when they are uploaded via the web as per [DataUploadServlet],
-                         * if they haven't already been uploaded that way.
+                             * if they haven't already been uploaded that way.
-                         */
+                             */
-                        node.smm.add("fixing", FixQueryHandler(this, req))
+                            node.smm.add("fixing", FixQueryHandler(this, req as RatesFixProtocol.QueryRequest))
-                        return@addMessageHandler
+                            Unit
+                        }
                     },
                     { message, e -> logger.error("Exception during interest rate oracle request processing", e) }
             )
         }
 
-        private class FixQueryHandler(val service: Service, val request: RatesFixProtocol.QueryRequest) : ProtocolLogic<Unit>() {
+        private class FixQueryHandler(val service: Service,
+                                      val request: RatesFixProtocol.QueryRequest) : ProtocolLogic<Unit>() {
+
             companion object {
                 object RECEIVED : ProgressTracker.Step("Received fix request")
-
                 object SENDING : ProgressTracker.Step("Sending fix response")
             }
+
+            override val topic: String get() = RatesFixProtocol.TOPIC
             override val progressTracker = ProgressTracker(RECEIVED, SENDING)
 
             init {
@@ -84,9 +89,8 @@ object NodeInterestRates {
             override fun call(): Unit {
                 val answers = service.oracle.query(request.queries, request.deadline)
                 progressTracker.currentStep = SENDING
-                send("${RatesFixProtocol.TOPIC}.query", request.replyToParty, request.sessionID, answers)
+                send(request.replyToParty, request.sessionID, answers)
             }
-
         }
 
         // File upload support

node/src/main/kotlin/com/r3corda/node/services/statemachine/StateMachineManager.kt

@@ -253,8 +253,9 @@ class StateMachineManager(val serviceHub: ServiceHubInternal, tokenizableService
         request.payload?.let {
             val topic = "${request.topic}.${request.sessionIDForSend}"
             psm.logger.trace { "Sending message of type ${it.javaClass.name} using topic $topic to ${request.destination} (${it.toString().abbreviate(50)})" }
-            val address = serviceHub.networkMapCache.getNodeByLegalName(request.destination!!.name)!!.address
+            val node = serviceHub.networkMapCache.getNodeByLegalName(request.destination!!.name)
-            serviceHub.networkService.send(topic, it, address)
+            requireNotNull(node) { "Don't know about ${request.destination}" }
+            serviceHub.networkService.send(topic, it, node!!.address)
         }
         if (request is FiberRequest.NotExpectingResponse) {
             // We sent a message, but don't expect a response, so re-enter the continuation to let it keep going.

node/src/main/kotlin/com/r3corda/node/services/transactions/NotaryService.kt

@@ -32,7 +32,7 @@ abstract class NotaryService(val smm: StateMachineManager,
     abstract val protocolFactory: NotaryProtocol.Factory
 
     init {
-        addMessageHandler(NotaryProtocol.TOPIC_INITIATE,
+        addMessageHandler(NotaryProtocol.TOPIC,
                 { req: NotaryProtocol.Handshake -> processRequest(req) }
         )
     }

node/src/test/kotlin/com/r3corda/node/messaging/TwoPartyTradeProtocolTests.kt

@@ -58,14 +58,14 @@ class TwoPartyTradeProtocolTests {
                           otherSide: Party, assetToSell: StateAndRef<OwnableState>, price: Amount<Issued<Currency>>,
                           myKeyPair: KeyPair, buyerSessionID: Long): ListenableFuture<SignedTransaction> {
         val seller = TwoPartyTradeProtocol.Seller(otherSide, notary, assetToSell, price, myKeyPair, buyerSessionID)
-        return smm.add("${TwoPartyTradeProtocol.TRADE_TOPIC}.seller", seller)
+        return smm.add("${TwoPartyTradeProtocol.TOPIC}.seller", seller)
     }
 
     private fun runBuyer(smm: StateMachineManager, notaryNode: NodeInfo,
                          otherSide: Party, acceptablePrice: Amount<Issued<Currency>>, typeToBuy: Class<out OwnableState>,
                          sessionID: Long): ListenableFuture<SignedTransaction> {
         val buyer = TwoPartyTradeProtocol.Buyer(otherSide, notaryNode.identity, acceptablePrice, typeToBuy, sessionID)
-        return smm.add("${TwoPartyTradeProtocol.TRADE_TOPIC}.buyer", buyer)
+        return smm.add("${TwoPartyTradeProtocol.TOPIC}.buyer", buyer)
     }
 
     @Before

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

@@ -68,50 +68,45 @@ class InMemoryNetworkMapServiceTest {
         assert(!service.processRegistrationChangeRequest(NetworkMapService.RegistrationRequest(removeWireChange, mapServiceNode.info.address, Long.MIN_VALUE)).success)
     }
 
-    class TestAcknowledgePSM(val server: NodeInfo, val hash: SecureHash)
+    class TestAcknowledgePSM(val server: NodeInfo, val hash: SecureHash) : ProtocolLogic<Unit>() {
-    : ProtocolLogic<Unit>() {
+        override val topic: String get() = NetworkMapService.PUSH_ACK_PROTOCOL_TOPIC
         @Suspendable
         override fun call() {
             val req = NetworkMapService.UpdateAcknowledge(hash, serviceHub.networkService.myAddress)
-            send(NetworkMapService.PUSH_ACK_PROTOCOL_TOPIC, server.identity, 0, req)
+            send(server.identity, 0, req)
         }
     }
 
     class TestFetchPSM(val server: NodeInfo, val subscribe: Boolean, val ifChangedSinceVersion: Int? = null)
     : ProtocolLogic<Collection<NodeRegistration>?>() {
+        override val topic: String get() = NetworkMapService.FETCH_PROTOCOL_TOPIC
         @Suspendable
         override fun call(): Collection<NodeRegistration>? {
             val sessionID = random63BitValue()
             val req = NetworkMapService.FetchMapRequest(subscribe, ifChangedSinceVersion, serviceHub.networkService.myAddress, sessionID)
-            return sendAndReceive<NetworkMapService.FetchMapResponse>(
+            return sendAndReceive<NetworkMapService.FetchMapResponse>(server.identity, 0, sessionID, req).validate { it.nodes }
-                    NetworkMapService.FETCH_PROTOCOL_TOPIC, server.identity, 0, sessionID, req)
-                    .validate { it.nodes }
         }
     }
 
     class TestRegisterPSM(val server: NodeInfo, val reg: NodeRegistration, val privateKey: PrivateKey)
     : ProtocolLogic<NetworkMapService.RegistrationResponse>() {
+        override val topic: String get() = NetworkMapService.REGISTER_PROTOCOL_TOPIC
         @Suspendable
         override fun call(): NetworkMapService.RegistrationResponse {
             val sessionID = random63BitValue()
             val req = NetworkMapService.RegistrationRequest(reg.toWire(privateKey), serviceHub.networkService.myAddress, sessionID)
-
+            return sendAndReceive<NetworkMapService.RegistrationResponse>(server.identity, 0, sessionID, req).validate { it }
-            return sendAndReceive<NetworkMapService.RegistrationResponse>(
-                    NetworkMapService.REGISTER_PROTOCOL_TOPIC, server.identity, 0, sessionID, req)
-                    .validate { it }
         }
     }
 
     class TestSubscribePSM(val server: NodeInfo, val subscribe: Boolean)
     : ProtocolLogic<NetworkMapService.SubscribeResponse>() {
+        override val topic: String get() = NetworkMapService.SUBSCRIPTION_PROTOCOL_TOPIC
         @Suspendable
         override fun call(): NetworkMapService.SubscribeResponse {
             val sessionID = random63BitValue()
             val req = NetworkMapService.SubscribeRequest(subscribe, serviceHub.networkService.myAddress, sessionID)
-
+            return sendAndReceive<NetworkMapService.SubscribeResponse>(server.identity, 0, sessionID, req).validate { it }
-            return sendAndReceive<NetworkMapService.SubscribeResponse>(
-                    NetworkMapService.SUBSCRIPTION_PROTOCOL_TOPIC, server.identity, 0, sessionID, req)
-                    .validate { it }
         }
     }
 

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

@@ -97,6 +97,7 @@ class NodeSchedulerServiceTest : SingletonSerializeAsToken() {
             (serviceHub as TestReference).testReference.calls += increment
             (serviceHub as TestReference).testReference.countDown.countDown()
         }
+        override val topic: String get() = throw UnsupportedOperationException()
     }
 
     class Command : TypeOnlyCommandData()

node/src/test/kotlin/com/r3corda/node/services/statemachine/StateMachineManagerTests.kt

@@ -59,6 +59,8 @@ class StateMachineManagerTests {
             protocolStarted = true
             Fiber.park()
         }
+
+        override val topic: String get() = throw UnsupportedOperationException()
     }
 
 
@@ -68,6 +70,8 @@ class StateMachineManagerTests {
 
         @Suspendable
         override fun call() {}
+
+        override val topic: String get() = throw UnsupportedOperationException()
     }
 
 

node/src/test/kotlin/node/services/NotaryChangeTests.kt

@@ -11,10 +11,10 @@ import com.r3corda.node.services.network.NetworkMapService
 import com.r3corda.node.services.transactions.SimpleNotaryService
 import org.junit.Before
 import org.junit.Test
-import protocols.StateReplacementException
-import protocols.StateReplacementRefused
 import protocols.NotaryChangeProtocol
 import protocols.NotaryChangeProtocol.Instigator
+import protocols.StateReplacementException
+import protocols.StateReplacementRefused
 import java.util.concurrent.ExecutionException
 import kotlin.test.assertEquals
 import kotlin.test.assertFailsWith
@@ -46,7 +46,7 @@ class NotaryChangeTests {
         val state = issueState(clientNodeA)
         val newNotary = newNotaryNode.info.identity
         val protocol = Instigator(state, newNotary)
-        val future = clientNodeA.smm.add(NotaryChangeProtocol.TOPIC_CHANGE, protocol)
+        val future = clientNodeA.smm.add(NotaryChangeProtocol.TOPIC, protocol)
 
         net.runNetwork()
 
@@ -59,7 +59,7 @@ class NotaryChangeTests {
         val state = issueMultiPartyState(clientNodeA, clientNodeB)
         val newNotary = newNotaryNode.info.identity
         val protocol = Instigator(state, newNotary)
-        val future = clientNodeA.smm.add(NotaryChangeProtocol.TOPIC_CHANGE, protocol)
+        val future = clientNodeA.smm.add(NotaryChangeProtocol.TOPIC, protocol)
 
         net.runNetwork()
 
@@ -75,7 +75,7 @@ class NotaryChangeTests {
         val state = issueMultiPartyState(clientNodeA, clientNodeB)
         val newEvilNotary = Party("Evil Notary", generateKeyPair().public)
         val protocol = Instigator(state, newEvilNotary)
-        val future = clientNodeA.smm.add(NotaryChangeProtocol.TOPIC_CHANGE, protocol)
+        val future = clientNodeA.smm.add(NotaryChangeProtocol.TOPIC, protocol)
 
         net.runNetwork()
 

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

@@ -36,6 +36,7 @@ import java.nio.file.Paths
 import java.security.PublicKey
 import java.time.Instant
 import java.util.*
+import java.util.concurrent.CountDownLatch
 import kotlin.system.exitProcess
 import kotlin.test.assertEquals
 
@@ -202,77 +203,64 @@ private fun runBuyer(node: Node, amount: Amount<Issued<Currency>>) {
         it.storePath
     }
 
-    val future = if (node.isPreviousCheckpointsPresent) {
+    // Self issue some cash.
-        node.smm.findStateMachines(TraderDemoProtocolBuyer::class.java).single().second
+    //
-    } else {
+    // TODO: At some point this demo should be extended to have a central bank node.
+    node.services.fillWithSomeTestCash(3000.DOLLARS, node.info.identity)
+
+    // Wait around until a node asks to start a trade with us. In a real system, this part would happen out of band
+    // via some other system like an exchange or maybe even a manual messaging system like Bloomberg. But for the
+    // next stage in our building site, we will just auto-generate fake trades to give our nodes something to do.
+    //
+    // As the seller initiates the two-party trade protocol, here, we will be the buyer.
+    node.services.networkService.addMessageHandler("$DEMO_TOPIC.0") { message, registration ->
         // We use a simple scenario-specific wrapper protocol to make things happen.
-        val buyer = TraderDemoProtocolBuyer(attachmentsPath, node.info.identity, amount)
+        val otherSide = message.data.deserialize<Party>()
+        val buyer = TraderDemoProtocolBuyer(otherSide, attachmentsPath, amount)
         node.smm.add("demo.buyer", buyer)
     }
 
-    future.get()   // This thread will halt forever here.
+    CountDownLatch(1).await()  // Prevent the application from terminating
 }
 
 // We create a couple of ad-hoc test protocols that wrap the two party trade protocol, to give us the demo logic.
 
 val DEMO_TOPIC = "initiate.demo.trade"
 
-private class TraderDemoProtocolBuyer(private val attachmentsPath: Path,
+private class TraderDemoProtocolBuyer(val otherSide: Party,
-                              val notary: Party,
+                                      private val attachmentsPath: Path,
-                              val amount: Amount<Issued<Currency>>) : ProtocolLogic<Unit>() {
+                                      val amount: Amount<Issued<Currency>>,
-    companion object {
+                                      override val progressTracker: ProgressTracker = ProgressTracker(STARTING_BUY)) : ProtocolLogic<Unit>() {
-        object WAITING_FOR_SELLER_TO_CONNECT : ProgressTracker.Step("Waiting for seller to connect to us")
 
-        object STARTING_BUY : ProgressTracker.Step("Seller connected, purchasing commercial paper asset")
+    object STARTING_BUY : ProgressTracker.Step("Seller connected, purchasing commercial paper asset")
-    }
 
-    override val progressTracker = ProgressTracker(WAITING_FOR_SELLER_TO_CONNECT, STARTING_BUY)
+    override val topic: String get() = DEMO_TOPIC
 
     @Suspendable
     override fun call() {
-        // Self issue some cash.
+        // The session ID disambiguates the test trade.
-        //
+        val sessionID = random63BitValue()
-        // TODO: At some point this demo should be extended to have a central bank node.
+        progressTracker.currentStep = STARTING_BUY
-        serviceHub.fillWithSomeTestCash(3000.DOLLARS, notary)
+        send(otherSide, 0, sessionID)
 
-        while (true) {
+        val notary = serviceHub.networkMapCache.notaryNodes[0]
-            // Wait around until a node asks to start a trade with us. In a real system, this part would happen out of band
+        val buyer = TwoPartyTradeProtocol.Buyer(
-            // via some other system like an exchange or maybe even a manual messaging system like Bloomberg. But for the
+                otherSide,
-            // next stage in our building site, we will just auto-generate fake trades to give our nodes something to do.
+                notary.identity,
-            //
+                amount,
-            // As the seller initiates the two-party trade protocol, here, we will be the buyer.
+                CommercialPaper.State::class.java,
-            try {
+                sessionID)
-                progressTracker.currentStep = WAITING_FOR_SELLER_TO_CONNECT
-                val newPartnerParty = receive<Party>(DEMO_TOPIC, 0).validate {
-                    val ourVersionOfParty = serviceHub.networkMapCache.getNodeByLegalName(it.name)!!.identity
-                    require(ourVersionOfParty == it)
-                    it
-                }
 
-                // The session ID disambiguates the test trade.
+        // This invokes the trading protocol and out pops our finished transaction.
-                val sessionID = random63BitValue()
+        val tradeTX: SignedTransaction = subProtocol(buyer)
-                progressTracker.currentStep = STARTING_BUY
+        // TODO: This should be moved into the protocol itself.
-                send(DEMO_TOPIC, newPartnerParty, 0, sessionID)
+        serviceHub.recordTransactions(listOf(tradeTX))
 
-                val notary = serviceHub.networkMapCache.notaryNodes[0]
+        logger.info("Purchase complete - we are a happy customer! Final transaction is: " +
-                val buyer = TwoPartyTradeProtocol.Buyer(newPartnerParty, notary.identity, amount,
+                "\n\n${Emoji.renderIfSupported(tradeTX.tx)}")
-                        CommercialPaper.State::class.java, sessionID)
 
-                // This invokes the trading protocol and out pops our finished transaction.
+        logIssuanceAttachment(tradeTX)
-                val tradeTX: SignedTransaction = subProtocol(buyer)
+        logBalance()
-                // TODO: This should be moved into the protocol itself.
-                serviceHub.recordTransactions(listOf(tradeTX))
-
-                logger.info("Purchase complete - we are a happy customer! Final transaction is: " +
-                        "\n\n${Emoji.renderIfSupported(tradeTX.tx)}")
-
-                logIssuanceAttachment(tradeTX)
-                logBalance()
-
-            } catch(e: Exception) {
-                logger.error("Something went wrong whilst trading!", e)
-            }
-        }
     }
 
     private fun logBalance() {
@@ -318,11 +306,13 @@ private class TraderDemoProtocolSeller(val otherSide: Party,
         fun tracker() = ProgressTracker(ANNOUNCING, SELF_ISSUING, TRADING)
     }
 
+    override val topic: String get() = DEMO_TOPIC
+
     @Suspendable
     override fun call() {
         progressTracker.currentStep = ANNOUNCING
 
-        val sessionID = sendAndReceive<Long>(DEMO_TOPIC, otherSide, 0, 0, serviceHub.storageService.myLegalIdentity).validate { it }
+        val sessionID = sendAndReceive<Long>(otherSide, 0, 0, serviceHub.storageService.myLegalIdentity).validate { it }
 
         progressTracker.currentStep = SELF_ISSUING
 

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

@@ -81,6 +81,7 @@ object AutoOfferProtocol {
             fun tracker() = ProgressTracker(RECEIVED, ANNOUNCING, DEALING)
         }
 
+        override val topic: String get() = TOPIC
         override val progressTracker = tracker()
 
         init {
@@ -95,17 +96,14 @@ object AutoOfferProtocol {
             val notary = serviceHub.networkMapCache.notaryNodes.first().identity
             // need to pick which ever party is not us
             val otherParty = notUs(*dealToBeOffered.parties).single()
-            val otherNode = (serviceHub.networkMapCache.getNodeByLegalName(otherParty.name))
-            requireNotNull(otherNode) { "Cannot identify other party " + otherParty.name + ", know about: " + serviceHub.networkMapCache.partyNodes.map { it.identity } }
-            val otherSide = otherNode!!.identity
             progressTracker.currentStep = ANNOUNCING
-            send(TOPIC, otherSide, 0, AutoOfferMessage(serviceHub.storageService.myLegalIdentity, notary, ourSessionID, dealToBeOffered))
+            send(otherParty, 0, AutoOfferMessage(serviceHub.storageService.myLegalIdentity, notary, ourSessionID, dealToBeOffered))
             progressTracker.currentStep = DEALING
-            val stx = subProtocol(TwoPartyDealProtocol.Acceptor(otherSide, notary, dealToBeOffered, ourSessionID, progressTracker.getChildProgressTracker(DEALING)!!))
+            val stx = subProtocol(TwoPartyDealProtocol.Acceptor(otherParty, notary, dealToBeOffered, ourSessionID, progressTracker.getChildProgressTracker(DEALING)!!))
             return stx
         }
 
-        fun notUs(vararg parties: Party): List<Party> {
+        private fun notUs(vararg parties: Party): List<Party> {
             val notUsParties: MutableList<Party> = arrayListOf()
             for (party in parties) {
                 if (serviceHub.storageService.myLegalIdentity != party) {

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

@@ -39,6 +39,8 @@ object ExitServerProtocol {
      */
     class Broadcast(@Suppress("PLATFORM_CLASS_MAPPED_TO_KOTLIN") val exitCode: Integer) : ProtocolLogic<Boolean>() {
 
+        override val topic: String get() = TOPIC
+
         @Suspendable
         override fun call(): Boolean {
             if (enabled) {
@@ -60,10 +62,7 @@ object ExitServerProtocol {
             if (recipient.address is MockNetworkMapCache.MockAddress) {
                 // Ignore
             } else {
-                // TODO: messaging ourselves seems to trigger a bug for the time being and we continuously receive messages
+                send(recipient.identity, 0, message)
-                if (recipient.identity != serviceHub.storageService.myLegalIdentity) {
-                    send(TOPIC, recipient.identity, 0, message)
-                }
             }
         }
     }

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

@@ -38,6 +38,8 @@ object UpdateBusinessDayProtocol {
             fun tracker() = ProgressTracker(NOTIFYING)
         }
 
+        override val topic: String get() = TOPIC
+
         @Suspendable
         override fun call(): Unit {
             progressTracker.currentStep = NOTIFYING
@@ -52,7 +54,7 @@ object UpdateBusinessDayProtocol {
             if (recipient.address is MockNetworkMapCache.MockAddress) {
                 // Ignore
             } else {
-                send(TOPIC, recipient.identity, 0, message)
+                send(recipient.identity, 0, message)
             }
         }
     }