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

CORDA-1122 Switch AMQP bridge to use auto-commit sends and acks (#2661)

Browse Source 
* CORDA-1122 Switch AMQP bridge to use auto-commit sends and acks

* CORDA-1122 Codereview: Add comment for extra commit to rollback and add better message sequence logging to tricky test
This commit is contained in:
Christian Sailer 2018-02-27 15:25:58 +00:00 committed by GitHub
parent 4695cb0810
commit 0ff37c0437
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 21 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -108,7 +108,7 @@ class AMQPBridgeManager(config: NodeSSLConfiguration, val artemisMessageClientFa
if (connected) {
log.info("Bridge Connected")
val sessionFactory = artemis.started!!.sessionFactory
val session = sessionFactory.createSession(NODE_USER, NODE_USER, false, false, false, false, DEFAULT_ACK_BATCH_SIZE)
val session = sessionFactory.createSession(NODE_USER, NODE_USER, false, true, true, false, DEFAULT_ACK_BATCH_SIZE)
this.session = session
val consumer = session.createConsumer(queueName)
this.consumer = consumer
@ -146,9 +146,11 @@ class AMQPBridgeManager(config: NodeSSLConfiguration, val artemisMessageClientFa
lock.withLock {
if (sendableMessage.onComplete.get() == MessageStatus.Acknowledged) {
artemisMessage.acknowledge()
session?.commit()
} else {
log.info("Rollback rejected message uuid: ${artemisMessage.getObjectProperty("_AMQ_DUPL_ID")}")
// We need to commit any acknowledged messages before rolling back the failed
// (unacknowledged) message.
session?.commit()
session?.rollback(false)
}
}

20
node/src/integration-test/kotlin/net/corda/node/amqp/AMQPBridgeTest.kt
View File

@ -68,23 +68,33 @@ class AMQPBridgeTest {
val receive = amqpServer.onReceive.toBlocking().iterator
amqpServer.start()
val receivedSequence = mutableListOf<Int>()
fun formatMessage(expected: String, actual: Int, received: List<Int>): String {
return "Expected message with id $expected, got $actual, previous message receive sequence: "
"${received.joinToString(", ", "[", "]")}."
}
val received1 = receive.next()
val messageID1 = received1.applicationProperties["CountProp"] as Int
assertArrayEquals("Test$messageID1".toByteArray(), received1.payload)
assertEquals(0, messageID1)
received1.complete(true) // Accept first message
receivedSequence.add(messageID1)
val received2 = receive.next()
val messageID2 = received2.applicationProperties["CountProp"] as Int
assertArrayEquals("Test$messageID2".toByteArray(), received2.payload)
assertEquals(1, messageID2)
assertEquals(1, messageID2, formatMessage("1", messageID2, receivedSequence))
received2.complete(false) // Reject message
receivedSequence.add(messageID2)
while (true) {
val received3 = receive.next()
val messageID3 = received3.applicationProperties["CountProp"] as Int
assertArrayEquals("Test$messageID3".toByteArray(), received3.payload)
assertNotEquals(0, messageID3)
assertNotEquals(0, messageID3, formatMessage("< 1", messageID3, receivedSequence))
receivedSequence.add(messageID3)
if (messageID3 != 1) { // keep rejecting any batched items following rejection
received3.complete(false)
} else { // beginnings of replay so accept again
@ -97,6 +107,7 @@ class AMQPBridgeTest {
val received4 = receive.next()
val messageID4 = received4.applicationProperties["CountProp"] as Int
assertArrayEquals("Test$messageID4".toByteArray(), received4.payload)
receivedSequence.add(messageID4)
if (messageID4 != 1) { // we may get a duplicate of the rejected message, in which case skip
assertEquals(2, messageID4) // next message should be in order though
break
@ -118,13 +129,16 @@ class AMQPBridgeTest {
val received5 = receive.next()
val messageID5 = received5.applicationProperties["CountProp"] as Int
if (messageID5 != 2) { // we may get a duplicate of the interrupted message, in which case skip
assertEquals(-1, messageID5) // next message should be in order though
assertEquals(-1, messageID5, formatMessage("-1", messageID5, receivedSequence)) // next message should be in order though
assertArrayEquals("Test_end".toByteArray(), received5.payload)
receivedSequence.add(messageID5)
break
}
receivedSequence.add(messageID5)
received5.complete(true)
}
println("Message sequence: ${receivedSequence.joinToString(", ", "[", "]")}")
bridgeManager.stop()
amqpServer.stop()
artemisClient.stop()