ExternalVendorCode/corda
1
0
Fork 0
mirror of https://github.com/corda/corda.git synced 2024-12-20 05:28:21 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Make the AMQP bridge individually ACK artemis messages (so that we don't have to wait for an ack flush or end up with ack'd messages queued up). (#4568)

Browse Source
This commit is contained in:
Christian Sailer 2019-01-14 15:32:08 +00:00 committed by GitHub
parent b505f770c1
commit 3ba2ef3092
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 2 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
node-api/src/main/kotlin/net/corda/nodeapi/internal/bridging/AMQPBridgeManager.kt
View File

@ -158,7 +158,7 @@ class AMQPBridgeManager(config: MutualSslConfiguration, maxMessageSize: Int,
logWarnWithMDC(msg) logWarnWithMDC(msg)
bridgeMetricsService?.packetDropEvent(artemisMessage, msg) bridgeMetricsService?.packetDropEvent(artemisMessage, msg)
// Ack the message to prevent same message being sent to us again. // Ack the message to prevent same message being sent to us again.
artemisMessage.acknowledge() artemisMessage.individualAcknowledge()
return return
} }
val data = ByteArray(artemisMessage.bodySize).apply { artemisMessage.bodyBuffer.readBytes(this) } val data = ByteArray(artemisMessage.bodySize).apply { artemisMessage.bodyBuffer.readBytes(this) }
@ -181,7 +181,7 @@ class AMQPBridgeManager(config: MutualSslConfiguration, maxMessageSize: Int,
logDebugWithMDC { "Bridge ACK ${sendableMessage.onComplete.get()}" } logDebugWithMDC { "Bridge ACK ${sendableMessage.onComplete.get()}" }
lock.withLock { lock.withLock {
if (sendableMessage.onComplete.get() == MessageStatus.Acknowledged) { if (sendableMessage.onComplete.get() == MessageStatus.Acknowledged) {
artemisMessage.acknowledge() artemisMessage.individualAcknowledge()
} else { } else {
logInfoWithMDC("Rollback rejected message uuid: ${artemisMessage.getObjectProperty("_AMQ_DUPL_ID")}") logInfoWithMDC("Rollback rejected message uuid: ${artemisMessage.getObjectProperty("_AMQ_DUPL_ID")}")
// We need to commit any acknowledged messages before rolling back the failed // We need to commit any acknowledged messages before rolling back the failed