ExternalVendorCode/corda
1
0
Fork 0
mirror of https://github.com/corda/corda.git synced 2025-05-02 08:43:15 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Adding optional delay to AMQ message sending via system properties (#185)

Browse Source
This commit is contained in:
Richard Green 2017-01-30 14:39:40 +00:00 committed by GitHub
parent 7b57cbe351
commit e54d6388fd
2 changed files with 39 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
node/src/main/kotlin/net/corda/node/driver/Driver.kt
View File

@ -468,18 +468,30 @@ open class DriverDSL(
else else
"" ""
val javaArgs = listOf( val additionalKeys = listOf("amq.delivery.delay.ms")
path,
"-Dname=${nodeConf.myLegalName}", val systemArgs = mutableMapOf(
"-javaagent:$quasarJarPath", "name" to nodeConf.myLegalName,
debugPortArg, "visualvm.display.name" to "Corda"
"-Dvisualvm.display.name=Corda", )
"-Xmx200m",
"-XX:+UseG1GC", for (key in additionalKeys) {
"-cp", classpath, if (System.getProperty(key) != null) {
className, systemArgs.set(key, System.getProperty(key))
"--base-directory=${nodeConf.baseDirectory}" }
).filter(String::isNotEmpty) }
val javaArgs = listOf(path) +
systemArgs.map { "-D${it.key}=${it.value}" } +
listOf(
"-javaagent:$quasarJarPath",
debugPortArg,
"-Xmx200m",
"-XX:+UseG1GC",
"-cp", classpath,
className,
"--base-directory=${nodeConf.baseDirectory}"
).filter(String::isNotEmpty)
val builder = ProcessBuilder(javaArgs) val builder = ProcessBuilder(javaArgs)
builder.redirectError(Paths.get("error.$className.log").toFile()) builder.redirectError(Paths.get("error.$className.log").toFile())
builder.inheritIO() builder.inheritIO()

23
node/src/main/kotlin/net/corda/node/services/messaging/NodeMessagingClient.kt
View File

@ -15,10 +15,10 @@ import net.corda.node.services.RPCUserService
import net.corda.node.services.api.MessagingServiceInternal import net.corda.node.services.api.MessagingServiceInternal
import net.corda.node.services.config.NodeConfiguration import net.corda.node.services.config.NodeConfiguration
import net.corda.node.services.messaging.ArtemisMessagingComponent.ConnectionDirection.Outbound import net.corda.node.services.messaging.ArtemisMessagingComponent.ConnectionDirection.Outbound
import net.corda.node.services.statemachine.StateMachineManager
import net.corda.node.utilities.* import net.corda.node.utilities.*
import org.apache.activemq.artemis.api.core.ActiveMQObjectClosedException import org.apache.activemq.artemis.api.core.ActiveMQObjectClosedException
import org.apache.activemq.artemis.api.core.Message.HDR_DUPLICATE_DETECTION_ID import org.apache.activemq.artemis.api.core.Message.*
import org.apache.activemq.artemis.api.core.Message.HDR_VALIDATED_USER
import org.apache.activemq.artemis.api.core.SimpleString import org.apache.activemq.artemis.api.core.SimpleString
import org.apache.activemq.artemis.api.core.client.* import org.apache.activemq.artemis.api.core.client.*
import org.bouncycastle.asn1.x500.X500Name import org.bouncycastle.asn1.x500.X500Name
@ -67,6 +67,7 @@ class NodeMessagingClient(override val config: NodeConfiguration,
// confusion. // confusion.
const val TOPIC_PROPERTY = "platform-topic" const val TOPIC_PROPERTY = "platform-topic"
const val SESSION_ID_PROPERTY = "session-id" const val SESSION_ID_PROPERTY = "session-id"
val AMQ_DELAY = Integer.valueOf(System.getProperty("amq.delivery.delay.ms", "0"))
} }
private class InnerState { private class InnerState {
@ -102,12 +103,12 @@ class NodeMessagingClient(override val config: NodeConfiguration,
} }
private val processedMessages: MutableSet<UUID> = Collections.synchronizedSet( private val processedMessages: MutableSet<UUID> = Collections.synchronizedSet(
object : AbstractJDBCHashSet<UUID, Table>(Table, loadOnInit = true) { object : AbstractJDBCHashSet<UUID, Table>(Table, loadOnInit = true) {
override fun elementFromRow(row: ResultRow): UUID = row[table.uuid] override fun elementFromRow(row: ResultRow): UUID = row[table.uuid]
override fun addElementToInsert(insert: InsertStatement, entry: UUID, finalizables: MutableList<() -> Unit>) { override fun addElementToInsert(insert: InsertStatement, entry: UUID, finalizables: MutableList<() -> Unit>) {
insert[table.uuid] = entry insert[table.uuid] = entry
} }
}) })
fun start(rpcOps: RPCOps, userService: RPCUserService) { fun start(rpcOps: RPCOps, userService: RPCUserService) {
state.locked { state.locked {
@ -368,6 +369,11 @@ class NodeMessagingClient(override val config: NodeConfiguration,
writeBodyBufferBytes(message.data) writeBodyBufferBytes(message.data)
// Use the magic deduplication property built into Artemis as our message identity too // Use the magic deduplication property built into Artemis as our message identity too
putStringProperty(HDR_DUPLICATE_DETECTION_ID, SimpleString(message.uniqueMessageId.toString())) putStringProperty(HDR_DUPLICATE_DETECTION_ID, SimpleString(message.uniqueMessageId.toString()))
// For demo purposes - if set then add a delay to messages in order to demonstrate that the flows are doing as intended
if (AMQ_DELAY > 0 && message.topicSession.topic == StateMachineManager.sessionTopic.topic) {
putLongProperty(HDR_SCHEDULED_DELIVERY_TIME, System.currentTimeMillis() + AMQ_DELAY);
}
} }
log.info("Send to: $mqAddress topic: ${message.topicSession.topic} sessionID: ${message.topicSession.sessionID} " + log.info("Send to: $mqAddress topic: ${message.topicSession.topic} sessionID: ${message.topicSession.sessionID} " +
"uuid: ${message.uniqueMessageId}") "uuid: ${message.uniqueMessageId}")
@ -376,6 +382,7 @@ class NodeMessagingClient(override val config: NodeConfiguration,
} }
} }
private fun getMQAddress(target: MessageRecipients): String { private fun getMQAddress(target: MessageRecipients): String {
return if (target == myAddress) { return if (target == myAddress) {
// If we are sending to ourselves then route the message directly to our P2P queue. // If we are sending to ourselves then route the message directly to our P2P queue.