Skip to content

BITMAG-1276-refactor JMS resource #93

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Knud-Aage wants to merge 2 commits into
base: master
Choose a base branch
from
Open

BITMAG-1276-refactor JMS resource #93

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b12a99d
Replaced manual JMS resouce cleanup with try-with-resources.
Knud-Aage Jul 13, 2026
482b4a6
Merge remote-tracking branch 'origin/master' into BITMAG-1276-refacto…
Knud-Aage Jul 15, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 26 additions & 13 deletions bitrepository-core/src/main/java/org/bitrepository/protocol/activemq/ActiveMQMessageBus.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,8 +35,6 @@
import jakarta.jms.TextMessage;
import jakarta.jms.Topic;
import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;

import java.io.ByteArrayInputStream;
import org.bitrepository.bitrepositorymessages.Message;
import org.bitrepository.bitrepositorymessages.MessageRequest;
import org.bitrepository.common.DefaultThreadFactory;
Expand Down Expand Up @@ -67,6 +65,7 @@
import org.slf4j.LoggerFactory;
import org.xml.sax.SAXException;

import java.io.ByteArrayInputStream;
import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.HashMap;
Expand Down Expand Up @@ -174,18 +173,21 @@ public ActiveMQMessageBus(Settings settings, SecurityManager securityManager) {
jaxbHelper = new JaxbHelper("xsd/", schemaLocation);
ActiveMQConnectionFactory connectionFactory = ArtemisConnectionFactoryProvider.create(configuration);
registerCustomMessageLoggers();
Connection newConnection = null;
try {
connection = connectionFactory.createConnection();
connection.setClientID(clientID);
connection.setExceptionListener(new MessageBusExceptionListener());
newConnection = connectionFactory.createConnection();
newConnection.setClientID(clientID);
newConnection.setExceptionListener(new MessageBusExceptionListener());

producerSession = connection.createSession(TRANSACTED, Session.AUTO_ACKNOWLEDGE);
consumerSession = connection.createSession(TRANSACTED, Session.AUTO_ACKNOWLEDGE);
producerSession = newConnection.createSession(TRANSACTED, Session.AUTO_ACKNOWLEDGE);
consumerSession = newConnection.createSession(TRANSACTED, Session.AUTO_ACKNOWLEDGE);
producer = producerSession.createProducer(null);
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
connection = newConnection;

startListeningForMessages();
} catch (JMSException e) {
closeQuietly(newConnection);
throw new CoordinationLayerException("Unable to initialise connection to message bus", e);
}
log.debug("ActiveMQConnection initialized for '{}'", configuration);
Expand All @@ -212,6 +214,20 @@ private void startListeningForMessages() {
connectionStarter.start();
}

/**
* Closes the connection, suppressing any JMSException, so it can be safely used for cleanup after a
* failed initialisation without masking the original error.
*/
private void closeQuietly(Connection connection) {

@ole-v-v ole-v-v Jul 27, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You have this method in two places, here and in RawMessagebus. Can we have it only once and reuse it?

if (connection != null) {
try {
connection.close();
} catch (JMSException e) {
log.warn("Failed to close connection after initialisation failure", e);
}
}
}

@Override
public synchronized void addListener(String destinationID, final MessageListener listener) {
addListener(destinationID, listener, false);
Expand Down Expand Up @@ -251,12 +267,9 @@ public synchronized void removeListener(String destinationID, MessageListener li
public void close() throws JMSException {
receivedMessageHandler.close();
log.info("Closing message bus: {}", configuration);
producerSession.close();
log.debug("Producer session closed.");
consumerSession.close();
log.debug("Consumer session closed.");
connection.close();
log.debug("Connection closed.");
try (connection; consumerSession; producerSession) {
log.debug("Closing producer session, consumer session and connection.");
}
}

@Override
Expand Down
39 changes: 38 additions & 1 deletion bitrepository-core/src/test/java/org/bitrepository/protocol/bus/ActiveMQMessageBusTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
import org.bitrepository.bitrepositorymessages.DeleteFileRequest;
import org.bitrepository.bitrepositorymessages.IdentifyPillarsForDeleteFileRequest;
import org.bitrepository.bitrepositorymessages.IdentifyPillarsForDeleteFileResponse;
import org.bitrepository.protocol.CoordinationLayerException;
import org.bitrepository.protocol.ProtocolComponentFactory;
import org.bitrepository.protocol.activemq.ActiveMQMessageBus;
import org.bitrepository.protocol.message.ExampleMessageFactory;
Expand Down Expand Up @@ -181,4 +182,40 @@ final void toFilterTest() throws Exception {
rawMessagebus.sendMessage(settingsForTestClient.getCollectionDestination(), rq);
collectionReceiver.waitForMessage(DeleteFileRequest.class);
}
}

@Test

@ole-v-v ole-v-v Jul 27, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nice tests.

@Tag("regressiontest")
final void closeReleasesJmsResourcesTest() throws Exception {
addDescription("Test that closing a message bus releases its JMS resources, so it can no longer " +
"be used to send messages afterwards.");
addStep("Create a dedicated message bus instance and close it",
"No exception should be thrown while closing.");
ActiveMQMessageBus dedicatedMessageBus = new ActiveMQMessageBus(settingsForTestClient, securityManager);
dedicatedMessageBus.close();

addStep("Attempt to send a message on the closed message bus",
"The send should fail, since the underlying JMS session and connection have been closed.");

@ole-v-v ole-v-v Jul 27, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nitpicking, English usually hasn’t got any comma before “since”. Here and in line 217.

IdentifyPillarsForDeleteFileRequest message =
ExampleMessageFactory.createMessage(IdentifyPillarsForDeleteFileRequest.class);
message.setDestination(settingsForTestClient.getCollectionDestination());
Assertions.assertThrows(CoordinationLayerException.class, () -> dedicatedMessageBus.sendMessage(message));
}

@Test
@Tag("regressiontest")
final void rawMessagebusCloseReleasesJmsResourcesTest() throws Exception {
addDescription("Test that closing a RawMessagebus releases its JMS resources, so it can no longer " +
"be used afterwards.");
addStep("Create a raw message bus instance and close it",
"No exception should be thrown while closing.");
RawMessagebus rawMessagebus = new RawMessagebus(
settingsForTestClient.getMessageBusConfiguration(),
securityManager);
rawMessagebus.close();

addStep("Attempt to create a producer on the closed raw message bus",
"The call should fail, since the underlying JMS session and connection have been closed.");
Assertions.assertThrows(CoordinationLayerException.class,
() -> rawMessagebus.getProducer(settingsForTestClient.getCollectionDestination()));
}
}
38 changes: 33 additions & 5 deletions bitrepository-core/src/test/java/org/bitrepository/protocol/bus/RawMessagebus.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,10 +30,10 @@
import jakarta.jms.MessageConsumer;
import jakarta.jms.MessageProducer;
import jakarta.jms.Session;
import org.bitrepository.protocol.activemq.ArtemisConnectionFactoryProvider;
import org.bitrepository.common.JaxbHelper;
import org.bitrepository.protocol.CoordinationLayerException;
import org.bitrepository.protocol.activemq.ActiveMQMessageBus;
import org.bitrepository.protocol.activemq.ArtemisConnectionFactoryProvider;
import org.bitrepository.protocol.security.SecurityManager;
import org.bitrepository.settings.repositorysettings.MessageBusConfiguration;
import org.slf4j.Logger;
Expand All @@ -57,19 +57,47 @@ public RawMessagebus(MessageBusConfiguration messageBusConfiguration, SecurityMa
this.securityManager = securityManager;

var connectionFactory = ArtemisConnectionFactoryProvider.create(messageBusConfiguration);
Connection newConnection = null;
try {
connection = connectionFactory.createConnection();
connection.setExceptionListener(new MessageBusExceptionListener());
newConnection = connectionFactory.createConnection();
newConnection.setExceptionListener(new MessageBusExceptionListener());

producerSession = connection.createSession(TRANSACTED, Session.AUTO_ACKNOWLEDGE);
consumerSession = connection.createSession(TRANSACTED, Session.AUTO_ACKNOWLEDGE);
producerSession = newConnection.createSession(TRANSACTED, Session.AUTO_ACKNOWLEDGE);
consumerSession = newConnection.createSession(TRANSACTED, Session.AUTO_ACKNOWLEDGE);
connection = newConnection;

connection.start();
} catch (JMSException e) {
closeQuietly(newConnection);
throw new CoordinationLayerException("Unable to initialise connection to message bus", e);
}
}

/**
* Closes the connection, suppressing any JMSException, so it can be safely used for cleanup after a
* failed initialisation without masking the original error.
*/
private void closeQuietly(Connection connection) {
if (connection != null) {
try {
connection.close();
} catch (JMSException e) {
log.warn("Failed to close connection after initialisation failure", e);
}
}
}

/**
* Closes the producer session, consumer session and connection. Declared in reverse close order so

@ole-v-v ole-v-v Jul 27, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nitpicking: The declaration order is an implementation detail and does not belong in the Javadoc. Put it in a comment inside the method instead. The guarantee that all are attempted regardless of exceptions does belong here.

* try-with-resources closes the sessions before the connection, guaranteeing all are attempted even
* if one throws.
*/
public void close() throws JMSException {
try (connection; consumerSession; producerSession) {
log.debug("Closing raw message bus connection.");
}
}

public void addHeader(Message msg,
String messageClass,
String replyTo,
Expand Down