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

ARTEMIS-1935: Close of openwire connection should closes all open sessions

This commit is contained in:
Benjamin Graf 2018-06-14 20:34:04 +02:00 committed by Clebert Suconic
parent f52fefb42e
commit 39871157b4
2 changed files with 20 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -183,12 +183,6 @@ public class OpenWireConnection extends AbstractRemotingConnection implements Se
*/
private ServerSession internalSession;
/**
* Used for proper closing of internal sessions like OpenWire advisory
* session at disconnect.
*/
private final Set<SessionId> internalSessionIds = new ConcurrentHashSet<>();
private final OperationContext operationContext;
private static final AtomicLongFieldUpdater<OpenWireConnection> LAST_SENT_UPDATER = AtomicLongFieldUpdater.newUpdater(OpenWireConnection.class, "lastSent");
@ -616,8 +610,11 @@ public class OpenWireConnection extends AbstractRemotingConnection implements Se
state.shutdown();
try {
for (SessionId sessionId : internalSessionIds) {
sessions.get(sessionId).close();
for (SessionId sessionId : sessionIdMap.values()) {
AMQSession session = sessions.get(sessionId);
if (session != null) {
session.close();
}
}
internalSession.close(false);
} catch (Exception e) {
@ -993,7 +990,6 @@ public class OpenWireConnection extends AbstractRemotingConnection implements Se
public void addSessions(Set<SessionId> sessionSet) {
for (SessionId sid : sessionSet) {
addSession(getState().getSessionState(sid).getInfo(), true);
internalSessionIds.add(sid);
}
}

15
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/SessionHandlingOpenWireTest.java
View File

@ -56,4 +56,19 @@ public class SessionHandlingOpenWireTest extends BasicOpenWireTest {
assertFalse(AssertionLoggerHandler.findText("Client connection failed, clearing up resources for session"));
assertFalse(AssertionLoggerHandler.findText("Cleared up resources for session"));
}
@Test
public void testInternalSessionHandlingNoSessionClose() throws Exception {
try (Connection conn = factory.createConnection()) {
conn.start();
Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
Destination dest = createDestination(session,ActiveMQDestination.QUEUE_TYPE);
sendMessages(session, dest, 1);
MessageConsumer consumer = session.createConsumer(dest);
Message m = consumer.receive(2000);
assertNotNull(m);
}
assertFalse(AssertionLoggerHandler.findText("Client connection failed, clearing up resources for session"));
assertFalse(AssertionLoggerHandler.findText("Cleared up resources for session"));
}
}