azure-servicebus-ts

SERVICEBUS_NAMESPACE= .servicebus.windows.net
SERVICEBUS_QUEUE_NAME=my-queue
SERVICEBUS_TOPIC_NAME=my-topic
SERVICEBUS_SUBSCRIPTION_NAME=my-subscription
AZURE_TOKEN_CREDENTIALS=prod # Required only if DefaultAzureCredential is used in production

import { ServiceBusClient } from "@azure/service-bus";
import { DefaultAzureCredential, ManagedIdentityCredential } from "@azure/identity";

// Local dev: DefaultAzureCredential. Production: set AZURE_TOKEN_CREDENTIALS=prod or AZURE_TOKEN_CREDENTIALS= 
const credential = new DefaultAzureCredential({requiredEnvVars: ["AZURE_TOKEN_CREDENTIALS"]});
// Or use a specific credential directly in production:
// See https://learn.microsoft.com/javascript/api/overview/azure/identity-readme?view=azure-node-latest#credential-classes
// const credential = new ManagedIdentityCredential();

const fullyQualifiedNamespace = process.env.SERVICEBUS_NAMESPACE!;
const client = new ServiceBusClient(fullyQualifiedNamespace, credential);

const sender = client.createSender("my-queue");

// Single message
await sender.sendMessages({
 body: { orderId: "12345", amount: 99.99 },
 contentType: "application/json",
});

// Batch messages
const batch = await sender.createMessageBatch();
batch.tryAddMessage({ body: "Message 1" });
batch.tryAddMessage({ body: "Message 2" });
await sender.sendMessages(batch);

await sender.close();

const receiver = client.createReceiver("my-queue");

// Receive batch
const messages = await receiver.receiveMessages(10, { maxWaitTimeInMs: 5000 });
for (const message of messages) {
 console.log(`Received: ${message.body}`);
 await receiver.completeMessage(message);
}

await receiver.close();

const receiver = client.createReceiver("my-queue");

const subscription = receiver.subscribe({
 processMessage: async (message) => {
 console.log(`Processing: ${message.body}`);
 // Message auto-completed on success
 },
 processError: async (args) => {
 console.error(`Error: ${args.error}`);
 },
});

// Stop after some time
setTimeout(async () => {
 await subscription.close();
 await receiver.close();
}, 60000);

// Send to topic
const topicSender = client.createSender("my-topic");
await topicSender.sendMessages({
 body: { event: "order.created", data: { orderId: "123" } },
 applicationProperties: { eventType: "order.created" },
});

// Receive from subscription
const subscriptionReceiver = client.createReceiver("my-topic", "my-subscription");
const messages = await subscriptionReceiver.receiveMessages(10);

// Send session message
const sender = client.createSender("session-queue");
await sender.sendMessages({
 body: { step: 1, data: "First step" },
 sessionId: "workflow-123",
});

// Receive session messages
const sessionReceiver = await client.acceptSession("session-queue", "workflow-123");
const messages = await sessionReceiver.receiveMessages(10);

// Get/set session state
const state = await sessionReceiver.getSessionState();
await sessionReceiver.setSessionState(Buffer.from(JSON.stringify({ progress: 50 })));

await sessionReceiver.close();

// Move to dead-letter
await receiver.deadLetterMessage(message, {
 deadLetterReason: "Validation failed",
 deadLetterErrorDescription: "Missing required field: orderId",
});

// Process dead-letter queue
const dlqReceiver = client.createReceiver("my-queue", { subQueueType: "deadLetter" });
const dlqMessages = await dlqReceiver.receiveMessages(10);
for (const msg of dlqMessages) {
 console.log(`DLQ Reason: ${msg.deadLetterReason}`);
 // Reprocess or log
 await dlqReceiver.completeMessage(msg);
}

const sender = client.createSender("my-queue");

// Schedule for future delivery
const scheduledTime = new Date(Date.now() + 60000); // 1 minute from now
const sequenceNumber = await sender.scheduleMessages(
 { body: "Delayed message" },
 scheduledTime
);

// Cancel scheduled message
await sender.cancelScheduledMessages(sequenceNumber);

// Defer message for later
await receiver.deferMessage(message);

// Receive deferred message by sequence number
const deferredMessage = await receiver.receiveDeferredMessages(message.sequenceNumber!);
await receiver.completeMessage(deferredMessage[0]);

const receiver = client.createReceiver("my-queue");

// Peek without removing
const peekedMessages = await receiver.peekMessages(10);
for (const msg of peekedMessages) {
 console.log(`Peeked: ${msg.body}`);
}

import {
 ServiceBusClient,
 ServiceBusSender,
 ServiceBusReceiver,
 ServiceBusSessionReceiver,
 ServiceBusMessage,
 ServiceBusReceivedMessage,
 ProcessMessageCallback,
 ProcessErrorCallback,
} from "@azure/service-bus";

// Peek-Lock (default) - message locked until completed/abandoned
const receiver = client.createReceiver("my-queue", { receiveMode: "peekLock" });
await receiver.completeMessage(message); // Remove from queue
await receiver.abandonMessage(message); // Return to queue
await receiver.deferMessage(message); // Defer for later
await receiver.deadLetterMessage(message); // Move to DLQ

// Receive-and-Delete - message removed immediately
const receiver = client.createReceiver("my-queue", { receiveMode: "receiveAndDelete" });