You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cs...@apache.org on 2026/10/17 20:28:37 UTC
(activemq) branch activemq-5.19.x updated: Ensure TransportConnection validates connection id (#2682) (#2694)
This is an automated email from the ASF dual-hosted git repository.
cshannon pushed a commit to branch activemq-5.19.x
in repository https://gitbox.apache.org/repos/asf/activemq.git
The following commit(s) were added to refs/heads/activemq-5.19.x by this push:
new ce5794bd44 Ensure TransportConnection validates connection id (#2682) (#2694)
ce5794bd44 is described below
commit ce5794bd44b7636b63ea4736eb5419472d3cb4c0
Author: Christopher L. Shannon <cs...@apache.org>
AuthorDate: Sat Oct 17 16:28:32 2026 -0400
Ensure TransportConnection validates connection id (#2682) (#2694)
Update the Single transport connection state register to check that the
passed connection id matches what the register is tracking before
returning. If a mismatch is found an error is thrown.
(cherry picked from commit afdbf624c92d60a5adef62d7fc584bb5d09fb33b)
---
.../SingleTransportConnectionStateRegister.java | 98 +++++---
.../activemq/broker/TransportConnection.java | 4 +
...SingleTransportConnectionStateRegisterTest.java | 270 ++++++++++++++++++++
.../activemq/broker/XARecoveryBrokerTest.java | 3 +
.../TransportConnectionStateValidationTest.java | 275 +++++++++++++++++++++
5 files changed, 609 insertions(+), 41 deletions(-)
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/SingleTransportConnectionStateRegister.java b/activemq-broker/src/main/java/org/apache/activemq/broker/SingleTransportConnectionStateRegister.java
index 159c6e6d09..4f5743addf 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/SingleTransportConnectionStateRegister.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/SingleTransportConnectionStateRegister.java
@@ -23,6 +23,7 @@ import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
+import java.util.Objects;
import org.apache.activemq.command.ConnectionId;
import org.apache.activemq.command.ConsumerId;
import org.apache.activemq.command.ProducerId;
@@ -32,35 +33,35 @@ import org.apache.activemq.command.SessionId;
*
*/
-public class SingleTransportConnectionStateRegister implements TransportConnectionStateRegister{
+public class SingleTransportConnectionStateRegister implements TransportConnectionStateRegister{
- private TransportConnectionState connectionState;
- private ConnectionId connectionId;
+ // Only store connectionState because it contains the connectionId inside
+ // ConnectionInfo
+ private TransportConnectionState connectionState;
- public TransportConnectionState registerConnectionState(ConnectionId connectionId,
+ public synchronized TransportConnectionState registerConnectionState(ConnectionId connectionId,
TransportConnectionState state) {
- TransportConnectionState rc = connectionState;
- connectionState = state;
- this.connectionId = connectionId;
- return rc;
+ Objects.requireNonNull(connectionId, "connectionId must not be null");
+ Objects.requireNonNull(state, "state must not be null");
+ Objects.requireNonNull(state.getInfo(), "state must not have a null ConnectionInfo");
+ if (!connectionId.equals(state.getInfo().getConnectionId())) {
+ throw new IllegalArgumentException("connectionId and stateInfo must have same connectionId");
+ }
+ this.connectionState = state;
+ return connectionState;
}
public synchronized TransportConnectionState unregisterConnectionState(ConnectionId connectionId) {
- TransportConnectionState rc = null;
-
-
- if (connectionId != null && connectionState != null && this.connectionId!=null){
- if (this.connectionId.equals(connectionId)){
- rc = connectionState;
- connectionState = null;
- connectionId = null;
- }
+ TransportConnectionState existing = findConnectionState(connectionId);
+ if (existing != null) {
+ this.connectionState = null;
+ return existing;
}
- return rc;
+ return null;
}
public synchronized List<TransportConnectionState> listConnectionStates() {
- List<TransportConnectionState> rc = new ArrayList<TransportConnectionState>();
+ List<TransportConnectionState> rc = new ArrayList<>();
if (connectionState != null) {
rc.add(connectionState);
}
@@ -68,7 +69,7 @@ public class SingleTransportConnectionStateRegister implements TransportConnect
}
public synchronized TransportConnectionState lookupConnectionState(String connectionId) {
- TransportConnectionState cs = connectionState;
+ TransportConnectionState cs = findConnectionState(connectionId);
if (cs == null) {
throw new IllegalStateException(
"Cannot lookup a connectionId for a connection that had not been registered: "
@@ -78,7 +79,7 @@ public class SingleTransportConnectionStateRegister implements TransportConnect
}
public synchronized TransportConnectionState lookupConnectionState(ConsumerId id) {
- TransportConnectionState cs = connectionState;
+ TransportConnectionState cs = findConnectionState(id.getConnectionId());
if (cs == null) {
throw new IllegalStateException(
"Cannot lookup a consumer from a connection that had not been registered: "
@@ -88,7 +89,7 @@ public class SingleTransportConnectionStateRegister implements TransportConnect
}
public synchronized TransportConnectionState lookupConnectionState(ProducerId id) {
- TransportConnectionState cs = connectionState;
+ TransportConnectionState cs = findConnectionState(id.getConnectionId());
if (cs == null) {
throw new IllegalStateException(
"Cannot lookup a producer from a connection that had not been registered: "
@@ -98,7 +99,7 @@ public class SingleTransportConnectionStateRegister implements TransportConnect
}
public synchronized TransportConnectionState lookupConnectionState(SessionId id) {
- TransportConnectionState cs = connectionState;
+ TransportConnectionState cs = findConnectionState(id.getConnectionId());
if (cs == null) {
throw new IllegalStateException(
"Cannot lookup a session from a connection that had not been registered: "
@@ -108,8 +109,11 @@ public class SingleTransportConnectionStateRegister implements TransportConnect
}
public synchronized TransportConnectionState lookupConnectionState(ConnectionId connectionId) {
- TransportConnectionState cs = connectionState;
- return cs;
+ // If null then this will just return a null value, this is expected in some cases
+ // (such as duplex bridging)
+ // This could be improved in the future to error if not registered, but the important
+ // thing is to not return an existing registered state, unless it matches the connectionId
+ return findConnectionState(connectionId);
}
public synchronized boolean doesHandleMultipleConnectionStates() {
@@ -120,32 +124,44 @@ public class SingleTransportConnectionStateRegister implements TransportConnect
return connectionState == null;
}
- public void intialize(TransportConnectionStateRegister other) {
-
+ public synchronized void intialize(TransportConnectionStateRegister other) {
if (other.isEmpty()){
clear();
- }else{
- Map map = other.mapStates();
- Iterator i = map.entrySet().iterator();
- Map.Entry<ConnectionId, TransportConnectionState> entry = (Entry<ConnectionId, TransportConnectionState>) i.next();
- connectionId = entry.getKey();
- connectionState =entry.getValue();
+ } else {
+ Map<ConnectionId, TransportConnectionState> map = other.mapStates();
+ Iterator<Entry<ConnectionId, TransportConnectionState>> i = map.entrySet().iterator();
+ Map.Entry<ConnectionId, TransportConnectionState> entry = i.next();
+ connectionState = entry.getValue();
}
-
}
- public Map<ConnectionId, TransportConnectionState> mapStates() {
- Map<ConnectionId, TransportConnectionState> map = new HashMap<ConnectionId, TransportConnectionState>();
+ public synchronized Map<ConnectionId, TransportConnectionState> mapStates() {
+ Map<ConnectionId, TransportConnectionState> map = new HashMap<>();
if (!isEmpty()) {
- map.put(connectionId, connectionState);
+ map.put(getConnectionId(), connectionState);
}
return map;
}
- public void clear() {
- connectionState=null;
- connectionId=null;
-
+ public synchronized void clear() {
+ connectionState = null;
}
+ private ConnectionId getConnectionId() {
+ return !isEmpty() && connectionState.getInfo() != null ?
+ connectionState.getInfo().getConnectionId() : null;
+ }
+
+ private TransportConnectionState findConnectionState(String connectionId) {
+ return findConnectionState(new ConnectionId(connectionId));
+ }
+
+ private TransportConnectionState findConnectionState(ConnectionId connectionId) {
+ // Only return the state if provided id is not null, and it matches the
+ // existing connectionId of the state
+ if (connectionId != null && connectionId.equals(getConnectionId())) {
+ return connectionState;
+ }
+ return null;
+ }
}
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnection.java b/activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnection.java
index e753c1712f..40d1f69cc2 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnection.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnection.java
@@ -1705,6 +1705,10 @@ public class TransportConnection implements Connection, Task, CommandVisitor {
@Override
public Response processConsumerControl(ConsumerControl control) throws Exception {
ConsumerBrokerExchange consumerExchange = getConsumerBrokerExchange(control.getConsumerId());
+ if (consumerExchange == null) {
+ throw new IllegalStateException(broker.getBrokerName()
+ + " Cannot process ConsumerControl for a consumer that had not been registered: " + control.getConsumerId());
+ }
broker.processConsumerControl(consumerExchange, control);
return null;
}
diff --git a/activemq-broker/src/test/java/org/apache/activemq/broker/SingleTransportConnectionStateRegisterTest.java b/activemq-broker/src/test/java/org/apache/activemq/broker/SingleTransportConnectionStateRegisterTest.java
new file mode 100644
index 0000000000..8f5334b80a
--- /dev/null
+++ b/activemq-broker/src/test/java/org/apache/activemq/broker/SingleTransportConnectionStateRegisterTest.java
@@ -0,0 +1,270 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.broker;
+
+import java.util.List;
+import java.util.Map;
+
+import org.apache.activemq.command.ConnectionId;
+import org.apache.activemq.command.ConnectionInfo;
+import org.apache.activemq.command.ConsumerId;
+import org.apache.activemq.command.ProducerId;
+import org.apache.activemq.command.SessionId;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class SingleTransportConnectionStateRegisterTest {
+
+ private static final String CONNECTION_ID = "test-connection-id";
+ private final ConnectionId connectionId = new ConnectionId(CONNECTION_ID);
+ private SingleTransportConnectionStateRegister register;
+
+ @Before
+ public void setUp() {
+ register = new SingleTransportConnectionStateRegister();
+ }
+
+ @Test
+ public void testRegisterConnectionState() {
+ TransportConnectionState state = createState(connectionId);
+ TransportConnectionState result = register.registerConnectionState(connectionId, state);
+
+ assertEquals("Should return the registered state", state, result);
+ assertEquals("Should have one connection state", 1, register.listConnectionStates().size());
+ }
+
+ @Test(expected = NullPointerException.class)
+ public void testRegisterConnectionStateNullConnectionId() {
+ TransportConnectionState state = createState(connectionId);
+ register.registerConnectionState(null, state);
+ }
+
+ @Test(expected = NullPointerException.class)
+ public void testRegisterConnectionStateNullState() {
+ register.registerConnectionState(connectionId, null);
+ }
+
+ @Test(expected = NullPointerException.class)
+ public void testRegisterConnectionStateNullInfo() {
+ register.registerConnectionState(connectionId,
+ new TransportConnectionState(null, null));
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testRegisterConnectionStateMismatch() {
+ register.registerConnectionState(connectionId, createState(new ConnectionId("connection-id-2")));
+ }
+
+ @Test
+ public void testUnregisterConnectionState() {
+ TransportConnectionState state = createState(connectionId);
+
+ register.registerConnectionState(connectionId, state);
+ TransportConnectionState result = register.unregisterConnectionState(connectionId);
+
+ assertEquals("Should return the unregistered state", state, result);
+ assertTrue("Should be empty after unregistering", register.isEmpty());
+ }
+
+ @Test
+ public void testUnregisterConnectionStateNotFound() {
+ TransportConnectionState result = register.unregisterConnectionState(connectionId);
+ assertNull("Should return null when unregistering non-existent connection", result);
+ }
+
+ @Test
+ public void testListConnectionStatesEmpty() {
+ List<TransportConnectionState> result = register.listConnectionStates();
+ assertTrue("List should be empty", result.isEmpty());
+ }
+
+ @Test
+ public void testListConnectionStates() {
+ TransportConnectionState state = createState(connectionId);
+
+ register.registerConnectionState(connectionId, state);
+ List<TransportConnectionState> result = register.listConnectionStates();
+
+ assertEquals("List should contain one state", 1, result.size());
+ assertEquals("State should be the registered one", state, result.get(0));
+ }
+
+ @Test
+ public void testLookupConnectionStateString() {
+ TransportConnectionState state = createState(connectionId);
+
+ register.registerConnectionState(connectionId, state);
+ TransportConnectionState result = register.lookupConnectionState(CONNECTION_ID);
+
+ assertEquals("Should return the registered state", state, result);
+ }
+
+ @Test(expected = IllegalStateException.class)
+ public void testLookupConnectionStateStringNotFound() {
+ register.lookupConnectionState(CONNECTION_ID);
+ }
+
+ @Test
+ public void testLookupConnectionStateConsumerId() {
+ TransportConnectionState state = createState(connectionId);
+ ConsumerId consumerId = new ConsumerId();
+ consumerId.setConnectionId(CONNECTION_ID);
+ consumerId.setSessionId(1);
+ consumerId.setValue(1);
+
+ register.registerConnectionState(connectionId, state);
+ TransportConnectionState result = register.lookupConnectionState(consumerId);
+
+ assertEquals("Should return the registered state", state, result);
+ }
+
+ @Test(expected = IllegalStateException.class)
+ public void testLookupConnectionStateConsumerIdNotFound() {
+ ConsumerId consumerId = new ConsumerId();
+ consumerId.setConnectionId(CONNECTION_ID);
+
+ register.lookupConnectionState(consumerId);
+ }
+
+ @Test
+ public void testLookupConnectionStateProducerId() {
+ TransportConnectionState state = createState(connectionId);
+ ProducerId producerId = new ProducerId();
+ producerId.setConnectionId(CONNECTION_ID);
+ producerId.setSessionId(1);
+ producerId.setValue(1);
+
+ register.registerConnectionState(connectionId, state);
+ TransportConnectionState result = register.lookupConnectionState(producerId);
+
+ assertEquals("Should return the registered state", state, result);
+ }
+
+ @Test(expected = IllegalStateException.class)
+ public void testLookupConnectionStateProducerIdNotFound() {
+ ProducerId producerId = new ProducerId();
+ producerId.setConnectionId(CONNECTION_ID);
+
+ register.lookupConnectionState(producerId);
+ }
+
+ @Test
+ public void testLookupConnectionStateSessionId() {
+ TransportConnectionState state = createState(connectionId);
+ SessionId sessionId = new SessionId();
+ sessionId.setConnectionId(CONNECTION_ID);
+ sessionId.setValue(1);
+
+ register.registerConnectionState(connectionId, state);
+ TransportConnectionState result = register.lookupConnectionState(sessionId);
+
+ assertEquals("Should return the registered state", state, result);
+ }
+
+ @Test(expected = IllegalStateException.class)
+ public void testLookupConnectionStateSessionIdNotFound() {
+ SessionId sessionId = new SessionId();
+ sessionId.setConnectionId(CONNECTION_ID);
+
+ register.lookupConnectionState(sessionId);
+ }
+
+ @Test
+ public void testLookupConnectionStateConnectionId() {
+ TransportConnectionState state = createState(connectionId);
+
+ register.registerConnectionState(connectionId, state);
+ TransportConnectionState result = register.lookupConnectionState(connectionId);
+
+ assertEquals("Should return the registered state", state, result);
+ }
+
+ @Test
+ public void testLookupConnectionStateConnectionIdNotFound() {
+ TransportConnectionState result = register.lookupConnectionState(connectionId);
+ assertNull("Should return null when not registered", result);
+ }
+
+ @Test
+ public void testDoesHandleMultipleConnectionStates() {
+ assertFalse("Should not handle multiple connection states", register.doesHandleMultipleConnectionStates());
+ }
+
+ @Test
+ public void testIsEmpty() {
+ assertTrue("Should be empty initially", register.isEmpty());
+ register.registerConnectionState(connectionId, createState(connectionId));
+ assertFalse("Should not be empty after registering", register.isEmpty());
+ }
+
+ @Test
+ public void testClear() {
+ register.registerConnectionState(connectionId, createState(connectionId));
+ register.clear();
+
+ assertTrue("Should be empty after clear", register.isEmpty());
+ }
+
+ @Test
+ public void testInitializeEmpty() {
+ SingleTransportConnectionStateRegister other = new SingleTransportConnectionStateRegister();
+ register.intialize(other);
+
+ assertTrue("Should be empty after initializing from empty register", register.isEmpty());
+ }
+
+ @Test
+ public void testInitializeFromOther() {
+ SingleTransportConnectionStateRegister other = new SingleTransportConnectionStateRegister();
+ TransportConnectionState state = createState(connectionId);
+
+ other.registerConnectionState(connectionId, state);
+ register.intialize(other);
+
+ assertEquals("Should have the state from other register", state,
+ register.lookupConnectionState(connectionId));
+ assertEquals("Should have one connection state", 1, register.listConnectionStates().size());
+ }
+
+ @Test
+ public void testMapStatesEmpty() {
+ Map<ConnectionId, TransportConnectionState> result = register.mapStates();
+ assertTrue("Map should be empty", result.isEmpty());
+ }
+
+ @Test
+ public void testMapStatesWithConnection() {
+ TransportConnectionState state = createState(connectionId);
+
+ register.registerConnectionState(connectionId, state);
+ Map<ConnectionId, TransportConnectionState> result = register.mapStates();
+
+ assertEquals("Map should contain one entry", 1, result.size());
+ assertEquals("State should be the registered one", state, result.get(connectionId));
+ }
+
+ private TransportConnectionState createState(ConnectionId connectionId) {
+ return new TransportConnectionState(new ConnectionInfo(connectionId), null);
+ }
+
+}
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/broker/XARecoveryBrokerTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/XARecoveryBrokerTest.java
index 9e174c2ff0..fe51fcd7cd 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/broker/XARecoveryBrokerTest.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/XARecoveryBrokerTest.java
@@ -1497,6 +1497,7 @@ public class XARecoveryBrokerTest extends BrokerRestartTestSupport {
// unsubscribe
connection.request(consumerInfo.createRemoveCommand());
RemoveSubscriptionInfo removeSubscriptionInfo = new RemoveSubscriptionInfo();
+ removeSubscriptionInfo.setConnectionId(connectionInfo.getConnectionId());
removeSubscriptionInfo.setClientId(connectionInfo.getClientId());
removeSubscriptionInfo.setSubscriptionName(consumerInfo.getSubscriptionName());
connection.request(removeSubscriptionInfo);
@@ -1579,6 +1580,7 @@ public class XARecoveryBrokerTest extends BrokerRestartTestSupport {
// unsubscribe
connection.request(consumerInfo.createRemoveCommand());
RemoveSubscriptionInfo removeSubscriptionInfo = new RemoveSubscriptionInfo();
+ removeSubscriptionInfo.setConnectionId(connectionInfo.getConnectionId());
removeSubscriptionInfo.setClientId(connectionInfo.getClientId());
removeSubscriptionInfo.setSubscriptionName(consumerInfo.getSubscriptionName());
connection.request(removeSubscriptionInfo);
@@ -1671,6 +1673,7 @@ public class XARecoveryBrokerTest extends BrokerRestartTestSupport {
// unsubscribe
connection.request(consumerInfo.createRemoveCommand());
RemoveSubscriptionInfo removeSubscriptionInfo = new RemoveSubscriptionInfo();
+ removeSubscriptionInfo.setConnectionId(connectionInfo.getConnectionId());
removeSubscriptionInfo.setClientId(connectionInfo.getClientId());
removeSubscriptionInfo.setSubscriptionName(consumerInfo.getSubscriptionName());
connection.request(removeSubscriptionInfo);
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/transport/TransportConnectionStateValidationTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/transport/TransportConnectionStateValidationTest.java
new file mode 100644
index 0000000000..a823ba6864
--- /dev/null
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/transport/TransportConnectionStateValidationTest.java
@@ -0,0 +1,275 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.transport;
+
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNotSame;
+import static org.junit.Assert.assertTrue;
+
+import javax.jms.Connection;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Queue;
+import javax.jms.Session;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.activemq.ActiveMQConnection;
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.ActiveMQMessageConsumer;
+import org.apache.activemq.ActiveMQMessageProducer;
+import org.apache.activemq.ActiveMQSession;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.TransportConnection;
+import org.apache.activemq.broker.TransportConnectionState;
+import org.apache.activemq.broker.TransportConnector;
+import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.command.ActiveMQMessage;
+import org.apache.activemq.command.ActiveMQQueue;
+import org.apache.activemq.command.ConsumerControl;
+import org.apache.activemq.command.ConsumerId;
+import org.apache.activemq.command.ConsumerInfo;
+import org.apache.activemq.command.ExceptionResponse;
+import org.apache.activemq.command.MessageAck;
+import org.apache.activemq.command.ProducerId;
+import org.apache.activemq.command.ProducerInfo;
+import org.apache.activemq.command.SessionId;
+import org.apache.activemq.command.SessionInfo;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.URI;
+
+public class TransportConnectionStateValidationTest {
+
+ private static final Logger LOG = LoggerFactory.getLogger(
+ TransportConnectionStateValidationTest.class);
+
+ private TransportConnector connector;
+ private URI connectionURI;
+ private BrokerService brokerService;
+ private final AtomicInteger counter = new AtomicInteger();
+
+ @Before
+ public void setUp() throws Exception {
+ startBroker();
+ }
+
+ @After
+ public void stopBroker() throws Exception {
+ if (brokerService != null) {
+ brokerService.stop();
+ }
+ }
+
+ @Test
+ public void testMismatchedConsumerProducer() throws Exception {
+
+ ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory(connectionURI);
+ try (ActiveMQConnection conn1 = (ActiveMQConnection) cf.createConnection()) {
+ // Create a connection with a consumer
+ conn1.start();
+ TransportConnection conn1Tc = findTransportConnection(conn1);
+
+ // Create a second connection and grab its state
+ ActiveMQConnection conn2 = (ActiveMQConnection) cf.createConnection();
+ conn2.start();
+ TransportConnection conn2Tc = findTransportConnection(conn2);
+ TransportConnectionState secondState = conn2Tc.lookupConnectionState(conn2.getConnectionInfo().getConnectionId());
+ assertNotSame(conn1Tc, conn2Tc);
+
+ // Try to add a session to connection 2 that uses connection 1's connection id
+ SessionId conn1SessionId = new SessionId();
+ conn1SessionId.setConnectionId(conn1Tc.getConnectionId());
+ conn1SessionId.setValue(counter.incrementAndGet());
+ SessionInfo conn1SessionInfo = new SessionInfo(conn1SessionId);
+
+ // processAddSession should skip this because it doesn't match
+ conn2.getTransport().request(conn1SessionInfo);
+ // The second connection should not be tracking the session because it didn't match
+ // and was skipped
+ assertTrue(secondState.getSessionStates().stream()
+ .noneMatch(ss -> ss.getInfo().getSessionId().equals(conn1SessionId)));
+
+ // Attempting to add a consumer using the wrong connection id should throw an exception
+ ExceptionResponse response = (ExceptionResponse) conn2.getTransport().request(
+ new ConsumerInfo(conn1SessionInfo, counter.incrementAndGet()));
+ assertTrue(response.getException() instanceof IllegalStateException);
+ assertTrue(response.getException().getMessage().contains(
+ "Cannot add a consumer to a connection that had not been registered"));
+
+ // Attempting to add a consumer using the wrong connection id should throw an exception
+ response = (ExceptionResponse) conn2.getTransport().request(
+ new ProducerInfo(conn1SessionInfo, counter.incrementAndGet()));
+ assertTrue(response.getException() instanceof IllegalStateException);
+ assertTrue(response.getException().getMessage().contains(
+ "Cannot add a producer to a connection that had not been registered"));
+ }
+ }
+
+ @Test
+ public void testMismatchedConsumerAck() throws Exception {
+ ActiveMQQueue queue = new ActiveMQQueue("test");
+
+ ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory(connectionURI);
+ try (ActiveMQConnection conn1 = (ActiveMQConnection) cf.createConnection()) {
+ conn1.start();
+ Session session = conn1.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ // Create a consumer/producer on a test queue and send a single message
+ ActiveMQMessageConsumer consumer = (ActiveMQMessageConsumer) session.createConsumer(queue);
+ MessageProducer producer = session.createProducer(queue);
+ ActiveMQMessage message = (ActiveMQMessage) session.createTextMessage("test");
+ producer.send(message);
+ ConsumerId conn1ConsumerId = consumer.getConsumerId();
+
+ // create a second connection
+ ActiveMQConnection conn2 = (ActiveMQConnection) cf.createConnection();
+ conn2.start();
+ TransportConnection conn2Tc = findTransportConnection(conn2);
+ TransportConnectionState secondState = conn2Tc.lookupConnectionState(conn2.getConnectionInfo().getConnectionId());
+
+ // copy session id from connection 1 and attempt to add to connection 2
+ SessionInfo sessionInfo = new SessionInfo(conn1ConsumerId.getParentId());
+ conn2.getTransport().request(sessionInfo);
+ // The second connection should not be tracking the session because it didn't match
+ // and was skipped
+ assertTrue(secondState.getSessionStates().stream()
+ .noneMatch(ss -> ss.getInfo().getSessionId().equals(sessionInfo.getSessionId())));
+
+ // copy consumer from connection 1 and try to add to connection 2
+ ConsumerInfo ci = new ConsumerInfo(conn1ConsumerId);
+ ci.setDestination(new ActiveMQQueue("test"));
+
+ // This should throw an error
+ ExceptionResponse response = (ExceptionResponse) conn2.getTransport().request(ci);
+ assertTrue(response.getException() instanceof IllegalStateException);
+ assertTrue(response.getException().getMessage().contains(
+ "Cannot add a consumer to a connection that had not been registered"));
+
+ // Try to ack on connection 2 with a consumer 1 for connection 1
+ // This should be ignored
+ MessageAck ack = new MessageAck();
+ ack.setConsumerId(conn1ConsumerId);
+ ack.setDestination(new ActiveMQQueue("test"));
+ ack.setAckType(org.apache.activemq.command.MessageAck.STANDARD_ACK_TYPE);
+ ack.setMessageID(message.getMessageId());
+ ack.setCommandId(counter.incrementAndGet());
+ conn2.getTransport().request(ack);
+
+ // Message should not be acked and still pending
+ assertEquals(1, brokerService.getDestination(queue)
+ .getDestinationStatistics().getMessages().getCount());
+ }
+ }
+
+ @Test
+ public void testMismatchedConsumerControl() throws Exception {
+ ActiveMQQueue queue = new ActiveMQQueue("test");
+
+ ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory(connectionURI);
+ try (ActiveMQConnection conn1 = (ActiveMQConnection) cf.createConnection()) {
+ conn1.start();
+ Session session = conn1.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ // Create a consumer/producer on a test queue and send a single message
+ ActiveMQMessageConsumer consumer = (ActiveMQMessageConsumer) session.createConsumer(queue);
+ MessageProducer producer = session.createProducer(queue);
+ ActiveMQMessage message = (ActiveMQMessage) session.createTextMessage("test");
+ producer.send(message);
+ ConsumerId conn1ConsumerId = consumer.getConsumerId();
+
+ // create a second connection
+ ActiveMQConnection conn2 = (ActiveMQConnection) cf.createConnection();
+ conn2.start();
+ TransportConnection conn2Tc = findTransportConnection(conn2);
+ TransportConnectionState secondState = conn2Tc.lookupConnectionState(conn2.getConnectionInfo().getConnectionId());
+
+ // copy session id from connection 1 and attempt to add to connection 2
+ SessionInfo sessionInfo = new SessionInfo(conn1ConsumerId.getParentId());
+ conn2.getTransport().request(sessionInfo);
+ // The second connection should not be tracking the session because it didn't match
+ // and was skipped
+ assertTrue(secondState.getSessionStates().stream()
+ .noneMatch(ss -> ss.getInfo().getSessionId().equals(sessionInfo.getSessionId())));
+
+ ConsumerControl control = new ConsumerControl();
+ control.setConsumerId(conn1ConsumerId);
+ control.setDestination(queue);
+
+ // This should throw an error
+ ExceptionResponse response = (ExceptionResponse) conn2.getTransport().request(control);
+ assertTrue(response.getException() instanceof IllegalStateException);
+ assertTrue(response.getException().getMessage().contains(
+ "Cannot process ConsumerControl for a consumer that had not been registered"));
+
+ }
+ }
+
+ @Test
+ public void testMismatchedSend() throws Exception {
+ ActiveMQQueue queue = new ActiveMQQueue("test");
+
+ ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory(connectionURI);
+ try (ActiveMQConnection conn1 = (ActiveMQConnection) cf.createConnection()) {
+ conn1.start();
+ ActiveMQSession session = (ActiveMQSession) conn1.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ ActiveMQMessage message = (ActiveMQMessage) session.createTextMessage("test");
+
+ // create a second connection
+ ActiveMQConnection conn2 = (ActiveMQConnection) cf.createConnection();
+ conn2.start();
+
+ // Create a session/producer id that use connection 1's ID info and set on message
+ SessionId conn1SessionId = new SessionId();
+ conn1SessionId.setConnectionId(conn1.getConnectionInfo().getConnectionId().toString());
+ conn1SessionId.setValue(counter.incrementAndGet());
+ message.setProducerId(new ProducerId(conn1SessionId, counter.incrementAndGet()));
+
+ // This should throw an error because we are trying to send a message on a transport
+ // for connection 2 but the message contains a producer id for connection 1
+ ExceptionResponse response = (ExceptionResponse) conn2.getTransport().request(message);
+ assertTrue(response.getException() instanceof IllegalStateException);
+ assertTrue(response.getException().getMessage().contains(
+ "Cannot lookup a producer from a connection that had not been registered"));
+ }
+ }
+
+ protected BrokerService createBroker() throws Exception {
+ BrokerService broker = new BrokerService();
+ broker.setPersistent(false);
+ broker.setUseJmx(false);
+ connector = broker.addConnector("tcp://0.0.0.0:0");
+ connectionURI = connector.getPublishableConnectURI();
+ return broker;
+ }
+
+ public void startBroker() throws Exception {
+ brokerService = createBroker();
+ brokerService.start();
+ brokerService.waitUntilStarted();
+ }
+
+ private TransportConnection findTransportConnection(ActiveMQConnection connection) {
+ return connector.getConnections().stream().filter(
+ tc -> tc.lookupConnectionState(
+ connection.getConnectionInfo().getConnectionId()) != null).findFirst().orElseThrow();
+ }
+}
+
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@activemq.apache.org
For additional commands, e-mail: commits-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact