ExternalVendorCode/corda
1
0
Fork 0
mirror of https://github.com/corda/corda.git synced 2025-06-22 17:09:00 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

node: change InMemoryMessagingNetwork so that the exposed allMessages stream is in sync with iterate()

Browse Source
This commit is contained in:
Andras Slemmer
2016-06-08 17:28:41 +01:00
parent 53bd5c2287
commit 3bf95429e9
2 changed files with 4 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -214,7 +214,8 @@ abstract class Simulation(val runAsync: Boolean,
} }
} }
val networkInitialisationFinished: ListenableFuture<Unit> = Futures.transform(Futures.allAsList(network.nodes.map { it.networkMapRegistrationFuture }), Function { }) val networkInitialisationFinished: ListenableFuture<Unit> =
Futures.transform(Futures.allAsList(network.nodes.map { it.networkMapRegistrationFuture }), Function { })
fun start(): ListenableFuture<Unit> { fun start(): ListenableFuture<Unit> {
network.startNodes() network.startNodes()

4
node/src/main/kotlin/com/r3corda/node/services/network/InMemoryMessagingNetwork.kt
View File

@ -104,7 +104,6 @@ class InMemoryMessagingNetwork() : SingletonSerializeAsToken() {
} else { } else {
msgSendInternal(transfer) msgSendInternal(transfer)
} }
_allMessages.onNext(MessageTransfer(from, message, recipients))
} }
private fun msgSendInternal(transfer: MessageTransfer) { private fun msgSendInternal(transfer: MessageTransfer) {
@ -256,7 +255,7 @@ class InMemoryMessagingNetwork() : SingletonSerializeAsToken() {
private fun pumpInternal(block: Boolean): MessageTransfer? { private fun pumpInternal(block: Boolean): MessageTransfer? {
val q = getQueueForHandle(handle) val q = getQueueForHandle(handle)
val transfer = (if (block) q.take() else q.poll()) ?: return null val transfer = (if (block) q.take() else q.poll()) ?: return null
System.err.println("T " + transfer.hashCode())
val deliverTo = state.locked { val deliverTo = state.locked {
val h = handlers.filter { if (it.topic.isBlank()) true else transfer.message.topic == it.topic } val h = handlers.filter { if (it.topic.isBlank()) true else transfer.message.topic == it.topic }
@ -277,6 +276,7 @@ class InMemoryMessagingNetwork() : SingletonSerializeAsToken() {
// Now deliver via the requested executor, or on this thread if no executor was provided at registration time. // Now deliver via the requested executor, or on this thread if no executor was provided at registration time.
(handler.executor ?: MoreExecutors.directExecutor()).execute { (handler.executor ?: MoreExecutors.directExecutor()).execute {
try { try {
_allMessages.onNext(transfer)
handler.callback(transfer.message, handler) handler.callback(transfer.message, handler)
} catch(e: Exception) { } catch(e: Exception) {
loggerFor<InMemoryMessagingNetwork>().error("Caught exception in handler for $this/${handler.topic}", e) loggerFor<InMemoryMessagingNetwork>().error("Caught exception in handler for $this/${handler.topic}", e)