DOCSEARCHbyAlgolia
Submit Docs

stomp.py Documentation

repository·dev·Indexed 19 days ago

https://github.com/jasonrbriggs/stomp.py

A Python STOMP client library and standalone CLI tool supporting versions 1.0, 1.1, and 1.2 of the protocol. It enables interaction with messaging servers such as ActiveMQ and RabbitMQ, providing features for sending and receiving messages, manual acknowledgement (ack/nack), atomic transactions, and failover connection configurations.

Tokens
5.9K
Snippets
24
Records
26
Agent score
68%

What's inside stomp.py

  1. Overview of the stomp.py package structure

    dev

    The stomp.py package is organized into several submodules and subpackages that handle different aspects of the STOMP protocol implementation. Key components include:

    • Core Logic: Found in the main stomp module and stomp.protocol.
    • Connection Management: Handled by stomp.connect and stomp.transport.
    • Event Handling: Managed via stomp.listener for reacting to STOMP frames.
    • Constants and Utilities: stomp.constants provides protocol-specific values, while stomp.utils and stomp.colours provide helper functions.
    • Error Handling: stomp.exception defines the exception hierarchy.
    • Extensibility: The stomp.adapter subpackage allows for different transport or protocol adaptations.
    • Testing: The stomp.test subpackage provides tools for testing STOMP implementations.

  2. What is Stomp.py and what protocols does it support?

    dev

    Stomp.py is a Python client library for the STOMP (Simple Text Oriented Messaging Protocol). It is designed for inter-application communication via message brokers.

    Key features include:

    • Support for all STOMP protocol versions: 1.0, 1.1, and 1.2.
    • A command-line client installed via pip for testing purposes.
    • Compatibility with brokers such as ActiveMQ, RabbitMQ, and stompserver.

  3. Use interactive commands in the stomp client

    dev

    After establishing a connection, you can issue commands directly in the prompt. Common commands include:

    • subscribe <destination>: Listen for messages on a specific queue or topic.
    • send <destination> <message>: Send a message to a destination.
    • unsubscribe <destination>: Stop listening to a destination.
    • exit or quit: Terminate the session.

    To see a list of all available commands within the application, type help.

    subscribe /queue/test
send /queue/test hello world

  4. Acknowledge and Nack messages

    dev

    To control message consumption flow, use manual acknowledgement modes. When subscribing, set the ack parameter to 'client' or 'client-individual'.

    Once a message is received, use the message-id and the subscription id to signal the server:

    • ack(message_id, subscription_id): Confirms the message was successfully processed.
    • nack(message_id, subscription_id): Signals that the message was not processed (allowing the broker to potentially redeliver or failover).
    # Subscribe with client-side acknowledgement
conn.subscribe('/queue/test', id=4, ack='client')

# Inside your listener's on_message method:
# To acknowledge:
conn.ack('mybroker-14aa2', 4)

# To negatively acknowledge:
conn.nack('mybroker-14ab2', 4)

  6. Disconnect gracefully

    dev
    Use conn.disconnect() to shut down the connection. By default, stomp.py uses a receipt parameter to ensure a graceful shutdown. The client will wait for the server to send back a response to the DISCONNECT frame before the connection is fully dropped.
    conn.disconnect()

  7. Run stomp.py unit tests locally using Docker or Podman

    dev

    To run the project's unit tests in a containerized environment, follow these steps using make and either Docker or Podman:

    1. Install dependencies: poetry install
    2. Create the image: make docker-image (or make podman-image)
    3. Run the container: make run-docker (or make run-podman)
    4. Run tests: make test
    5. Cleanup: make remove-docker (or make remove-podman)
    poetry install
make docker-image
make run-docker
make test
make remove-docker

  8. Handle disconnects and implement reconnection

    dev

    To handle connection failures or heartbeats timeouts, implement the on_disconnected method in a ConnectionListener subclass. This method can be used to trigger a reconnection logic.

    Note on Heartbeats: If your message processing (e.g., inside on_message) takes longer than the configured heartbeat interval, the connection may time out and disconnect. You should ensure your processing logic doesn't block the heartbeat mechanism or implement reconnection in on_disconnected to recover.

    import stomp
import time

def connect_and_subscribe(conn):
    conn.connect('guest', 'guest', wait=True)
    conn.subscribe(destination='/queue/test', id=1, ack='auto')

class MyListener(stomp.ConnectionListener):
    def __init__(self, conn):
        self.conn = conn

    def on_error(self, frame):
        print(f'received an error "{frame.body}"')

    def on_message(self, frame):
        print(f'received a message "{frame.body}"')
        # Simulate long processing
        time.sleep(5)

    def on_disconnected(self):
        print('disconnected - attempting to reconnect...')
        connect_and_subscribe(self.conn)

conn = stomp.Connection([('localhost', 62613)], heartbeats=(4000, 4000))
conn.set_listener('', MyListener(conn))
connect_and_subscribe(conn)

  9. Run the stomp.py command-line client

    dev

    You can launch the stomp.py command-line client using either the Python module syntax or the direct stomp executable (if installed in your bin directory).

    To connect to a local server on the default port (61613):

    python -m stomp -H localhost -P 61613
# OR
stomp -H localhost -P 61613

    Once connected, you can interact with the messaging system using inline commands like subscribe and send.

    python -m stomp -H localhost -P 61613

  10. Quick Start: Send a message with stomp.py

    dev

    To connect to a local message broker and send a message, instantiate a stomp.Connection, connect using credentials, and use the send method specifying a destination (e.g., a queue).

    import stomp
import sys

conn = stomp.Connection()
conn.connect('admin', 'password', wait=True)
conn.send(body=' '.join(sys.argv[1:]), destination='/queue/test')
conn.disconnect()

  11. Send and receive messages

    dev

    Once connected, you can interact with the broker using send, subscribe, and unsubscribe.

    • Sending: Use c.send(destination, body, ...) to transmit a message to a specific destination.
    • Receiving: To receive messages, you must:
      1. Implement a subclass of ConnectionListener to handle incoming data.
      2. Register the listener using c.set_listener(name, listener_instance). Multiple listeners can be registered with unique names.
      3. Subscribe to a destination using c.subscribe(destination, id, ...). Note that for STOMP 1.1+, the id parameter is required.
    • Unsubscribing: Use c.unsubscribe(id) with the unique subscription ID used during the subscribe call.
    from stomp import Connection, PrintingListener

c = Connection([('127.0.0.1', 62613)])

# 1. Set up a listener (PrintingListener is a built-in for debugging)
c.set_listener('my_listener', PrintingListener())

# 2. Connect
c.connect('admin', 'password', wait=True)

# 3. Subscribe (id is required for STOMP 1.1+)
c.subscribe('/queue/test', id=123)

# 4. Send a message
c.send('/queue/test', 'a test message')

# 5. Unsubscribe
c.unsubscribe(123)

  12. Use transactions for atomic message sending

    dev

    Transactions allow you to group multiple send operations so they are only visible to the broker once committed.

    1. Call conn.begin() to start a transaction. This returns a transaction_id.
    2. Pass this transaction_id to the transaction parameter of the send method.
    3. Call conn.commit(transaction_id) to finalize the messages, or conn.abort(transaction_id) to discard them.
    conn.subscribe('/queue/test', id=5)

# Start transaction
txid = conn.begin()

# Send messages associated with the transaction
conn.send('/queue/test', 'test1', transaction=txid)
conn.send('/queue/test', 'test2', transaction=txid)
conn.send('/queue/test', 'test3', transaction=txid)

# Commit the transaction
conn.commit(txid)

# OR to discard:
# conn.abort(txid)