fix: close subscription consumers on GossClient.close() (leak from #33)

pnnl.goss.core/src/pnnl/goss/core/client/DefaultClientConsumer.java

@@ -81,10 +81,18 @@ public DefaultClientConsumer(Session session, Destination destination) {
     }
 
     public void close() {
+        // messageConsumer can be null when the constructor's session.createConsumer()
+        // call failed: the constructor logs and swallows the exception rather than
+        // rethrowing, so a partially-constructed consumer with a null messageConsumer
+        // can still end up tracked and later closed. Guard against that null so close()
+        // cannot NPE on a consumer that never finished construction.
+        if (getMessageConsumer() == null) {
+            return;
+        }
         try {
             getMessageConsumer().close();
         } catch (JMSException e) {
-            e.printStackTrace();
+            log.error("Failed to close message consumer", e);
         }
     }
 

pnnl.goss.core/src/pnnl/goss/core/client/GossClient.java

@@ -104,6 +104,11 @@ public class GossClient implements Client {
     private String trustStore;
     private String trustStorePassword;
     private List<Thread> threads = new ArrayList<Thread>();
+    // Consumers created by subscribe() are long-lived (unlike the getResponse()
+    // consumer, which is closed in its own try/finally). Track them here so
+    // close() can deregister every subscription's MessageConsumer from the
+    // broker instead of leaking it when the reference goes out of scope.
+    private final List<ClientConsumer> subscriptionConsumers = new ArrayList<ClientConsumer>();
     private PROTOCOL protocol;
     private Credentials credentials = null;
     private String token = null;
@@ -366,8 +371,9 @@ public Client subscribe(String topicName, GossResponseEvent event)
             if (this.protocol.equals(PROTOCOL.OPENWIRE) || this.protocol.equals(PROTOCOL.STOMP)) {
                 // Both OPENWIRE and STOMP use the same JMS patterns with ActiveMQ
                 destination = getDestination(topicName);
-                new DefaultClientConsumer(new DefaultClientListener(event),
-                        session, destination);
+                ClientConsumer clientConsumer = new DefaultClientConsumer(
+                        new DefaultClientListener(event), session, destination);
+                subscriptionConsumers.add(clientConsumer);
             }
         } finally {
 
@@ -399,8 +405,9 @@ public Client subscribe(String destinationName, GossResponseEvent event, DESTINA
             if (this.protocol.equals(PROTOCOL.OPENWIRE) || this.protocol.equals(PROTOCOL.STOMP)) {
                 // Both OPENWIRE and STOMP use the same JMS patterns with ActiveMQ
                 destination = getDestination(destinationName, destinationType);
-                new DefaultClientConsumer(new DefaultClientListener(event),
-                        session, destination);
+                ClientConsumer clientConsumer = new DefaultClientConsumer(
+                        new DefaultClientListener(event), session, destination);
+                subscriptionConsumers.add(clientConsumer);
             }
         } finally {
 
@@ -603,6 +610,23 @@ public void publish(Destination destination, Serializable data) throws SystemExc
     public void close() {
         try {
             log.debug("Client closing!");
+
+            // Close every consumer created by subscribe() before the session closes,
+            // so the MessageConsumer (and its listener thread) is deregistered from
+            // the broker instead of being abandoned. Each close() is individually
+            // guarded: one consumer throwing must not skip the remaining consumers,
+            // the list clear below, or the session/connection teardown that follows,
+            // otherwise a single bad consumer strands the authenticated session.
+            for (ClientConsumer clientConsumer : subscriptionConsumers) {
+                try {
+                    clientConsumer.close();
+                } catch (Exception e) {
+                    log.warn("Failed to close subscription consumer {}; continuing to close remaining consumers",
+                            clientConsumer, e);
+                }
+            }
+            subscriptionConsumers.clear();
+
             if (session != null) {
                 session.close();
                 session = null;

pnnl.goss.core/test/pnnl/goss/core/client/test/GossClientConsumerLeakTest.java

@@ -0,0 +1,134 @@
+package pnnl.goss.core.client.test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.List;
+
+import jakarta.jms.MessageConsumer;
+import jakarta.jms.Session;
+import jakarta.jms.Topic;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import pnnl.goss.core.Client.PROTOCOL;
+import pnnl.goss.core.ClientConsumer;
+import pnnl.goss.core.client.GossClient;
+
+/**
+ * Regression coverage for GADP-018: GossClient.subscribe() created a
+ * DefaultClientConsumer and dropped the reference immediately after the method
+ * returned, so the underlying JMS MessageConsumer (and its listener thread) was
+ * never closed and stayed registered on the broker for the lifetime of the
+ * connection.
+ *
+ * GossClient has no broker-backed constructor path suitable for a unit test
+ * (createSession() dials a real ActiveMQConnectionFactory), so these tests
+ * inject a mocked Session via reflection to exercise subscribe() and close()
+ * without a live broker. This covers the unit-level contract (consumer tracked,
+ * closed, list cleared); it does not cover the broker-observable consequence
+ * (the consumer count on the ActiveMQ destination actually drops to zero),
+ * which needs a live-broker integration test.
+ */
+public class GossClientConsumerLeakTest {
+
+    private GossClient client;
+    private Session mockSession;
+    private MessageConsumer mockConsumer;
+
+    @BeforeEach
+    void setUp() throws Exception {
+        client = new GossClient(PROTOCOL.OPENWIRE, null, "tcp://localhost:61616",
+                "stomp://localhost:61613");
+
+        mockSession = mock(Session.class);
+        Topic mockTopic = mock(Topic.class);
+        mockConsumer = mock(MessageConsumer.class);
+        when(mockSession.createTopic("test.topic")).thenReturn(mockTopic);
+        when(mockSession.createConsumer(mockTopic)).thenReturn(mockConsumer);
+
+        setPrivateField(client, "session", mockSession);
+    }
+
+    @Test
+    @DisplayName("subscribe() tracks the created consumer instead of dropping the reference")
+    void subscribeTracksConsumerForLifecycleManagement() throws Exception {
+        client.subscribe("test.topic", response -> {
+        });
+
+        List<?> tracked = getPrivateField(client, "subscriptionConsumers", List.class);
+        assertThat(tracked).hasSize(1);
+        assertThat(tracked.get(0)).isInstanceOf(ClientConsumer.class);
+    }
+
+    @Test
+    @DisplayName("close() closes every MessageConsumer created by subscribe(), deregistering it from the broker")
+    void closeClosesTrackedConsumers() throws Exception {
+        client.subscribe("test.topic", response -> {
+        });
+
+        client.close();
+
+        verify(mockConsumer, times(1)).close();
+    }
+
+    @Test
+    @DisplayName("close() clears the tracked-consumer list so a second close() is a no-op, not a double-close")
+    void closeClearsTrackedConsumers() throws Exception {
+        client.subscribe("test.topic", response -> {
+        });
+        client.close();
+
+        List<?> tracked = getPrivateField(client, "subscriptionConsumers", List.class);
+        assertThat(tracked).isEmpty();
+    }
+
+    @Test
+    @DisplayName("close() tolerates a consumer whose close() throws: remaining consumers still close, "
+            + "the tracked list still clears, and session teardown still runs")
+    void closeSurvivesAThrowingConsumerAndStillTearsDownSession() throws Exception {
+        // Regression guard for the HIGH finding: a consumer left in a bad state
+        // (e.g. a null messageConsumer from a failed subscribe()) must not abort
+        // the close() loop and strand the remaining consumers, the tracked list,
+        // or the session/connection teardown that follows.
+        ClientConsumer throwingConsumer = mock(ClientConsumer.class);
+        doThrow(new RuntimeException("boom")).when(throwingConsumer).close();
+        ClientConsumer healthyConsumer = mock(ClientConsumer.class);
+
+        List<ClientConsumer> tracked = new ArrayList<>();
+        tracked.add(throwingConsumer);
+        tracked.add(healthyConsumer);
+        setPrivateField(client, "subscriptionConsumers", tracked);
+
+        client.close();
+
+        verify(throwingConsumer, times(1)).close();
+        verify(healthyConsumer, times(1)).close();
+
+        List<?> trackedAfter = getPrivateField(client, "subscriptionConsumers", List.class);
+        assertThat(trackedAfter).isEmpty();
+
+        verify(mockSession, times(1)).close();
+    }
+
+    @SuppressWarnings("unchecked")
+    private static <T> T getPrivateField(Object target, String name, Class<T> type) throws Exception {
+        Field field = target.getClass().getDeclaredField(name);
+        field.setAccessible(true);
+        return (T) field.get(target);
+    }
+
+    private static void setPrivateField(Object target, String name, Object value) throws Exception {
+        Field field = target.getClass().getDeclaredField(name);
+        field.setAccessible(true);
+        field.set(target, value);
+    }
+}