Contour is the message bus implementation which provides communication between .NET services via different transport protocols.
Contour builds all the topology (exchanges, queues with all properties set up) for you. At this moment it supports only AMQP/RabbitMQ.
Configuration can be set in .config file, declaring the 'endpoints' section inside the 'serviceBus' group:
<sectionGroup name="serviceBus">
<section name="endpoints" type="Contour.Configurator.EndpointsSection, Contour" />
</sectionGroup>Each endpoint represents separate service bus instance using which messages are sent and / or received. For example, if application has two endpoints, then from service bus point of view it looks like two different participants (producer or / and consumer). And vice versa, if two different applications use the same endpoint name, then it’s like one interaction participant for service bus (when default routing is used), so you can create concurrent consumers.
Endpoints configuration is defined in 'serviceBus' section:
<serviceBus>
<endpoints>
<endpoint name="point1" connectionString="amqp://localhost:5672/ ">
<!-- ... -->
</endpoint>
<endpoint name="point2" connectionString="amqp://localhost:5672/ ">
<!-- ... -->
</endpoint>
</endpoints>
</serviceBus>It is possible to define the configuration for multiple endpoints. Each endpoint has mandatory parameters: unique name and connection string to the broker.
Additionally, we can pass component name of IBusLifecycleHandler type in the attribute 'lifecycleHandler', which will be invoked when the state of service bus client is changed. For example, when it starts or stops.
<serviceBus>
<endpoints>
<endpoint name="point1" connectionString="amqp://localhost:5672/" lifecycleHandler="SpecialLifecycleHandler">
<!-- ... -->
</endpoint>
</endpoints>
</serviceBus>It may be helpful, if your component needs to know when service bus was started or stopped.
Every message, that failed to be processed, is added to the fault queue. Fault queue has the same name as the endpoint with additional '.Fault' postfix. faultQueueTtl attribute controls TTL of fault messages in such queues. faultQueueLimit attribute limits maximum number of messages. If number of fault messages reaches this limit, old messages will be substituted.
<serviceBus>
<endpoints>
<endpoint name="point1" connectionString="amqp://localhost:5672/" faultQueueTtl="3.00:00:00" faultQueueLimit="500">
<!-- ... -->
</endpoint>
</endpoints>
</serviceBus>Default value for TTL - 21 days, for maximum size of the fault queue - no limits.
All incoming messages are declared in the 'incoming' collection section. Each message label declares with an individual element 'on'. 'key' and 'label' attributes are mandatory. For each incoming message topology elements are created if they don't exist:
- exchange with name equal to label attribute value;
- queue with name "<endpoint name>.<label>"
- binding from exchange to queue.
Additionally, you can set the parameters (as attributes of the element 'on'):
- requiresAccept – event processing must be explicitly confirmed in the handler,otherwise message will be returned to the queue (at-least once delivery). False by default (at-most once delivery);
- react – message handler name (for example, the name of the registered handler in the IoC-container);
- type – CLR-type of the event body. Can be represented by fully qualified name. In this case, default mechanism of type searching is used. Or you can use a short name type. In this case, it will search for all loaded assemblies in the application domain.If type is not specified, ExpandoObject will be used by default.
- validate – message validator name (class that implements IMessageValidator), which operates only within a declared subscription;
- lifestyle – allows you to specify a handler lifestyle.
Lifestyle possible values:
- Normal – handler is requested, when you create a bus client (default value);
- Lazy – handler is requested once, when you get the first message;
- Delegated – handler is requested every time you get the message, allowing you to manage handler lifetime.
<incoming>
<on key="message" label="message.label" react="MessageHandler" />
<on key="request" label="request.label" react="RequestHandler" type="RequestPayload" validate="InputValidator" requiresAccept="true" />
<on key="lazy" label="lazy.label" react="LazyHandler" lifestyle="Lazy" />
</incoming>Each incoming message passes through global validators, so they can be optionally defined in 'validators' section. Each class of validator must implement IMessageValidator interface (for more convenience you can inherit from class FluentPayloadValidatorOf, which has already implemented this interface).
Also you can set a value of 'group' parameter to 'true'. This would mean that you don’t need to enumerate all validators in component, and just add the name of a validators group, which will include all classes, implementing interface IMessageValidator.
<endpoint name="point1" connectionString="amqp://localhost:5672/ ">
<validators>
<add name="NameValidator" group="true"/>
</validators>
<!-- ... -->
</endpoint>Tag 'caching' controls the caching of incoming Request/Reply messages
You can enable caching by setting attribute to ‘enabled’. At the requested side response is cached by the key, which is generated on the basis of the request body. Caching time is defined by the responder when calling 'Reply' method.
'Publisher/Subscriber' requests are not cached.
<endpoint name="point1" connectionString="amqp://localhost:5672/">
<caching enabled="true"/>
<!-- ... -->
</endpoint>Endpoint provides ability to setup the necessary number of handler threads, processing incoming messages, in 'parallelismLevel' attribute. Messages are distributed between the handlers by the Round-Robin principle.
<endpoints>
<endpoint name="Emitter" connectionString="amqp://localhost:5672/" parallelismLevel="4">
</endpoint>
</endpoints>By default, one handler for incoming messages is used.
Endpoint also provides ability to setup a certain number of incoming messages by one access to the broker. This field specifies the prefetch window size. When the handler finishes processing a message, the following message is already held locally, rather than needing to be sent down the channel. Prefetching gives a performance improvement.
<endpoints>
<endpoint name="Consumer" connectionString="amqp://localhost:5672/">
<qos prefetchCount="8" />
<incoming />
</endpoint>
</endpoints>By default, uses the value of 50 messages, which will be read by one access. We strongly recommend starting with 1 message, increasing the number of messages after performance testing.
All outgoing messages are declared in the 'outgoing' collection section. Each message label is declared with an individual tag 'route'. 'key' and 'label' attributes are mandatory.
Key is a label alias, which can be used from the code, and allows you to separate concrete label (exchange in RabbitMQ) of the message from its representation in code. Outgoing route may be referenced from the application
- by key, adding a colon (:) as a prefix.
- directly by label.
Additionally, you can set the parameters (as attributes of the 'route' element):
- persist – broker saves the message for reliable delivery (false by default);
- confirm – broker must confirm message receipt for the processing (message will be stored if persist="true", false by default);
- ttl – message lifetime (unlimited, if the value is not specified);
- timeout – response time for requests (default value – 30 seconds).
To support request/reply pattern, you must declare a subscription point for awaiting response messages. At the moment you can declare only default subscription point. For each request / reply route unique incoming queue per application is created.
<outgoing>
<route key="message" label="message.label" persist="true" ttl="00:01:00" />
<route key="request" label="request.label" confirm="true" timeout="00:00:50">
<callbackEndpoint default="true" />
</route>
</outgoing>On the sender’s side you can turn on dynamic routing. It allows you to send messages with labels, which were not configured during the endpoint creation.
Dynamic routing can be useful in cases, when you don’t have enough information about all the labels at the moment of endpoint creation.
<endpoints>
<endpoint name="point1" connectionString="amqp://localhost:5672/">
<dynamic outgoing="true" />
</endpoint>
</endpoints>AppConfigConfigurator class is used for applying xml-configuration.
Class constructor can take an object, implementing IDependencyResolver interface, which is used for getting bus client dependencies. It is used for searching message handlers, validators or specifying life cycle handler. If concrete client configuration doesn’t require external dependencies, this parameter can be omitted. For simplicity, instead of implementing a particular class, you can pass to the constructor delegate of DependencyResolverFunc type.
For example, typical creation of AppConfigConfigurator object using Ninject container:
var configurator = new AppConfigConfigurator((name, type) => kernel.Get(type, name));To apply configuration ‘Configure’ method is used, which must be invoked during the bus instance creation.
var bus = new BusFactory().Create(cfg =>
{
configurator.Configure("point1", cfg);
});Using this method you can combine configuration via code and configuration file.
If endpoints names are unknown (or should not be known) at compile time, they can be accessed through the Endpoints property. Example of creating and configuring all bus clients:
var busFactory = new BusFactory();
var busInstances = configurator
.Endpoints
.Select(e => busFactory.Create(cfg => configurator.Configure(e, cfg)))
.ToList();Endpoint can be completely configured via code except incoming message handler lifestyle.
To use dynamic routing set message label to MessageLabel.Any value.
IBus bus1 = new BusFactory().Create(
cfg =>
{
cfg.UseRabbitMq();
cfg.SetEndpoint("point1");
cfg.SetConnectionString("amqp://localhost:5672/ ");
cfg.HandleLifecycleWith(new SpecialLifecycleHandler());
cfg.RegisterValidator(new NameValidator());
cfg.EnableCaching();
cfg.UseParallelismLevel(4);
cfg.SetDefaultQoS(8);
cfg.Route("out.message.label")
.Persistently()
.WithTtl(new TimeSpan(0, 1, 0));
cfg.Route("out.request.label")
.WithRequestTimeout(new TimeSpan(0, 0, 50))
.WithConfirmation();
cfg.On("in.message.label")
.ReactWith(MessageHandler());
cfg.On<RequestPayload>("in.request.label")
.ReactWith(RequestHandler())
.WhenVerifiedBy(new InputValidator())
.RequiresAccept();
cfg.On<RequestPayload>("in.lazy.label")
.ReactWith(LazyHandler());
});
IBus bus2 = new BusFactory().Create(
cfg =>
{
cfg.UseRabbitMq();
cfg.SetEndpoint("point2");
cfg.SetConnectionString("amqp://localhost:5672/");
});List of used headers is represented in table below.
For message tracking in chain of application components interaction through service bus several incoming messages headers are copied in the outgoing messages.
| Header field name | Description | Copying |
|---|---|---|
| x-correlation-id | The correlation identifier is used to track a set of messages as a single group. It is used to match reply message with the request. New value for header is generated for every request (if not supplied in headers parameter). | Only for one-way messages. |
| x-expires | Header, which contains the rules of data deterioration. For example: x-expires: at 2016-04-01T22:00:33Z or x-expires: in 100 | No |
| x-message-type | Message label with which it was sent. Using this header is not recommended | No |
| x-persist | Marks message persistance (saved on disk) or not | No |
| x-reply-route | Reply message address for the request. For example, x-reply-route: direct:///amq.gen-n9DsUj1qm4vgCq0MHHPoBQ | No |
| x-timeout | Response timeout for the request | No |
| x-ttl | Message TTL | No |
| x-breadcrumbs | List of all endpoints, through which message has passed, separated by semicolon (;) | Yes (appending new value) |
| x-original-message-id | Message identifier, that started the messages exchange | Yes |
In Contour you can organize message processing as a set of sequential atomic message transformations based on Pipes and Filters template.
For example, you need to forward messages, that has field 'Tick' and its value is odd. Other messages should be filtered. It can be done with the following set of filters:
var configurator = new AppConfigConfigurator((name, type) => kernel.Get(type, name));
configuration.On("document.systemtick")
.ReactWith(new PipelineConsumerOf<ExpandoObject>(
new IMessageOperator[]
{
new JsonPathFilter("Tick"),
new Filter(request => ((dynamic)request.Payload).Tick % 2 == 1),
new StaticRouter("document.oddsystemtick")
}));Most of the filters (messages transformers) are universal and can be used for processing different types of messages.
Contour comprises the following set of the universal message filters:
- Acceptor,
- ContentBasedRouter,
- Filter,
- JsonPathFilter,
- RecipientList,
- Reply,
- Splitter,
- StaticRouter,
- Translator,
- TransparentReply,
- WireTap.
Confirms the message processing and forwards it to the next filter.
Routes each message to the correct recipient based on a message content (template in EIP).
Behavior is defined by the signature:
public ContentBasedRouter(Func<IMessage, MessageLabel> routeResolverFunc)
Filters messages by the predicate (template in EIP).
Behavior is defined by the signature:
public Filter(Predicate<IMessage> predicateFunc)Filter checks node availability at the specified JSONPath.
Behavior is defined by the signature:
public JsonPathFilter(string jsonPath)
Inspects an incoming message, determines the list of desired recipients, and forwards the message to all channels associated with the recipients in the list. (template in EIP).
Behavior is defined by the signature:
public RecipientList(Func<IMessage, MessageLabel[]> determineRecipientList)Sends an incoming message as a reply and breaks the message flow. It is terminal filter in the chain.
Break out the composite message into a series of individual messages, each of them contains data related to one item. (template in EIP).
Behavior is defined by the signature:
public Splitter(Func<IMessage, IEnumerable<object>> splitterFunc)Forwards the incoming messages to the specified recipient.
Behavior is defined by the signature:
public StaticRouter(string label)
Translate one data format into another (template in EIP).
Behavior is defined by the signature:
public Translator(Func<IMessage, object> translationFunc)Sends incoming message as a reply and forwards it to the incoming flow.
A special case of RecipientList, which allows you to listen to the channel (template in EIP).
Behavior is defined by the signature:
public WireTap(MessageLabel messageLabel)- All outgoing messages of the last operation are published to the bus.
- Synchronous messages processing.
- Broker is not used to transmit messages between operators.
- Operations that produce multiple messages per message, publish them in the same outgoing channel.
- Operations are executed in the same address space and can affect each other. It is not recommended to use Reflection and mutable objects.
- clone the repository
- run "build.cmd" to make sure all unit tests are passing.
- run "build.cmd RunAllTests" to make sure all integration and unit tests are passing. In this case you have to configure access to the RabbitMQ broker.
The library is available under MIT. For more information see the License file in the GitHub repository.