ExternalVendorCode/corda
1
0
Fork 0
mirror of https://github.com/corda/corda.git synced 2025-01-31 00:24:59 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

This fixes the node startup issue - https://github.com/corda/corda/issues/37 (#48)

Browse Source 
* Create missing artemis bridge when node is added to the network map
* Added integration test for this issue
* addressed PR issues
This commit is contained in:
Patrick Kuo 2016-12-12 13:29:31 +00:00 committed by GitHub
parent fc69624f41
commit a4201c7152
2 changed files with 49 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
node/src/integration-test/kotlin/net/corda/services/messaging/ArtemisMessagingServerTest.kt Normal file
View File

@ -0,0 +1,29 @@
package net.corda.services.messaging
import net.corda.node.driver.driver
import org.junit.Test
import java.nio.file.Paths
import java.time.Instant
import java.time.ZoneOffset
import java.time.format.DateTimeFormatter
import java.util.concurrent.TimeUnit
class ArtemisMessagingServerTest {
@Test
fun `network map will work after restart`() {
val dir = Paths.get("build", getTimestampAsDirectoryName())
// Start the network map.
driver(driverDirectory = dir) {
arrayOf(startNode("NodeA"), startNode("NodeB"), startNode("Notary")).forEach { it.get() }
}
// Start the network map second time, this will restore message queues from the journal.
// This will hang and fail prior the fix. https://github.com/corda/corda/issues/37
driver(driverDirectory = dir) {
arrayOf(startNode("NodeA"), startNode("NodeB"), startNode("Notary")).forEach { it.get(5, TimeUnit.MINUTES) }
}
}
private fun getTimestampAsDirectoryName(): String {
return DateTimeFormatter.ofPattern("yyyyMMddHHmmss").withZone(ZoneOffset.UTC).format(Instant.now())
}
}

27
node/src/main/kotlin/net/corda/node/services/messaging/ArtemisMessagingServer.kt
View File

@ -85,10 +85,14 @@ class ArtemisMessagingServer(override val config: NodeConfiguration,
config.basedir.expectedOnDefaultFileSystem()
}
/**
* The server will make sure the bridge exists on network map changes, see method [destroyOrCreateBridge]
* We assume network map will be updated accordingly when the client node register with the network map server.
*/
fun start() = mutex.locked {
if (!running) {
configureAndStartServer()
networkChangeHandle = networkMapCache.changed.subscribe { destroyPossibleStaleBridge(it) }
networkChangeHandle = networkMapCache.changed.subscribe { destroyOrCreateBridge(it) }
running = true
}
}
@ -108,13 +112,22 @@ class ArtemisMessagingServer(override val config: NodeConfiguration,
maybeDeployBridgeForAddress(networkMapService)
}
private fun destroyPossibleStaleBridge(change: MapChange) {
val staleNodeInfo = when (change) {
is MapChange.Modified -> change.previousNode
is MapChange.Removed -> change.node
is MapChange.Added -> return
/**
* The bridge will be created automatically when the queues are created, however, this is not the case when the network map restarted.
* The queues are restored from the journal, and because the queues are added before we register the callback handler, this method will never get called for existing queues.
* This results in message queues up and never get send out. (https://github.com/corda/corda/issues/37)
*
* We create the bridges indirectly now because the network map is not persisted and there are no ways to obtain host and port information on startup.
* TODO : Create the bridge directly from the list of queues on start up when we have a persisted network map service.
*/
private fun destroyOrCreateBridge(change: MapChange) {
val (newNode, staleNode) = when (change) {
is MapChange.Modified -> change.node to change.previousNode
is MapChange.Removed -> null to change.node
is MapChange.Added -> change.node to null
}
(staleNodeInfo.address as? ArtemisAddress)?.let { maybeDestroyBridge(it.queueName) }
(staleNode?.address as? ArtemisAddress)?.let { maybeDestroyBridge(it.queueName) }
(newNode?.address as? ArtemisAddress)?.let { if (activeMQServer.queueQuery(it.queueName).isExists) maybeDeployBridgeForAddress(it) }
}
private fun configureAndStartServer() {