From f4be37b8eb9fe7536277dbd8df968ea7b472da94 Mon Sep 17 00:00:00 2001 From: Marc DiPasquale Date: Thu, 30 Jul 2026 14:55:31 -0400 Subject: [PATCH] Elevate patterns/ samples to documented JCSMP best practices Applies best-practice deltas in-place to the four pattern samples that have skill reference versions, per the JCSMP Best Practices guide: https://docs.solace.com/API/API-Developer-Guide-JCSMP/JCSMP-API-Best-Practices.md GuaranteedPublisher: - Fix correlation key: use an immutable per-send id instead of the reused, mutated message object (setCorrelationKey(message) bug) - Act on session events (pause/resume/shutdown) instead of only logging - Add publish-only reconnect consumer to surface transport events - Pause publishing while transport is down - Graceful SIGINT shutdown hook + teardown in main's finally (drains ACKs) GuaranteedSubscriber: - Provision the durable queue + topic subscription in-process (capability checked) so a fresh broker works out of the box - Act on session and flow events instead of only logging - IGNORE_DUPLICATE_SUBSCRIPTION_ERROR for re-run safety - Graceful SIGINT shutdown hook + teardown in main's finally - Render broker subcode in the flow onException handler DirectPublisher: - Act on session events; add publish-only reconnect consumer - Pause publishing while transport is down; explicit DIRECT delivery mode - Graceful SIGINT shutdown hook + teardown in main's finally DirectSubscriber: - Act on session events; graceful SIGINT shutdown hook + teardown Compile-verified with ./gradlew compileJava. Not exercised against a live broker. Request-reply stubs (GuaranteedReplier, GuaranteedRequestorAsync) left as-is; they are empty // TBD placeholders. Co-Authored-By: Claude Opus 4.8 --- .../jcsmp/patterns/DirectPublisher.java | 104 ++++++++-- .../jcsmp/patterns/DirectSubscriber.java | 62 +++++- .../jcsmp/patterns/GuaranteedPublisher.java | 131 ++++++++++--- .../jcsmp/patterns/GuaranteedSubscriber.java | 180 ++++++++++++++---- 4 files changed, 406 insertions(+), 71 deletions(-) diff --git a/src/main/java/com/solace/samples/jcsmp/patterns/DirectPublisher.java b/src/main/java/com/solace/samples/jcsmp/patterns/DirectPublisher.java index 2b9e133..ba260f3 100644 --- a/src/main/java/com/solace/samples/jcsmp/patterns/DirectPublisher.java +++ b/src/main/java/com/solace/samples/jcsmp/patterns/DirectPublisher.java @@ -24,32 +24,39 @@ import java.util.concurrent.TimeUnit; import com.solacesystems.jcsmp.BytesMessage; +import com.solacesystems.jcsmp.DeliveryMode; import com.solacesystems.jcsmp.JCSMPChannelProperties; import com.solacesystems.jcsmp.JCSMPErrorResponseException; import com.solacesystems.jcsmp.JCSMPErrorResponseSubcodeEx; import com.solacesystems.jcsmp.JCSMPException; import com.solacesystems.jcsmp.JCSMPFactory; import com.solacesystems.jcsmp.JCSMPProperties; +import com.solacesystems.jcsmp.JCSMPReconnectEventHandler; import com.solacesystems.jcsmp.JCSMPSession; import com.solacesystems.jcsmp.JCSMPStreamingPublishCorrelatingEventHandler; import com.solacesystems.jcsmp.JCSMPTransportException; import com.solacesystems.jcsmp.SessionEventArgs; import com.solacesystems.jcsmp.SessionEventHandler; +import com.solacesystems.jcsmp.Topic; import com.solacesystems.jcsmp.XMLMessageProducer; /** * A more performant sample that shows an application that publishes. */ public class DirectPublisher { - + private static final String SAMPLE_NAME = DirectPublisher.class.getSimpleName(); private static final String TOPIC_PREFIX = "solace/samples/"; // used as the topic "root" private static final String API = "JCSMP"; private static final int APPROX_MSG_RATE_PER_SEC = 100; private static final int PAYLOAD_SIZE = 100; - + private static volatile int msgSentCounter = 0; // num messages sent private static volatile boolean isShutdown = false; + private static volatile boolean isConnected = true; // tracks transport state via reconnect events + private static JCSMPSession session; + private static XMLMessageProducer producer; + private static ScheduledExecutorService statsPrintingThread; /** Main method. */ public static void main(String... args) throws JCSMPException, IOException, InterruptedException { @@ -58,7 +65,30 @@ public static void main(String... args) throws JCSMPException, IOException, Inte System.exit(-1); } System.out.println(API + " " + SAMPLE_NAME + " initializing..."); + try { + setupSolace(args); + // graceful shutdown: a SIGINT (Ctrl-C) signals the main loop to stop, then the + // hook joins the main thread so the cleanup in teardownSolace() (stop publishing, + // close the session) runs to completion before the JVM halts. The JVM exits as + // soon as all shutdown hooks return, so the hook must wait, not just set a flag. + final Thread mainThread = Thread.currentThread(); + Runtime.getRuntime().addShutdownHook(new Thread(() -> { + System.out.println("Shutdown signal received, stopping publisher..."); + isShutdown = true; + try { + mainThread.join(5000); // wait for the main thread's session close + } catch (InterruptedException e) { + // nothing more we can do; the JVM is halting + } + })); + runPublishLoop(); + } finally { + teardownSolace(); + System.out.println("Main thread quitting."); + } + } + private static void setupSolace(String[] args) throws JCSMPException { final JCSMPProperties properties = new JCSMPProperties(); properties.setProperty(JCSMPProperties.HOST, args[0]); // host:port properties.setProperty(JCSMPProperties.VPN_NAME, args[1]); // message-vpn @@ -72,17 +102,33 @@ public static void main(String... args) throws JCSMPException, IOException, Inte channelProps.setConnectRetriesPerHost(5); // recommended settings // https://docs.solace.com/Solace-PubSub-Messaging-APIs/API-Developer-Guide/Configuring-Connection-T.htm properties.setProperty(JCSMPProperties.CLIENT_CHANNEL_PROPERTIES, channelProps); - final JCSMPSession session; + // best practice: register a session event handler at session creation and handle + // each event appropriately, rather than only logging it session = JCSMPFactory.onlyInstance().createSession(properties, null, new SessionEventHandler() { @Override - public void handleEvent(SessionEventArgs event) { // could be reconnecting, connection lost, etc. + public void handleEvent(SessionEventArgs event) { System.out.printf("### Received a Session event: %s%n", event); + switch (event.getEvent()) { + case RECONNECTING: // session went down, automatic reconnect attempt in progress + isConnected = false; + break; + case RECONNECTED: // automatic reconnect succeeded, session re-established + isConnected = true; + break; + case DOWN_ERROR: // session was up and then went down; reconnects exhausted + System.out.println("Session is down and reconnect attempts are exhausted, quitting."); + // the session will not recover; end the main loop so teardownSolace() + // runs in main's finally + isShutdown = true; + break; + default: + break; + } } }); session.connect(); // connect to the broker // Simple anonymous inner-class for handling publishing events - final XMLMessageProducer producer; producer = session.getMessageProducer(new JCSMPStreamingPublishCorrelatingEventHandler() { // unused in Direct Messaging application, only for Guaranteed/Persistent publishing application @Override public void responseReceivedEx(Object key) { @@ -103,23 +149,48 @@ public void handleErrorEx(Object key, JCSMPException cause, long timestamp) { } }); - ScheduledExecutorService statsPrintingThread = Executors.newSingleThreadScheduledExecutor(); + // best practice for publish-only applications: transport reconnect events are only + // exposed through a consumer, so acquire an empty one (never started) and register a + // reconnect event handler to pause publishing while the connection is re-established + session.getMessageConsumer(new JCSMPReconnectEventHandler() { + @Override + public boolean preReconnect() throws JCSMPException { + System.out.println("### preReconnect(): transport down, pausing publishing"); + isConnected = false; + return true; // true means let the API proceed with its reconnect attempts + } + + @Override + public void postReconnect() throws JCSMPException { + System.out.println("### postReconnect(): transport restored, resuming publishing"); + isConnected = true; + } + }, null); // no message listener: this consumer exists only to surface transport events + } + + private static void runPublishLoop() throws IOException, InterruptedException { + statsPrintingThread = Executors.newSingleThreadScheduledExecutor(); statsPrintingThread.scheduleAtFixedRate(() -> { System.out.printf("%s %s Published msgs/s: %,d%n",API,SAMPLE_NAME,msgSentCounter); // simple way of calculating message rates msgSentCounter = 0; }, 1, 1, TimeUnit.SECONDS); - - System.out.println(API + " " + SAMPLE_NAME + " connected, and running. Press [ENTER] to quit."); + + System.out.println(API + " " + SAMPLE_NAME + " connected, and running. Press [ENTER] or Ctrl-C to quit."); // preallocate a binary message, reuse it each loop, for performance final BytesMessage message = JCSMPFactory.onlyInstance().createMessage(BytesMessage.class); byte[] payload = new byte[PAYLOAD_SIZE]; // preallocate memory, for reuse, for performance // loop the main thread, waiting for a quit signal while (System.in.available() == 0 && !isShutdown) { + if (!isConnected) { // transport is down: wait for the API's automatic reconnect + Thread.sleep(100); + continue; + } try { // each loop, change the payload, less trivial example than static payload char chosenCharacter = (char)(Math.round(msgSentCounter % 26) + 65); // rotate through letters [A-Z] Arrays.fill(payload,(byte)chosenCharacter); // fill the payload completely with that char message.setData(payload); + message.setDeliveryMode(DeliveryMode.DIRECT); // at-most-once; DIRECT is also the API default message.setApplicationMessageId(UUID.randomUUID().toString()); // as an example of a header // dynamic topics!! "solace/samples/jcsmp/direct/pub/A" String topicString = new StringBuilder(TOPIC_PREFIX).append(API.toLowerCase()) @@ -137,13 +208,24 @@ public void handleErrorEx(Object key, JCSMPException cause, long timestamp) { Thread.sleep(1000 / APPROX_MSG_RATE_PER_SEC); // do Thread.sleep(0) for max speed // Note: STANDARD Edition Solace PubSub+ broker is limited to 10k msg/s max ingress } catch (InterruptedException e) { + Thread.currentThread().interrupt(); // restore the interrupt status, do not swallow it isShutdown = true; } } } isShutdown = true; - statsPrintingThread.shutdown(); // stop printing stats - session.closeSession(); // will also close producer object - System.out.println("Main thread quitting."); + } + + private static void teardownSolace() { + // teardownSolace() runs in main's finally on EVERY exit path (normal quit, ENTER, + // SIGINT, DOWN_ERROR, or an exception), so Solace teardown is never skipped. + isShutdown = true; + if (statsPrintingThread != null) { + statsPrintingThread.shutdown(); // stop printing stats + } + // direct is at-most-once: no outstanding broker ACKs to drain, so close immediately + if (session != null) { + session.closeSession(); // will also close producer object + } } } diff --git a/src/main/java/com/solace/samples/jcsmp/patterns/DirectSubscriber.java b/src/main/java/com/solace/samples/jcsmp/patterns/DirectSubscriber.java index 0719b2a..b34efd1 100644 --- a/src/main/java/com/solace/samples/jcsmp/patterns/DirectSubscriber.java +++ b/src/main/java/com/solace/samples/jcsmp/patterns/DirectSubscriber.java @@ -39,6 +39,7 @@ public class DirectSubscriber { private static volatile int msgRecvCounter = 0; // num messages received private static volatile boolean hasDetectedDiscard = false; // detected any discards yet? private static volatile boolean isShutdown = false; // are we done yet? + private static JCSMPSession session; /** the main method. */ public static void main(String... args) throws JCSMPException, IOException, InterruptedException { @@ -47,7 +48,30 @@ public static void main(String... args) throws JCSMPException, IOException, Inte System.exit(-1); } System.out.println(API + " " + SAMPLE_NAME + " initializing..."); + try { + setupSolace(args); + // graceful shutdown: a SIGINT (Ctrl-C) signals the main loop to stop, then the + // hook joins the main thread so the cleanup in teardownSolace() (close the + // session) runs to completion before the JVM halts. The JVM exits as soon as all + // shutdown hooks return, so the hook must wait, not just set a flag. + final Thread mainThread = Thread.currentThread(); + Runtime.getRuntime().addShutdownHook(new Thread(() -> { + System.out.println("Shutdown signal received, stopping subscriber..."); + isShutdown = true; + try { + mainThread.join(5000); // wait for the main thread's session close + } catch (InterruptedException e) { + // nothing more we can do; the JVM is halting + } + })); + awaitMessages(); + } finally { + teardownSolace(); + System.out.println("Main thread quitting."); + } + } + private static void setupSolace(String[] args) throws JCSMPException { final JCSMPProperties properties = new JCSMPProperties(); properties.setProperty(JCSMPProperties.HOST, args[0]); // host:port properties.setProperty(JCSMPProperties.VPN_NAME, args[1]); // message-vpn @@ -61,11 +85,28 @@ public static void main(String... args) throws JCSMPException, IOException, Inte channelProps.setConnectRetriesPerHost(5); // recommended settings // https://docs.solace.com/Solace-PubSub-Messaging-APIs/API-Developer-Guide/Configuring-Connection-T.htm properties.setProperty(JCSMPProperties.CLIENT_CHANNEL_PROPERTIES, channelProps); - final JCSMPSession session; + // best practice: register a session event handler at session creation and handle + // each event appropriately, rather than only logging it session = JCSMPFactory.onlyInstance().createSession(properties, null, new SessionEventHandler() { @Override - public void handleEvent(SessionEventArgs event) { // could be reconnecting, connection lost, etc. + public void handleEvent(SessionEventArgs event) { System.out.printf("### Received a Session event: %s%n", event); + switch (event.getEvent()) { + case RECONNECTING: // session went down, automatic reconnect attempt in progress + System.out.println("Session reconnecting; direct delivery is paused until re-established"); + break; + case RECONNECTED: // automatic reconnect succeeded, session re-established + System.out.println("Session reconnected; the subscription is restored and delivery resumes"); + break; + case DOWN_ERROR: // session was up and then went down; reconnects exhausted + System.out.println("Session is down and reconnect attempts are exhausted, quitting."); + // the session will not recover; end the main loop so teardownSolace() + // runs in main's finally + isShutdown = true; + break; + default: + break; + } } }); session.connect(); // connect to the broker @@ -100,7 +141,10 @@ public void onException(JCSMPException e) { // uh oh! session.addSubscription(JCSMPFactory.onlyInstance().createTopic(TOPIC_PREFIX + "*/direct/>")); // add more subscriptions here if you want consumer.start(); - System.out.println(API + " " + SAMPLE_NAME + " connected, and running. Press [ENTER] to quit."); + System.out.println(API + " " + SAMPLE_NAME + " connected, and running. Press [ENTER] or Ctrl-C to quit."); + } + + private static void awaitMessages() throws IOException, InterruptedException { while (System.in.available() == 0 && !isShutdown) { Thread.sleep(1000); // wait 1 second System.out.printf("%s %s Received msgs/s: %,d%n",API,SAMPLE_NAME,msgRecvCounter); // simple way of calculating message rates @@ -112,7 +156,15 @@ public void onException(JCSMPException e) { // uh oh! } } isShutdown = true; - session.closeSession(); // will also close consumer object - System.out.println("Main thread quitting."); + } + + private static void teardownSolace() { + // teardownSolace() runs in main's finally on EVERY exit path (normal quit, ENTER, + // SIGINT, DOWN_ERROR, or an exception), so Solace teardown is never skipped. + isShutdown = true; + // direct is at-most-once: no acknowledgements to drain before exit, so close directly + if (session != null) { + session.closeSession(); // will also close consumer object + } } } diff --git a/src/main/java/com/solace/samples/jcsmp/patterns/GuaranteedPublisher.java b/src/main/java/com/solace/samples/jcsmp/patterns/GuaranteedPublisher.java index cc0f50f..2602148 100644 --- a/src/main/java/com/solace/samples/jcsmp/patterns/GuaranteedPublisher.java +++ b/src/main/java/com/solace/samples/jcsmp/patterns/GuaranteedPublisher.java @@ -27,7 +27,6 @@ import org.apache.logging.log4j.Logger; import com.solacesystems.jcsmp.BytesMessage; -import com.solacesystems.jcsmp.BytesXMLMessage; import com.solacesystems.jcsmp.DeliveryMode; import com.solacesystems.jcsmp.JCSMPChannelProperties; import com.solacesystems.jcsmp.JCSMPErrorResponseException; @@ -36,6 +35,7 @@ import com.solacesystems.jcsmp.JCSMPFactory; import com.solacesystems.jcsmp.JCSMPProducerEventHandler; import com.solacesystems.jcsmp.JCSMPProperties; +import com.solacesystems.jcsmp.JCSMPReconnectEventHandler; import com.solacesystems.jcsmp.JCSMPSession; import com.solacesystems.jcsmp.JCSMPStreamingPublishCorrelatingEventHandler; import com.solacesystems.jcsmp.JCSMPTransportException; @@ -47,19 +47,23 @@ import com.solacesystems.jcsmp.XMLMessageProducer; public class GuaranteedPublisher { - + private static final String SAMPLE_NAME = GuaranteedPublisher.class.getSimpleName(); static final String TOPIC_PREFIX = "solace/samples/"; // used as the topic "root" private static final String API = "JCSMP"; private static final int PUBLISH_WINDOW_SIZE = 50; private static final int APPROX_MSG_RATE_PER_SEC = 100; private static final int PAYLOAD_SIZE = 512; - + // remember to add log4j2.xml to your classpath private static final Logger logger = LogManager.getLogger(); // log4j2, but could also use SLF4J, JCL, etc. private static volatile int msgSentCounter = 0; // num messages sent private static volatile boolean isShutdown = false; + private static volatile boolean isConnected = true; // tracks transport state via reconnect events + private static JCSMPSession session; + private static XMLMessageProducer producer; + private static ScheduledExecutorService statsPrintingThread; /** Main. */ public static void main(String... args) throws JCSMPException, IOException, InterruptedException { @@ -68,7 +72,31 @@ public static void main(String... args) throws JCSMPException, IOException, Inte System.exit(-1); } System.out.println(API + " " + SAMPLE_NAME + " initializing..."); + try { + setupSolace(args); + // graceful shutdown: a SIGINT (Ctrl-C) signals the main loop to stop, then the + // hook joins the main thread so the cleanup in teardownSolace() (finish + // outstanding ACKs, close the session) runs to completion before the JVM halts. + // The JVM exits as soon as all shutdown hooks return, so the hook must wait, + // not just set a flag. + final Thread mainThread = Thread.currentThread(); + Runtime.getRuntime().addShutdownHook(new Thread(() -> { + System.out.println("Shutdown signal received, stopping publisher..."); + isShutdown = true; + try { + mainThread.join(5000); // wait for the main thread's ACK drain and session close + } catch (InterruptedException e) { + // nothing more we can do; the JVM is halting + } + })); + runPublishLoop(); + } finally { + teardownSolace(); + System.out.println("Main thread quitting."); + } + } + private static void setupSolace(String[] args) throws JCSMPException { final JCSMPProperties properties = new JCSMPProperties(); properties.setProperty(JCSMPProperties.HOST, args[0]); // host:port properties.setProperty(JCSMPProperties.VPN_NAME, args[1]); // message-vpn @@ -82,35 +110,76 @@ public static void main(String... args) throws JCSMPException, IOException, Inte channelProps.setConnectRetriesPerHost(5); // recommended settings // https://docs.solace.com/Solace-PubSub-Messaging-APIs/API-Developer-Guide/Configuring-Connection-T.htm properties.setProperty(JCSMPProperties.CLIENT_CHANNEL_PROPERTIES, channelProps); - final JCSMPSession session; + // best practice: register a session event handler at session creation and handle + // each event appropriately, rather than only logging it session = JCSMPFactory.onlyInstance().createSession(properties, null, new SessionEventHandler() { @Override - public void handleEvent(SessionEventArgs event) { // could be reconnecting, connection lost, etc. + public void handleEvent(SessionEventArgs event) { logger.info("### Received a Session event: " + event); + switch (event.getEvent()) { + case RECONNECTING: // session went down, automatic reconnect attempt in progress + isConnected = false; + break; + case RECONNECTED: // automatic reconnect succeeded, session re-established + isConnected = true; + break; + case DOWN_ERROR: // session was up and then went down; reconnects exhausted + logger.error("Session is down and reconnect attempts are exhausted, quitting."); + // the session will not recover; end the main loop so teardownSolace() + // runs in main's finally + isShutdown = true; + break; + default: + break; + } } }); session.connect(); - - XMLMessageProducer producer = session.getMessageProducer(new PublishCallbackHandler(), new JCSMPProducerEventHandler() { + + producer = session.getMessageProducer(new PublishCallbackHandler(), new JCSMPProducerEventHandler() { @Override public void handleEvent(ProducerEventArgs event) { // as of JCSMP v10.10, this event only occurs when republishing unACKed messages on an unknown flow (DR failover) logger.info("*** Received a producer event: " + event); } }); - - ScheduledExecutorService statsPrintingThread = Executors.newSingleThreadScheduledExecutor(); + + // best practice for publish-only applications: transport reconnect events are only + // exposed through a consumer, so acquire an empty one (never started) and register a + // reconnect event handler to pause publishing while the connection is re-established + session.getMessageConsumer(new JCSMPReconnectEventHandler() { + @Override + public boolean preReconnect() throws JCSMPException { + logger.info("### preReconnect(): transport down, pausing publishing"); + isConnected = false; + return true; // true means let the API proceed with its reconnect attempts + } + + @Override + public void postReconnect() throws JCSMPException { + logger.info("### postReconnect(): transport restored, resuming publishing"); + isConnected = true; + } + }, null); // no message listener: this consumer exists only to surface transport events + } + + private static void runPublishLoop() throws JCSMPException, IOException, InterruptedException { + statsPrintingThread = Executors.newSingleThreadScheduledExecutor(); statsPrintingThread.scheduleAtFixedRate(() -> { System.out.printf("%s %s Published msgs/s: %,d%n",API,SAMPLE_NAME,msgSentCounter); // simple way of calculating message rates msgSentCounter = 0; }, 1, 1, TimeUnit.SECONDS); - - System.out.println(API + " " + SAMPLE_NAME + " connected, and running. Press [ENTER] to quit."); + + System.out.println(API + " " + SAMPLE_NAME + " connected, and running. Press [ENTER] or Ctrl-C to quit."); byte[] payload = new byte[PAYLOAD_SIZE]; // preallocate BytesMessage message = JCSMPFactory.onlyInstance().createMessage(BytesMessage.class); // preallocate - System.out.println("Publishing to topic '"+ TOPIC_PREFIX + API.toLowerCase() + - "/pers/pub/...', please ensure queue has matching subscription."); + System.out.println("Publishing to topic '"+ TOPIC_PREFIX + API.toLowerCase() + + "/pers/pub/...', please ensure queue has matching subscription."); while (System.in.available() == 0 && !isShutdown) { // loop until ENTER pressed, or shutdown flag + if (!isConnected) { // transport is down: wait for the API's automatic reconnect + Thread.sleep(100); + continue; + } message.reset(); // ready for reuse // each loop, change the payload as an example char chosenCharacter = (char)(Math.round(msgSentCounter % 26) + 65); // choose a "random" letter [A-Z] @@ -118,12 +187,16 @@ public void handleEvent(ProducerEventArgs event) { // use a BytesMessage this sample, instead of TextMessage message.setData(payload); message.setDeliveryMode(DeliveryMode.PERSISTENT); // required for Guaranteed - message.setApplicationMessageId(UUID.randomUUID().toString()); // as an example + String msgId = UUID.randomUUID().toString(); + message.setApplicationMessageId(msgId); // as an example // as another example, let's define a user property! SDTMap map = JCSMPFactory.onlyInstance().createMap(); map.putString("sample",API + "_" + SAMPLE_NAME); message.setProperties(map); - message.setCorrelationKey(message); // used for ACK/NACK correlation locally within the API + // correlation key for local ACK/NACK correlation: use an immutable per-send + // identifier, NOT the reused message object, which this loop keeps mutating + // while up to PUBLISH_WINDOW_SIZE sends are still awaiting their ACK + message.setCorrelationKey(msgId); String topicString = new StringBuilder(TOPIC_PREFIX).append(API.toLowerCase()) .append("/pers/pub/").append(chosenCharacter).toString(); // NOTE: publishing to topic, so make sure GuaranteedSubscriber queue is subscribed to same topic, @@ -142,15 +215,31 @@ public void handleEvent(ProducerEventArgs event) { Thread.sleep(1000 / APPROX_MSG_RATE_PER_SEC); // do Thread.sleep(0) for max speed // Note: STANDARD Edition Solace PubSub+ broker is limited to 10k msg/s max ingress } catch (InterruptedException e) { + Thread.currentThread().interrupt(); // restore the interrupt status, do not swallow it isShutdown = true; } } } isShutdown = true; - statsPrintingThread.shutdown(); // stop printing stats - Thread.sleep(1500); // give time for the ACKs to arrive from the broker - session.closeSession(); - System.out.println("Main thread quitting."); + } + + private static void teardownSolace() { + // teardownSolace() runs in main's finally on EVERY exit path (normal quit, ENTER, + // SIGINT, DOWN_ERROR, or an exception), so Solace teardown is never skipped. + isShutdown = true; + if (statsPrintingThread != null) { + statsPrintingThread.shutdown(); // stop printing stats + } + if (session != null) { + // gracefully finish outstanding ACKs before exit: give the broker time to + // confirm in-flight PERSISTENT messages before the session is closed + try { + Thread.sleep(1500); // give time for the ACKs to arrive from the broker + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); // preserve interrupt; still close the session below + } + session.closeSession(); + } } //////////////////////////////////////////////////////////////////////////// @@ -161,14 +250,12 @@ private static class PublishCallbackHandler implements JCSMPStreamingPublishCorr @Override public void responseReceivedEx(Object key) { assert key != null; // this shouldn't happen, this should only get called for an ACK - assert key instanceof BytesXMLMessage; logger.debug(String.format("ACK for Message %s", key)); // good enough, the broker has it now } - + @Override public void handleErrorEx(Object key, JCSMPException cause, long timestamp) { if (key != null) { // NACK - assert key instanceof BytesXMLMessage; logger.warn(String.format("NACK for Message %s - %s", key, cause)); // probably want to do something here. some error handling possibilities: // - send the message again diff --git a/src/main/java/com/solace/samples/jcsmp/patterns/GuaranteedSubscriber.java b/src/main/java/com/solace/samples/jcsmp/patterns/GuaranteedSubscriber.java index b8c26e7..bd6f835 100644 --- a/src/main/java/com/solace/samples/jcsmp/patterns/GuaranteedSubscriber.java +++ b/src/main/java/com/solace/samples/jcsmp/patterns/GuaranteedSubscriber.java @@ -16,24 +16,29 @@ package com.solace.samples.jcsmp.patterns; +import com.solacesystems.jcsmp.BytesMessage; import com.solacesystems.jcsmp.BytesXMLMessage; +import com.solacesystems.jcsmp.CapabilityType; import com.solacesystems.jcsmp.ConsumerFlowProperties; +import com.solacesystems.jcsmp.EndpointProperties; import com.solacesystems.jcsmp.FlowEventArgs; import com.solacesystems.jcsmp.FlowEventHandler; import com.solacesystems.jcsmp.FlowReceiver; import com.solacesystems.jcsmp.JCSMPChannelProperties; import com.solacesystems.jcsmp.JCSMPErrorResponseException; +import com.solacesystems.jcsmp.JCSMPErrorResponseSubcodeEx; import com.solacesystems.jcsmp.JCSMPException; import com.solacesystems.jcsmp.JCSMPFactory; import com.solacesystems.jcsmp.JCSMPProperties; import com.solacesystems.jcsmp.JCSMPSession; import com.solacesystems.jcsmp.JCSMPTransportException; -import com.solacesystems.jcsmp.OperationNotSupportedException; import com.solacesystems.jcsmp.Queue; import com.solacesystems.jcsmp.SessionEventArgs; import com.solacesystems.jcsmp.SessionEventHandler; +import com.solacesystems.jcsmp.Topic; import com.solacesystems.jcsmp.XMLMessageListener; import java.io.IOException; +import java.nio.charset.StandardCharsets; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; @@ -41,12 +46,15 @@ public class GuaranteedSubscriber { private static final String SAMPLE_NAME = GuaranteedSubscriber.class.getSimpleName(); private static final String QUEUE_NAME = "q_jcsmp_sub"; + // topic to map onto the queue; matches the publisher sample's topic root + private static final String TOPIC_NAME = "solace/samples/jcsmp/pers/pub/>"; private static final String API = "JCSMP"; - + private static volatile int msgRecvCounter = 0; // num messages received private static volatile boolean hasDetectedRedelivery = false; // detected any messages being redelivered? private static volatile boolean isShutdown = false; // are we done? private static FlowReceiver flowQueueReceiver; + private static JCSMPSession session; // remember to add log4j2.xml to your classpath private static final Logger logger = LogManager.getLogger(); // log4j2, but could also use SLF4J, JCL, etc. @@ -58,64 +66,146 @@ public static void main(String... args) throws JCSMPException, InterruptedExcept System.exit(-1); } System.out.println(API + " " + SAMPLE_NAME + " initializing..."); + try { + if (!setupSolace(args)) { + return; + } + // graceful shutdown: a SIGINT (Ctrl-C) signals the main loop to stop, then the + // hook joins the main thread so the cleanup in teardownSolace() (stop the flow, + // finish outstanding ACKs, close the session) runs to completion before the JVM + // halts. The JVM exits as soon as all shutdown hooks return, so the hook must + // wait, not just set a flag. + final Thread mainThread = Thread.currentThread(); + Runtime.getRuntime().addShutdownHook(new Thread(() -> { + System.out.println("Shutdown signal received, stopping consumer..."); + isShutdown = true; + try { + mainThread.join(5000); // wait for the main thread's flow stop, ACK drain, and session close + } catch (InterruptedException e) { + // nothing more we can do; the JVM is halting + } + })); + awaitMessages(); + } finally { + teardownSolace(); + System.out.println("Main thread quitting."); + } + } + private static boolean setupSolace(String[] args) throws JCSMPException { final JCSMPProperties properties = new JCSMPProperties(); properties.setProperty(JCSMPProperties.HOST, args[0]); // host:port properties.setProperty(JCSMPProperties.VPN_NAME, args[1]); // message-vpn properties.setProperty(JCSMPProperties.USERNAME, args[2]); // client-username if (args.length > 3) { properties.setProperty(JCSMPProperties.PASSWORD, args[3]); // client-password - } + } JCSMPChannelProperties channelProps = new JCSMPChannelProperties(); channelProps.setReconnectRetries(20); // recommended settings channelProps.setConnectRetriesPerHost(5); // recommended settings // https://docs.solace.com/Solace-PubSub-Messaging-APIs/API-Developer-Guide/Configuring-Connection-T.htm properties.setProperty(JCSMPProperties.CLIENT_CHANNEL_PROPERTIES, channelProps); - final JCSMPSession session; + // re-run idempotency: addSubscription() on the queue below throws + // "Subscription Already Exists" on every run after the first unless the session + // ignores duplicate-subscription errors (this property defaults to false) + properties.setProperty(JCSMPProperties.IGNORE_DUPLICATE_SUBSCRIPTION_ERROR, true); + // best practice: register a session event handler at session creation and handle + // each event appropriately, rather than only logging it session = JCSMPFactory.onlyInstance().createSession(properties, null, new SessionEventHandler() { @Override - public void handleEvent(SessionEventArgs event) { // could be reconnecting, connection lost, etc. + public void handleEvent(SessionEventArgs event) { logger.info("### Received a Session event: " + event); + switch (event.getEvent()) { + case RECONNECTING: // session went down, automatic reconnect attempt in progress + logger.warn("Session reconnecting; message delivery is paused until re-established"); + break; + case RECONNECTED: // automatic reconnect succeeded, session re-established + logger.info("Session reconnected; the flow rebinds and delivery resumes"); + break; + case DOWN_ERROR: // session was up and then went down; reconnects exhausted + logger.error("Session is down and reconnect attempts are exhausted, quitting."); + // the session will not recover; end the main loop so teardownSolace() + // runs in main's finally + isShutdown = true; + break; + default: + break; + } } }); session.connect(); // configure the queue API object locally final Queue queue = JCSMPFactory.onlyInstance().createQueue(QUEUE_NAME); + + // provision the durable queue in-process and map the topic onto it at startup, so a + // fresh broker works out of the box (rather than erroring out if the queue is absent). + // best practice: confirm the broker allows client-side endpoint management first + if (!session.isCapable(CapabilityType.ENDPOINT_MANAGEMENT)) { + System.err.println("This client/broker does not allow client-side endpoint management; " + + "provision the queue out-of-band or grant the capability. Exiting."); + return false; // teardownSolace() in main's finally closes the session + } + EndpointProperties endpointProps = new EndpointProperties(); + endpointProps.setAccessType(EndpointProperties.ACCESSTYPE_EXCLUSIVE); // single-consumer sample + endpointProps.setPermission(EndpointProperties.PERMISSION_CONSUME); + // provision the durable queue idempotently so a re-run is safe + session.provision(queue, endpointProps, JCSMPSession.FLAG_IGNORE_ALREADY_EXISTS); + // map the topic onto the queue with a subscription (pub/sub onto a queue); + // WAIT_FOR_CONFIRM blocks until the broker confirms the subscription is in place + Topic topic = JCSMPFactory.onlyInstance().createTopic(TOPIC_NAME); + session.addSubscription(queue, topic, JCSMPSession.WAIT_FOR_CONFIRM); + // Create a Flow be able to bind to and consume messages from the Queue. final ConsumerFlowProperties flow_prop = new ConsumerFlowProperties(); flow_prop.setEndpoint(queue); flow_prop.setAckMode(JCSMPProperties.SUPPORTED_MESSAGE_ACK_CLIENT); // best practice - flow_prop.setActiveFlowIndication(true); // Flow events will advise when + flow_prop.setActiveFlowIndication(true); // Flow events will advise when active/inactive System.out.printf("Attempting to bind to queue '%s' on the broker.%n", QUEUE_NAME); - try { - // see bottom of file for QueueFlowListener class, which receives the messages from the queue - flowQueueReceiver = session.createFlow(new QueueFlowListener(), flow_prop, null, new FlowEventHandler() { - @Override - public void handleEvent(Object source, FlowEventArgs event) { - // Flow events are usually: active, reconnecting (i.e. unbound), reconnected, active - logger.info("### Received a Flow event: " + event); - // try disabling and re-enabling the queue to see in action + // best practice: register a flow event handler at flow creation and handle each + // event appropriately, rather than only logging it + // see bottom of file for QueueFlowListener class, which receives the messages from the queue + flowQueueReceiver = session.createFlow(new QueueFlowListener(), flow_prop, null, new FlowEventHandler() { + @Override + public void handleEvent(Object source, FlowEventArgs event) { + logger.info("### Received a Flow event: " + event); + switch (event.getEvent()) { + case FLOW_ACTIVE: // bound and actively receiving from the queue + logger.info("Flow active: consuming from queue " + QUEUE_NAME); + break; + case FLOW_INACTIVE: // flow is up but this consumer is not the active one + logger.warn("Flow inactive: bound but not receiving (another consumer holds the active flow on this exclusive queue)"); + break; + case FLOW_RECONNECTING: // flow unbound (e.g. queue disabled); API is rebinding + logger.warn("Flow reconnecting: delivery paused while the API rebinds"); + break; + case FLOW_RECONNECTED: // flow rebind succeeded, delivery resumes + logger.info("Flow reconnected: delivery resumed"); + break; + case FLOW_UP: // the flow is established + logger.info("Flow up: bound to queue " + QUEUE_NAME); + break; + case FLOW_DOWN: // the flow was established and then went down + logger.error("Flow down: delivery from queue " + QUEUE_NAME + " has stopped"); + // decide here whether to recreate the flow or shut down; this sample + // shuts down so teardownSolace() runs in main's finally + isShutdown = true; + break; + default: + break; } - }); - } catch (OperationNotSupportedException e) { // not allowed to do this - throw e; - } catch (JCSMPErrorResponseException e) { // something else went wrong: queue not exist, queue shutdown, etc. - logger.error(e); - System.err.printf("%n*** Could not establish a connection to queue '%s': %s%n", QUEUE_NAME, e.getMessage()); - System.err.println("Create queue using PubSub+ Manager WebGUI, and add subscription "+ - GuaranteedPublisher.TOPIC_PREFIX+"*/pers/>"); - System.err.println(" or see the SEMP CURL scripts inside the 'semp-rest-api' directory."); - // could also try to retry, loop and retry until successfully able to connect to the queue - System.err.println("NOTE: see QueueProvision sample for how to construct queue with consumer app."); - System.err.println("Exiting."); - return; - } + // try disabling and re-enabling the queue to see these events in action + } + }); + return true; + } + + private static void awaitMessages() throws JCSMPException, IOException, InterruptedException { // tell the broker to start sending messages on this queue receiver flowQueueReceiver.start(); // async queue receive working now, so time to wait until done... - System.out.println(SAMPLE_NAME + " connected, and running. Press [ENTER] to quit."); + System.out.println(SAMPLE_NAME + " connected, and running. Press [ENTER] or Ctrl-C to quit."); while (System.in.available() == 0 && !isShutdown) { Thread.sleep(1000); // wait 1 second System.out.printf("%s %s Received msgs/s: %,d%n",API,SAMPLE_NAME,msgRecvCounter); // simple way of calculating message rates @@ -126,10 +216,24 @@ public void handleEvent(Object source, FlowEventArgs event) { } } isShutdown = true; - flowQueueReceiver.stop(); - Thread.sleep(1000); - session.closeSession(); // will also close consumer object - System.out.println("Main thread quitting."); + } + + private static void teardownSolace() { + // teardownSolace() runs in main's finally on EVERY exit path (normal quit, ENTER, + // SIGINT, DOWN_ERROR, or an exception), so Solace teardown is never skipped. + isShutdown = true; + if (flowQueueReceiver != null) { + // gracefully stop the flow and finish outstanding ACKs before exit + flowQueueReceiver.stop(); + try { + Thread.sleep(1000); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); // preserve interrupt; still close the session below + } + } + if (session != null) { + session.closeSession(); // will also close consumer object + } } //////////////////////////////////////////////////////////////////////////// @@ -140,6 +244,12 @@ private static class QueueFlowListener implements XMLMessageListener { @Override public void onReceive(BytesXMLMessage msg) { msgRecvCounter++; + // Payload lives in the binary attachment: the publisher writes it with setData(), + // so read it with getData(), NOT getBytes() (the separate, empty XML-content part). + if (msg instanceof BytesMessage) { + logger.debug(API + " " + SAMPLE_NAME + " received: " + + new String(((BytesMessage) msg).getData(), StandardCharsets.UTF_8)); + } if (msg.getRedelivered()) { // useful check // this is the broker telling the consumer that this message has been sent and not ACKed before. // this can happen if an exception is thrown, or the broker restarts, or the netowrk disconnects @@ -158,6 +268,10 @@ public void onException(JCSMPException e) { if (e instanceof JCSMPTransportException) { // all reconnect attempts failed isShutdown = true; // let's quit; or, could initiate a new connection attempt } else { + if (e instanceof JCSMPErrorResponseException) { // broker error response carries extra detail + JCSMPErrorResponseException ere = (JCSMPErrorResponseException) e; + logger.warn("Specifics: " + JCSMPErrorResponseSubcodeEx.getSubcodeAsString(ere.getSubcodeEx()) + ": " + ere.getResponsePhrase()); + } // Generally unrecoverable exception, probably need to recreate and restart the flow flowQueueReceiver.close(); // add logic in main thread to restart FlowReceiver, or can exit the program