Acknowledge all artemis messages

experimental/flow-worker/src/main/kotlin/net/corda/flowworker/FlowWorker.kt

@@ -56,7 +56,10 @@ class FlowWorker(flowWorkerId: String, private val flowWorkerServiceHub: FlowWor
         val consumer = session.createConsumer(queueName)
         val producer = session.createProducer()
 
-        consumer.setMessageHandler { message -> handleFlowWorkerMessage(message, session, producer) }
+        consumer.setMessageHandler { message ->
+            handleFlowWorkerMessage(message, session, producer)
+            message.acknowledge()
+        }
 
         thread {
             (flowWorkerServiceHub.networkService as P2PMessagingClient).run()

experimental/rpc-worker/src/main/kotlin/net/corda/rpcWorker/CordaRpcWorkerOps.kt

@@ -84,7 +84,10 @@ class CordaRpcWorkerOps(
         }
 
         val consumer = session.createConsumer(rpcWorkerQueueName)
-        consumer.setMessageHandler { message -> handleFlowWorkerMessage(message) }
+        consumer.setMessageHandler { message ->
+            handleFlowWorkerMessage(message)
+            message.acknowledge()
+        }
 
         networkMapFeed().updates.subscribe { mapChange: NetworkMapCache.MapChange? ->
             val networkMapUpdateMessage = NetworkMapUpdate(services.myInfo.legalIdentities.first().name, mapChange!!)

node-api/src/main/kotlin/net/corda/nodeapi/internal/bridging/BridgeControlListener.kt

@@ -91,6 +91,7 @@ class BridgeControlListener(val config: MutualSslConfiguration,
             } catch (ex: Exception) {
                 log.error("Unable to process bridge control message", ex)
             }
+            msg.acknowledge()
         }
     }
 
@@ -113,6 +114,7 @@ class BridgeControlListener(val config: MutualSslConfiguration,
             } catch (ex: Exception) {
                 log.error("Unable to process bridge notification message", ex)
             }
+            msg.acknowledge()
         }
     }
 

node/src/main/kotlin/net/corda/node/services/messaging/RPCServer.kt

@@ -292,6 +292,7 @@ class RPCServer<OPS : RPCOps>(
         val clientAddress = artemisMessage.getStringProperty(ManagementHelper.HDR_ROUTING_NAME)
         log.warn("Detected RPC client disconnect on address $clientAddress, scheduling for reaping")
         invalidateClient(SimpleString(clientAddress))
+        artemisMessage.acknowledge()
     }
 
     private fun bindingAdditionArtemisMessageHandler(artemisMessage: ClientMessage) {
@@ -303,6 +304,7 @@ class RPCServer<OPS : RPCOps>(
 
         val buffer = stopBuffering(clientAddress)
         buffer?.let { drainBuffer(it) }
+        artemisMessage.acknowledge()
     }
 
     /**