Azure Service Bus

The Azure Service Bus SDKs, unchanged, with a connection string: queues, topics, subscriptions with filters, sessions, scheduling and dead-lettering over AMQP 1.0.

Connection string

Endpoint=sb://amqp.powerqueue.link/;SharedAccessKeyName=<your-identifier>;SharedAccessKey=anything

• SharedAccessKeyName is your tenant identifier. The key can be anything: tokens are signed with it, but signatures are not checked.
• The SDKs connect with AMQP over TLS to port 5671 of amqp.powerqueue.link, as they do to Azure. The administration API (ServiceBusAdministrationClient) uses HTTPS on port 443 of the same host.
• Do not add UseDevelopmentEmulator=true: it turns TLS off. Use the default AMQP transport; AMQP over WebSockets is not supported.

Python (azure-servicebus)

pip install azure-servicebus. A queue, with no setup:

from azure.servicebus import ServiceBusClient, ServiceBusMessage

CONNECTION_STRING = (
    "Endpoint=sb://amqp.powerqueue.link/;"
    "SharedAccessKeyName=<your-identifier>;SharedAccessKey=anything"
)

with ServiceBusClient.from_connection_string(CONNECTION_STRING) as client:
    with client.get_queue_sender("invoices") as sender:
        sender.send_messages([ServiceBusMessage(f"invoice {i}") for i in range(3)])
    with client.get_queue_receiver("invoices", max_wait_time=5) as receiver:
        for message in receiver:
            print(str(message))
            receiver.complete_message(message)

A topic with a filtered subscription, created through the administration API:

from azure.servicebus import ServiceBusClient, ServiceBusMessage
from azure.servicebus.management import ServiceBusAdministrationClient, SqlRuleFilter

admin = ServiceBusAdministrationClient.from_connection_string(CONNECTION_STRING)
admin.create_topic("alerts")
admin.create_subscription("alerts", "errors")
admin.delete_rule("alerts", "errors", "$Default")
admin.create_rule("alerts", "errors", "only-errors", filter=SqlRuleFilter("level = 'error'"))

with ServiceBusClient.from_connection_string(CONNECTION_STRING) as client:
    with client.get_topic_sender("alerts") as sender:
        sender.send_messages(ServiceBusMessage("disk full", application_properties={"level": "error"}))
        sender.send_messages(ServiceBusMessage("all good", application_properties={"level": "info"}))
    with client.get_subscription_receiver("alerts", "errors", max_wait_time=5) as receiver:
        for message in receiver:
            print(str(message))
            receiver.complete_message(message)

JavaScript (@azure/service-bus)

npm install @azure/service-bus, then run as an ES module:

import { ServiceBusClient } from "@azure/service-bus";

const connectionString =
  "Endpoint=sb://amqp.powerqueue.link/;SharedAccessKeyName=<your-identifier>;SharedAccessKey=anything";
const client = new ServiceBusClient(connectionString);

const sender = client.createSender("invoices");
await sender.sendMessages({ body: "hello", messageId: "invoice-1" });
await sender.close();

const receiver = client.createReceiver("invoices");
const messages = await receiver.receiveMessages(10, { maxWaitTimeInMs: 5000 });
for (const message of messages) {
  console.log(message.body);
  await receiver.completeMessage(message);
}
await receiver.close();
await client.close();

C# (.NET, Azure.Messaging.ServiceBus)

dotnet add package Azure.Messaging.ServiceBus:

using Azure.Messaging.ServiceBus;

var connectionString =
    "Endpoint=sb://amqp.powerqueue.link/;SharedAccessKeyName=<your-identifier>;SharedAccessKey=anything";
await using var client = new ServiceBusClient(connectionString);

await using ServiceBusSender sender = client.CreateSender("invoices");
await sender.SendMessageAsync(new ServiceBusMessage("hello"));

await using ServiceBusReceiver receiver = client.CreateReceiver("invoices");
var messages = await receiver.ReceiveMessagesAsync(maxMessages: 10, maxWaitTime: TimeSpan.FromSeconds(5));
foreach (var message in messages)
{
    Console.WriteLine(message.Body.ToString());
    await receiver.CompleteMessageAsync(message);
}

Supported

• Queues, topics and subscriptions; send, receive (peek-lock and receive-and-delete), complete, abandon, dead-letter and defer; the dead-letter subqueues.
• Management operations: peek, renew lock, schedule and cancel scheduled messages, receive deferred messages by sequence number, sessions (session state, session lock renewal, listing sessions), and subscription rules.
• Subscription rules with SQL filters, correlation filters, true and false filters, and SQL rule actions (SET, REMOVE).
• Auto-forwarding (ForwardTo) within your tenant, auto-delete on idle, and local transactions across the entities of one connection.
• The administration REST API: create, get, update, delete and list queues, topics, subscriptions and rules, runtime properties, and namespace information.

Differences from Azure Service Bus

• Sending to or receiving from an unknown queue creates it rather than failing with "entity not found". Queues created that way are deleted after an hour empty and unused, and not while a link to them is open; creating one through the administration API keeps it.
• No AMQP over WebSockets, no Azure AD (Entra ID) tokens, no premium features such as large messages; partitioning is always reported off.
• Transactions are atomic in what other clients see, but not across a crash, and cannot span connections.
• Message time to live, default time to live and scheduled enqueue times are limited to 1 hour; messages are limited to 256 KiB.
• ForwardTo naming another namespace is refused with 400.
• Each connection keeps the tenant of its first token; a token for a different key name on the same connection is refused.

Messages from other interfaces

Bodies sent through the other interfaces arrive as a single data section of their bytes. A message you send whose body is an AMQP value (a number, map, list and so on) or sequence reads back the same through Service Bus, and as JSON text, marked x-pq-body-encoding = json where the reader can carry a mark, through the others; a string or binary value crosses as its plain bytes. AMQP 0-9-1 headers and SQS and Pub/Sub attributes arrive as application properties with their types; an AMQP table or array header as JSON text, a decimal as its decimal text. Property names starting x-pq- are the service's and are not passed on.

The full rules, with examples: Using one queue from several interfaces.

Errors