Getting started

This guide is a step by step introduction. If you want to know more about how Tapeti works, for example how it determines the exchange and routing keys, see In-depth.

Install packages

I’ll assume you are familiar with installing NuGet.org packages into your project.

Find and install the Tapeti package. This will also install Tapeti.Annotations, which contains the various attributes.

You will need an integration package as well for your IoC (Inversion of Control) container of choice. Various containers are supported by default:

• SimpleInjector (Tapeti.SimpleInjector)
• Autofac (Tapeti.Autofac)
• Castle Windsor (Tapeti.CastleWindsor)
• Ninject (Tapeti.Ninject)
• Unity (Tapeti.UnityContainer)

SimpleInjector is used in all examples. The “01-PublishSubscribe” example included in the source shows how the other integration packages can be used.

Note

If you need support for your favourite library, implement IDependencyContainer using the existing packages’ source as a reference and replace SimpleInjectorDependencyResolver with your class name in the example code.

Configuring Tapeti

First create an instance of TapetiConfig, tell it which controllers to register and have it gather all the information by calling Build. Then pass the result to a TapetiConnection.

using SimpleInjector;
using Tapeti;

internal class Program
{
    private static void Main()
    {
        var container = new Container();
        var config = new TapetiConfig(new SimpleInjectorDependencyResolver(container))
            .RegisterAllControllers()
            .Build();

        using (var connection = new TapetiConnection(config)
        {
            Params = new TapetiConnectionParams
            {
                Host = "localhost"
            }
        })
        {
            connection.SubscribeSync();

            // Start service
        }
    }
}

Note

RegisterAllControllers without parameters searches the entry assembly. Pass an Assembly parameter to register other or additional controllers. You can call RegisterAllControllers multiple times with different assemblies.

Caution

Tapeti attempts to register it’s default implementations in the IoC container during configuration, as well as when starting the connection (to register IPublisher). If your container is immutable after the initial configuration, like SimpleInjector is, make sure that you run the Tapeti configuration before requesting any instances from the container.

Defining a message

A message is simply a plain object which can be serialized using Json.NET.

public class RabbitEscapedMessage
{
    public string Name { get; set; }
    public string LastKnownHutch { get; set; }
}

Creating a message controller

To handle messages you need what Tapeti refers to as a “message controller”. It is similar to an ASP.NET controller if you’re familiar with those, but it handles RabbitMQ messages instead of HTTP requests.

All you need to do is create a new class and annotate it with the MessageController attribute and a queue attribute. The name and folder of the class is not important to Tapeti, though you might want to agree on a standard in your team.

The queue attribute can be either DynamicQueue or DurableQueue. The attribute can be set for the entire controller (which is considered the default scenario) or specified / overridden per message handler.

DynamicQueue will create a queue with a name generated by RabbitMQ which is automatically deleted when your service stops. Bindings will be added for the messages handled by the controller. You will typically use dynamic queues for scenarios where handling the message is only relevant while the service is running (for example, updating a service’s cache or performing live queries).

DurableQueue requires a queue name as the parameter. By default, the queue is assumed to be present already and Tapeti will throw an exception if it does not. If you want Tapeti to create and update the durable queues as well, see Durable queues in In-depth.

[MessageController]
[DynamicQueue("monitoring")]
public class MonitoringController
{
}

Note

Notice the parameter to DynamicQueue. This defines the prefix. If specified, the queue name will begin with the supplied value, followed by a unique identifier, so it can be more easily recognized in the RabbitMQ management interface.

Handling incoming messages

Any public method in a message controller is considered a message handler. There are a few requirements which are enforced by Tapeti. Below are the default requirements, although some extension packages (like the Flow extension) add their own or alter these requirements.

• The first parameter must be the message class.
• The return type can be void, Task, Task<message class> or a message class.

The name of the method is not important to Tapeti. Any parameter other than the first will be resolved in two ways:

1. Registered middleware can alter the behaviour of parameters. Tapeti includes one by default for CancellationToken parameters, see Parameter binding in In-depth.
2. Any remaining parameters are resolved using the IoC container, although it is considered best practice to use the constructor for dependency injection instead.

A new controller is instantiated for each message, so it is safe to use public or private fields to store state while handling the message. Just don’t expect it to be there for the next message. If you need this behaviour, take a look at the Flow extension!

[MessageController]
[DynamicQueue]
public class MonitoringController
{
    public void LogEscape(RabbitEscapedMessage message)
    {
        Logger.Warning($"This is a beige alert. {message.Name} has escaped." +
                       $"It was last seen in {message.LastKnownHutch}.");
    }
}

Note

If you’re doing anything asynchronous in the message handler, make it async as well! Simply change the return type to “Task” or “async Task”.

If the method returns a message object, that object is published as if it was a reply to the incoming message, maintaining the correlationId and respecting the replyTo header. See In-depth for request-response requirements.

Publishing messages

To send a message, get a reference to IPublisher using dependency injection and call the Publish method. For example, to broadcast another message from a message handler:

public class LogMessage
{
    public string Level { get; set; }
    public string Description { get; set; }
}


[MessageController]
[DynamicQueue]
public class MonitoringController
{
    private readonly IPublisher publisher;

    public MonitoringController(IPublisher publisher)
    {
        this.publisher = publisher;
    }

    public async Task LogEscape(RabbitEscapedMessage message)
    {
        await publisher.Publish(new LogMessage
        {
            Level = "Beige",
            Description = $"{message.Name} has escaped." +
                          $"It was last seen in {message.LastKnownHutch}."
        });
    }
}

Connection parameters

If you don’t want to use the default connection parameters, which is probably a good idea in a production environment, you can manually specify the properties for TapetiConnectionParams or get them from your configuration of choice. Tapeti provides with two helpers.

App.config / Web.config

The TapetiAppSettingsConnectionParams class can be used to load the connection parameters from the AppSettings:

using (var connection = new TapetiConnection(config)
{
    Params = new TapetiAppSettingsConnectionParams()
})

The constructor accepts a prefix parameter, which defaults to “rabbitmq:”. You can then specify the values in the appSettings block of your App.config or Web.config. Any omitted parameters will use the default value.

<?xml version="1.0" encoding="utf-8" ?>
<configuration>
  <appSettings>
    <add key="rabbitmq:hostname" value="localhost" />
    <add key="rabbitmq:port" value="5672" />
    <add key="rabbitmq:virtualhost" value="/" />
    <add key="rabbitmq:username" value="guest" />
    <add key="rabbitmq:password" value="guest" />
    <add key="rabbitmq:prefetchcount" value="50" />
    <add key="rabbitmq:managementport" value="15672" />
    <add key="rabbitmq:clientproperty:application" value="Example" />
  </appSettings>
</configuration>

The last entry is special: any setting which starts with “clientproperty:”, after the configured prefix, will be added to the ClientProperties set. These properties are visible in the RabbitMQ Management interface and can be used to identify the connection.

ConnectionString

Tapeti also includes a helper which can parse a connection string style value which is mainly for compatibility with EasyNetQ. It made porting our applications slightly easier. EasyNetQ is a very capable library which includes high- and low-level wrappers for the RabbitMQ Client as well as a Management API client, and is worth checking out if you have a use case that is not suited to Tapeti.

To parse a connection string, use the ConnectionStringParser.Parse method. You can of course still load the value from the AppSettings easily:

using (var connection = new TapetiConnection(config)
{
    Params = Tapeti.Helpers.ConnectionStringParser.Parse(
      ConfigurationManager.AppSettings["RabbitMQ.ConnectionString"])
})

An example connection string:

host=localhost;username=guest;password=prefetchcount=5

Supported keys are:

• hostname
• port
• virtualhost
• username
• password
• prefetchcount
• managementport

Any keys in the connection string which are not supported will be silently ignored.