diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5b89401..ca19207 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -7,7 +7,7 @@ name: ci # + post-merge version backstop on pushes to main # link-check: CI-02: the tools/check-links.sh sweep # compile-check: EVAL-02 compile the baked JCSMP samples verbatim -# trigger-evals: lightweight skill-triggering check; self-skips green without a key +# trigger-evals: lightweight skill-triggering check; disabled until CI may hold an API key on: # Deliberately no `edited` type: running on title/body edits either re-runs # the paid eval matrix for a no-code change or, if jobs are gated, lets the @@ -133,7 +133,11 @@ jobs: # run summary. The secret is legal in the job env but illegal in a job-level # if, so the guard reads HAS_KEY at step level and the real key is scoped to # the run step only. + # Disabled: no policy permits an API key on this public repo, so the job can + # only self-skip. Run the trigger evals locally instead; remove this `if` to + # re-enable the job once a key is allowed. trigger-evals: + if: false runs-on: ubuntu-latest timeout-minutes: 120 # per model leg; 57 cases x 3 runs (~50 min observed) strategy: diff --git a/.gitignore b/.gitignore index 7a26224..34d846d 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,5 @@ # GSD planning directory (local-only) .planning/ + +# Python bytecode cache +__pycache__/ diff --git a/plugins/solace-messaging-skills/evals/README.md b/plugins/solace-messaging-skills/evals/README.md index e8b4661..f02ec65 100644 --- a/plugins/solace-messaging-skills/evals/README.md +++ b/plugins/solace-messaging-skills/evals/README.md @@ -8,15 +8,15 @@ To run the evals manually, you need the `claude` CLI and `jq` on your PATH, plus ```shell export ANTHROPIC_API_KEY= # or CLAUDE_CODE_OAUTH_TOKEN -./tools/run-trigger-evals.sh # defaults to claude-haiku-4-5 -./tools/run-trigger-evals.sh --model claude-sonnet-5 +./tools/run-trigger-evals.sh # defaults to a Haiku model +./tools/run-trigger-evals.sh --model sonnet # the latest Sonnet ``` Each prompt runs three times and the verdict is the majority result. Set `TRIGGER_EVAL_RUNS=1` for a quicker, cheaper local check, and `TRIGGER_EVAL_MODEL` to change the default model. A run passes when at least 90% of cases match their expectation, and any infrastructure failure (such as a missing credential, a malformed corpus, or zero discovered cases) fails the run regardless of the rate. A case may also set `"must_pass": true`, and any must-pass failure fails the run regardless of the pooled rate. Note that a negative case asserts only that the named skill stays silent, so another skill in the plugin may legitimately fire on the same prompt. ## Continuous integration -In GitHub Actions, the `trigger-evals` job in `.github/workflows/ci.yml` runs this same script on every pull request as a two model matrix (`claude-haiku-4-5` and `claude-sonnet-5`), each model an independent check, and self-skips green when the `ANTHROPIC_API_KEY` secret is absent. +The `trigger-evals` job in `.github/workflows/ci.yml` is disabled (`if: false`), because no policy permits an API key on this public repository. Run the trigger evals locally as shown above. To re-enable the job once a key is allowed, remove the `if`. It then runs this same script on every pull request as a two model matrix (Haiku and Sonnet), each model an independent check, and self-skips green when the `ANTHROPIC_API_KEY` secret is absent. # Output evals @@ -60,12 +60,12 @@ Run from the repository root: ```shell export ANTHROPIC_API_KEY= # or CLAUDE_CODE_OAUTH_TOKEN -./tools/run-output-evals.sh # defaults to claude-sonnet-5 -./tools/run-output-evals.sh --model claude-opus-5 +./tools/run-output-evals.sh # defaults to a Sonnet model +./tools/run-output-evals.sh --model opus # the latest Opus ./tools/run-output-evals.sh --case appdev-quickstart-implement-full ``` -Run the suite for both `claude-sonnet-5` and `claude-opus-5` before a PR that affects skill content, and record both results in the PR description, including whether live verify ran or skipped (the runner prints this in its summary). A full leg is a long run; keep the machine awake for it (on macOS, prefix the command with `caffeinate -i`), because a sleep mid-run surfaces as INFRA failures. A permission denial of a tool the runner grants (Bash, Write, and the rest of its allowlist) is also INFRA: managed settings, hooks, or local command shims can deny a command, the model then improvises, and the result no longer measures the skill. The case line names the denied tool and command; adjust the policy for the run instead of reading the result as a skill failure. Each case runs once by default (`OUTPUT_EVAL_RUNS=1`), because a full implement-flow case is expensive; raise it for a majority-vote stability study. `OUTPUT_EVAL_JUDGE_MODEL` (default `claude-sonnet-5`) stays fixed across subject models so leg differences are attributable to the subject. `OUTPUT_EVAL_WORKDIR` receives transcripts and generated projects, and the work directory is kept and printed when the run fails. +Run the suite for both the latest Sonnet and the latest Opus (`--model sonnet` and `--model opus`) before a PR that affects skill content, and record both results in the PR description, including whether live verify ran or skipped (the runner prints this in its summary). A full leg is a long run; keep the machine awake for it (on macOS, prefix the command with `caffeinate -i`), because a sleep mid-run surfaces as INFRA failures. A permission denial of a tool the runner grants (Bash, Write, and the rest of its allowlist) is also INFRA: managed settings, hooks, or local command shims can deny a command, the model then improvises, and the result no longer measures the skill. The case line names the denied tool and command; adjust the policy for the run instead of reading the result as a skill failure. Each case runs once by default (`OUTPUT_EVAL_RUNS=1`), because a full implement-flow case is expensive; raise it for a majority-vote stability study. `OUTPUT_EVAL_JUDGE_MODEL` (default a pinned Sonnet model) stays fixed across subject models so leg differences are attributable to the subject. `OUTPUT_EVAL_WORKDIR` receives transcripts and generated projects, and the work directory is kept and printed when the run fails. The gate matches the trigger evals: a run passes when at least 90% of cases pass, any infrastructure failure fails the run, and any `must_pass` failure fails the run regardless of the pooled rate. The forbidden-behavior negatives and the compile case are `must_pass`. diff --git a/plugins/solace-messaging-skills/skills/solace-application-development/references/jcsmp/implement-mode.md b/plugins/solace-messaging-skills/skills/solace-application-development/references/jcsmp/implement-mode.md index 76b78ee..a13aa14 100644 --- a/plugins/solace-messaging-skills/skills/solace-application-development/references/jcsmp/implement-mode.md +++ b/plugins/solace-messaging-skills/skills/solace-application-development/references/jcsmp/implement-mode.md @@ -126,7 +126,7 @@ Hold every invariant while adapting: keep the shared connection helper, the sess **Preserve the marker contract.** A variant adapts the SHAPE, never the marker strings. It MUST still emit its leaf's `VERIFY:` markers, character for character, so the fixed `verify.sh` stages drive it unchanged (this is exactly why a durable-reply-queue requestor still prints `VERIFY: REPLY_RECEIVED`). Never rename or drop a marker to fit a variant. -**Demo harness is not application logic.** The samples carry demo-harness elements that exist only to make a standalone run observable: the ENTER-to-quit `System.in` loop, the once-per-second stats-printing thread, the pacing `Thread.sleep` between sends, and the rotating example payload. When the developer's request describes a real application domain, do NOT carry these into the generated classes: replace the example payload and send cadence with the application's real messages and triggers, and keep only the lifecycle pieces the leaf's steps call for (the SIGINT hook on the long-running role, the self-exit on the foreground role). When the request IS a demo or a try-it run, keeping the harness is fine. Two things are NOT demo harness: the `trace(...)` narration method (keep it; applications replace its single body to route narration to their logging or reporting system) and the `VERIFY:` markers, which are emitted through `trace(...)` like every other status line. Because `verify.sh` greps captured stdout for the markers and for the shutdown hook's `Shutdown signal received` proof line (both flow through `trace(...)`), the generated `trace(...)` body MUST stay `System.out.println` through the Step 5 verification stages; rerouting that body to a logger or reporting system is the developer's step AFTER the verify stages pass, never part of generation. Never treat the logging backend (Step 3) as demo harness: the backend and its config file ship in every generated app. +**Demo harness is not application logic.** The samples carry demo-harness elements that exist only to make a standalone run observable: the ENTER-to-quit `System.in` loop, the once-per-second stats-printing thread, the continuous publish loop, and the rotating example payload. When the developer's request describes a real application domain, do NOT carry these into the generated classes: replace the example payload and the publish loop with the application's real messages and triggers, and never add a delay or rate limit between sends that the application does not ask for, and keep only the lifecycle pieces the leaf's steps call for (the SIGINT hook on the long-running role, the self-exit on the foreground role). When the request IS a demo or a try-it run, keeping the harness is fine. Two things are NOT demo harness: the `trace(...)` narration method (keep it; applications replace its single body to route narration to their logging or reporting system) and the `VERIFY:` markers, which are emitted through `trace(...)` like every other status line. Because `verify.sh` greps captured stdout for the markers and for the shutdown hook's `Shutdown signal received` proof line (both flow through `trace(...)`), the generated `trace(...)` body MUST stay `System.out.println` through the Step 5 verification stages; rerouting that body to a logger or reporting system is the developer's step AFTER the verify stages pass, never part of generation. Never treat the logging backend (Step 3) as demo harness: the backend and its config file ship in every generated app. **Comments follow their code.** The samples' comments are part of the reference, not decoration: they carry the documented best practices (the session-independent `createMessage`, the binary payload and the legacy `XML` naming, `getData()` versus `getBytes()`, registering a session event handler at session creation, the ACK/NACK handling options, the correlation-key rule) into the code a developer reads long after this session. Generation carries them under ONE contract, on every door and every app shape: diff --git a/plugins/solace-messaging-skills/skills/solace-application-development/references/jcsmp/jcsmp-direct-publisher-sample.java b/plugins/solace-messaging-skills/skills/solace-application-development/references/jcsmp/jcsmp-direct-publisher-sample.java index 93c618f..69beb5f 100644 --- a/plugins/solace-messaging-skills/skills/solace-application-development/references/jcsmp/jcsmp-direct-publisher-sample.java +++ b/plugins/solace-messaging-skills/skills/solace-application-development/references/jcsmp/jcsmp-direct-publisher-sample.java @@ -57,7 +57,6 @@ public class DirectPublisher { private static final String APP_NAME = DirectPublisher.class.getSimpleName(); 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 = 512; // remember to add log4j2.xml to your classpath @@ -189,15 +188,10 @@ private static void runPublishLoop() throws IOException { if (e instanceof JCSMPTransportException) { // all reconnect attempts failed isShutdown = true; // let's quit; or, could initiate a new connection attempt } - } finally { // add a delay between messages - try { - 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; - } } + // no delay between messages: the loop publishes as fast as send() returns, the way an + // application publishes each message as soon as it has one + // Note: STANDARD Edition Solace broker is limited to 10k msg/s max ingress } isShutdown = true; } diff --git a/plugins/solace-messaging-skills/skills/solace-application-development/references/jcsmp/jcsmp-direct-requestor-sample.java b/plugins/solace-messaging-skills/skills/solace-application-development/references/jcsmp/jcsmp-direct-requestor-sample.java index 3bfe9a2..a564530 100644 --- a/plugins/solace-messaging-skills/skills/solace-application-development/references/jcsmp/jcsmp-direct-requestor-sample.java +++ b/plugins/solace-messaging-skills/skills/solace-application-development/references/jcsmp/jcsmp-direct-requestor-sample.java @@ -60,8 +60,7 @@ public class DirectRequestor { private static final String API = "JCSMP"; // POSITIVE blocking-request timeout in milliseconds: requestor.request(...) blocks up // to this long for the correlated reply before throwing JCSMPRequestTimeoutException. - // A positive timeout is the direct reliability knob (NOT a burst); 3000 ms matches the - // upstream DirectRequestorBlocking sample. + // A positive timeout is the direct reliability knob (NOT a burst). private static final int REQUEST_TIMEOUT_MS = 3000; private static JCSMPSession session; diff --git a/plugins/solace-messaging-skills/skills/solace-application-development/references/jcsmp/jcsmp-guaranteed-publisher-sample.java b/plugins/solace-messaging-skills/skills/solace-application-development/references/jcsmp/jcsmp-guaranteed-publisher-sample.java index a4d0ef1..e0e8a5f 100644 --- a/plugins/solace-messaging-skills/skills/solace-application-development/references/jcsmp/jcsmp-guaranteed-publisher-sample.java +++ b/plugins/solace-messaging-skills/skills/solace-application-development/references/jcsmp/jcsmp-guaranteed-publisher-sample.java @@ -58,7 +58,6 @@ public class GuaranteedPublisher { 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 = 255; // max guaranteed-publish window (range 1-255) - 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 @@ -220,15 +219,10 @@ private static void runPublishLoop() throws IOException, InterruptedException { if (e instanceof JCSMPTransportException) { // all reconnect attempts failed isShutdown = true; // let's quit; or, could initiate a new connection attempt } - } finally { // add a delay between messages - try { - 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; - } } + // no delay between messages: the loop publishes as fast as send() returns, the way an + // application publishes each message as soon as it has one + // Note: STANDARD Edition Solace broker is limited to 10k msg/s max ingress } isShutdown = true; } diff --git a/plugins/solace-messaging-skills/skills/solace-application-development/references/jcsmp/jcsmp-guaranteed-replier-sample.java b/plugins/solace-messaging-skills/skills/solace-application-development/references/jcsmp/jcsmp-guaranteed-replier-sample.java index 471a19c..bfbacd2 100644 --- a/plugins/solace-messaging-skills/skills/solace-application-development/references/jcsmp/jcsmp-guaranteed-replier-sample.java +++ b/plugins/solace-messaging-skills/skills/solace-application-development/references/jcsmp/jcsmp-guaranteed-replier-sample.java @@ -157,8 +157,7 @@ public void handleEvent(SessionEventArgs event) { final Queue requestQueue = JCSMPFactory.onlyInstance().createQueue(REQUEST_QUEUE_NAME); // ELEVATION: provision the durable request queue in-process and map the request - // topic onto it at startup, so a fresh broker works out of the box (the upstream - // sample provisions with flag 0, which throws "already exists" on a re-run). + // topic onto it at startup, so a fresh broker works out of the box. // 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; " @@ -168,8 +167,8 @@ public void handleEvent(SessionEventArgs event) { EndpointProperties endpointProps = new EndpointProperties(); endpointProps.setAccessType(EndpointProperties.ACCESSTYPE_EXCLUSIVE); // single-replier sample endpointProps.setPermission(EndpointProperties.PERMISSION_CONSUME); - // provision the durable request queue IDEMPOTENTLY so a re-run is safe (NOT the - // upstream flag 0, which throws on the second run) + // provision the durable request queue IDEMPOTENTLY so a re-run is safe (flag 0 would + // throw "already exists" on the second run) session.provision(requestQueue, endpointProps, JCSMPSession.FLAG_IGNORE_ALREADY_EXISTS); // map the request topic onto the queue with a subscription (PERSISTENT pub/sub onto // a queue); WAIT_FOR_CONFIRM blocks until the broker confirms the subscription diff --git a/plugins/solace-messaging-skills/skills/solace-application-development/references/jcsmp/jcsmp-guaranteed-subscriber-sample.java b/plugins/solace-messaging-skills/skills/solace-application-development/references/jcsmp/jcsmp-guaranteed-subscriber-sample.java index 6f13cbd..b2471c2 100644 --- a/plugins/solace-messaging-skills/skills/solace-application-development/references/jcsmp/jcsmp-guaranteed-subscriber-sample.java +++ b/plugins/solace-messaging-skills/skills/solace-application-development/references/jcsmp/jcsmp-guaranteed-subscriber-sample.java @@ -153,8 +153,7 @@ public void handleEvent(SessionEventArgs event) { final Queue queue = JCSMPFactory.onlyInstance().createQueue(QUEUE_NAME); // ELEVATION: provision the durable queue in-process and map the topic onto it - // at startup, so a fresh broker works out of the box (the upstream sample expects - // the queue to pre-exist and errors out otherwise). + // at startup, so a fresh broker works out of the box. // 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; " diff --git a/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-direct-publisher-async-sample.py b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-direct-publisher-async-sample.py new file mode 100644 index 0000000..1ea90bd --- /dev/null +++ b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-direct-publisher-async-sample.py @@ -0,0 +1,329 @@ +""" +Derived from patterns/direct_publisher.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 Direct Messages: + https://docs.solace.com/API/API-Developer-Guide-Python/Python-DM-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 direct (at-most-once) pub/sub journey: +basic-auth connect with a reconnection retry strategy, a continuous DIRECT binary +publish loop with a rotating payload to a topic, a publish failure listener for the +asynchronous failures the API reports, capacity-bounded back pressure that blocks +publish() while the buffer is full, reconnection and service interruption listeners +registered BEFORE connect(), and a graceful SIGINT/SIGTERM shutdown. Direct +messaging is at-most-once: there is no broker receipt, so there is NO publish +receipt listener and NO user context here (the inverse of the guaranteed +publisher, which acts on each ACK/NACK), and there are no outstanding receipts to +drain at shutdown. Messages published while the API reconnects are lost, and each one +is reported through the publish failure listener. Publishes the payload directly, with +no OutboundMessageBuilder. 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. + +Only practices documented in canonical Solace sources are encoded here. + +Copied into a generated project as direct_publisher_async.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 +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.publisher.direct_message_publisher import ( + DirectMessagePublisher, + FailedPublishEvent, + PublishFailureListener, +) +from solace.messaging.resources.topic import Topic + +from solace_connection_config import load_service_properties + +APP_NAME = "DirectPublisher" +TOPIC_PREFIX = "solace/samples/" # used as the topic "root" +API = "Python" +PAYLOAD_SIZE = 512 +# 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: publish() blocks once this many messages wait unsent, 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. +PUBLISH_BUFFER_CAPACITY = 1000 +# terminate(grace_period): how long to wait for buffered sends to leave before the +# publisher stops; terminate() raises IncompleteMessageDeliveryError when messages are +# still buffered after that. 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 buffered sends more time to leave, shorter exits sooner. +TERMINATE_GRACE_PERIOD_MS = 10_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: DirectMessagePublisher + 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 sent + + +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() + ) + # DIRECT publisher with capacity-bounded back pressure: publish() blocks once + # PUBLISH_BUFFER_CAPACITY messages wait unsent, and returns when there is room again. + # The API default (on_back_pressure_elastic) buffers without bound and the developer + # guide cautions it can exhaust memory; on_back_pressure_reject(n) raises + # PublisherOverflowError instead of blocking. Building the publisher needs only the + # built service; start() is what requires the connection. One service can serve + # several publishers. + publisher = ( + messaging_service.create_direct_message_publisher_builder() + .on_back_pressure_wait(PUBLISH_BUFFER_CAPACITY) + .build() + ) + # direct messaging has no broker receipt, so failures the API detects after publish() + # returned (a buffered message that could not be sent) arrive asynchronously through + # this listener; without it they are silent + publisher.set_publish_failure_listener(PublishFailureHandler()) + 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 + + +def run_publish_loop(state: PublisherState) -> None: + threading.Thread(target=print_stats, args=(state,), name="stats", daemon=True).start() + + trace(f"{API} {APP_NAME} connected, and running. Press Ctrl-C to quit.") + trace(f"Publishing to topic '{TOPIC_PREFIX}{API.lower()}/direct/pub/...', " + "at-most-once (no broker receipt, no redelivery).") + while not state.shutdown.is_set(): + # each loop, change the payload as an example + chosen_character = chr(state.msg_sent_counter % 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) + # NO user context: direct is at-most-once with no broker receipt, so there is + # nothing to correlate (the inverse of the guaranteed publisher, which passes the + # message id as user_context for local ACK/NACK correlation) + topic = Topic.of(f"{TOPIC_PREFIX}{API.lower()}/direct/pub/{chosen_character}") + try: + # blocks while the buffer is full (back pressure) on a connected transport; during + # a reconnect it does not block (see ServiceEventHandler.on_reconnecting). A + # shutdown signal that arrives during that wait takes effect once publish() + # returns: when the buffer drains, or when the service goes down and the API + # releases the wait. + # builderless publish: the payload goes to publish() directly, and the API builds the + # message. Set a per-message header with additional_message_properties, for example + # additional_message_properties={message_properties.APPLICATION_MESSAGE_ID: msg_id} + # with `from solace.messaging.config.solace_properties import message_properties`. + # Note: an OutboundMessageBuilder (messaging_service.message_builder()) is very + # useful for "templated" messages: set the headers that are the same on every + # message once on the builder, and build each message with only its few + # differences (see the guaranteed publisher sample). + state.publisher.publish(payload, topic) + state.msg_sent_counter += 1 + except PubSubPlusClientError as error: + # publish() raises when the message cannot be sent and retrying would not help + logger.warning("publish() failed, quitting: %s", error) + state.exit_code = 1 + break # let's quit; or, could initiate a new connection attempt + # no delay between messages: the loop publishes as fast as publish() returns, the way + # an application publishes each message as soon as it has one + # Note: STANDARD Edition Solace broker is limited to 10k msg/s max ingress + state.shutdown.set() + + +def print_stats(state: PublisherState) -> None: + # simple way of calculating message rates; wait() returns True (ending the loop) on shutdown + while not state.shutdown.wait(1.0): + trace(f"{API} {APP_NAME} Published msgs/s: {state.msg_sent_counter:,}") + state.msg_sent_counter = 0 + + +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, service + # interruption, or an exception), so Solace teardown and application-side cleanup are + # never skipped. + state.shutdown.set() + if state.publisher.is_running(): + # direct is at-most-once: there are no broker receipts to drain, so terminate() only + # gives the messages still in the publisher's buffer up to the grace period to leave. + # It raises IncompleteMessageDeliveryError when messages are still buffered after + # that; catch it so the disconnect below still runs. Those messages are lost, which + # direct (at-most-once) messaging allows, so this is a warning and the exit code stays + # 0. An application that must not lose messages at shutdown can raise the level (or + # use guaranteed messaging). + try: + state.publisher.terminate(TERMINATE_GRACE_PERIOD_MS) + except IncompleteMessageDeliveryError as error: + logger.warning("Publisher stopped with messages still buffered after %d ms: %s", + TERMINATE_GRACE_PERIOD_MS, error) + 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: + 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. Direct messaging is + # at-most-once: while the transport is down, publish() does not block, and the API + # fails each message it cannot send and reports it through the publish failure + # listener (PublishFailureHandler below). An application that must not lose messages + # uses guaranteed messaging instead (see the guaranteed publisher sample); one that + # must not produce messages during an outage can pause its publish loop between this + # event and on_reconnected(). + logger.warning("Service reconnecting: %s (%s)", event.get_message(), event.get_cause()) + + def on_reconnected(self, event: ServiceEvent) -> None: + # automatic reconnect succeeded: messages published from now on are sent again + 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() + + +class PublishFailureHandler(PublishFailureListener): + """Very simple class, used for the asynchronous publish failures the API reports for + direct messages (there is no broker receipt to act on). Runs on an API thread: + return promptly and do not block in it (C API Best Practices).""" + + def on_failed_publish(self, failed_publish_event: FailedPublishEvent) -> None: + # a buffered message could not be sent (the developer guide names an invalid topic or a + # termination of the service as causes); direct has no NACK, so this never fires for one + logger.warning("Failed to publish to %s: %s", + failed_publish_event.get_destination().get_name(), failed_publish_event.get_exception()) + + +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-direct-subscriber-async-sample.py b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-direct-subscriber-async-sample.py new file mode 100644 index 0000000..4f1319c --- /dev/null +++ b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-direct-subscriber-async-sample.py @@ -0,0 +1,309 @@ +""" +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: +basic-auth connect with a reconnection retry strategy, a plain topic subscription +(no queue, no provisioning), an async DIRECT receiver with a capacity-bounded +buffer that detects discards, 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). +Structured as module-level functions, setup_solace(...), connect_solace(...), +await_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. + +Only practices documented in canonical Solace sources are encoded here. + +Copied into a generated project as direct_subscriber_async.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 +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.receiver.message_receiver import MessageHandler +from solace.messaging.resources.topic_subscription import TopicSubscription + +from solace_connection_config import load_service_properties + +APP_NAME = "DirectSubscriber" +# topic to subscribe to directly; matches the direct publisher sample's topic root +TOPIC_NAME = "solace/samples/python/direct/pub/>" +API = "Python" +# 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 handler, 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 handler to drain the messages the API +# has already received before the receiver stops; terminate() raises +# IncompleteMessageDeliveryError when messages remain after that. 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 lets the handler finish more of the +# received messages, shorter exits sooner. +TERMINATE_GRACE_PERIOD_MS = 10_000 + +# 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, the flags the listeners flip, 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 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 main + # loop exits, 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) + await_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 + # handler, 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 + # set the message handler BEFORE start(), so the handler is in place when delivery + # begins and received messages do not wait in the API's buffer for it. receive_async() + # needs a connected service, so it runs here after connect(), not in setup_solace(). + # see bottom of file for DirectMessageHandler, which receives the messages from the topic + state.receiver.receive_async(DirectMessageHandler(state)) + trace(f"Adding direct topic subscription '{TOPIC_NAME}'.") + state.receiver.start() + trace(f"{APP_NAME} subscribed and consuming. Press Ctrl-C to quit.") + + +def await_messages(state: SubscriberState) -> None: + # async direct receive working now, so time to wait until done... + while not state.shutdown.wait(1.0): # wait 1 second; the wait returns True once shutdown is requested + trace(f"{API} {APP_NAME} Received msgs/s: {state.msg_recv_counter:,}") # simple way of calculating message rates + 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 second + + +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(): + # direct is at-most-once: there are no acknowledgements to drain before exit, so + # terminate() stops delivery and gives the handler up to the grace period to drain + # the messages the API has already received. It raises IncompleteMessageDeliveryError + # when messages remain after that; catch it so the disconnect below still runs + try: + state.receiver.terminate(TERMINATE_GRACE_PERIOD_MS) + except IncompleteMessageDeliveryError as error: + logger.error("Receiver stopped with messages still undelivered to the handler after %d ms: %s", + TERMINATE_GRACE_PERIOD_MS, error) + state.exit_code = 1 + 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: + 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) + 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 main loop exits and + # teardown_solace() runs in main's finally. + self.state.exit_code = 1 + self.state.shutdown.set() + + +class DirectMessageHandler(MessageHandler): + """Very simple class, used for receiving direct messages from the topic. on_message runs + on an API-owned receiver thread, not the main thread: return promptly and do not block in + it (hand heavy work to the application's own thread or queue), since delivery stalls while + it runs (C API Best Practices).""" + + def __init__(self, state: SubscriberState) -> None: + self.state = state + + def on_message(self, message: InboundMessage) -> None: + self.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 callback + 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(): + self.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. + + +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-async-sample.py b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-publisher-async-sample.py new file mode 100644 index 0000000..a94e4b5 --- /dev/null +++ b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-publisher-async-sample.py @@ -0,0 +1,353 @@ +""" +Derived from patterns/guaranteed_publisher.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: 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, publish receipt (ACK/NACK) +handling correlated by user context, capacity-bounded back pressure that blocks +publish() while the buffer is full, reconnection and service interruption listeners +registered BEFORE connect(), and a graceful SIGINT/SIGTERM shutdown that drains +outstanding receipts before disconnecting. 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. + +Only practices documented in canonical Solace sources are encoded here. + +Copied into a generated project as guaranteed_publisher_async.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 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, PubSubPlusClientError +from solace.messaging.messaging_service import ( + MessagingService, + ReconnectionAttemptListener, + ReconnectionListener, + ServiceEvent, + ServiceInterruptionListener, +) +from solace.messaging.publisher.persistent_message_publisher import ( + MessagePublishReceiptListener, + PersistentMessagePublisher, + PublishReceipt, +) +from solace.messaging.resources.topic import Topic + +from solace_connection_config import load_service_properties + +APP_NAME = "GuaranteedPublisher" +TOPIC_PREFIX = "solace/samples/" # used as the topic "root" +API = "Python" +PAYLOAD_SIZE = 512 +# 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: publish() blocks once this many messages wait unsent, 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. +PUBLISH_BUFFER_CAPACITY = 1000 +# terminate(grace_period): how long to wait for buffered sends and their broker receipts +# to complete before the publisher stops; terminate() raises IncompleteMessageDeliveryError +# when sends or receipts are still outstanding after that. 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 sends and receipts 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 sent + + +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 drains the outstanding + # receipts 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 capacity-bounded back pressure: publish() blocks once + # PUBLISH_BUFFER_CAPACITY messages wait unsent, and returns when there is room again. + # The API default (on_back_pressure_elastic) buffers without bound and the developer + # guide cautions it can exhaust memory; on_back_pressure_reject(n) raises + # PublisherOverflowError instead of blocking. 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() + .on_back_pressure_wait(PUBLISH_BUFFER_CAPACITY) + .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 + # best practice: check every publish receipt; a receipt that carries an exception is a + # NACK or a failure to persist (see PublishReceiptHandler at the bottom of the file). + # The listener is set AFTER start(): the API raises IllegalStateError on a publisher + # that is not started yet. + state.publisher.set_message_publish_receipt_listener(PublishReceiptHandler()) + + +def run_publish_loop(state: PublisherState) -> None: + threading.Thread(target=print_stats, args=(state,), name="stats", daemon=True).start() + # 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.") + while not state.shutdown.is_set(): + # each loop, change the payload as an example + chosen_character = chr(state.msg_sent_counter % 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: + # user_context for local ACK/NACK correlation: the API hands it back in the + # PublishReceipt, so an immutable per-send identifier (the message id) lets the + # receipt handler name the exact message the broker accepted or rejected. + # publish() blocks while the buffer is full (back pressure). During an automatic + # reconnect the messages wait in the buffer for the transport and go out after + # it, and publish() blocks only once the buffer fills; the loop needs no pause of + # its own (the direct publisher, whose messages fail instead of waiting, does need + # one). A shutdown signal that + # arrives during that wait takes effect once publish() returns: when the buffer + # drains, or when the service goes down and the API releases the wait. + state.publisher.publish(message, topic, user_context=msg_id) + state.msg_sent_counter += 1 + except PubSubPlusClientError as error: + # publish() raises when the message cannot be sent and retrying would not help + logger.warning("publish() failed, quitting: %s", error) + state.exit_code = 1 + break # let's quit; or, could initiate a new connection attempt + # no delay between messages: the loop publishes as fast as publish() returns, the way + # an application publishes each message as soon as it has one + # Note: STANDARD Edition Solace broker is limited to 10k msg/s max ingress + state.shutdown.set() + + +def print_stats(state: PublisherState) -> None: + # simple way of calculating message rates; wait() returns True (ending the loop) on shutdown + while not state.shutdown.wait(1.0): + trace(f"{API} {APP_NAME} Published msgs/s: {state.msg_sent_counter:,}") + state.msg_sent_counter = 0 + + +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, service + # interruption, or an exception), so Solace teardown and application-side cleanup are + # never skipped. + state.shutdown.set() + if state.publisher.is_running(): + # gracefully finish outstanding receipts before exit: terminate(grace_period) first + # sends what is still buffered, then waits up to the grace period for the broker's + # receipt of every published message, so no in-flight PERSISTENT message is + # abandoned unacknowledged. This is the explicit-condition shutdown; no fixed sleep. + # It raises IncompleteMessageDeliveryError when sends or receipts are 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 sends or receipts 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, receipt) + 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: + 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: new messages wait in the + # buffer (publish() blocks once it is full) and go out after the reconnect, so no + # guaranteed message is lost + logger.warning("Service reconnecting: %s (%s)", event.get_message(), event.get_cause()) + + def on_reconnected(self, event: ServiceEvent) -> None: + # automatic reconnect succeeded: the waiting messages go 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() + + +class PublishReceiptHandler(MessagePublishReceiptListener): + """Very simple class, used for handling publish receipts (ACKs/NACKs) from the broker. + Runs on an API thread: + return promptly and do not block in it (C API Best Practices).""" + + def on_publish_receipt(self, publish_receipt: PublishReceipt) -> None: + msg_id = publish_receipt.user_context # the message id passed to publish() + if publish_receipt.exception is None: # ACK + logger.debug("ACK for Message ID %s", msg_id) # good enough, the broker has it now + return + # NACK: the exception says why (for example MessageDestinationDoesNotExistError when no + # queue subscribes to the topic and the client-profile rejects on no subscription match) + logger.warning("NACK for Message ID %s - %s", msg_id, publish_receipt.exception) + # probably want to do something here. some error handling possibilities: + # - look the message up by this ID in your application's outbound store and send it again + # - send it somewhere else (error handling queue?) + # - log and continue + # - pause and retry (backoff) - maybe set a flag to slow down the publisher + + +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-async-sample.py b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-subscriber-async-sample.py new file mode 100644 index 0000000..dd0936e --- /dev/null +++ b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-subscriber-async-sample.py @@ -0,0 +1,372 @@ +""" +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: 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 that ACKs only after processing, receiver active/passive state +and termination handling, reconnection and service interruption listeners +registered BEFORE connect(), and a graceful SIGINT/SIGTERM shutdown. Structured as +module-level functions, setup_solace(...), connect_solace(...), await_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. + +Only practices documented in canonical Solace sources are encoded here. + +Copied into a generated project as guaranteed_subscriber_async.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 +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.message_receiver import MessageHandler +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 = "GuaranteedSubscriber" +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" +# 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 handler to drain the messages the API +# has already received before the receiver stops; terminate() raises +# IncompleteMessageDeliveryError when messages remain after that. 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 lets the handler finish more of the +# received messages, shorter exits sooner. +TERMINATE_GRACE_PERIOD_MS = 10_000 + +# 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, the flags the listeners flip, 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 failed terminate()) + 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 main + # loop exits, 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 + await_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 handler 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 main 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 + # set the message handler BEFORE start(), so the handler is in place when delivery + # begins and received messages do not wait in the API's buffer for it. receive_async() + # needs a connected service, so it runs here after connect(), not in setup_solace(). + # see bottom of file for QueueMessageHandler, which receives the messages from the queue + state.receiver.receive_async(QueueMessageHandler(state, state.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 on this receiver + 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 await_messages(state: SubscriberState) -> None: + # async queue receive working now, so time to wait until done... + trace(f"{APP_NAME} connected, and running. Press Ctrl-C to quit.") + while not state.shutdown.wait(1.0): # wait 1 second; the wait returns True once shutdown is requested + trace(f"{API} {APP_NAME} Received msgs/s: {state.msg_recv_counter:,}") # simple way of calculating message rates + 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 second + + +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(): + # gracefully stop delivery before exit: terminate(grace_period) stops the receiver + # and gives the handler up to the grace period to drain (and ACK) the messages the + # API has already received. The receiver turns TERMINATED once its buffer is empty, + # while the handler may still hold the last message, so that message's ack() can + # fail (the API logs a warning). A message received but not yet ACKed stays on the + # queue and is redelivered. terminate() raises IncompleteMessageDeliveryError when + # messages remain after the grace period; catch it so the disconnect below still runs + try: + state.receiver.terminate(TERMINATE_GRACE_PERIOD_MS) + except IncompleteMessageDeliveryError as error: + logger.error("Receiver stopped with messages still undelivered to the handler after %d ms: %s", + TERMINATE_GRACE_PERIOD_MS, error) + state.exit_code = 1 + 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: + 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 + 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 main 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 main loop exits and teardown_solace() runs in + # main's finally. + self.state.exit_code = 1 + self.state.shutdown.set() + + +class QueueMessageHandler(MessageHandler): + """Very simple class, used for receiving messages from the queue. on_message runs on an + API-owned receiver thread, not the main thread: return promptly and do not block in it + (hand heavy work to the application's own thread or queue), since delivery from the queue + stalls while it runs (C API Best Practices).""" + + def __init__(self, state: SubscriberState, receiver: PersistentMessageReceiver) -> None: + self.state = state + self.receiver = receiver + + def on_message(self, message: InboundMessage) -> None: + self.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 + self.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 API swallows the exception + # with one warning and the message stays unacknowledged on the flow. + 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). + self.receiver.ack(message) # ACKs are asynchronous + + +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-solace-connection-config.py b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-solace-connection-config.py new file mode 100644 index 0000000..a0e4343 --- /dev/null +++ b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-solace-connection-config.py @@ -0,0 +1,124 @@ +""" +Shared connection-config helper for the Python reference samples. + +Loads connection properties into the dict that +MessagingService.builder().from_properties(...) takes. Source precedence: + 1. config.json in the working directory, when present. Its keys are Solace + Python API property names (the values of the solace_properties constants): + "solace.messaging.transport.host" + "solace.messaging.service.vpn-name" + "solace.messaging.authentication.basic.username" + "solace.messaging.authentication.basic.password" + Every key passes through to from_properties() unchanged, so other service + properties need no parser change. A later builder call overrides the same + key, and each sample sets its reconnection strategy before from_properties(), + so reconnection keys in the file override the sample's default. The native library rejects an invalid value at build() or + connect(); the API silently ignores an unknown key. config.json holds broker + credentials, so it MUST be gitignored. + 2. otherwise the command-line arguments: + [password] + +The password is optional. The API requires the basic-auth password key at build(), +so an absent or blank password is sent as an empty string. config.json must be a +single flat JSON object; the standard-library json module reads it. + +Grounding: the "load a JSON file, then from_properties()" idiom and the required +host and vpn-name keys: + https://docs.solace.com/API/API-Developer-Guide-Python/Python-API-Messaging-Service.md + +Copied into a generated project as solace_connection_config.py (a module name +needs underscores); every sample imports it with + from solace_connection_config import load_service_properties + +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 json +import os +import sys + +from solace.messaging.config.solace_properties import ( + authentication_properties, + service_properties, + transport_layer_properties, +) + +CONFIG_FILE = "config.json" + +# the three connection values every sample needs; the password is optional +REQUIRED_KEYS = ( + transport_layer_properties.HOST, + service_properties.VPN_NAME, + authentication_properties.SCHEME_BASIC_USER_NAME, +) + + +def load_service_properties(args: list[str], app_name: str) -> dict[str, object]: + """ + Load connection details from config.json when present in the working directory, + otherwise from the command-line arguments, log which source was used, and return the + from_properties() dict with every loaded property (generic pass-through). Prints a + usage line and exits when neither source supplies host, message-vpn, and + client-username; rejects a blank host, message-vpn, or client-username from either + source. The basic-auth password key is always present: the API requires it at build(), + so an absent or blank password is passed as an empty string. + + Args: + args: the app's command-line arguments without the script name (the fallback source) + app_name: the application name, used only in the usage message + """ + properties = _load_config_file() + if properties is not None: + source = CONFIG_FILE + elif len(args) >= 3: + source = "command-line arguments" + properties = { + transport_layer_properties.HOST: args[0], + service_properties.VPN_NAME: args[1], + authentication_properties.SCHEME_BASIC_USER_NAME: args[2], + } + if len(args) > 3: + properties[authentication_properties.SCHEME_BASIC_PASSWORD] = args[3] + else: + trace(f"No {CONFIG_FILE} in the working directory; provide connection details on the command line.") + trace(f"Usage: {app_name} [password]\n") + sys.exit(1) + for key in REQUIRED_KEYS: + _require_non_blank(properties.get(key), key) + # password is optional: absent or blank means an empty password is sent + if not properties.get(authentication_properties.SCHEME_BASIC_PASSWORD): + properties[authentication_properties.SCHEME_BASIC_PASSWORD] = "" + host = properties[transport_layer_properties.HOST] + trace(f"{app_name}: using connection details from {source} (host={host}).") + return properties + + +def _load_config_file() -> dict[str, object] | None: + if not os.path.exists(CONFIG_FILE): + return None # no config file: fall back to the command-line arguments + # present but unreadable (for example a permission-locked credentials file): open() + # raises PermissionError, so this fails loud rather than silently falling back to + # different command-line connection args. A malformed file raises JSONDecodeError. + with open(CONFIG_FILE, encoding="utf-8") as config_file: + parsed = json.load(config_file) + if not isinstance(parsed, dict): + raise ValueError(f"{CONFIG_FILE} must hold a single JSON object") + for required in REQUIRED_KEYS: + if required not in parsed: + raise ValueError(f'{CONFIG_FILE} is missing required key "{required}"') + return parsed + + +def _require_non_blank(value: object, key: str) -> None: + """Reject a missing, non-string, or blank required connection value, whatever its source.""" + if not isinstance(value, str) or not value.strip(): + raise ValueError(f'connection value for "{key}" must be a non-blank string') + + +def trace(message: str) -> None: + """Demo narration sink: every status line in this helper funnels through this one + function. An application replaces this single body to route narration to its + logger or reporting system.""" + print(message, flush=True) diff --git a/tools/check-links.sh b/tools/check-links.sh index 9bccbc5..e0dd888 100755 --- a/tools/check-links.sh +++ b/tools/check-links.sh @@ -6,8 +6,8 @@ # Complements the CI validate job (gh skill publish --dry-run), which checks the # Agent Skills spec but not links. # This sweep additionally: -# * covers the WHOLE skills/ tree — SKILL.md, references/*.md, the .java sample -# headers, and scripts/*.sh (not just *.md) — while excluding the repo-root +# * covers the WHOLE skills/ tree — SKILL.md, references/*.md, the .java and .py +# sample headers, and scripts/*.sh (not just *.md) — while excluding the repo-root # README (D-11: the sweep globs the skills/ tree only, never the repo root), # * detects Solace "soft 404s" — pages that return HTTP 200/30x but actually land # on a not-found page (docs.solace.com redirects bad paths to Not-Found.htm; @@ -74,7 +74,7 @@ should_skip() { return 1 } -# Collect unique http(s) links from the whole skills/ tree (*.md, *.java, *.sh). +# Collect unique http(s) links from the whole skills/ tree (*.md, *.java, *.py, *.sh). # README at the repo root is excluded because ROOT is the skills/ tree, never the # repo root (D-11). Doc links are now public docs.solace.com .md URLs, checked inline. # @@ -89,7 +89,7 @@ while IFS= read -r line; do links+=("$line") done < <( grep -rhoE 'https?://[^ )"`'"'"'>]+' "$ROOT" \ - --include='*.md' --include='*.java' --include='*.sh' \ + --include='*.md' --include='*.java' --include='*.py' --include='*.sh' \ --exclude-dir='docs' \ | sed -E 's/[.,;:*]+$//' \ | sort -u