Anixe.IO.Messaging.AspCore 4.0.0

Anixe.IO.Messaging.AspCore

Anixe.IO.Messaging.AspCore allows for quick integration of ASP.Core app with Anixe Messaging backed by RabbitMQ.

How to integrate

Add in csproj as a package reference

<PackageReference Include="Anixe.IO.Messaging.AspCore" Version="1.1.0" />

Register implementaion of IMessageFormatter, default formatter can be found in Anixe.IO.Messaging.Serializer

<PackageReference Include="Anixe.IO.Messaging.Serializer" Version="1.0.0" />

Producer config

Add section to appsettings.json

{
  ...,
  "Notifications": {
    "ProducerConfiguration": {
      "RabbitMqConfiguration": {
        "Url": "amqp://guest:guest@localhost:5672/",
        "AppName": "app_name"
      },
      "ExchangeConfiguration": {
        "Name": "example.topic",
        "Type": "topic",
        "Durable" : true, // or false, the exchange will survive a broker restart
        "ExchangeArguments": {  } // custom arguments passed to the exchange declaration
      },
      "TenantFieldName": "tenant",
      "Environment": "dev",
      "AppName": "app_name",
      "ContentEncoding": "gzip", // it will automatically gzip all messages.
      "RetryPolicy": {
        "Enabled": true,
        "RetryCount": 3,
        "SleepDurationMs": 200
      }
    }
  }
}

Register services in Startup.cs

public void ConfigureServices(IServiceCollection services)
{
    services.ConfigureProducerService(
        this.config, cfg => {
            // override NotificationsProducerConfiguration here
        }
    );
}

then in you business logic use INotificationService, provided by DI container

using Anixe.IO.Messaging.AspCore;

this.notificationService.SendNotification("routing.key", msg =>
    {
        msg.EntityId = "123";
        msg.EntityType = "entityType";
        msg.EntityUrl = "appName/entityType/123";
        msg.Data = new { id = 123 };
        // if operation is not provided, then based on REST convention by HTTP metod
        msg.Operation = Anixe.IO.Messaging.Core.OperationType.Create;
    });

In case of publish operation fail, exception is thrown.

Subscriber config

Add section to appsettings.json

{
  ...,
  "Notifications": {
    "SubscriberConfiguration": {
      "RabbitMqConfiguration": {
        "Url": "amqp://guest:guest@localhost:5672/",
        "AppName": "app_name",
        "Async": true // optional, to receive messages via AsyncEventingBasicConsumer
      },
      "QueueConfiguration": {
        "QueueName": "", // unnamed queue
        "Durable" : true, // or false, the queue will survive a broker restart
        "Exclusive" : true, // or false, queue that has had at least one consumer is deleted when last consumer unsubscribes
        "AutoDelete": true,  // or false, Optional, used by plugins and broker-specific features such as message TTL, queue length limit, etc
        "Bindings": [
          {
            "Exchange": "distro.topic",
            "RoutingKeys": ["tenant.*.resource.sales_profile"],
            "ExchangeArguments": {  } // custom arguments passed to the exchange declaration
          }
        ]
      },

    }
  }
}

Also build-in rabbitmq channels can be setup

{
  ...,
  "Notifications": {
    "SubscriberConfiguration": {
      "RabbitMqConfiguration": {
        "Url": "amqp://guest:guest@localhost:5672/",
        "AppName": "app_name",
        "Async": true // optional, to receive messages via AsyncEventingBasicConsumer
      },
      "QueueConfiguration": {
        "QueueName": "", // unnamed queue
        "Bindings": [
          {
            "Exchange": "amq.fanout",
            "ExchangeType": "fanout",
            "RoutingKeys": []
          }
        ]
      }
    }
  }
}

Register services in Startup.cs.

Each ConfigureServices registers the following dependencies:

• ISubscriberConnection singleton instance
• ListenerHostedService hosted service
• ISubscriberReloader singleton instance

The following subscrubtion scenarios are possible:

Register single subscription for common NotificationMessage model

public void ConfigureServices(IServiceCollection services)
{
  services.ConfigureSubscriberService(this.config);
  services.AddSingleton<IMessageReceiver<NotificationMessage>, MyNotificationHandler>(); // handle in my application
}

Optinally provide override configuration:

public void ConfigureServices(IServiceCollection services)
{
  services.ConfigureSubscriberService(this.config, opts => {
    opts.RabbitMqConfiguration.Url = "amqp://example.com";
  });
  services.AddSingleton<IMessageReceiver<NotificationMessage>, MyNotificationHandler>(); // handle in my application
}

Register single subscription for generic NotificationMessage<Geolocation> model

public void ConfigureServices(IServiceCollection services)
{
  services.ConfigureSubscriberService<NotificationMessage<Geolocation>>(this.config);
  services.AddSingleton<IMessageReceiver<NotificationMessage<Geolocation>>, MyNotificationHandler>(); // handle in my application
}

Register single subscription for many models, each per AMQP routing key

This scenario is usefull when many instances of the same subscriber must receive exact copy of published message (classic pub-sub)

public void ConfigureServices(IServiceCollection services)
{
  services.AddSingleton<IMessageReceiver<NotificationMessage<Person>>, PersonModelHandler>();
  services.AddSingleton<IMessageReceiver<NotificationMessage<List<Child>>>, ChildrenModelHandler>();
  services.AddSingleton<IMessageReceiver<NotificationMessage<Child>>, ChildModelHandler>();

  services.ConfigureSubscriberService(this.config, (builder) => {
    builder
      .AddMessageHandler<NotificationMessage<Person>>($"env.*.tenant.*.resource.person")
      .AddMessageHandler<NotificationMessage<List<Child>>>($"env.*.tenant.*.resource.children")
      .AddMessageHandler<NotificationMessage<Child>>($"env.*.tenant.*.resource.child");
  });
}

or

public void ConfigureServices(IServiceCollection services)
{
  services.ConfigureSubscriberService(this.config, (builder) => {
    builder
      .AddMessageHandler<NotificationMessage<Person>, PersonModelHandler>($"env.*.tenant.*.resource.person")
      .AddMessageHandler<NotificationMessage<List<Child>>, ChildrenModelHandler>($"env.*.tenant.*.resource.children")
      .AddMessageHandler<NotificationMessage<Child>, ChildModelHandler>($"env.*.tenant.*.resource.child");
  });
}

Register single subscription for many models with named queue, each per AMQP routing key

This scenario is usefull when many instances of the same subscriber must receive only one published message in round-robin manner. Here all instances of this subscriber will share the same queue queue1, so rabbitmq will distribute message in round-robin manner.

public void ConfigureServices(IServiceCollection services)
{
  services.ConfigureSubscriberService(this.config, (builder) => {
    builder
      .AddMessageHandler<NotificationMessage<Person>, PersonModelHandler>("queue1", $"env.*.tenant.*.resource.person")
      .AddMessageHandler<NotificationMessage<List<Child>>, ChildrenModelHandler>("queue1", $"env.*.tenant.*.resource.children")
      .AddMessageHandler<NotificationMessage<Child>, ChildModelHandler>("queue1", $"env.*.tenant.*.resource.child");
  });
}

Register many subscriptions for many models, each per AMQP routing key

public void ConfigureServices(IServiceCollection services)
{
  var geolinkSection = config.GetSection("GeolinkImporterNotifications");

  services.ConfigureSubscriberService<GeoLocation>(rescSection);
  services.AddSingleton<IMessageReceiver<GeoLocation>, GeoLocationNotificationHandler>();

  var rescSection = config.GetSection("RESCNotifications");
  services.ConfigureSubscriberService(this.config, (builder) => {
    builder
      .AddMessageHandler<NotificationMessage<Person>, PersonModelHandler>($"env.*.tenant.*.resource.person")
      .AddMessageHandler<NotificationMessage<List<Child>>, ChildrenModelHandler>($"env.*.tenant.*.resource.children")
      .AddMessageHandler<NotificationMessage<Child>, ChildModelHandler>($"env.*.tenant.*.resource.child");
  });
}

Routing keys from AddMessageHandler will replace these from appsettings.json

⚠️ Configuration field Notifications:SubscriberConfiguration:QueueConfiguration:QueueName must remain empty. Setting it can lead to unexpected behaviour. If you want named queues, add the names as parameters of .AddMessageHandler("my_queue_name", "my_routing_key"); and remember that the queue names must be distinct.

Register custom connection

public void ConfigureServices(IServiceCollection services)
{
  var conn = new MyCustomSubscriberConnection();
  services.ConfigureSubscriberService(this.config, (builder) => {
    builder
      .AddCustomConnection(conn)
      /*chain other registrations*/
      ;
  });
}

Register NewtonsoftJson deserializer

Add Anixe.IO.Messaging.Serializer assembly to csproj then:

public void ConfigureServices(IServiceCollection services)
{
  var conn = new MyCustomSubscriberConnection();
  services.ConfigureSubscriberService(this.config, (builder) => {
    builder
      .AddCustomFormatter<NewtonsoftMessageFormatter>()
      /*chain other registrations*/
      ;
  });
}

Deserializing notification messages

By default Anixe.IO.Messaging provides NotificationMessage and NotificationMessage<T> build-in classes which implelement Resfinity Entity Consistent System convention. These classes are envelopes for Data property which can provide custom data:

• In NotificationMessage the Data is object and deserializes into .Net primitives like Dictionary<string, object>, List<T>, string and value typed
• In NotificationMessage<T> the Data is instance of T and and deserializes into generic model

But you don't have to use NotificationMessage class, you can provide your own type.

Check src/Anixe.IO.Messaging.Test/Core/MessageFormatterTest.cs for serialization/deserialization examples

Example for NotificationMessage registration (implicit)

public void ConfigureServices(IServiceCollection services)
{
  services.ConfigureSubscriberService(this.config); // receive and deserialize
  services.AddSingleton<IMessageReceiver<NotificationMessage>, MyNotificationHandler>(); // handle in my application
}

Example for NotificationMessage<T> registration

public void ConfigureServices(IServiceCollection services)
{
  services.ConfigureSubscriberService<NotificationMessage<Person>>(this.config); // receive and deserialize
  services.AddSingleton<IMessageReceiver<NotificationMessage<Person>>, MyNotificationHandler>(); // handle in my application
}

Example for any model registration

public void ConfigureServices(IServiceCollection services)
{
  services.ConfigureSubscriberService<GeoLocation>(this.config); // receive and deserialize
  services.AddSingleton<IMessageReceiver<GeoLocation>, GeoLocationNotificationHandler>(); // handle in my application
}