You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cl...@apache.org on 2023/05/19 20:30:49 UTC
[activemq-artemis] branch main updated (c47d15c12a -> 47d6dfaad2)
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
from c47d15c12a ARTEMIS-4283 Fail fast CORE client connect on closing
new 3a48258f7d ARTEMIS-2824 clientID not set on some notifications
new a57c48ec55 ARTEMIS-4275 add ID to consumer notifications
new 47d6dfaad2 This closes #4468
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../core/server/impl/ServerConsumerImpl.java | 8 +-
.../core/server/impl/ServerSessionImpl.java | 6 +
.../jms/multiprotocol/JMSNotificationTest.java | 167 +++++++++++++++++++++
.../MultiprotocolJMSClientTestSupport.java | 22 +--
4 files changed, 191 insertions(+), 12 deletions(-)
create mode 100644 tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/multiprotocol/JMSNotificationTest.java
[activemq-artemis] 03/03: This closes #4468
Posted by cl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
commit 47d6dfaad2e43e492acf499275dfb190c7edfb34
Merge: c47d15c12a a57c48ec55
Author: Clebert Suconic <cl...@apache.org>
AuthorDate: Fri May 19 16:27:52 2023 -0400
This closes #4468
.../core/server/impl/ServerConsumerImpl.java | 8 +-
.../core/server/impl/ServerSessionImpl.java | 6 +
.../jms/multiprotocol/JMSNotificationTest.java | 167 +++++++++++++++++++++
.../MultiprotocolJMSClientTestSupport.java | 22 +--
4 files changed, 191 insertions(+), 12 deletions(-)
[activemq-artemis] 01/03: ARTEMIS-2824 clientID not set on some notifications
Posted by cl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
commit 3a48258f7d816f28968e581a65df994b442d01d0
Author: Justin Bertram <jb...@apache.org>
AuthorDate: Mon May 8 11:58:05 2023 -0500
ARTEMIS-2824 clientID not set on some notifications
---
.../core/server/impl/ServerConsumerImpl.java | 6 +-
.../core/server/impl/ServerSessionImpl.java | 4 +
.../multiprotocol/JMSClientIdNotificationTest.java | 144 +++++++++++++++++++++
.../MultiprotocolJMSClientTestSupport.java | 22 ++--
4 files changed, 164 insertions(+), 12 deletions(-)
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerConsumerImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerConsumerImpl.java
index 6630604425..42295aaa8b 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerConsumerImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerConsumerImpl.java
@@ -607,10 +607,14 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
// HORNETQ-946
props.putSimpleStringProperty(ManagementHelper.HDR_USER, SimpleString.toSimpleString(session.getUsername()));
- props.putSimpleStringProperty(ManagementHelper.HDR_REMOTE_ADDRESS, SimpleString.toSimpleString(((ServerSessionImpl) session).getRemotingConnection().getRemoteAddress()));
+ props.putSimpleStringProperty(ManagementHelper.HDR_REMOTE_ADDRESS, SimpleString.toSimpleString(session.getRemotingConnection().getRemoteAddress()));
props.putSimpleStringProperty(ManagementHelper.HDR_SESSION_NAME, SimpleString.toSimpleString(session.getName()));
+ if (session.getRemotingConnection().getClientID() != null) {
+ props.putSimpleStringProperty(ManagementHelper.HDR_CLIENT_ID, SimpleString.toSimpleString(session.getRemotingConnection().getClientID()));
+ }
+
Notification notification = new Notification(null, CoreNotificationType.CONSUMER_CLOSED, props);
managementService.sendNotification(notification);
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java
index 6f7fb3c897..1984dd8099 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java
@@ -620,6 +620,10 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
props.putSimpleStringProperty(ManagementHelper.HDR_FILTERSTRING, filterString);
}
+ if (remotingConnection.getClientID() != null) {
+ props.putSimpleStringProperty(ManagementHelper.HDR_CLIENT_ID, SimpleString.toSimpleString(remotingConnection.getClientID()));
+ }
+
Notification notification = new Notification(null, CoreNotificationType.CONSUMER_CREATED, props);
if (logger.isDebugEnabled()) {
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/multiprotocol/JMSClientIdNotificationTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/multiprotocol/JMSClientIdNotificationTest.java
new file mode 100644
index 0000000000..713483613e
--- /dev/null
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/multiprotocol/JMSClientIdNotificationTest.java
@@ -0,0 +1,144 @@
+/*
+ * 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.artemis.tests.integration.jms.multiprotocol;
+
+import javax.jms.Connection;
+import javax.jms.MessageConsumer;
+import javax.jms.Session;
+import javax.jms.Topic;
+
+import org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration;
+import org.apache.activemq.artemis.api.core.ActiveMQException;
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.api.core.QueueConfiguration;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.api.core.client.ClientConsumer;
+import org.apache.activemq.artemis.api.core.client.ClientMessage;
+import org.apache.activemq.artemis.api.core.client.ClientSession;
+import org.apache.activemq.artemis.api.core.client.ClientSessionFactory;
+import org.apache.activemq.artemis.api.core.client.ServerLocator;
+import org.apache.activemq.artemis.api.core.management.CoreNotificationType;
+import org.apache.activemq.artemis.api.core.management.ManagementHelper;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.tests.util.RandomUtil;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.apache.activemq.artemis.api.core.management.ManagementHelper.HDR_CLIENT_ID;
+
+public class JMSClientIdNotificationTest extends MultiprotocolJMSClientTestSupport {
+
+ private ClientConsumer notificationConsumer;
+ private String clientID;
+
+ @Before
+ public void setClientID() {
+ clientID = RandomUtil.randomString();
+ }
+
+ @Before
+ public void createNotificationConsumer() throws Exception {
+ ServerLocator locator = addServerLocator(createInVMNonHALocator());
+ ClientSessionFactory sf = addSessionFactory(locator.createSessionFactory());
+ ClientSession session = addClientSession(sf.createSession(false, true, true));
+ session.start();
+ SimpleString notificationQueue = RandomUtil.randomSimpleString();
+ session.createQueue(new QueueConfiguration(notificationQueue).setAddress(ActiveMQDefaultConfiguration.getDefaultManagementNotificationAddress()).setDurable(false));
+ notificationConsumer = addClientConsumer(session.createConsumer(notificationQueue));
+ }
+
+ private void flush() throws ActiveMQException {
+ ClientMessage message;
+ do {
+ message = notificationConsumer.receiveImmediate();
+ }
+ while (message != null);
+ }
+
+ @Override
+ protected void addConfiguration(ActiveMQServer server) throws Exception {
+ server.getConfiguration().addAcceptorConfiguration("invm", "vm://0");
+ }
+
+ @Test(timeout = 30000)
+ public void testConsumerNotificationAMQP() throws Exception {
+ testConsumerNotification(createConnection(getBrokerQpidJMSConnectionURI(), null, null, clientID, true));
+ }
+
+ @Test(timeout = 30000)
+ public void testConsumerNotificationCore() throws Exception {
+ testConsumerNotification(createCoreConnection(getBrokerCoreJMSConnectionString(), null, null, clientID, true));
+ }
+
+ @Test(timeout = 30000)
+ public void testConsumerNotificationOpenWire() throws Exception {
+ testConsumerNotification(createOpenWireConnection(getBrokerOpenWireJMSConnectionString(), null, null, clientID, true));
+ }
+
+ private void testConsumerNotification(Connection connection) throws Exception {
+ final String subscriptionName = "mySub";
+ try {
+ flush();
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Topic topic = session.createTopic(getTopicName());
+ flush();
+ MessageConsumer consumer = session.createDurableSubscriber(topic, subscriptionName);
+ notificationConsumer.receiveImmediate(); // clear the BINDING_ADDED notification for the subscription queue
+ validateClientIdOnNotification(CoreNotificationType.CONSUMER_CREATED);
+ consumer.close();
+ validateClientIdOnNotification(CoreNotificationType.CONSUMER_CLOSED);
+ session.unsubscribe(subscriptionName);
+ } finally {
+ connection.close();
+ }
+ }
+
+ @Test(timeout = 30000)
+ public void testSessionNotificationAMQP() throws Exception {
+ testSessionNotification(createConnection(getBrokerQpidJMSConnectionURI(), null, null, clientID, true));
+ }
+
+ @Test(timeout = 30000)
+ public void testSessionNotificationCore() throws Exception {
+ testSessionNotification(createCoreConnection(getBrokerCoreJMSConnectionString(), null, null, clientID, true));
+ }
+
+ @Test(timeout = 30000)
+ public void testSessionNotificationOpenWire() throws Exception {
+ testSessionNotification(createOpenWireConnection(getBrokerOpenWireJMSConnectionString(), null, null, clientID, true));
+ }
+
+ private void testSessionNotification(Connection connection) throws Exception {
+ try {
+ flush();
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ validateClientIdOnNotification(CoreNotificationType.SESSION_CREATED);
+ session.close();
+ validateClientIdOnNotification(CoreNotificationType.SESSION_CLOSED);
+ } finally {
+ connection.close();
+ }
+ }
+
+ private void validateClientIdOnNotification(CoreNotificationType notificationType) throws ActiveMQException {
+ Message m = notificationConsumer.receive(1000);
+ assertNotNull(m);
+ assertEquals(notificationType.toString(), m.getStringProperty(ManagementHelper.HDR_NOTIFICATION_TYPE));
+ assertTrue(m.getPropertyNames().contains(HDR_CLIENT_ID));
+ assertEquals(clientID, m.getStringProperty(HDR_CLIENT_ID));
+ }
+}
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/multiprotocol/MultiprotocolJMSClientTestSupport.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/multiprotocol/MultiprotocolJMSClientTestSupport.java
index 59b49fc3b5..c4eb134469 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/multiprotocol/MultiprotocolJMSClientTestSupport.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/multiprotocol/MultiprotocolJMSClientTestSupport.java
@@ -173,7 +173,7 @@ public abstract class MultiprotocolJMSClientTestSupport extends ActiveMQTestBase
return server;
}
- protected void addConfiguration(ActiveMQServer server) {
+ protected void addConfiguration(ActiveMQServer server) throws Exception {
}
@@ -368,11 +368,11 @@ public abstract class MultiprotocolJMSClientTestSupport extends ActiveMQTestBase
return createCoreConnection(getBrokerCoreJMSConnectionString(), null, null, null, start);
}
- private Connection createCoreConnection(String connectionString,
- String username,
- String password,
- String clientId,
- boolean start) throws JMSException {
+ protected Connection createCoreConnection(String connectionString,
+ String username,
+ String password,
+ String clientId,
+ boolean start) throws JMSException {
ActiveMQJMSConnectionFactory factory = new ActiveMQJMSConnectionFactory(connectionString);
Connection connection = trackJMSConnection(factory.createConnection(username, password));
@@ -414,11 +414,11 @@ public abstract class MultiprotocolJMSClientTestSupport extends ActiveMQTestBase
return createOpenWireConnection(getBrokerOpenWireJMSConnectionString(), null, null, null, start);
}
- private Connection createOpenWireConnection(String connectionString,
- String username,
- String password,
- String clientId,
- boolean start) throws JMSException {
+ protected Connection createOpenWireConnection(String connectionString,
+ String username,
+ String password,
+ String clientId,
+ boolean start) throws JMSException {
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(connectionString);
Connection connection = trackJMSConnection(factory.createConnection(username, password));
[activemq-artemis] 02/03: ARTEMIS-4275 add ID to consumer notifications
Posted by cl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
commit a57c48ec556158a9f28c8fc9c318a81b7295b24e
Author: Justin Bertram <jb...@apache.org>
AuthorDate: Wed May 10 12:08:53 2023 -0500
ARTEMIS-4275 add ID to consumer notifications
---
.../core/server/impl/ServerConsumerImpl.java | 2 +
.../core/server/impl/ServerSessionImpl.java | 2 +
...ificationTest.java => JMSNotificationTest.java} | 51 ++++++++++++++++------
3 files changed, 41 insertions(+), 14 deletions(-)
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerConsumerImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerConsumerImpl.java
index 42295aaa8b..b720baf64c 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerConsumerImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerConsumerImpl.java
@@ -615,6 +615,8 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
props.putSimpleStringProperty(ManagementHelper.HDR_CLIENT_ID, SimpleString.toSimpleString(session.getRemotingConnection().getClientID()));
}
+ props.putLongProperty(ManagementHelper.HDR_CONSUMER_NAME, getID());
+
Notification notification = new Notification(null, CoreNotificationType.CONSUMER_CLOSED, props);
managementService.sendNotification(notification);
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java
index 1984dd8099..3284654049 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java
@@ -624,6 +624,8 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
props.putSimpleStringProperty(ManagementHelper.HDR_CLIENT_ID, SimpleString.toSimpleString(remotingConnection.getClientID()));
}
+ props.putLongProperty(ManagementHelper.HDR_CONSUMER_NAME, consumer.getID());
+
Notification notification = new Notification(null, CoreNotificationType.CONSUMER_CREATED, props);
if (logger.isDebugEnabled()) {
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/multiprotocol/JMSClientIdNotificationTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/multiprotocol/JMSNotificationTest.java
similarity index 66%
rename from tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/multiprotocol/JMSClientIdNotificationTest.java
rename to tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/multiprotocol/JMSNotificationTest.java
index 713483613e..32e897ff21 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/multiprotocol/JMSClientIdNotificationTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/multiprotocol/JMSNotificationTest.java
@@ -39,8 +39,9 @@ import org.junit.Before;
import org.junit.Test;
import static org.apache.activemq.artemis.api.core.management.ManagementHelper.HDR_CLIENT_ID;
+import static org.apache.activemq.artemis.api.core.management.ManagementHelper.HDR_CONSUMER_NAME;
-public class JMSClientIdNotificationTest extends MultiprotocolJMSClientTestSupport {
+public class JMSNotificationTest extends MultiprotocolJMSClientTestSupport {
private ClientConsumer notificationConsumer;
private String clientID;
@@ -76,20 +77,20 @@ public class JMSClientIdNotificationTest extends MultiprotocolJMSClientTestSuppo
@Test(timeout = 30000)
public void testConsumerNotificationAMQP() throws Exception {
- testConsumerNotification(createConnection(getBrokerQpidJMSConnectionURI(), null, null, clientID, true));
+ testConsumerNotifications(createConnection(getBrokerQpidJMSConnectionURI(), null, null, clientID, true));
}
@Test(timeout = 30000)
public void testConsumerNotificationCore() throws Exception {
- testConsumerNotification(createCoreConnection(getBrokerCoreJMSConnectionString(), null, null, clientID, true));
+ testConsumerNotifications(createCoreConnection(getBrokerCoreJMSConnectionString(), null, null, clientID, true));
}
@Test(timeout = 30000)
public void testConsumerNotificationOpenWire() throws Exception {
- testConsumerNotification(createOpenWireConnection(getBrokerOpenWireJMSConnectionString(), null, null, clientID, true));
+ testConsumerNotifications(createOpenWireConnection(getBrokerOpenWireJMSConnectionString(), null, null, clientID, true));
}
- private void testConsumerNotification(Connection connection) throws Exception {
+ private void testConsumerNotifications(Connection connection) throws Exception {
final String subscriptionName = "mySub";
try {
flush();
@@ -97,16 +98,32 @@ public class JMSClientIdNotificationTest extends MultiprotocolJMSClientTestSuppo
Topic topic = session.createTopic(getTopicName());
flush();
MessageConsumer consumer = session.createDurableSubscriber(topic, subscriptionName);
- notificationConsumer.receiveImmediate(); // clear the BINDING_ADDED notification for the subscription queue
- validateClientIdOnNotification(CoreNotificationType.CONSUMER_CREATED);
+ Message m = receiveNotification(CoreNotificationType.CONSUMER_CREATED, notificationConsumer);
+ validateClientIdOnNotification(m, CoreNotificationType.CONSUMER_CREATED);
+ String consumerID = validatePropertyOnNotification(m, CoreNotificationType.CONSUMER_CREATED, HDR_CONSUMER_NAME, null, false);
consumer.close();
- validateClientIdOnNotification(CoreNotificationType.CONSUMER_CLOSED);
+ m = receiveNotification(CoreNotificationType.CONSUMER_CLOSED, notificationConsumer);
+ validateClientIdOnNotification(m, CoreNotificationType.CONSUMER_CLOSED);
+ validatePropertyOnNotification(m, CoreNotificationType.CONSUMER_CLOSED, HDR_CONSUMER_NAME, consumerID, true);
session.unsubscribe(subscriptionName);
} finally {
connection.close();
}
}
+ ClientMessage receiveNotification(CoreNotificationType notificationType, ClientConsumer consumer) throws Exception {
+ for (;;) {
+ ClientMessage message = consumer.receive(1000);
+ if (message == null) {
+ return null;
+ }
+ String receivedType = message.getStringProperty(ManagementHelper.HDR_NOTIFICATION_TYPE);
+ if (String.valueOf(receivedType).equals(notificationType.toString())) {
+ return message;
+ }
+ }
+ }
+
@Test(timeout = 30000)
public void testSessionNotificationAMQP() throws Exception {
testSessionNotification(createConnection(getBrokerQpidJMSConnectionURI(), null, null, clientID, true));
@@ -126,19 +143,25 @@ public class JMSClientIdNotificationTest extends MultiprotocolJMSClientTestSuppo
try {
flush();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
- validateClientIdOnNotification(CoreNotificationType.SESSION_CREATED);
+ validateClientIdOnNotification(notificationConsumer.receive(1000), CoreNotificationType.SESSION_CREATED);
session.close();
- validateClientIdOnNotification(CoreNotificationType.SESSION_CLOSED);
+ validateClientIdOnNotification(notificationConsumer.receive(1000), CoreNotificationType.SESSION_CLOSED);
} finally {
connection.close();
}
}
- private void validateClientIdOnNotification(CoreNotificationType notificationType) throws ActiveMQException {
- Message m = notificationConsumer.receive(1000);
+ private void validateClientIdOnNotification(Message m, CoreNotificationType notificationType) {
+ validatePropertyOnNotification(m, notificationType, HDR_CLIENT_ID, clientID, true);
+ }
+
+ private String validatePropertyOnNotification(Message m, CoreNotificationType notificationType, SimpleString propertyName, String propertyValue, boolean checkValue) {
assertNotNull(m);
assertEquals(notificationType.toString(), m.getStringProperty(ManagementHelper.HDR_NOTIFICATION_TYPE));
- assertTrue(m.getPropertyNames().contains(HDR_CLIENT_ID));
- assertEquals(clientID, m.getStringProperty(HDR_CLIENT_ID));
+ assertTrue(m.getPropertyNames().contains(propertyName));
+ if (checkValue) {
+ assertEquals(propertyValue, m.getStringProperty(propertyName));
+ }
+ return m.getStringProperty(propertyName);
}
}