1import { ServiceBusClient, ServiceBusMessage } from '@azure/service-bus';
2
3const sbClient = new ServiceBusClient(process.env.SERVICE_BUS_CONNECTION!);
4
5// Producer — send messages to a queue
6async function sendOrder(order: { id: string; total: number; items: string[] }) {
7 const sender = sbClient.createSender('orders');
8 try {
9 const message: ServiceBusMessage = {
10 body: order,
11 messageId: order.id, // Deduplication key
12 sessionId: order.id, // Session for ordered processing
13 contentType: 'application/json',
14 };
15 await sender.sendMessages(message);
16 console.log(`Sent order ${order.id}`);
17 } finally {
18 await sender.close();
19 }
20}
21
22// Consumer — process messages with sessions
23async function startOrderProcessor() {
24 const receiver = sbClient.createReceiver('orders', {
25 receiveMode: 'peekLock', // Lock message while processing
26 });
27
28 receiver.subscribe({
29 processMessage: async (msg) => {
30 const order = msg.body;
31 console.log(`Processing: ${order.id}`);
32 await processOrder(order);
33 await receiver.completeMessage(msg); // Remove from queue
34 },
35 processError: async (err) => {
36 console.error('Receiver error:', err);
37 },
38 });
39}
40
41// Dead-letter queue processor for failed messages
42async function processDeadLetters() {
43 const dlqReceiver = sbClient.createReceiver('orders', {
44 subQueueType: 'deadLetter',
45 receiveMode: 'peekLock',
46 });
47 const messages = await dlqReceiver.receiveMessages(10, { maxWaitTimeInMs: 5000 });
48 for (const msg of messages) {
49 console.log(`DLQ: ${msg.deadLetterReason} — ${msg.messageId}`);
50 await dlqReceiver.completeMessage(msg);
51 }
52}
53
54async function processOrder(order: any) { /* business logic */ }
