Apache
/
activemq-artemis
mirror of https://github.com/apache/activemq-artemis.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

ARTEMIS-4575 Only start the consumers that were added 

Change from forcing a session start cycle on each consumer add
event and start only those consumers that were added which will
trigger a prompt delivery action on each. The session should be
marked started on create to account for the remove of the start
on each consumer add event.
This commit is contained in:
Timothy Bish 2024-01-19 15:37:15 -05:00 committed by Justin Bertram
parent a70c421493
commit 54f5dae458
2 changed files with 10 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireConnection.java
View File

@ -982,7 +982,7 @@ public class OpenWireConnection extends AbstractRemotingConnection implements Se
return;
}
amqSession.start();
consumersList.forEach((c) -> c.start());
if (AdvisorySupport.isAdvisoryTopic(info.getDestination())) {
//advisory for temp destinations
@ -1122,6 +1122,7 @@ public class OpenWireConnection extends AbstractRemotingConnection implements Se
public AMQSession addSession(SessionInfo ss) {
AMQSession amqSession = new AMQSession(getState().getInfo(), ss, server, this, protocolManager, coreMessageObjectPools);
amqSession.initialize();
amqSession.start();
sessions.put(ss.getSessionId(), amqSession);
sessionIdMap.put(amqSession.getCoreSession().getName(), ss.getSessionId());

8
artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/amq/AMQConsumer.java
View File

@ -137,6 +137,14 @@ public class AMQConsumer {
return filterString;
}
public void start() {
if (serverConsumer == null) {
throw new IllegalStateException("Cannot start the AMQConsumer until it has been initialized");
}
serverConsumer.setStarted(true);
}
public void init(SlowConsumerDetectionListener slowConsumerDetectionListener, long nativeId) throws Exception {
SimpleString selector = info.getSelector() == null ? null : new SimpleString(convertOpenWireToActiveMQFilterString(info.getSelector()));