ARTEMIS-2073 - make sure connection gets set for interceptors

https://issues.apache.org/jira/browse/ARTEMIS-2073 (cherry picked from commit f4b77a2)
apache · Sep 4, 2018 · 170fa4c · 170fa4c
1 parent 1b9062b
commit 170fa4c
Show file tree

Hide file tree

Showing 2 changed files with 47 additions and 0 deletions.
diff --git a/...va/org/apache/activemq/artemis/protocol/amqp/broker/ActiveMQProtonRemotingConnection.java b/...va/org/apache/activemq/artemis/protocol/amqp/broker/ActiveMQProtonRemotingConnection.java
@@ -50,6 +50,7 @@ public ActiveMQProtonRemotingConnection(ProtonProtocolManager manager,
       super(transportConnection, executor);
       this.manager = manager;
       this.amqpConnection = amqpConnection;
+      transportConnection.setProtocolConnection(this);
    }
 
    public Executor getExecutor() {

diff --git a/...va/org/apache/activemq/artemis/tests/integration/amqp/AmqpSendReceiveInterceptorTest.java b/...va/org/apache/activemq/artemis/tests/integration/amqp/AmqpSendReceiveInterceptorTest.java
@@ -176,4 +176,50 @@ public boolean intercept(AMQPMessage packet, RemotingConnection connection) thro
       receiver.close();
       connection.close();
    }
+
+   @Test(timeout = 60000)
+   public void testCheckRemotingConnection() throws Exception {
+      final CountDownLatch latch = new CountDownLatch(1);
+      final boolean[] passed = {false};
+      server.getRemotingService().addIncomingInterceptor(new AmqpInterceptor() {
+         @Override
+         public boolean intercept(AMQPMessage message, RemotingConnection connection) throws ActiveMQException {
+            passed[0] = connection != null;
+            latch.countDown();
+            return true;
+         }
+      });
+
+      AmqpClient client = createAmqpClient();
+      AmqpConnection connection = addConnection(client.connect());
+      AmqpSession session = connection.createSession();
+
+      AmqpSender sender = session.createSender(getTestName());
+      AmqpMessage message = new AmqpMessage();
+      message.setMessageId("msg" + 1);
+      message.setText("Test-Message");
+      sender.send(message);
+
+      assertTrue(latch.await(2, TimeUnit.SECONDS));
+      assertTrue("connection not set", passed[0]);
+
+      final CountDownLatch latch2 = new CountDownLatch(1);
+      server.getRemotingService().addOutgoingInterceptor(new AmqpInterceptor() {
+         @Override
+         public boolean intercept(AMQPMessage packet, RemotingConnection connection) throws ActiveMQException {
+            passed[0] = connection != null;
+            latch2.countDown();
+            return true;
+         }
+      });
+      AmqpReceiver receiver = session.createReceiver(getTestName());
+      receiver.flow(2);
+      AmqpMessage amqpMessage = receiver.receive(5, TimeUnit.SECONDS);
+      assertNotNull(amqpMessage);
+      assertEquals(latch2.getCount(), 0);
+      assertTrue("connection not set", passed[0]);
+      sender.close();
+      receiver.close();
+      connection.close();
+   }
 }