From f550c3b98fcdbc6414e6fa08487881e32e476d86 Mon Sep 17 00:00:00 2001 From: adiel-sammak <71227923+aelsammak@users.noreply.github.com> Date: Thu, 22 Oct 2026 21:02:30 -0400 Subject: [PATCH 1/4] EBP-3125: Add blocking Python publish and receive reference samples - Add python-guaranteed-publisher-blocking-sample.py: publish_await_acknowledgement() with a positive timeout, no receipt listener; a timeout, NACK, overflow, or other publish failure stops the loop with exit code 1 - Add python-guaranteed-subscriber-blocking-sample.py: receive_message(timeout), process, then ack() on the main thread - Add python-direct-subscriber-blocking-sample.py: receive_message(timeout) loop with discard detection - Blocking subscribers call terminate(0): no handler thread drains the buffer after the loop stops; unacked guaranteed messages are redelivered, buffered direct messages drop - Keep the async twins' structure; no time.sleep(), no receive_async(), rate reports inline instead of a stats thread - Add the three samples to the Python sample-check roster --- .../evals/sample-check/check-samples.sh | 3 + ...ython-direct-subscriber-blocking-sample.py | 335 +++++++++++++++ ...on-guaranteed-publisher-blocking-sample.py | 366 ++++++++++++++++ ...n-guaranteed-subscriber-blocking-sample.py | 402 ++++++++++++++++++ 4 files changed, 1106 insertions(+) create mode 100644 plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-direct-subscriber-blocking-sample.py create mode 100644 plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-publisher-blocking-sample.py create mode 100644 plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-subscriber-blocking-sample.py diff --git a/plugins/solace-messaging-skills/skills/solace-application-development/references/python/evals/sample-check/check-samples.sh b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/evals/sample-check/check-samples.sh index 5de6288..766bbc4 100755 --- a/plugins/solace-messaging-skills/skills/solace-application-development/references/python/evals/sample-check/check-samples.sh +++ b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/evals/sample-check/check-samples.sh @@ -24,8 +24,11 @@ SAMPLES=( "python-solace-connection-config.py:solace_connection_config.py" "python-guaranteed-publisher-sample.py:guaranteed_publisher.py" "python-guaranteed-subscriber-sample.py:guaranteed_subscriber.py" + "python-guaranteed-publisher-blocking-sample.py:guaranteed_publisher_blocking.py" + "python-guaranteed-subscriber-blocking-sample.py:guaranteed_subscriber_blocking.py" "python-direct-publisher-sample.py:direct_publisher.py" "python-direct-subscriber-sample.py:direct_subscriber.py" + "python-direct-subscriber-blocking-sample.py:direct_subscriber_blocking.py" "python-direct-requestor-blocking-sample.py:direct_requestor_blocking.py" "python-direct-requestor-sample.py:direct_requestor.py" "python-direct-replier-sample.py:direct_replier.py" diff --git a/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-direct-subscriber-blocking-sample.py b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-direct-subscriber-blocking-sample.py new file mode 100644 index 0000000..debb8f3 --- /dev/null +++ b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-direct-subscriber-blocking-sample.py @@ -0,0 +1,335 @@ +""" +Derived from patterns/direct_receiver.py + SolaceSamples/solace-samples-python + https://github.com/SolaceSamples/solace-samples-python + +Elevated to documented best practices per the Python API developer guide: + Messaging Service: + https://docs.solace.com/API/API-Developer-Guide-Python/Python-API-Messaging-Service.md + Consuming Direct Messages: + https://docs.solace.com/API/API-Developer-Guide-Python/Python-DM-Receive.md + the Python API reference (units, defaults, listener contracts): + https://docs.solace.com/API-Developer-Online-Ref-Documentation/python/index.html + and the C API Best Practices (the Python API wraps the C API, so its callback, + reconnect, acknowledgement, and time-to-live practices apply): + https://docs.solace.com/API/API-Developer-Guide/C-API-Best-Practices.md + +Reference sample. Wired for a basic direct (at-most-once) pub/sub journey, BLOCKING +variant: basic-auth connect with a reconnection retry strategy, a plain topic +subscription (no queue, no provisioning), a DIRECT receiver with a capacity-bounded +buffer that detects discards and whose main loop calls receive_message(timeout), +reconnection and service interruption listeners registered BEFORE connect(), and a +graceful SIGINT/SIGTERM shutdown. Direct messaging is at-most-once: there is no +broker ACK and no redelivery, so there is NO client acknowledgement here (the inverse +of the guaranteed subscriber). The non-blocking variant is +python-direct-subscriber-sample.py. Structured as module-level functions, +setup_solace(...), connect_solace(...), receive_messages(...), and +teardown_solace(...), that share one SubscriberState; setup_solace() creates the +service and the receiver before any connection exists, and main() runs teardown from +its finally on every exit path after that. + +Why blocking: the application's own thread is the single thread of control for the +messaging data path. The main thread takes each message from the receiver and +processes it, so no message handler runs on an API thread and no application state is +shared with one. In Python the GIL lets only one thread run Python code at a time, so +the message-handler design gains less from its extra thread than it does in other +languages. This sample makes no throughput claim. The cost: the loop processes one +message at a time, so a slow loop fills the receive buffer and the receiver discards +messages, and a shutdown signal takes effect only when receive_message() returns. The +service event listeners still run on API threads. + +Only practices documented in canonical Solace sources are encoded here. + +Copied into a generated project as direct_subscriber_blocking.py, next to the shared +helper python-solace-connection-config.py copied as solace_connection_config.py. + +Any generated adaptation of this sample MUST begin with the exact line: + AI-assisted code. Review before production use. +(This reference sample itself carries no such header by design.) +""" + +import logging +import signal +import sys +import threading +import time +from dataclasses import dataclass + +from solace.messaging.config.retry_strategy import RetryStrategy +from solace.messaging.errors.pubsubplus_client_error import IncompleteMessageDeliveryError, PubSubPlusClientError +from solace.messaging.messaging_service import ( + MessagingService, + ReconnectionAttemptListener, + ReconnectionListener, + ServiceEvent, + ServiceInterruptionListener, +) +from solace.messaging.receiver.direct_message_receiver import DirectMessageReceiver +from solace.messaging.receiver.inbound_message import InboundMessage +from solace.messaging.resources.topic_subscription import TopicSubscription + +from solace_connection_config import load_service_properties + +APP_NAME = "DirectSubscriberBlocking" +# topic to subscribe to directly; matches the direct publisher sample's topic root +TOPIC_NAME = "solace/samples/python/direct/pub/>" +API = "Python" +# receive_message(timeout) blocks up to this many MILLISECONDS for the next message, then +# returns None. The loop checks the shutdown event each time the call returns, so this value +# bounds how long a shutdown signal waits during an empty wait. 1 s keeps Ctrl-C prompt and +# matches the rate report interval; a longer timeout wakes the loop less often but stops +# slower. Never pass None: the call then waits without a limit and a shutdown signal never +# takes effect while no message arrives. +RECEIVE_TIMEOUT_MS = 1000 +# how often the receive loop reports its message rate, in seconds +STATS_INTERVAL_S = 1.0 +# reconnect budget: 20 attempts 3 s apart, a minute of reconnect attempts, a reasonable +# default when the design says nothing about high availability. The interval is in +# MILLISECONDS per the RetryStrategy API reference. +RECONNECT_RETRIES = 20 +RECONNECT_RETRY_INTERVAL_MS = 3000 +# For HA failover resilience, use the C API Best Practices values +# (https://docs.solace.com/API/API-Developer-Guide/C-API-Best-Practices.md): +# HA failover: with_connection_retry_strategy(RetryStrategy.parametrized_retry(1, 3000)), +# with_reconnection_retry_strategy(RetryStrategy.parametrized_retry(20, 3000)), +# properties[transport_layer_properties.CONNECTION_RETRIES_PER_HOST] = 5 +# back pressure: once this many received messages wait for the receive loop, the receiver +# discards each new incoming message until there is room +# The capacity is a count of messages, not a size in bytes: the memory it holds depends on +# the size of each message. Tune it to the memory the application's deployment can give +# the buffer. The API has no default capacity (its default, on_back_pressure_elastic, +# buffers without bound); 1000 is the value the Python developer guide uses in its back +# pressure examples. +RECEIVE_BUFFER_CAPACITY = 1000 +# terminate(grace_period): how long to wait for the messages the API has already received +# to be taken from its buffer before the receiver stops. In this blocking sample only the +# receive loop takes them, and that loop has stopped by teardown, so a grace period only +# delays the exit. 0 stops at once: the buffered messages are dropped, which direct +# (at-most-once) messaging permits. +TERMINATE_GRACE_PERIOD_MS = 0 + +# API events go through standard logging; the trace() narration below is a separate channel +logger = logging.getLogger(APP_NAME) + + +@dataclass +class SubscriberState: + """What the main loop, the signal handler, and the API listener threads share: the + service, the receiver, the shutdown event, and the counters. The service and the + receiver are required, so every SubscriberState holds both (built, though not + necessarily connected or started). Use this type of app for receiving Direct + (at-most-once) messages from a topic.""" + messaging_service: MessagingService + receiver: DirectMessageReceiver + shutdown: threading.Event + connect_attempted: bool = False + exit_code: int = 0 # set to 1 on a failure exit (service interruption, a receive failure, or a failed terminate()) + msg_recv_counter: int = 0 # num messages received + has_detected_discard: bool = False # any discards seen? + + +def main(args: list[str]) -> int: + trace(f"{API} {APP_NAME} initializing...") + shutdown = threading.Event() + + def on_shutdown_signal(signum: int, _frame) -> None: + trace(f"Shutdown signal received ({signal.Signals(signum).name}), stopping subscriber...") + shutdown.set() + + # graceful shutdown: SIGINT (Ctrl-C) or SIGTERM sets the shutdown event, the receive + # loop exits once receive_message() returns, and teardown_solace() in the finally below + # stops the receiver and disconnects. Python runs signal handlers on the main thread, so + # the handler only flags; the cleanup runs on the main thread's normal exit path. + # Registered before connect, so a SIGTERM during the blocking connect() still reaches + # teardown. + signal.signal(signal.SIGINT, on_shutdown_signal) + signal.signal(signal.SIGTERM, on_shutdown_signal) + # basic username/password connection details, built by the shared connection-config + # helper: read from a config.json in the working directory if present, else from the + # command line ( [password]). Turning the + # application's input into Solace properties is the application's job, so it stays out + # of setup_solace(), which works from the properties alone. + properties = load_service_properties(args, APP_NAME) + # setup_solace() raises before any connection exists, so there is nothing to tear down + # if it fails + state = setup_solace(properties, shutdown) + try: + connect_solace(state) + receive_messages(state) + finally: + teardown_solace(state) + trace("Main thread quitting.") + return state.exit_code # non-zero after a failure, so scripts and supervisors see it + + +def setup_solace(properties: dict, shutdown: threading.Event) -> SubscriberState: + # build() creates the native session and resolves the host, so an unresolvable host + # fails here, before connect() + messaging_service = ( + MessagingService.builder() + # the builder merges each call in order and a later call wins, so the reconnection + # strategy comes first: it is the default, and reconnection keys in the properties + # (for example from config.json) override it + .with_reconnection_retry_strategy( + RetryStrategy.parametrized_retry(RECONNECT_RETRIES, RECONNECT_RETRY_INTERVAL_MS)) + .from_properties(properties) + .build() + ) + # DIRECT receiver: a topic subscription on the receiver (no queue, no provisioning). + # Direct messaging is at-most-once: messages flow straight to the subscriber with no + # broker ACK and no redelivery. start() applies the subscription and blocks until the + # broker confirms it, so the route is live before this subscriber reports ready. + # Capacity-bounded back pressure: once RECEIVE_BUFFER_CAPACITY messages wait for the + # receive loop, on_back_pressure_drop_latest(n) discards each new incoming message, and + # the next message delivered carries an internal discard indication. The API default + # (on_back_pressure_elastic) buffers without bound; on_back_pressure_drop_oldest(n) + # discards the oldest buffered message instead. Building the receiver needs only the + # built service; start() is what requires the connection. + receiver = ( + messaging_service.create_direct_message_receiver_builder() + .with_subscriptions([TopicSubscription.of(TOPIC_NAME)]) + .on_back_pressure_drop_latest(RECEIVE_BUFFER_CAPACITY) + .build() + ) + state = SubscriberState(messaging_service=messaging_service, receiver=receiver, shutdown=shutdown) + # best practice: register the service event listeners BEFORE connect(), so no + # reconnection or interruption event raised during or right after the connect is lost, + # and handle each event appropriately rather than only logging it + service_events = ServiceEventHandler(state) + messaging_service.add_reconnection_attempt_listener(service_events) + messaging_service.add_reconnection_listener(service_events) + messaging_service.add_service_interruption_listener(service_events) + return state + + +def connect_solace(state: SubscriberState) -> None: + state.connect_attempted = True + state.messaging_service.connect() # blocking connect + # no receive_async(): the receive loop takes each message with receive_message(). The + # API reference states the two receive styles are mutually exclusive on one receiver. + trace(f"Adding direct topic subscription '{TOPIC_NAME}'.") + state.receiver.start() + trace(f"{APP_NAME} subscribed and consuming. Press Ctrl-C to quit.") + + +def receive_messages(state: SubscriberState) -> None: + next_report = time.monotonic() + STATS_INTERVAL_S + while not state.shutdown.is_set(): + try: + # blocks up to RECEIVE_TIMEOUT_MS for the next message. None means a timeout, or + # that the receiver or the service is shutting down; either way the loop goes + # round and the shutdown check above decides whether to stop + message = state.receiver.receive_message(RECEIVE_TIMEOUT_MS) + except PubSubPlusClientError as error: + # raised once the service is down or the receiver is terminated. A service + # interruption has already set shutdown and exit_code + if not state.shutdown.is_set(): + logger.error("receive_message() failed, quitting: %s", error) + state.exit_code = 1 + break + if message is not None: + process_message(state, message) + # simple way of calculating message rates + now = time.monotonic() + if now >= next_report: + trace(f"{API} {APP_NAME} Received msgs/s: {state.msg_recv_counter:,}") + state.msg_recv_counter = 0 + if state.has_detected_discard: # at least one direct message was dropped before this subscriber + trace("*** Discard detected (at-most-once: a direct message was dropped) ***") + state.has_detected_discard = False # only show the warning once per report + next_report = now + STATS_INTERVAL_S + + +def process_message(state: SubscriberState, message: InboundMessage) -> None: + state.msg_recv_counter += 1 + # the publisher sample sends a bytearray payload (the binary attachment), read back with + # get_payload_as_bytes(); a str payload arrives via get_payload_as_string() instead + # best practice: handle an unexpected message format without raising (C API Best + # Practices); errors="replace" keeps a non-UTF-8 payload from raising in the receive loop + payload = message.get_payload_as_bytes() + text = payload.decode("utf-8", errors="replace") if payload is not None else message.get_payload_as_string() + trace(f"{API} {APP_NAME} received: {text}") + # at-most-once discard evidence (API reference): the broker sets the broker discard + # indication when it discarded direct messages destined for this subscriber for any + # reason, and the API sets the internal one when its own receiver buffer dropped + # messages. There is NO redelivery in direct messaging. + discard = message.get_message_discard_notification() + if discard.has_broker_discard_indication() or discard.has_internal_discard_indication(): + state.has_detected_discard = True + # Direct messaging is at-most-once: there is no consumer-side acknowledgement here (the + # inverse of the guaranteed CLIENT-ack subscriber). The message is consumed as it + # arrives; the broker holds no copy and expects no ack. + + +def teardown_solace(state: SubscriberState) -> None: + # Application cleanup belongs here: once setup_solace() returns, teardown_solace() runs + # in main's finally on EVERY exit path (normal quit, SIGINT/SIGTERM, service + # interruption, or an exception), so Solace teardown and application-side cleanup are + # never skipped. + state.shutdown.set() + if state.receiver.is_running(): + # stop delivery at once (TERMINATE_GRACE_PERIOD_MS is 0): the receive loop has + # stopped, so nothing takes more messages from the API's buffer. terminate() raises + # IncompleteMessageDeliveryError when the buffer still holds messages. Direct is + # at-most-once, so dropping them is a normal stop, not a failure. Catch it so the + # disconnect below still runs + try: + state.receiver.terminate(TERMINATE_GRACE_PERIOD_MS) + except IncompleteMessageDeliveryError as error: + logger.warning("Receiver stopped with unprocessed direct messages in its buffer; " + "they are dropped (at-most-once): %s", error) + except PubSubPlusClientError as error: + logger.error("Receiver terminate() failed: %s", error) + state.exit_code = 1 + if state.connect_attempted: + # disconnect() raises IllegalStateError on a service that never attempted to connect, + # hence the flag; on a service that is already down it returns quietly + state.messaging_service.disconnect() # will also release the receiver + + +def trace(message: str) -> None: + """Demo narration sink: every status line in this sample funnels through this one + function. An application replaces this single body to route narration to its + logger or reporting system. The logger calls for API events (service) are a + separate channel and stay as they are.""" + print(message, flush=True) + + +############################################################################ + + +class ServiceEventHandler(ReconnectionAttemptListener, ReconnectionListener, ServiceInterruptionListener): + """The three service event listeners in one class. Callbacks run on an API thread, even + in this blocking sample: return promptly and do not block in them, since events and + delivery stall while a callback runs (C API Best Practices).""" + + def __init__(self, state: SubscriberState) -> None: + self.state = state + + def on_reconnecting(self, event: ServiceEvent) -> None: + # connection lost, automatic reconnect attempt in progress: direct delivery is paused, + # and the broker does not store direct messages for this subscriber while it is + # disconnected (at-most-once). receive_message() keeps timing out meanwhile + logger.warning("Service reconnecting; direct delivery is paused until re-established: %s (%s)", + event.get_message(), event.get_cause()) + + def on_reconnected(self, event: ServiceEvent) -> None: + # automatic reconnect succeeded; the API re-applies the topic subscription and delivery + # resumes (unlike JCSMP, no reapply-subscriptions property is needed) + logger.info("Service reconnected to %s; the subscription is restored and delivery resumes", + event.get_broker_uri()) + + def on_service_interrupted(self, event: ServiceEvent) -> None: + # the connection went down and cannot be restored (reconnect attempts exhausted) + logger.error("Service interrupted and cannot be restored, quitting: %s (%s)", + event.get_message(), event.get_cause()) + # Application cleanup signal: the service will not recover. Trigger application-side + # cleanup from here. This sample sets shutdown, so the receive loop exits and + # teardown_solace() runs in main's finally. + self.state.exit_code = 1 + self.state.shutdown.set() + + +if __name__ == "__main__": + logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s") + sys.exit(main(sys.argv[1:])) diff --git a/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-publisher-blocking-sample.py b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-publisher-blocking-sample.py new file mode 100644 index 0000000..84aeff7 --- /dev/null +++ b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-publisher-blocking-sample.py @@ -0,0 +1,366 @@ +""" +Derived from patterns/guaranteed_publisher.py and howtos/pubsub/how_to_publish_persistent_message.py + SolaceSamples/solace-samples-python + https://github.com/SolaceSamples/solace-samples-python + +Elevated to documented best practices per the Python API developer guide: + Messaging Service: + https://docs.solace.com/API/API-Developer-Guide-Python/Python-API-Messaging-Service.md + Publishing Persistent Messages: + https://docs.solace.com/API/API-Developer-Guide-Python/Python-PM-Publish.md + the Python API reference (units, defaults, listener contracts): + https://docs.solace.com/API-Developer-Online-Ref-Documentation/python/index.html + and the C API Best Practices (the Python API wraps the C API, so its callback, + reconnect, acknowledgement, and time-to-live practices apply): + https://docs.solace.com/API/API-Developer-Guide/C-API-Best-Practices.md + +Reference sample. Wired for a basic publish-subscribe journey, BLOCKING variant: +basic-auth connect with a reconnection retry strategy, PERSISTENT binary publish to a +topic from a templated message builder that sets the time-to-live once, one blocking +publish_await_acknowledgement() per message with a POSITIVE acknowledgement timeout, +reconnection and service interruption listeners registered BEFORE connect(), and a +graceful SIGINT/SIGTERM shutdown. There is no publish receipt listener: the call +returns once the broker ACKs the message and raises on a NACK or a timeout, and any +publish failure stops the publisher with exit code 1. The non-blocking variant is +python-guaranteed-publisher-sample.py. Structured as module-level functions, +setup_solace(...), connect_solace(...), run_publish_loop(...), and teardown_solace(...), +that share one PublisherState; setup_solace() creates the service and the publisher +before any connection exists, and main() runs teardown from its finally on every exit +path after that. + +Why blocking: the application's own thread is the single thread of control for the +messaging data path. The main thread publishes each message and handles its ACK or +NACK itself, so no receipt callback runs on an API thread and no application state is +shared with one. In Python the GIL lets only one thread run Python code at a time, so +the receipt-listener design gains less from its extra thread than it does in other +languages. This sample makes no throughput claim. The cost: only one persistent +message is in flight at a time, so each message waits a full broker round trip before +the next one goes out. The service event listeners still run on API threads. + +Only practices documented in canonical Solace sources are encoded here. + +Copied into a generated project as guaranteed_publisher_blocking.py, next to the shared +helper python-solace-connection-config.py copied as solace_connection_config.py. + +Any generated adaptation of this sample MUST begin with the exact line: + AI-assisted code. Review before production use. +(This reference sample itself carries no such header by design.) +""" + +import logging +import signal +import sys +import threading +import time +import uuid +from dataclasses import dataclass + +from solace.messaging.config.retry_strategy import RetryStrategy +from solace.messaging.config.solace_properties import message_properties +from solace.messaging.errors.pubsubplus_client_error import ( + IncompleteMessageDeliveryError, + MessageDestinationDoesNotExistError, + MessageNotAcknowledgedByBrokerError, + MessageRejectedByBrokerError, + PublisherOverflowError, + PubSubPlusClientError, + PubSubTimeoutError, +) +from solace.messaging.messaging_service import ( + MessagingService, + ReconnectionAttemptListener, + ReconnectionListener, + ServiceEvent, + ServiceInterruptionListener, +) +from solace.messaging.publisher.persistent_message_publisher import PersistentMessagePublisher +from solace.messaging.resources.topic import Topic + +from solace_connection_config import load_service_properties + +APP_NAME = "GuaranteedPublisherBlocking" +TOPIC_PREFIX = "solace/samples/" # used as the topic "root" +API = "Python" +PAYLOAD_SIZE = 512 +# POSITIVE acknowledgement timeout in MILLISECONDS: publish_await_acknowledgement() blocks +# up to this long for the broker's ACK or NACK, then raises PubSubTimeoutError. Never pass +# None: the API then waits for the ACK without a limit, so a lost ACK hangs the publisher +# and a shutdown signal never takes effect. An ACK normally comes back in milliseconds; +# 5 s also covers a slow broker or a short reconnect. A longer reconnect than that times +# out the message in flight, and the publisher stops. +PUBLISH_ACK_TIMEOUT_MS = 5000 +# how often the publish loop reports its message rate, in seconds +STATS_INTERVAL_S = 1.0 +# reconnect budget: 20 attempts 3 s apart, a minute of reconnect attempts, a reasonable +# default when the design says nothing about high availability. The interval is in +# MILLISECONDS per the RetryStrategy API reference. +RECONNECT_RETRIES = 20 +RECONNECT_RETRY_INTERVAL_MS = 3000 +# For HA failover resilience, use the C API Best Practices values +# (https://docs.solace.com/API/API-Developer-Guide/C-API-Best-Practices.md): +# HA failover: with_connection_retry_strategy(RetryStrategy.parametrized_retry(1, 3000)), +# with_reconnection_retry_strategy(RetryStrategy.parametrized_retry(20, 3000)), +# properties[transport_layer_properties.CONNECTION_RETRIES_PER_HOST] = 5 +# terminate(grace_period): how long to wait for a send or its broker receipt that is still +# outstanding before the publisher stops; terminate() raises +# IncompleteMessageDeliveryError when one is still outstanding after that. With a blocking +# publish, at most one message is outstanding at teardown: the one whose wait timed out +# or failed. A bounded grace period keeps shutdown time predictable, for example within +# the stop timeout of a process supervisor, so the disconnect still runs. The API default +# is 600000 ms (10 minutes). Adjust the value to suit the application: longer gives that +# message more time to complete, shorter exits sooner. +TERMINATE_GRACE_PERIOD_MS = 10_000 +# best practice for Guaranteed publishing (C API Best Practices): set a time-to-live so an +# unconsumed message does not sit on a queue forever. In MILLISECONDS; 0, the API default, +# never expires. The queue must have respect-ttl enabled (off by default), or the broker +# ignores the TTL. Adjust the value to how long a message stays useful to its consumers. +# For most applications, make it at least the longest time a consumer can take to come +# back (restart plus reconnect), or a multiple of that time for a safety margin, so a +# message does not expire while its consumer recovers; use a shorter or longer TTL only +# for a specific need. +MESSAGE_TTL_MS = 60_000 + +# API events go through standard logging; the trace() narration below is a separate channel +logger = logging.getLogger(APP_NAME) + + +@dataclass +class PublisherState: + """What the publish loop, the signal handler, and the API listener threads share: the + service, the publisher, the shutdown event, and the counters. The service and the + publisher are required, so every PublisherState holds both (built, though not + necessarily connected or started).""" + messaging_service: MessagingService + publisher: PersistentMessagePublisher + shutdown: threading.Event + connect_attempted: bool = False + exit_code: int = 0 # set to 1 on a failure exit (service interruption, a publish failure, or a failed terminate()) + msg_sent_counter: int = 0 # num messages ACKed since the last rate report + + +def main(args: list[str]) -> int: + trace(f"{API} {APP_NAME} initializing...") + shutdown = threading.Event() + + def on_shutdown_signal(signum: int, _frame) -> None: + trace(f"Shutdown signal received ({signal.Signals(signum).name}), stopping publisher...") + shutdown.set() + + # graceful shutdown: SIGINT (Ctrl-C) or SIGTERM sets the shutdown event, the publish + # loop exits, and teardown_solace() in the finally below stops the publisher and + # disconnects. Python runs signal handlers on the main thread, so the handler only + # flags; the cleanup runs on the main thread's normal exit path. Registered before + # connect, so a SIGTERM during the blocking connect() still reaches teardown. + signal.signal(signal.SIGINT, on_shutdown_signal) + signal.signal(signal.SIGTERM, on_shutdown_signal) + # basic username/password connection details, built by the shared connection-config + # helper: read from a config.json in the working directory if present, else from the + # command line ( [password]). Turning the + # application's input into Solace properties is the application's job, so it stays out + # of setup_solace(), which works from the properties alone. + properties = load_service_properties(args, APP_NAME) + # setup_solace() raises before any connection exists, so there is nothing to tear down + # if it fails + state = setup_solace(properties, shutdown) + try: + connect_solace(state) + run_publish_loop(state) + finally: + teardown_solace(state) + trace("Main thread quitting.") + return state.exit_code # non-zero after a failure, so scripts and supervisors see it + + +def setup_solace(properties: dict, shutdown: threading.Event) -> PublisherState: + # build() creates the native session and resolves the host, so an unresolvable host + # fails here, before connect() + messaging_service = ( + MessagingService.builder() + # the builder merges each call in order and a later call wins, so the reconnection + # strategy comes first: it is the default, and reconnection keys in the properties + # (for example from config.json) override it + .with_reconnection_retry_strategy( + RetryStrategy.parametrized_retry(RECONNECT_RETRIES, RECONNECT_RETRY_INTERVAL_MS)) + .from_properties(properties) + .build() + ) + # PERSISTENT publisher with no back-pressure setting: publish_await_acknowledgement() + # returns only after the broker's receipt, so the publish buffer never holds more than + # the one message in flight, and the API default (on_back_pressure_elastic) cannot grow. + # The non-blocking sample bounds the buffer with on_back_pressure_wait() because its + # publish() returns before the receipt. Building the publisher needs only the built + # service; start() is what requires the connection. One service can serve several + # publishers. + publisher = messaging_service.create_persistent_message_publisher_builder().build() + state = PublisherState(messaging_service=messaging_service, publisher=publisher, shutdown=shutdown) + # best practice: register the service event listeners BEFORE connect(), so no + # reconnection or interruption event raised during or right after the connect is lost, + # and handle each event appropriately rather than only logging it + service_events = ServiceEventHandler(state) + messaging_service.add_reconnection_attempt_listener(service_events) + messaging_service.add_reconnection_listener(service_events) + messaging_service.add_service_interruption_listener(service_events) + return state + + +def connect_solace(state: PublisherState) -> None: + state.connect_attempted = True + state.messaging_service.connect() # blocking connect + state.publisher.start() # raises IllegalStateError unless the service is connected + # no publish receipt listener: each publish_await_acknowledgement() call returns the + # broker's ACK as its normal return and the NACK as an exception, on the main thread + + +def run_publish_loop(state: PublisherState) -> None: + # one OutboundMessageBuilder as a message "template", reused for every message: set the + # headers that are the same on every message (here the TTL) once on the builder. + # build() starts each message as a copy of the template, so there is no set call per + # message for these headers + message_builder = ( + state.messaging_service.message_builder() + .with_property(message_properties.PERSISTENT_TIME_TO_LIVE, MESSAGE_TTL_MS) + ) + + trace(f"{API} {APP_NAME} connected, and running. Press Ctrl-C to quit.") + trace(f"Publishing to topic '{TOPIC_PREFIX}{API.lower()}/pers/pub/...', " + "please ensure queue has matching subscription.") + # the rate report runs inline on this thread, not on a stats thread, so the main + # thread stays the only application thread + next_report = time.monotonic() + STATS_INTERVAL_S + msg_count = 0 + while not state.shutdown.is_set(): + # each loop, change the payload as an example + chosen_character = chr(msg_count % 26 + 65) # choose a "random" letter [A-Z] + # a bytearray payload travels as the binary attachment; consumers read it back with + # get_payload_as_bytes() (a str payload would arrive via get_payload_as_string()) + payload = bytearray(chosen_character.encode("ascii") * PAYLOAD_SIZE) + msg_id = str(uuid.uuid4()) + # the per-message difference goes in build()'s additional_message_properties, which + # applies to this message only; a with_*() call on the builder would change the + # template, so every later message would carry it too + message = message_builder.build( + payload, additional_message_properties={message_properties.APPLICATION_MESSAGE_ID: msg_id}) + # NOTE: publishing to topic, so make sure the consumer's queue is subscribed to the same topic, + # or enable "Reject Message to Sender on No Subscription Match" in the client-profile + topic = Topic.of(f"{TOPIC_PREFIX}{API.lower()}/pers/pub/{chosen_character}") + try: + # blocks until the broker ACKs this message or PUBLISH_ACK_TIMEOUT_MS passes: a + # normal return IS the ACK, so the message id needs no receipt correlation. + # During an automatic reconnect the message waits for the transport and its ACK + # comes after the reconnect. A shutdown signal that arrives during the wait + # takes effect once the call returns, within PUBLISH_ACK_TIMEOUT_MS. + state.publisher.publish_await_acknowledgement(message, topic, PUBLISH_ACK_TIMEOUT_MS) + state.msg_sent_counter += 1 + msg_count += 1 + except PubSubTimeoutError: + # no ACK or NACK inside the timeout. The broker can still have the message (the + # ACK can be late, for example after a long reconnect), so sending it again can + # deliver a duplicate + logger.error("No ACK for Message ID %s within %d ms, quitting", msg_id, PUBLISH_ACK_TIMEOUT_MS) + state.exit_code = 1 + break + except (MessageRejectedByBrokerError, MessageNotAcknowledgedByBrokerError, + MessageDestinationDoesNotExistError) as error: + # NACK: the broker did not persist the message, and the exception says why (for + # example MessageDestinationDoesNotExistError when no queue subscribes to the + # topic and the client-profile rejects on no subscription match). Some error + # handling possibilities instead of quitting: + # - send it again, or send it somewhere else (error handling queue?) + # - log and continue + # - pause and retry (backoff) + logger.error("NACK for Message ID %s, quitting: %s", msg_id, error) + state.exit_code = 1 + break + except PublisherOverflowError as error: + # the API reference lists it for this call; with one message in flight and the + # default back-pressure setting, the publish buffer does not overflow + logger.error("Publish buffer overflow for Message ID %s, quitting: %s", msg_id, error) + state.exit_code = 1 + break + except PubSubPlusClientError as error: + # the message cannot be sent (for example the service is down or the publisher + # terminated) and retrying would not help + logger.error("publish_await_acknowledgement() failed, quitting: %s", error) + state.exit_code = 1 + break # let's quit; or, could initiate a new connection attempt + # simple way of calculating message rates + now = time.monotonic() + if now >= next_report: + trace(f"{API} {APP_NAME} Published msgs/s: {state.msg_sent_counter:,}") + state.msg_sent_counter = 0 + next_report = now + STATS_INTERVAL_S + # no delay between messages: the next message goes out as soon as the previous one is + # ACKed. Note: STANDARD Edition Solace broker is limited to 10k msg/s max ingress + state.shutdown.set() + + +def teardown_solace(state: PublisherState) -> None: + # Application cleanup belongs here: once setup_solace() returns, teardown_solace() runs + # in main's finally on EVERY exit path (normal quit, SIGINT/SIGTERM, a publish failure, + # service interruption, or an exception), so Solace teardown and application-side + # cleanup are never skipped. + state.shutdown.set() + if state.publisher.is_running(): + # after a normal quit every message is already ACKed, so terminate() returns at once. + # After a timeout or a failure, it gives the one outstanding message up to the grace + # period. It raises IncompleteMessageDeliveryError when that message is still + # outstanding after the grace period; catch it so the disconnect below still runs + try: + state.publisher.terminate(TERMINATE_GRACE_PERIOD_MS) + except IncompleteMessageDeliveryError as error: + logger.error("Publisher stopped with a send or receipt still outstanding after %d ms: %s", + TERMINATE_GRACE_PERIOD_MS, error) + state.exit_code = 1 + except PubSubPlusClientError as error: + logger.error("Publisher terminate() failed: %s", error) + state.exit_code = 1 + if state.connect_attempted: + # disconnect() raises IllegalStateError on a service that never attempted to connect, + # hence the flag; on a service that is already down it returns quietly + state.messaging_service.disconnect() + + +def trace(message: str) -> None: + """Demo narration sink: every status line in this sample funnels through this one + function. An application replaces this single body to route narration to its + logger or reporting system. The logger calls for API events (service, publish + failure) are a separate channel and stay as they are.""" + print(message, flush=True) + + +############################################################################ + + +class ServiceEventHandler(ReconnectionAttemptListener, ReconnectionListener, ServiceInterruptionListener): + """The three service event listeners in one class. Callbacks run on an API thread, even + in this blocking sample: return promptly and do not block in them, since events and + delivery stall while a callback runs (C API Best Practices).""" + + def __init__(self, state: PublisherState) -> None: + self.state = state + + def on_reconnecting(self, event: ServiceEvent) -> None: + # connection lost, automatic reconnect attempt in progress: the message in flight + # waits for the transport, and publish_await_acknowledgement() keeps waiting for its + # ACK up to PUBLISH_ACK_TIMEOUT_MS + logger.warning("Service reconnecting; the message in flight waits for its ACK: %s (%s)", + event.get_message(), event.get_cause()) + + def on_reconnected(self, event: ServiceEvent) -> None: + # automatic reconnect succeeded: the waiting message goes out + logger.info("Service reconnected to %s", event.get_broker_uri()) + + def on_service_interrupted(self, event: ServiceEvent) -> None: + # the connection went down and cannot be restored (reconnect attempts exhausted) + logger.error("Service interrupted and cannot be restored, quitting: %s (%s)", + event.get_message(), event.get_cause()) + # Application cleanup signal: the service will not recover. Trigger application-side + # cleanup from here. This sample sets shutdown, so the publish loop exits and + # teardown_solace() runs in main's finally. + self.state.exit_code = 1 + self.state.shutdown.set() + + +if __name__ == "__main__": + logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s") + sys.exit(main(sys.argv[1:])) diff --git a/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-subscriber-blocking-sample.py b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-subscriber-blocking-sample.py new file mode 100644 index 0000000..7403cad --- /dev/null +++ b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-subscriber-blocking-sample.py @@ -0,0 +1,402 @@ +""" +Derived from patterns/guaranteed_receiver.py + SolaceSamples/solace-samples-python + https://github.com/SolaceSamples/solace-samples-python + +Elevated to documented best practices per the Python API developer guide: + Messaging Service: + https://docs.solace.com/API/API-Developer-Guide-Python/Python-API-Messaging-Service.md + Consuming Persistent Messages: + https://docs.solace.com/API/API-Developer-Guide-Python/Python-PM-Receive.md + Creating Queues with the Solace Python API: + https://docs.solace.com/API/API-Developer-Guide-Python/Python-API-Create-Queues.md + the Python API reference (units, defaults, listener contracts): + https://docs.solace.com/API-Developer-Online-Ref-Documentation/python/index.html + and the C API Best Practices (the Python API wraps the C API, so its callback, + reconnect, acknowledgement, and time-to-live practices apply): + https://docs.solace.com/API/API-Developer-Guide/C-API-Best-Practices.md + +Reference sample. Wired for a basic publish-subscribe journey, BLOCKING variant: +basic-auth connect with a reconnection retry strategy, in-process provisioning of a +durable exclusive queue plus a topic subscription so a fresh broker works out of the +box, a CLIENT-ack receiver whose main loop calls receive_message(timeout), processes +the message, and then ACKs it, receiver active/passive state and termination +handling, reconnection and service interruption listeners registered BEFORE +connect(), and a graceful SIGINT/SIGTERM shutdown. The non-blocking variant is +python-guaranteed-subscriber-sample.py. Structured as module-level functions, +setup_solace(...), connect_solace(...), receive_messages(...), and +teardown_solace(...), that share one SubscriberState; setup_solace() creates the +service and the receiver before any connection exists, and main() runs teardown from +its finally on every exit path after that. + +Why blocking: the application's own thread is the single thread of control for the +messaging data path. The main thread takes each message from the receiver, processes +it, and ACKs it, so no message handler runs on an API thread and no application state +is shared with one. In Python the GIL lets only one thread run Python code at a time, +so the message-handler design gains less from its extra thread than it does in other +languages. This sample makes no throughput claim. The cost: the loop processes one +message at a time, and a shutdown signal takes effect only when receive_message() +returns. The service event, receiver state, and termination listeners still run on +API threads. + +Only practices documented in canonical Solace sources are encoded here. + +Copied into a generated project as guaranteed_subscriber_blocking.py, next to the +shared helper python-solace-connection-config.py copied as solace_connection_config.py. + +Any generated adaptation of this sample MUST begin with the exact line: + AI-assisted code. Review before production use. +(This reference sample itself carries no such header by design.) +""" + +import logging +import signal +import sys +import threading +import time +from dataclasses import dataclass + +from solace.messaging.config.missing_resources_creation_configuration import MissingResourcesCreationStrategy +from solace.messaging.config.receiver_activation_passivation_configuration import ( + ReceiverState, + ReceiverStateChangeListener, +) +from solace.messaging.config.retry_strategy import RetryStrategy +from solace.messaging.errors.pubsubplus_client_error import IncompleteMessageDeliveryError, PubSubPlusClientError +from solace.messaging.messaging_service import ( + MessagingService, + ReconnectionAttemptListener, + ReconnectionListener, + ServiceEvent, + ServiceInterruptionListener, +) +from solace.messaging.receiver.inbound_message import InboundMessage +from solace.messaging.receiver.persistent_message_receiver import PersistentMessageReceiver +from solace.messaging.resources.queue import Queue +from solace.messaging.resources.topic_subscription import TopicSubscription +from solace.messaging.utils.life_cycle_control import TerminationEvent, TerminationNotificationListener + +from solace_connection_config import load_service_properties + +APP_NAME = "GuaranteedSubscriberBlocking" +QUEUE_NAME = "q_python_sub" +# topic to map onto the queue; matches the publisher sample's topic root +TOPIC_NAME = "solace/samples/python/pers/pub/>" +API = "Python" +# receive_message(timeout) blocks up to this many MILLISECONDS for the next message, then +# returns None. The loop checks the shutdown event each time the call returns, so this value +# bounds how long a shutdown signal waits during an empty wait. 1 s keeps Ctrl-C prompt and +# matches the rate report interval; a longer timeout wakes the loop less often but stops +# slower. Never pass None: the call then waits without a limit and a shutdown signal never +# takes effect while the queue is empty. +RECEIVE_TIMEOUT_MS = 1000 +# how often the receive loop reports its message rate, in seconds +STATS_INTERVAL_S = 1.0 +# reconnect budget: 20 attempts 3 s apart, a minute of reconnect attempts, a reasonable +# default when the design says nothing about high availability. The interval is in +# MILLISECONDS per the RetryStrategy API reference. +RECONNECT_RETRIES = 20 +RECONNECT_RETRY_INTERVAL_MS = 3000 +# For HA failover resilience, use the C API Best Practices values +# (https://docs.solace.com/API/API-Developer-Guide/C-API-Best-Practices.md): +# HA failover: with_connection_retry_strategy(RetryStrategy.parametrized_retry(1, 3000)), +# with_reconnection_retry_strategy(RetryStrategy.parametrized_retry(20, 3000)), +# properties[transport_layer_properties.CONNECTION_RETRIES_PER_HOST] = 5 +# terminate(grace_period): how long to wait for the messages the API has already received +# to be taken from its buffer before the receiver stops. In this blocking sample only the +# receive loop takes them, and that loop has stopped by teardown, so a grace period only +# delays the exit. 0 stops at once: a message received but not yet ACKed stays on the queue +# and is redelivered, so nothing is lost. +TERMINATE_GRACE_PERIOD_MS = 0 + +# API events go through standard logging; the trace() narration below is a separate channel +logger = logging.getLogger(APP_NAME) + + +@dataclass +class SubscriberState: + """What the main loop, the signal handler, and the API listener threads share: the + service, the receiver, the shutdown event, and the counters. The service and the + receiver are required, so every SubscriberState holds both (built, though not + necessarily connected or started). Use this type of app for receiving Guaranteed + messages (e.g. via a queue endpoint).""" + messaging_service: MessagingService + receiver: PersistentMessageReceiver + shutdown: threading.Event + connect_attempted: bool = False + exit_code: int = 0 # set to 1 on a failure exit (service interruption, receiver termination, or a receive failure) + msg_recv_counter: int = 0 # num messages received + has_detected_redelivery: bool = False # detected any messages being redelivered? + + +def main(args: list[str]) -> int: + trace(f"{API} {APP_NAME} initializing...") + shutdown = threading.Event() + + def on_shutdown_signal(signum: int, _frame) -> None: + trace(f"Shutdown signal received ({signal.Signals(signum).name}), stopping consumer...") + shutdown.set() + + # graceful shutdown: SIGINT (Ctrl-C) or SIGTERM sets the shutdown event, the receive + # loop exits once receive_message() returns, and teardown_solace() in the finally below + # stops the receiver and disconnects. Python runs signal handlers on the main thread, so + # the handler only flags; the cleanup runs on the main thread's normal exit path. + # Registered before connect, so a SIGTERM during the blocking connect() still reaches + # teardown. + signal.signal(signal.SIGINT, on_shutdown_signal) + signal.signal(signal.SIGTERM, on_shutdown_signal) + # basic username/password connection details, built by the shared connection-config + # helper: read from a config.json in the working directory if present, else from the + # command line ( [password]). Turning the + # application's input into Solace properties is the application's job, so it stays out + # of setup_solace(), which works from the properties alone. + properties = load_service_properties(args, APP_NAME) + # setup_solace() raises before any connection exists, so there is nothing to tear down + # if it fails + state = setup_solace(properties, shutdown) + try: + if not connect_solace(state): + return 1 # the bind failed; the finally below still runs teardown + receive_messages(state) + finally: + teardown_solace(state) + trace("Main thread quitting.") + return state.exit_code # non-zero after a failure, so scripts and supervisors see it + + +def setup_solace(properties: dict, shutdown: threading.Event) -> SubscriberState: + # build() creates the native session and resolves the host, so an unresolvable host + # fails here, before connect() + messaging_service = ( + MessagingService.builder() + # the builder merges each call in order and a later call wins, so the reconnection + # strategy comes first: it is the default, and reconnection keys in the properties + # (for example from config.json) override it + .with_reconnection_retry_strategy( + RetryStrategy.parametrized_retry(RECONNECT_RETRIES, RECONNECT_RETRY_INTERVAL_MS)) + .from_properties(properties) + .build() + ) + # configure the queue API object locally: durable, exclusive (single-consumer sample) + queue = Queue.durable_exclusive_queue(QUEUE_NAME) + + # ELEVATION: provision the durable queue in-process and map the topic onto it at + # start, so a fresh broker works out of the box. CREATE_ON_START creates the queue + # passed to build() when start() runs, as long as the client has permission; the + # default, DO_NOT_CREATE, disables provisioning so an administrator provisions the + # queue instead. with_subscriptions() on a durable queue adds the topic subscription + # to the queue at start (pub/sub onto a queue). A re-run reuses the existing queue + # and subscription. + receiver_builder = ( + messaging_service.create_persistent_message_receiver_builder() + .with_missing_resources_creation_strategy(MissingResourcesCreationStrategy.CREATE_ON_START) + .with_subscriptions([TopicSubscription.of(TOPIC_NAME)]) + # best practice: client acknowledgement (also the API default), so a message leaves + # the queue only after the loop has processed it and called ack() + .with_message_client_acknowledgement() + # request active/passive state changes: on an exclusive queue only one bound receiver + # is ACTIVE; any other is PASSIVE (bound, not receiving) until the active one goes away + .with_activation_passivation_support(ReceiverStateHandler()) + ) + # building the receiver needs only the built service; start() is what requires the + # connection + receiver = receiver_builder.build(queue) + state = SubscriberState(messaging_service=messaging_service, receiver=receiver, shutdown=shutdown) + # a broker-initiated termination (for example the queue was deleted or shut down) + # surfaces here; the handler ends the receive loop (see ReceiverTerminationHandler below) + receiver.set_termination_notification_listener(ReceiverTerminationHandler(state)) + # best practice: register the service event listeners BEFORE connect(), so no + # reconnection or interruption event raised during or right after the connect is lost, + # and handle each event appropriately rather than only logging it + service_events = ServiceEventHandler(state) + messaging_service.add_reconnection_attempt_listener(service_events) + messaging_service.add_reconnection_listener(service_events) + messaging_service.add_service_interruption_listener(service_events) + return state + + +def connect_solace(state: SubscriberState) -> bool: + state.connect_attempted = True + state.messaging_service.connect() # blocking connect + # no receive_async(): the receive loop takes each message with receive_message(). The + # API reference states the two receive styles are mutually exclusive on one receiver. + trace(f"Attempting to bind to queue '{QUEUE_NAME}' on the broker.") + try: + # start() provisions the queue plus its subscription (CREATE_ON_START) and binds + # to the queue; the broker then starts sending messages, and the API buffers them + # until receive_message() takes them + state.receiver.start() + except PubSubPlusClientError as error: + logger.error("Could not bind to queue '%s': %s. If this client may not create endpoints, " + "provision the queue and its topic subscription out-of-band or grant the " + "capability. Exiting.", QUEUE_NAME, error) + return False # teardown_solace() in main's finally disconnects + return True + + +def receive_messages(state: SubscriberState) -> None: + trace(f"{APP_NAME} connected, and running. Press Ctrl-C to quit.") + next_report = time.monotonic() + STATS_INTERVAL_S + while not state.shutdown.is_set(): + try: + # blocks up to RECEIVE_TIMEOUT_MS for the next message. None means a timeout, or + # that the receiver or the service is shutting down; either way the loop goes + # round and the shutdown check above decides whether to stop + message = state.receiver.receive_message(RECEIVE_TIMEOUT_MS) + except PubSubPlusClientError as error: + # raised once the service is down or the receiver is terminated. A service + # interruption or a receiver termination has already set shutdown and exit_code + if not state.shutdown.is_set(): + logger.error("receive_message() failed, quitting: %s", error) + state.exit_code = 1 + break + if message is not None: + process_message(state, message) + # simple way of calculating message rates + now = time.monotonic() + if now >= next_report: + trace(f"{API} {APP_NAME} Received msgs/s: {state.msg_recv_counter:,}") + state.msg_recv_counter = 0 + if state.has_detected_redelivery: # try shutting -> enabling the queue on the broker to see this + trace("*** Redelivery detected ***") + state.has_detected_redelivery = False # only show the error once per report + next_report = now + STATS_INTERVAL_S + + +def process_message(state: SubscriberState, message: InboundMessage) -> None: + state.msg_recv_counter += 1 + try: + # the publisher sample sends a bytearray payload (the binary attachment), read back with + # get_payload_as_bytes(); a str payload arrives via get_payload_as_string() instead. + # errors="replace" keeps a non-UTF-8 payload from raising, but the payload read or the + # trace can still raise, hence the try/finally around the processing + payload = message.get_payload_as_bytes() + text = payload.decode("utf-8", errors="replace") if payload is not None else message.get_payload_as_string() + trace(f"{API} {APP_NAME} received: {text}") + if message.is_redelivered(): # 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 network disconnects + # perhaps an error in processing? Should do extra checks to avoid duplicate processing + state.has_detected_redelivery = True + except Exception: + # best practice: handle an unexpected message format without crashing, log it, and + # still ACK below (C API Best Practices). Without this, the exception would end the + # receive loop and the message would stay unacknowledged on the queue. + logger.exception("Processing failed for message id %s; acknowledging anyway", + message.get_application_message_id()) + finally: + # Messages are removed from the broker queue when the ACK is received. + # ACK only after processing is complete: DO NOT ACK until all processing/storing + # of this message is done. ACK even an unexpectedly formatted message (C API Best + # Practices): a message left unacknowledged stays on the queue and is redelivered. To + # NACK explicitly instead, build the receiver with + # with_required_message_outcome_support(Outcome.FAILED, Outcome.REJECTED) and call + # receiver.settle(message, outcome). + try: + state.receiver.ack(message) # ACKs are asynchronous + except PubSubPlusClientError as error: + # the ACK did not reach the API (for example the receiver is terminating): the + # message stays on the queue and is redelivered + logger.warning("ACK failed for message id %s; the broker redelivers it: %s", + message.get_application_message_id(), error) + + +def teardown_solace(state: SubscriberState) -> None: + # Application cleanup belongs here: once setup_solace() returns, teardown_solace() runs + # in main's finally on EVERY exit path (normal quit, SIGINT/SIGTERM, service + # interruption, receiver termination, or an exception), so Solace teardown and + # application-side cleanup are never skipped. + state.shutdown.set() + if state.receiver.is_running(): + # stop delivery at once (TERMINATE_GRACE_PERIOD_MS is 0): the receive loop has + # stopped, so nothing takes more messages from the API's buffer. terminate() raises + # IncompleteMessageDeliveryError when the buffer still holds messages. Those messages + # were never ACKed, so they stay on the queue and are redelivered: a normal stop, not + # a failure. Catch it so the disconnect below still runs + try: + state.receiver.terminate(TERMINATE_GRACE_PERIOD_MS) + except IncompleteMessageDeliveryError as error: + logger.warning("Receiver stopped with unprocessed messages in its buffer; the broker " + "redelivers them: %s", error) + except PubSubPlusClientError as error: + logger.error("Receiver terminate() failed: %s", error) + state.exit_code = 1 + if state.connect_attempted: + # disconnect() raises IllegalStateError on a service that never attempted to connect, + # hence the flag; on a service that is already down it returns quietly + state.messaging_service.disconnect() # will also release the receiver + + +def trace(message: str) -> None: + """Demo narration sink: every status line in this sample funnels through this one + function. An application replaces this single body to route narration to its + logger or reporting system. The logger calls for API events (service, receiver + state, termination) are a separate channel and stay as they are.""" + print(message, flush=True) + + +############################################################################ + + +class ServiceEventHandler(ReconnectionAttemptListener, ReconnectionListener, ServiceInterruptionListener): + """The three service event listeners in one class. Callbacks run on an API thread, even + in this blocking sample: return promptly and do not block in them, since events and + delivery stall while a callback runs (C API Best Practices).""" + + def __init__(self, state: SubscriberState) -> None: + self.state = state + + def on_reconnecting(self, event: ServiceEvent) -> None: + # connection lost, automatic reconnect attempt in progress; receive_message() keeps + # timing out and the loop keeps going + logger.warning("Service reconnecting; message delivery is paused until re-established: %s (%s)", + event.get_message(), event.get_cause()) + + def on_reconnected(self, event: ServiceEvent) -> None: + # automatic reconnect succeeded; the receiver rebinds to the queue and delivery resumes + logger.info("Service reconnected to %s; the receiver rebinds and delivery resumes", event.get_broker_uri()) + + def on_service_interrupted(self, event: ServiceEvent) -> None: + # the connection went down and cannot be restored (reconnect attempts exhausted) + logger.error("Service interrupted and cannot be restored, quitting: %s (%s)", + event.get_message(), event.get_cause()) + # Application cleanup signal: the service will not recover. Trigger application-side + # cleanup from here. This sample sets shutdown, so the receive loop exits and + # teardown_solace() runs in main's finally. + self.state.exit_code = 1 + self.state.shutdown.set() + + +class ReceiverStateHandler(ReceiverStateChangeListener): + """Reports whether this receiver is the one the broker delivers to. Runs on an API thread: + return promptly and do not block in it (C API Best Practices).""" + + def on_change(self, old_state: ReceiverState, new_state: ReceiverState, change_time_stamp: float) -> None: + if new_state == ReceiverState.ACTIVE: # bound and actively receiving from the queue + logger.info("Receiver active: consuming from queue %s", QUEUE_NAME) + else: # PASSIVE: bound but not receiving (another consumer is active on this exclusive queue) + logger.warning("Receiver passive: bound but not receiving (another consumer is active on " + "exclusive queue %s)", QUEUE_NAME) + + +class ReceiverTerminationHandler(TerminationNotificationListener): + """Handles a broker-initiated termination of the receiver. Runs on an API thread: + return promptly and do not block in it (C API Best Practices).""" + + def __init__(self, state: SubscriberState) -> None: + self.state = state + + def on_termination(self, event: TerminationEvent) -> None: + # the receiver was terminated by the broker: delivery from the queue has stopped + logger.error("Receiver terminated, delivery from queue %s has stopped: %s (%s)", + QUEUE_NAME, event.message, event.cause) + # Application cleanup signal: decide here whether to recreate the receiver or shut + # down. This sample shuts down: the receive loop exits and teardown_solace() runs in + # main's finally. + self.state.exit_code = 1 + self.state.shutdown.set() + + +if __name__ == "__main__": + logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s") + sys.exit(main(sys.argv[1:])) From baa49d566484c87eb169a3e7e4944e2d06897fcc Mon Sep 17 00:00:00 2001 From: adiel-sammak <71227923+aelsammak@users.noreply.github.com> Date: Thu, 22 Oct 2026 21:25:03 -0400 Subject: [PATCH 2/4] EBP-3125: Correct terminate() and ack() comments in the blocking samples - Reword the publisher's terminate() comments and log text: without a receipt listener the API waits only for a message still unsent in its publish buffer, and does not wait for or report a sent message whose ACK is late - Reword the guaranteed subscriber's ack() failure example: ack() still works while the receiver is terminating and raises only once the receiver is terminated by the broker or the service is down - Verified against the solace-pubsubplus publisher and receiver implementation during the PR review --- ...on-guaranteed-publisher-blocking-sample.py | 28 ++++++++++--------- ...n-guaranteed-subscriber-blocking-sample.py | 4 +-- 2 files changed, 17 insertions(+), 15 deletions(-) diff --git a/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-publisher-blocking-sample.py b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-publisher-blocking-sample.py index 84aeff7..f5f9e58 100644 --- a/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-publisher-blocking-sample.py +++ b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-publisher-blocking-sample.py @@ -101,14 +101,15 @@ # HA failover: with_connection_retry_strategy(RetryStrategy.parametrized_retry(1, 3000)), # with_reconnection_retry_strategy(RetryStrategy.parametrized_retry(20, 3000)), # properties[transport_layer_properties.CONNECTION_RETRIES_PER_HOST] = 5 -# terminate(grace_period): how long to wait for a send or its broker receipt that is still -# outstanding before the publisher stops; terminate() raises -# IncompleteMessageDeliveryError when one is still outstanding after that. With a blocking -# publish, at most one message is outstanding at teardown: the one whose wait timed out -# or failed. A bounded grace period keeps shutdown time predictable, for example within -# the stop timeout of a process supervisor, so the disconnect still runs. The API default -# is 600000 ms (10 minutes). Adjust the value to suit the application: longer gives that -# message more time to complete, shorter exits sooner. +# terminate(grace_period): how long to wait for a message that the API has not yet handed +# to the transport (still in its publish buffer) before the publisher stops; terminate() +# raises IncompleteMessageDeliveryError when that message is still unsent after that. It +# does not wait for, or report, a sent message whose ACK is late: a blocking publish tracks +# its own ACK, and after an ACK timeout the message is normally already with the transport. +# A bounded grace period keeps shutdown time predictable, for example within the stop +# timeout of a process supervisor, so the disconnect still runs. The API default is +# 600000 ms (10 minutes). Adjust the value to suit the application: longer gives an unsent +# message more time to go out, shorter exits sooner. TERMINATE_GRACE_PERIOD_MS = 10_000 # best practice for Guaranteed publishing (C API Best Practices): set a time-to-live so an # unconsumed message does not sit on a queue forever. In MILLISECONDS; 0, the API default, @@ -301,14 +302,15 @@ def teardown_solace(state: PublisherState) -> None: # cleanup are never skipped. state.shutdown.set() if state.publisher.is_running(): - # after a normal quit every message is already ACKed, so terminate() returns at once. - # After a timeout or a failure, it gives the one outstanding message up to the grace - # period. It raises IncompleteMessageDeliveryError when that message is still - # outstanding after the grace period; catch it so the disconnect below still runs + # after a normal quit, an ACK timeout, or a NACK, the publish buffer is empty, so + # terminate() returns at once. It waits up to the grace period only for a message the + # API has not yet handed to the transport, and raises IncompleteMessageDeliveryError + # when that message is still unsent after the grace period; catch it so the disconnect + # below still runs try: state.publisher.terminate(TERMINATE_GRACE_PERIOD_MS) except IncompleteMessageDeliveryError as error: - logger.error("Publisher stopped with a send or receipt still outstanding after %d ms: %s", + logger.error("Publisher stopped with a message still unsent after %d ms: %s", TERMINATE_GRACE_PERIOD_MS, error) state.exit_code = 1 except PubSubPlusClientError as error: diff --git a/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-subscriber-blocking-sample.py b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-subscriber-blocking-sample.py index 7403cad..b1543b0 100644 --- a/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-subscriber-blocking-sample.py +++ b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-subscriber-blocking-sample.py @@ -295,8 +295,8 @@ def process_message(state: SubscriberState, message: InboundMessage) -> None: try: state.receiver.ack(message) # ACKs are asynchronous except PubSubPlusClientError as error: - # the ACK did not reach the API (for example the receiver is terminating): the - # message stays on the queue and is redelivered + # the ACK did not reach the API (for example the broker terminated the receiver, + # or the service went down): the message stays on the queue and is redelivered logger.warning("ACK failed for message id %s; the broker redelivers it: %s", message.get_application_message_id(), error) From 5c464e43442071aa743587ec1d4f39a81c00a2de Mon Sep 17 00:00:00 2001 From: adiel-sammak <71227923+aelsammak@users.noreply.github.com> Date: Fri, 23 Oct 2026 14:41:01 -0400 Subject: [PATCH 3/4] EBP-3125: Discard failed publishes in the blocking publisher - Handle a timeout, a NACK, or a buffer overflow with the "Discard the Message" strategy from the Python developer guide (Strategies for Handling Publish Receipt Errors): log at INFO and continue with the next message - Keep ERROR, exit code 1, and stop only for any other PubSubPlusClientError, where sending cannot continue - Note that a message discarded after a timeout can still be delivered, and name the wait-and-retry and fixed-retry alternatives - Addresses PR review feedback on log levels and stopping at the first failure --- ...on-guaranteed-publisher-blocking-sample.py | 36 +++++++++---------- 1 file changed, 16 insertions(+), 20 deletions(-) diff --git a/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-publisher-blocking-sample.py b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-publisher-blocking-sample.py index f5f9e58..dc84fee 100644 --- a/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-publisher-blocking-sample.py +++ b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-publisher-blocking-sample.py @@ -20,8 +20,9 @@ publish_await_acknowledgement() per message with a POSITIVE acknowledgement timeout, reconnection and service interruption listeners registered BEFORE connect(), and a graceful SIGINT/SIGTERM shutdown. There is no publish receipt listener: the call -returns once the broker ACKs the message and raises on a NACK or a timeout, and any -publish failure stops the publisher with exit code 1. The non-blocking variant is +returns once the broker ACKs the message and raises on a NACK or a timeout. A timeout, +a NACK, or a buffer overflow discards that message and the publisher continues; any +other publish failure stops the publisher with exit code 1. The non-blocking variant is python-guaranteed-publisher-sample.py. Structured as module-level functions, setup_solace(...), connect_solace(...), run_publish_loop(...), and teardown_solace(...), that share one PublisherState; setup_solace() creates the service and the publisher @@ -87,7 +88,7 @@ # None: the API then waits for the ACK without a limit, so a lost ACK hangs the publisher # and a shutdown signal never takes effect. An ACK normally comes back in milliseconds; # 5 s also covers a slow broker or a short reconnect. A longer reconnect than that times -# out the message in flight, and the publisher stops. +# out the message in flight, and the publisher discards it. PUBLISH_ACK_TIMEOUT_MS = 5000 # how often the publish loop reports its message rate, in seconds STATS_INTERVAL_S = 1.0 @@ -252,38 +253,33 @@ def run_publish_loop(state: PublisherState) -> None: # takes effect once the call returns, within PUBLISH_ACK_TIMEOUT_MS. state.publisher.publish_await_acknowledgement(message, topic, PUBLISH_ACK_TIMEOUT_MS) state.msg_sent_counter += 1 - msg_count += 1 + # A timeout, a NACK, or an overflow is handled with the "Discard the Message" strategy + # from the Python developer guide ("Strategies for Handling Publish Receipt Errors"): + # log it and go on to the next message, so these are INFO, not errors. Do not use this + # strategy if the application cannot tolerate message loss; the guide's other + # strategies are to wait and retry, and to retry a predefined number of times. except PubSubTimeoutError: # no ACK or NACK inside the timeout. The broker can still have the message (the - # ACK can be late, for example after a long reconnect), so sending it again can - # deliver a duplicate - logger.error("No ACK for Message ID %s within %d ms, quitting", msg_id, PUBLISH_ACK_TIMEOUT_MS) - state.exit_code = 1 - break + # ACK can be late, for example after a long reconnect), so a discarded message can + # still be delivered + logger.info("No ACK for Message ID %s within %d ms; discarding it", msg_id, PUBLISH_ACK_TIMEOUT_MS) except (MessageRejectedByBrokerError, MessageNotAcknowledgedByBrokerError, MessageDestinationDoesNotExistError) as error: # NACK: the broker did not persist the message, and the exception says why (for # example MessageDestinationDoesNotExistError when no queue subscribes to the - # topic and the client-profile rejects on no subscription match). Some error - # handling possibilities instead of quitting: - # - send it again, or send it somewhere else (error handling queue?) - # - log and continue - # - pause and retry (backoff) - logger.error("NACK for Message ID %s, quitting: %s", msg_id, error) - state.exit_code = 1 - break + # topic and the client-profile rejects on no subscription match) + logger.info("NACK for Message ID %s; discarding it: %s", msg_id, error) except PublisherOverflowError as error: # the API reference lists it for this call; with one message in flight and the # default back-pressure setting, the publish buffer does not overflow - logger.error("Publish buffer overflow for Message ID %s, quitting: %s", msg_id, error) - state.exit_code = 1 - break + logger.info("Publish buffer overflow for Message ID %s; discarding it: %s", msg_id, error) except PubSubPlusClientError as error: # the message cannot be sent (for example the service is down or the publisher # terminated) and retrying would not help logger.error("publish_await_acknowledgement() failed, quitting: %s", error) state.exit_code = 1 break # let's quit; or, could initiate a new connection attempt + msg_count += 1 # simple way of calculating message rates now = time.monotonic() if now >= next_report: From 1d8ea799492980c34fa3f596d8fcb104fc78eb47 Mon Sep 17 00:00:00 2001 From: adiel-sammak <71227923+aelsammak@users.noreply.github.com> Date: Fri, 23 Oct 2026 15:01:36 -0400 Subject: [PATCH 4/4] EBP-3125: Base the blocking publisher ACK timeout comment on the C API ACK timer - Explain PUBLISH_ACK_TIMEOUT_MS from the C API publish ACK timer: an unACKed message is resent every 2000 ms, so 5000 ms allows three attempts - State the convention: keep the value a little above a multiple of 2000 ms - Addresses PR review feedback on how the ACK timeout is chosen --- .../python/python-guaranteed-publisher-blocking-sample.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-publisher-blocking-sample.py b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-publisher-blocking-sample.py index dc84fee..577d254 100644 --- a/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-publisher-blocking-sample.py +++ b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-publisher-blocking-sample.py @@ -86,9 +86,9 @@ # POSITIVE acknowledgement timeout in MILLISECONDS: publish_await_acknowledgement() blocks # up to this long for the broker's ACK or NACK, then raises PubSubTimeoutError. Never pass # None: the API then waits for the ACK without a limit, so a lost ACK hangs the publisher -# and a shutdown signal never takes effect. An ACK normally comes back in milliseconds; -# 5 s also covers a slow broker or a short reconnect. A longer reconnect than that times -# out the message in flight, and the publisher discards it. +# and a shutdown signal never takes effect. The C API resends an unACKed message every +# 2000 ms (its publish ACK timer), so 5000 ms allows three attempts. Keep the value a little +# above a multiple of 2000 ms. PUBLISH_ACK_TIMEOUT_MS = 5000 # how often the publish loop reports its message rate, in seconds STATS_INTERVAL_S = 1.0