Service Bus: Topic Filters

Overview

Domain events published to topics and often used for microservice integrations. The publisher defines the topic to which a given domain-event will be published. However, the publisher does not know of all the consumers of the topic. The consumers, knowing the topic name, defines subscriptions to which events will be received matching a filer. If a consumer is not executing, the messages will be queued and delivered the next time the subscriber starts.

Define Domain-Event

Add the following domain-event class to the Examples.ServiceBus.Domain/Events directory:

using NetFusion.Messaging.Types;

namespace Examples.ServiceBus.Domain.Events;

public class AutoSaleCompleted : DomainEvent
{
    public string Make { get; }
    public string Model { get; }
    public int Year { get; }
    public string? Color { get; init; }
    public bool? IsNew { get; init; }

    public AutoSaleCompleted(string make, string model, int year)
    {
        Make = make;
        Model = model;
        Year = year;
    }
}

Publishing Microservice

Add the following code to the ServiceBusRouter OnDefineEntities method to specify that a AutoSalesCompleted domain-event should be delivered to the CompletedAutoSales topic when published:

DefineTopic<AutoSaleCompleted>(topic =>
{
    topic.TopicName = "CompletedAutoSales";
    topic.SetMessageProperties((m, e) => m.ApplicationProperties["Make"] = e.Make);
});

The above routing completes the following:

• Creates a topic named CompletedAutoSales
• Sets the properties on the service-bus message that can be used in filters specified by subscribers

Subscribing Microservice

The subscribing microservice defines a route, specifying the method to be called, when a domain-event is published to the topic. Define the following handler:

using System;
using Examples.ServiceBus.Domain.Events;
using NetFusion.Common.Extensions;

namespace Examples.ServiceBus.App.Handlers;

public class AutoSalesHandler
{
    public void OnGermanAutoSales(AutoSaleCompleted completedSale)
    {
        Console.WriteLine(nameof(OnGermanAutoSales));
        Console.WriteLine(completedSale.ToIndentedJson());
    }
    
    public void OnAmericanAutoSales(AutoSaleCompleted completedSale)
    {
        Console.WriteLine(nameof(OnAmericanAutoSales));
        Console.WriteLine(completedSale.ToIndentedJson());
    }
}

Define the following routes to dispatch the received domain-events to the above handler methods when received on a specific subscription:

SubscribeToTopic<AutoSaleCompleted>("CompletedAutoSales", route =>
{
    route.ToConsumer<AutoSalesHandler>(c => c.OnAmericanAutoSales, meta =>
    {
        meta.SubscriptionName = "AmericanSales";
        meta.AddRule("AmericanMakeFiler", new SqlRuleFilter("Make IN  ('Ford', 'Buick')"));
    });
});

SubscribeToTopic<AutoSaleCompleted>("CompletedAutoSales", route =>
{
    route.ToConsumer<AutoSalesHandler>(c => c.OnGermanAutoSales, meta =>
    {
        meta.SubscriptionName = "GermanSales";
        meta.AddRule("GermanMakeFiler", new SqlRuleFilter("Make IN  ('VW', 'BMW')"));
    });
});

The above completes the following:

• Creates a subscription named AmericanSales on the CompletedAutoSales topic
• Specifies that only messages where the make is Ford or Buick should be sent to the subscription
• Creates a subscription named GermanSales on the CompletedAutoSales topic
• Specifies that only messages where the make is VW or BMW should be send to the subscription

Define Api Model

Add a corresponding model to the following directory: Examples.ServiceBus.WebApi/Models

using System.ComponentModel.DataAnnotations;

namespace Examples.ServiceBus.WebApi.Models;

public class AutoSalesModel
{
    [Required] public string Make { get; set; } = string.Empty;
    [Required] public string Model { get; set; } = string.Empty;
    [Required] public string Color { get; set; } = string.Empty;
    public int Year { get; set; } 
    public bool IsNew { get; set; }
}

Define Api Controller

[HttpPost("auto/sales")]
public async Task<IActionResult> AutoSales(AutoSalesModel model)
{
    if (!ModelState.IsValid)
    {
        return BadRequest(ModelState);
    }

    var domainEvent = new AutoSaleCompleted(model.Make, model.Model, model.Year)
    {
        Color = model.Color,
        IsNew = model.IsNew
    };

    await _messaging.PublishAsync(domainEvent);
    return Ok();
}

Execute Example

Complete the following to run the example microservice and send a HTTP Post request to the example controller:

Execute Microservice

cd ./Examples.ServiceBus/src/Examples.ServiceBus.WebApi/
dotnet run

Send Requests

Post the following requests to: http://localhost:5000/api/example/auto/sales

{
    "make": "BMW",
    "model": "325",
    "year": 2005,
    "color": "White",
    "isNew": true
}

{
    "make": "Ford",
    "model": "F150",
    "year": 2020,
    "color": "Silver",
    "isNew": true
}

Service Bus Configurations

Since the IsAutoCreateEnabled configuration property was set to true within the Program.cs file when bootstrapping the microservice, the specified topic and subscriptions are automatically created: