Kit SDK Azure Service Bus pour TypeScript

Messagerie d'entreprise avec files d'attente, rubriques et abonnements.

Installation

npm install @azure/service-bus @azure/identity

Variables d'environnement

SERVICEBUS_NAMESPACE=<namespace>.servicebus.windows.net
SERVICEBUS_QUEUE_NAME=my-queue
SERVICEBUS_TOPIC_NAME=my-topic
SERVICEBUS_SUBSCRIPTION_NAME=my-subscription
AZURE_TOKEN_CREDENTIALS=prod # Requis uniquement si DefaultAzureCredential est utilisé en production

Authentification

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

// Dev local : DefaultAzureCredential. Production : définir AZURE_TOKEN_CREDENTIALS=prod ou AZURE_TOKEN_CREDENTIALS=<specific_credential>
const credential = new DefaultAzureCredential({requiredEnvVars: ["AZURE_TOKEN_CREDENTIALS"]});
// Ou utiliser une authentification spécifique directement en production :
// Voir 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);

Flux de travail principal

Envoyer des messages à une file d'attente

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

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

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

await sender.close();

Recevoir des messages d'une file d'attente

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

// Recevoir par lot
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();

S'abonner aux messages (basé sur les événements)

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

const subscription = receiver.subscribe({
  processMessage: async (message) => {
    console.log(`Processing: ${message.body}`);
    // Message auto-complété en cas de succès
  },
  processError: async (args) => {
    console.error(`Error: ${args.error}`);
  },
});

// S'arrêter après un certain temps
setTimeout(async () => {
  await subscription.close();
  await receiver.close();
}, 60000);

Rubriques et abonnements

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

// Recevoir d'un abonnement
const subscriptionReceiver = client.createReceiver("my-topic", "my-subscription");
const messages = await subscriptionReceiver.receiveMessages(10);

Sessions de messages

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

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

// Obtenir/définir l'état de la session
const state = await sessionReceiver.getSessionState();
await sessionReceiver.setSessionState(Buffer.from(JSON.stringify({ progress: 50 })));

await sessionReceiver.close();

Gestion des messages non livrables

// Déplacer vers la file non livrable
await receiver.deadLetterMessage(message, {
  deadLetterReason: "Validation failed",
  deadLetterErrorDescription: "Missing required field: orderId",
});

// Traiter la file non livrable
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}`);
  // Retraiter ou enregistrer
  await dlqReceiver.completeMessage(msg);
}

Messages programmés

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

// Programmer pour une livraison future
const scheduledTime = new Date(Date.now() + 60000); // 1 minute à partir de maintenant
const sequenceNumber = await sender.scheduleMessages(
  { body: "Delayed message" },
  scheduledTime
);

// Annuler un message programmé
await sender.cancelScheduledMessages(sequenceNumber);

Report de messages

// Reporter un message pour plus tard
await receiver.deferMessage(message);

// Recevoir un message reporté par numéro de séquence
const deferredMessage = await receiver.receiveDeferredMessages(message.sequenceNumber!);
await receiver.completeMessage(deferredMessage[0]);

Consulter les messages (non-destructif)

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

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

Types clés

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

Modes de réception

// Peek-Lock (par défaut) - message verrouillé jusqu'à complétion/abandon
const receiver = client.createReceiver("my-queue", { receiveMode: "peekLock" });
await receiver.completeMessage(message);   // Supprimer de la file
await receiver.abandonMessage(message);    // Retourner à la file
await receiver.deferMessage(message);      // Reporter pour plus tard
await receiver.deadLetterMessage(message); // Déplacer vers DLQ

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

Bonnes pratiques

1. Utiliser Microsoft Entra Token Credential - Utiliser DefaultAzureCredential pour le développement local ; utiliser ManagedIdentityCredential ou WorkloadIdentityCredential pour la production
2. Réutiliser les clients - Créer ServiceBusClient une seule fois, partager entre les envoyeurs/récepteurs
3. Fermer les ressources - Toujours fermer les envoyeurs/récepteurs après utilisation
4. Gérer les erreurs - Implémenter le callback processError pour les récepteurs d'abonnement
5. Utiliser les sessions pour l'ordonnancement - Quand l'ordre des messages importe au sein d'un groupe
6. Configurer la file non livrable - Toujours traiter les messages de la DLQ
7. Envoyer par lots - Utiliser createMessageBatch() pour plusieurs messages

Documentation de référence

Pour des modèles détaillés, voir :

• Modèles Files vs Rubriques - Modèles de files/rubriques, sessions, modes de réception, règlement des messages
• Gestion des erreurs et fiabilité - Codes ServiceBusError, gestion DLQ, renouvellement de verrous, arrêt gracieux