DEPRECATED 
This repo is no longer supported and will not receive any updates nor bug fixes, This code has been moved to the main Liquid repository and will be maintained there.
Liquid Application Framework - Messaging
This repository is part of the Liquid Application Framework, a modern Dotnet Core Application Framework for building cloud native microservices.
The main repository contains the examples and documentation on how to use Liquid.
Liquid Messaging
This package contains the messaging subsystem of Liquid, for interacting with assynchronous services. It also features several implementations. In order to use it, add the main package (Liquid.Messaging) to your project, along with the specific implementation that you will need.
| Available Cartridges | Badges |
|---|---|
| Liquid.Messaging.Aws | |
| Liquid.Messaging.ServiceBus | |
| Liquid.Messaging.Gcp | |
| Liquid.Messaging.Kafka | |
| Liquid.Messaging.RabbitMq |
Getting Started
This is a sample usage with Azure Service Bus cartridge
To publish a message, you must:
Register a producer in service provider
//the value of the parameter must be the name of the appsettings section where the configuration of your producer is defined.
services.AddServiceBusProducer<SampleMessageEntity>("Liquid:Messaging:ServiceBus:SampleProducer");Include dependency in domain class constructor, and invoke this send method
using Liquid.Messaging;public class MySampleProducer
{
//Dependency must be a ILiquidProducer implementation
private ILiquidProducer<MySampleMessage> _producer;
public MySampleProducer(ILiquidProducer<MySampleMessage> producer)
{
_producer = producer;
}
public async Task Handle()
{
// Create a instance of message object defined as the type of the producer
var message = new MySampleMessage();
// Just invoke the Send method
await _producer.SendMessageAsync(message, new Dictionary<string, object> { { "headerTest", "value" } });
}
}To consume messages, implement LiquidWorker:
public class Worker : ILiquidWorker<SampleMessageEntity>
{
private readonly IMediator _mediator;
public Worker(IMediator mediator)
{
_mediator = mediator;
}
//the ProcessMessageAsync method is invoked when a message arrives
public async Task ProcessMessageAsync(ProcessMessageEventArgs<SampleMessageEntity> args, CancellationToken cancellationToken)
{
// Do what you need with the message, usually translate to a command and send it to the Mediator Service
await _mediator.Send(new PutCommandRequest(args.Data));
}
}The dependency injection method register the LiquidBackgroundService{TMessage}, that is the Liquid implementation of IHostedService and work as the host of Consumer.
//the value of the first parameter must be the name of the appsettings section where the configuration of your consumer
//is defined, and second parameter is an array of assemblies where domain handlers are defined.
services.AddLiquidServiceBusConsumer<Worker, SampleMessageEntity>("Liquid:Messaging:ServiceBus:SampleConsumer", typeof(PutCommandRequest).Assembly);Configurations sample:
"liquid": {
"messaging": {
"serviceBus": {
"sampleProducer": {
"ConnectionString": "",
"EntityPath": "sampleentity"
},
"sampleConsumer": {
"ConnectionString": "",
"EntityPath": "samplemessage"
}
}
}
}
