azure-service-bus.ts

import { Injectable } from '@nestjs/common';
import { ReceiveMode, ServiceBusClient } from '@azure/service-bus';

@Injectable()
export class ServiceBusService {
  constructor() {
    this.init();
  }

  async init() {
    const serviceBusClient = ServiceBusClient.createFromConnectionString(
      'xxxxxxxxxxxxx',
    );
    const queueClient = serviceBusClient.createQueueClient('myqueue1');
    const sender = queueClient.createSender();

    const receiver = queueClient.createReceiver(ReceiveMode.peekLock, { sessionId: 'my-session' });
    receiver.registerMessageHandler(
      async (message) => {
        console.log(message.deliveryCount);
        console.log(message.body);
      },
      (error) => {},
    );

    await sender.send({
      sessionId: 'my-session',
      body: {
        id: Date.now().toString(),
      },
    });

    const topicClient = serviceBusClient.createTopicClient('mytopic1');
    const topicSender = topicClient.createSender();
    await topicSender.send({
      sessionId: 'session1',
      body: {
        id: Date.now().toString(),
      },
    });

    // need to save ref to database: so we can remove the message if required
    const ref = await topicSender.scheduleMessage(new Date(Date.now() + 1000 * 600), {
      sessionId: 'session1',
      body: {
        id: Date.now().toString(),
      },
    });

    const topicSubNzClient = serviceBusClient.createSubscriptionClient('mytopic1', 'newzealand');

    const topicSubThaiClient = serviceBusClient.createSubscriptionClient('mytopic1', 'thailand');
    const topicSubAusClient = serviceBusClient.createSubscriptionClient('mytopic1', 'australia');
    const topicNzReceiver = topicSubNzClient.createReceiver(ReceiveMode.peekLock, { sessionId: 'session1' });
    topicNzReceiver.registerMessageHandler(
      async (message) => {
        console.log('nz sub');
        console.log(message.body);
      },
      (error) => {},
    );
    const topicThaiReceiver = topicSubThaiClient.createReceiver(ReceiveMode.peekLock, { sessionId: 'session1' });
    topicThaiReceiver.registerMessageHandler(
      async (message) => {
        console.log('thai sub');
        console.log(message.body);
      },
      (error) => {},
    );

    const topicAusReceiver = topicSubAusClient.createReceiver(ReceiveMode.peekLock, { sessionId: 'session1' });
    topicAusReceiver.registerMessageHandler(
      async (message) => {
        console.log('aus sub');
        console.log(message.body);
      },
      (error) => {},
    );

    await topicAusReceiver.receiveMessages(10).then(console.log)

    await topicSender.send({
      body: {
        id: Date.now().toString(),
      },
    });
  }
}