naas.ai Docs

PythonQueueAdapter

2 min read

PythonQueueAdapter

What it is

  • A process-local message bus adapter built on Python stdlib queue.Queue.
  • Supports topic + routing key filtering with wildcard matching:
    • * matches exactly one dot-separated segment
    • # matches zero or more segments

Public API

  • class PythonQueueAdapter(IBusAdapter)
    • __init__() -> None
      • Initializes internal subscriber registry and lock.
    • topic_publish(topic: str, routing_key: str, payload: bytes) -> None
      • Publishes payload to all subscribers whose topic pattern matches topic and whose routing_key pattern matches routing_key.
    • topic_consume(topic: str, routing_key: str, callback: Callable[[bytes], None]) -> Thread
      • Registers a subscriber for a topic and routing_key pattern.
      • Starts a daemon thread that pulls messages from an internal queue and invokes callback(payload).
      • Returns the started Thread.
    • _match_routing_key(pattern: str, routing_key: str) -> bool (static)
      • Implements dot-separated wildcard matching for topics and routing keys.

Configuration/Dependencies

  • No external dependencies; uses stdlib:
    • queue.Queue, threading.Thread, threading.RLock
  • Depends on naas_abi_core.services.bus.BusPorts.IBusAdapter for interface conformance.

Usage

from naas_abi_core.services.bus.adapters.secondary.PythonQueueAdapter import PythonQueueAdapter
 
bus = PythonQueueAdapter()
 
def on_message(payload: bytes) -> None:
    print("got:", payload.decode())
 
# Subscribe to topic "orders" and routing keys like "created.eu", "created.us", etc.
bus.topic_consume(topic="orders", routing_key="created.*", callback=on_message)
 
bus.topic_publish(topic="orders", routing_key="created.eu", payload=b"order-123")

Caveats

  • In-process only: no inter-process or network transport.
  • topic_consume(...) runs callbacks on a daemon thread.
  • To stop a consumer, the callback must raise StopIteration (breaks the consume loop).
  • If the callback raises any other exception:
    • the payload is re-queued for that subscriber
    • the consumer thread then raises and exits (message may be redelivered if a new consumer is created).

Graph View