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

ENT-2116: handle amqp client remote error, added test (#3636)

Browse Source 
* ENT-2116: handle amqp client remote error, added test

* ENT: 2116 rename test, added logging

* ENT-2116: rename test to indicate its purpose
This commit is contained in:
bpaunescu 2018-07-18 16:07:42 +01:00 committed by GitHub
parent 08b5cb6d5f
commit 53b398a460
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 37 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
node-api/src/main/kotlin/net/corda/nodeapi/internal/protonwrapper/engine/ConnectionStateMachine.kt
View File

@ -66,6 +66,8 @@ internal class ConnectionStateMachine(private val serverMode: Boolean,
private fun logInfoWithMDC(msg: String) = withMDC { log.info(msg) } private fun logInfoWithMDC(msg: String) = withMDC { log.info(msg) }
private fun logWarnWithMDC(msg: String, ex: Throwable? = null) = withMDC { log.warn(msg, ex) }
private fun logErrorWithMDC(msg: String, ex: Throwable? = null) = withMDC { log.error(msg, ex) } private fun logErrorWithMDC(msg: String, ex: Throwable? = null) = withMDC { log.error(msg, ex) }
val connection: Connection val connection: Connection
@ -308,6 +310,16 @@ internal class ConnectionStateMachine(private val serverMode: Boolean,
} }
} }
override fun onLinkRemoteClose(e: Event) {
val link = e.link
if(link.remoteCondition != null) {
logWarnWithMDC("Connection closed due to error on remote side: `${link.remoteCondition.description}`")
transport.condition = link.condition
transport.close_tail()
transport.pop(Math.max(0, transport.pending())) // Force generation of TRANSPORT_HEAD_CLOSE (not in C code)
}
}
override fun onLinkFinal(event: Event) { override fun onLinkFinal(event: Event) {
val link = event.link val link = event.link
if (link is Sender) { if (link is Sender) {

25
node/src/integration-test/kotlin/net/corda/node/amqp/ProtonWrapperTests.kt
View File

@ -362,6 +362,31 @@ class ProtonWrapperTests {
} }
} }
@Test
fun `Message sent from AMQP to non-existent Artemis inbox is rejected and client disconnects`() {
val (server, artemisClient) = createArtemisServerAndClient()
val amqpClient = createClient()
var connected = false
amqpClient.onConnection.subscribe { change ->
connected = change.connected
}
val clientConnected = amqpClient.onConnection.toFuture()
amqpClient.start()
assertEquals(true, clientConnected.get().connected)
assertEquals(CHARLIE_NAME, CordaX500Name.build(clientConnected.get().remoteCert!!.subjectX500Principal))
val sendAddress = P2P_PREFIX + "Test"
val testData = "Test".toByteArray()
val testProperty = mutableMapOf<String, Any?>()
testProperty["TestProp"] = "1"
val message = amqpClient.createMessage(testData, sendAddress, CHARLIE_NAME.toString(), testProperty)
amqpClient.write(message)
assertEquals(MessageStatus.Rejected, message.onComplete.get())
assertEquals(false, connected)
amqpClient.stop()
artemisClient.stop()
server.stop()
}
private fun createArtemisServerAndClient(maxMessageSize: Int = MAX_MESSAGE_SIZE): Pair<ArtemisMessagingServer, ArtemisMessagingClient> { private fun createArtemisServerAndClient(maxMessageSize: Int = MAX_MESSAGE_SIZE): Pair<ArtemisMessagingServer, ArtemisMessagingClient> {
val artemisConfig = rigorousMock<AbstractNodeConfiguration>().also { val artemisConfig = rigorousMock<AbstractNodeConfiguration>().also {
doReturn(temporaryFolder.root.toPath() / "artemis").whenever(it).baseDirectory doReturn(temporaryFolder.root.toPath() / "artemis").whenever(it).baseDirectory