Applying Correlation filters for targeted Service Bus messaging

To apply correlation filters for targeted messaging in Azure Service Bus, you'll need to create a Rule on a Subscription that filters messages sent to a Topic based on predefined conditions. A CorrelationFilter holds the properties against which the messages are matched.

Here's what we'll do:

1. We'll define an Azure Service Bus Topic. This is a publish-subscribe pattern where senders (publishers) send messages that are not directly sent to queues. Instead, they are sent to a Topic.
2. Next, we define a Subscription to the Topic. This allows receivers (subscribers) to receive messages. Subscribers can define rules on these subscriptions to filter messages.
3. We'll then create a Rule with a CorrelationFilter. The CorrelationFilter is a set of conditions that the Azure Service Bus service applies to the Message headers to determine if they are to be delivered to the Subscription.

Here is the TypeScript program that demonstrates these steps using Pulumi:

import * as pulumi from "@pulumi/pulumi";
import * as azure_native from "@pulumi/azure-native";

// Configuration for the Azure Service Bus namespace
const namespaceName = "myNamespace";
const topicName = "myTopic";
const subscriptionName = "mySubscription";
const ruleName = "myCorrelationRule";
const resourceGroupName = "myResourceGroup";

// Create an Azure Service Bus Namespace
const serviceBusNamespace = new azure_native.servicebus.Namespace(namespaceName, {
    namespaceName: namespaceName,
    resourceGroupName: resourceGroupName,
    sku: {
        name: "Standard", // Standard SKU for the namespace allows for Topics
    },
});

// Create a Service Bus Topic within the namespace
const topic = new azure_native.servicebus.Topic(topicName, {
    namespaceName: namespaceName,
    topicName: topicName,
    resourceGroupName: resourceGroupName,
}, { dependsOn: serviceBusNamespace }); // Ensure the namespace is created before the topic

// Create a Subscription to the Topic
const topicSubscription = new azure_native.servicebus.Subscription(subscriptionName, {
    subscriptionName: subscriptionName,
    topicName: topic.name,
    namespaceName: namespaceName,
    resourceGroupName: resourceGroupName,
}, { dependsOn: topic }); // Ensure the topic is created before the subscription

// Create a Rule with a CorrelationFilter
const ruleWithCorrelationFilter = new azure_native.servicebus.Rule(ruleName, {
    ruleName: ruleName,
    topicName: topic.name,
    subscriptionName: topicSubscription.name,
    namespaceName: namespaceName,
    resourceGroupName: resourceGroupName,
    // Here we specify the correlation filter properties
    correlationFilter: {
        contentType: "application/json",
        label: "important",
        // You can add more properties as needed
    },
}, { dependsOn: topicSubscription }); // Ensure the subscription is created before the rule

// Export the primary connection string to access the topic
const namespaceAuthRule = serviceBusNamespace.name.apply(namespace => 
    azure_native.servicebus.listNamespaceKeys({
        namespaceName: namespace,
        resourceGroupName: resourceGroupName,
        authorizationRuleName: "RootManageSharedAccessKey",
    })
);

export const primaryConnectionString = namespaceAuthRule.apply(keys => keys.primaryConnectionString);

Explanation:

• We use the @pulumi/azure-native package which provides native Azure resources.

• First, we create a Namespace since the Service Bus Topic, Subscription, and Rule must reside inside a namespace.

• Then, we define a Topic, which is the destination to which messages are sent.

• Next, we create a Subscription to this Topic. This subscription is essentially a "virtual queue" that receives copies of the messages sent to the topic.

• After creating the Subscription, we define a Rule with a CorrelationFilter. This filter specifies the properties a message must have to be selected for the subscription. In this case, we are specifying the contentType as application/json and the label as important.

• Finally, we export the primaryConnectionString which can be used to authorize clients to send and receive messages to and from the Service Bus Topic.

Deploying the Code:

Ensure you have Pulumi installed and configured with the appropriate access to your Azure account. You can then run the following commands:

pulumi up

This will prompt you with a summary of the resources to be created, modified, or deleted. Confirm the prompt, and Pulumi will proceed to apply your infrastructure code.

Note: Don't forget to replace "myNamespace", "myTopic", "mySubscription", "myCorrelationRule", and "myResourceGroup" with your own names. Also, note that azure_native.servicebus.listNamespaceKeys function is used to retrieve the Service Bus namespace's keys, which requires the RootManageSharedAccessKey that's typically created with a new Service Bus namespace. Adjust the authorization rule name if necessary.