integration servicebus topics

Service Bus: Topics

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 an optional 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.RabbitMQ.Domain/Events directory:

using NetFusion.Messaging.Types;

namespace Examples.ServiceBus.Domain.Events;

public class PropertySold : DomainEvent
{
    public string Address { get; }
    public string City { get; }
    public string State { get; }
    public string Zip { get; }
    public decimal AskingPrice { get; init; }
    public decimal SoldPrice { get; init; }

    public PropertySold(string address, string city, string state, string zip)
    {
        Address = address;
        City = city;
        State = state;
        Zip = zip;
    }
}

Publishing Microservice

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

DefineTopic<PropertySold>(topic =>
{
    topic.TopicName = "RealEstate";
    topic.WhenDomainEvent(e => e.Zip != "15068");
});

The above routing completes the following:

• Creates a topic named RealEstate
• Specifies a predicate so any published domain-event with a zip code of 15068 will not be delivered to the topic.

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 System.Threading;
using System.Threading.Tasks;
using Examples.ServiceBus.Domain.Events;
using NetFusion.Common.Extensions;

namespace Examples.ServiceBus.App.Handlers;

public class RealEstateHandler
{
    public async Task OnPropertySold(PropertySold domainEvent, CancellationToken token)
    {
        Console.WriteLine(nameof(OnPropertySold));
        
        await Task.Delay(TimeSpan.FromMilliseconds(1), token);
        Console.WriteLine(domainEvent.ToIndentedJson());
        
        token.ThrowIfCancellationRequested();
    }
}

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

SubscribeToTopic<PropertySold>("RealEstate", route =>
{
    route.ToConsumer<RealEstateHandler>(c => c.OnPropertySold, meta =>
    {
        meta.SubscriptionName = "SoldProperties";
    });
});

Define Api Model

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

using System.ComponentModel.DataAnnotations;

namespace Examples.ServiceBus.Infra.Plugin.Modules;

public class PropertySoldModel
{
    [Required] public string Address { get; set; } = string.Empty;
    [Required] public string City { get; set; } = string.Empty;
    [Required] public string State { get; set; } = string.Empty;
    [Required] public string Zip { get; set; } = string.Empty;
    
    public decimal AskingPrice { get; init; }
    public decimal SoldPrice { get; init; }
}

Define Api Controller

namespace Examples.ServiceBus.WebApi.Controllers;

[ApiController, Route("api/[controller]")]
public class ExampleController : ControllerBase
{
    private readonly IMessagingService _messaging;
    
    public ExampleController(
        IMessagingService messaging, )
    {
        _messaging = messaging;
    }
    
    [HttpPost("properties/sold")]
    public async Task<IActionResult> PropertySold(PropertySoldModel model)
    {
        if (!ModelState.IsValid)
        {
            return BadRequest(ModelState);
        }

        var domainEvent = new PropertySold(model.Address, model.City, model.State, model.Zip)
        {
            AskingPrice = model.AskingPrice,
            SoldPrice = model.SoldPrice
        };

        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/properties/sold

{
    "address": "444 Main Street",
    "city": "Cheshire",
    "state": "CT",
    "zip": "06410",
    "askingPrice": 560000,
    "soldPrice": 555000
}

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 subscription are automatically created: