UMessageBus Classes

Message bus worker of urgap.

The worker is agnostic of both the transport and of what a message asks for: the transport owns the topic and subscription lifecycle and message settlement, the handler performs the work, and this module owns the receive loop and the completion event.

A message is expected to look like:

{
    "uuid": "<correlation id>",
    "subscription_key": "<routing key the subscription filters on>",
    "consumer_kwargs": {...},   # whatever the handler needs
}

and the handler is called with the full message body, returning (ok, output_uris).

class urgap.umessagebus.worker.MessageHandler(*args, **kwargs)

Callable which performs the work a message asks for.

urgap.umessagebus.worker.handle_messages(message_bus: UMessageBusBase, handler: MessageHandler, exit_after_first: bool = False, max_autorenew: float = 0, max_empty_polls: int = 3) → None

Receive and handle messages until the worker is done.

Parameters:

  • message_bus – Transport to receive from.

  • handler – Callable performing the work a message asks for.

  • exit_after_first – If True, stop after one handled message.

  • max_autorenew – Requested lock duration in seconds, 0 disables renewal.

  • max_empty_polls – Number of consecutive empty polls after which the worker exits. 0 keeps the worker polling forever.

urgap.umessagebus.worker.init_message_bus(cred_key: str, subscription_key: str, topic_name: str | None = None, subscription_name: str | None = None, completion_topic: str | None = None) → UMessageBusBase

Initialize the message bus transport for a cred_key.

The scheme of the cred_key selects the implementation, e.g. azure-servicebus://<namespace>.servicebus.windows.net or gcp-pubsub://<project-id>.

Parameters:

  • cred_key – UUri of the message bus.

  • subscription_key – Routing key the worker accepts.

  • topic_name – Topic to receive from, defaults to config service_bus_topic.

  • subscription_name – Subscription to receive from.

  • completion_topic – Topic to publish completion events to, defaults to config service_bus_completion_topic.

Returns:

Transport instance for the requested message bus.

Raises:

ImportError – If no transport is available for the cred_key scheme.

urgap.umessagebus.worker.load_message_context(consumer_kwargs: dict[str, Any]) → None

Apply the config and ucredentials a message brought along.

Parameters:

consumer_kwargs – Consumer kwargs of a message.

urgap.umessagebus.worker.process_message(message: Any, message_bus: UMessageBusBase, handler: MessageHandler, exit_after_first: bool = False) → bool

Handle a single message.

Parameters:

  • message – Received transport specific message.

  • message_bus – Transport the message came from, used to settle it.

  • handler – Callable performing the work the message asks for.

  • exit_after_first – If True, stop the loop after one handled message.

Returns:

True if the worker loop should stop, False otherwise.

urgap.umessagebus.worker.run_subscription_worker(cred_key: str, subscription_key: str, handler: MessageHandler, subscription_name: str | None = None, topic_name: str | None = None, completion_topic: str | None = None, exit_after_first: bool | None = None, max_autorenew: float | None = None, max_empty_polls: int = 3, shutdown_event: EventClass | None = None) → None

Run a message bus subscription worker for one routing key.

Parameters:

  • cred_key – UUri of the message bus, its scheme selects the transport.

  • subscription_key – Routing key this worker accepts, also used as the subscription filter value.

  • handler – Callable performing the work a message asks for.

  • subscription_name – Subscription to receive from, defaults to a transport safe form of subscription_key.

  • topic_name – Topic to receive from, defaults to config service_bus_topic.

  • completion_topic – Topic to publish completion events to, defaults to config service_bus_completion_topic.

  • exit_after_first – If True, stop after one handled message. Defaults to config service_bus_exit_after_first_message.

  • max_autorenew – Requested lock duration in seconds. Defaults to config service_bus_max_autorenewal_minutes.

  • max_empty_polls – Number of consecutive empty polls after which the worker exits. 0 keeps the worker polling forever.

  • shutdown_event – Event to signal the parent process to terminate.

class urgap.umessagebus.umessagebus_manager.UMessageBusManager

Manager for message bus transports.

The UMessageBusManager detects which message bus implementations are available, mapping cred_key schemes to their respective implementations.

__init__() → None

Discover and register available implementations.

BASE_CLASS

alias of UMessageBusBase

class urgap.umessagebus.io._base.UMessageBusBase(cred_key: str, subscription_key: str, topic_name: str | None = None, subscription_name: str | None = None, completion_topic: str | None = None)

Base class for message bus transports in urgap.

A transport owns the topic and subscription lifecycle of one subscription and settles the messages it hands out. It is addressed by a cred_key UUri whose scheme selects the implementation, e.g. azure-servicebus://<namespace>.servicebus.windows.net or gcp-pubsub://<project-id>.

Subclasses implement the transport specific parts and are discovered by their SCHEMA attribute.

__init__(cred_key: str, subscription_key: str, topic_name: str | None = None, subscription_name: str | None = None, completion_topic: str | None = None) → None

Create a transport for one subscription.

Parameters:

  • cred_key – UUri of the message bus, its scheme selects this class and its remainder addresses the namespace, project or account.

  • subscription_key – Routing key this worker accepts, also used as the subscription filter value.

  • topic_name – Topic to receive from.

  • subscription_name – Subscription to receive from, defaults to a transport safe form of subscription_key.

  • completion_topic – Topic to publish completion events to, None disables completion events.

abandon(message: Any) → None

Release a message back to the subscription for redelivery.

Parameters:

message – Transport specific message.

property address: str

Get the transport address, i.e. the cred_key without its scheme.

Returns:

Namespace, project or account this transport talks to.

close() → None

Close the clients this transport opened.

complete(message: Any) → None

Settle a message as handled so it is not redelivered.

Parameters:

message – Transport specific message.

connect() → None

Open the clients this transport needs.

ensure_entities() → None

Create topic, subscription and routing filter if they do not exist.

get_body(message: Any) → str

Get the json body of a received message.

Parameters:

message – Transport specific message.

Returns:

Message body as a json string.

publish_completion(event: dict) → None

Publish a completion event, if a completion topic is configured.

Parameters:

event – Event payload to serialize.

receive(max_wait_time: int, max_messages: int = 1) → list

Receive up to max_messages messages.

Parameters:

  • max_wait_time – Seconds to wait for a message before giving up.

  • max_messages – Maximum number of messages to return.

Returns:

List of transport specific messages, empty if none arrived.

renew(message: Any, duration: float) → None

Keep a message locked while it is being handled.

Parameters:

  • message – Transport specific message.

  • duration – Requested lock duration in seconds.

static safe_subscription_name(subscription_key: str) → str

Turn a routing key into a name every transport accepts.

Pub/Sub allows letters, numbers and -_.~+% only, so the colon of a unode_full_identifier is replaced.

Parameters:

subscription_key – Routing key, e.g. a unode_full_identifier.

Returns:

Subscription name without characters the transports reject.

class urgap.umessagebus.io.azure_servicebus.UMessageBusAzureServiceBus(**kwargs: Any)

UMessageBus class interface for Azure Service Bus topics.

Messages are routed to a subscription by a SQL rule filter on the subscription_key application property and are received in PEEK_LOCK mode, so a failed message is abandoned and redelivered.

abandon(message: Any) → None

Abandon a Service Bus message so it is redelivered.

Parameters:

message – Received Service Bus message.

close() → None

Close receiver, renewer and client.

complete(message: Any) → None

Complete a Service Bus message.

Parameters:

message – Received Service Bus message.

connect() → None

Open the Service Bus client, receiver and completion sender.

ensure_entities() → None

Create the topics, the subscription and its routing filter.

get_body(message: Any) → str

Get the json body of a Service Bus message.

Parameters:

message – Received Service Bus message.

Returns:

Message body as a json string.

publish_completion(event: dict) → None

Publish a completion event to the completion topic.

Parameters:

event – Event payload to serialize.

receive(max_wait_time: int, max_messages: int = 1) → list

Receive messages from the subscription.

Parameters:

  • max_wait_time – Seconds to wait for a message before giving up.

  • max_messages – Maximum number of messages to return.

Returns:

List of received Service Bus messages.

renew(message: Any, duration: float) → None

Register a message with an AutoLockRenewer.

Parameters:

  • message – Received Service Bus message.

  • duration – Maximum lock renewal duration in seconds.

class urgap.umessagebus.io.gcp_pubsub.UMessageBusGCPPubSub(**kwargs: Any)

UMessageBus class interface for GCP Pub/Sub topics.

Messages are routed to a subscription by a subscription filter on the subscription_key attribute and are pulled synchronously, so a failed message has its ack deadline zeroed and is redelivered.

The cred_key carries the project, e.g. gcp-pubsub://my-gcp-project.

Note

A Pub/Sub subscription filter is immutable. If the subscription already exists with a different filter it is used as it is, which is logged as a warning because messages may then not be the ones this worker expects.

abandon(message: Any) → None

Nack a Pub/Sub message by zeroing its ack deadline.

Parameters:

message – Received Pub/Sub message.

close() → None

Stop the lease renewer and close the subscriber client.

complete(message: Any) → None

Acknowledge a Pub/Sub message.

Parameters:

message – Received Pub/Sub message.

connect() → None

Open the Pub/Sub subscriber and publisher clients.

ensure_entities() → None

Create the topics and the subscriptions if they are missing.

get_body(message: Any) → str

Get the json body of a Pub/Sub message.

Parameters:

message – Received Pub/Sub message.

Returns:

Message body as a json string.

property project: str

Get the GCP project this transport talks to.

Returns:

GCP project id.

publish_completion(event: dict) → None

Publish a completion event to the completion topic.

Parameters:

event – Event payload to serialize.

receive(max_wait_time: int, max_messages: int = 1) → list

Pull messages from the subscription.

Parameters:

  • max_wait_time – Seconds to wait for a message before giving up.

  • max_messages – Maximum number of messages to return.

Returns:

List of received Pub/Sub messages, empty if none arrived in time.

renew(message: Any, duration: float) → None

Keep a Pub/Sub message leased while it is being handled.

Pub/Sub caps a single extension at 10 minutes, so a longer duration is held by re-extending the deadline from a background thread until the duration is used up or the message is settled, which is what the Service Bus AutoLockRenewer does for that transport.

Parameters:

  • message – Received Pub/Sub message.

  • duration – Requested lock duration in seconds.