Skip to main content

Queue Sink: RabbitMQ

The RabbitMQ Sink Task​

Connecting a RabbitMQ broker​

Users of RavenDB 6.0 and on can create an ongoing Sink task that connects a RabbitMQ broker, retrieves messages from selected queues, runs a user-defined script to manipulate data and construct documents, and potentially stores the created documents in RavenDB collections.

In the message broker architecture, RavenDB sinks take the role of data consumers.
A sink would connect a RabbitMQ broker using a connection string, and retrieve messages from the broker's queues.

Read below about adding a connection string via API.
Read here about adding a connection string using Studio.

Retrieving messages from RabbitMQ queues​

While a queue sink task is running, it prevents the host database from being unloaded due to idle operations.
This ensures that message consumption from RabbitMQ queues continues uninterrupted, even if the database would otherwise go idle.

When a message is sent to a RabbitMQ broker by a producer, it is pushed to the tail of a queue. As preceding messages are pulled, the message advances up the queue until it reaches its head and can be consumed by RavenDB's sink.

Running user-defined scripts​

A sink task's script is a JavaScript segment. Its basic role is to retrieve selected RabbitMQ messages or message properties, and construct documents that will then be stored in RavenDB.

The script can store the whole message as a document, as in this segment:

// Add the document a metadata `@collection` property to keep it in
// this collection, or do not set it to store the document in @empty).
this['@metadata']['@collection'] = 'Orders';
// Store the message as is, using its Id property as its RavenDB Id as well.
put(this.Id.toString(), this)

The script can also retrieve selected information from the read message and construct a new document that does not resemble the original message.
Scripts often apply two sections: a section that creates a JSON object that defines the document's structure and contents, and a second section that stores the document.

E.g., for RabbitMQ messages of this format:

{
   "Id" : 13,
   "FirstName" : "John",
   "LastName" : "Doe"
}

We can create this script:

var item = {
    Id : this.Id,
    FirstName : this.FirstName,
    LastName : this.LastName,
    FullName : this.FirstName + ' ' + this.LastName,
    "@metadata" : {
        "@collection" : "Users"
    }
};

// Use .toString() to pass the Id as a string even if RabbitMQ provides it as a number
put(this.Id.toString(), item)

The script can also apply various other JavaScript commands, including load to load a RavenDB document (e.g. to construct a document that includes data from the retrieved message and complementing data from existing RavenDB documents), del to remove existing RavenDB documents, and many others.

Storing documents in RavenDB collections​

The sink task consumes batches of queued messages and stores them in RavenDB in a transactional manner, processing either the entire batch or none of it.

Exceptions to this rule

Some script processing errors are allowed; when such an error occurs RavenDB will skip the affected message, record the event in the logs, and alert the user in Studio, but continue processing the batch.

Once a batch is consumed, the task confirms it by sending _channel.BasicAck.

Note that the number of documents included in a batch is configurable.

Take care of duplicates

Producers may enqueue multiple instances of the same document.
If processing each message only once is important to the consumer, it is the consumer's responsibility to verify the uniqueness of each consumed message.

Note that as long as the Id property of RabbitMQ messages is preserved (so duplicate messages share an Id), the script's put(ID, { ... }) command will overwrite a previous document with the same Id and only one copy of it will remain.

Client API​

Adding a RabbitMQ connection string​

Before defining a sink task, add a RabbitMQ connection string for the task to use.
Create a QueueConnectionString object configured for RabbitMQ, and pass it to PutConnectionStringOperation.

// Add a RabbitMQ connection string
var res = store.Maintenance.Send(
    new PutConnectionStringOperation<QueueConnectionString>(
        new QueueConnectionString
        {
            Name = "RabbitMqConStr",
            BrokerType = QueueBrokerType.RabbitMq,
            RabbitMqConnectionSettings = new RabbitMqConnectionSettings()
                    { ConnectionString = "amqp://guest:guest@localhost:5672/" }
        }));

For the full property reference, see the Connection string class in the Syntax section.

Adding a RabbitMQ sink task​

To define the sink task, prepare a QueueSinkConfiguration object and pass it to AddQueueSinkOperation.
The configuration needs to name the connection string to use, set the broker type to RabbitMQ, and hold one or more QueueSinkScript objects.

  • Each script lists the RabbitMQ queues to consume from in its Queues property, where each entry is a queue name.
  • Each script's Script property holds the JavaScript that turns the consumed messages into RavenDB documents.

Example: Adding a RabbitMQ sink task​

// Add RabbitMQ connection string
var res = store.Maintenance.Send(
    new PutConnectionStringOperation<QueueConnectionString>(
        new QueueConnectionString
        {
            Name = "RabbitMqConStr",
            BrokerType = QueueBrokerType.RabbitMq,
            RabbitMqConnectionSettings = new RabbitMqConnectionSettings()
                    { ConnectionString = "amqp://guest:guest@localhost:5672/" }
        }));

// Define a Sink script
QueueSinkScript queueSinkScript = new QueueSinkScript
{
    // Script name
    Name = "orders",
    // A list of RabbitMQ queues to consume from
    Queues = new List<string>() { "orders" },
    // Apply this script
    Script = @"this['@metadata']['@collection'] = 'Orders';
               put(this.Id.toString(), this)"
};

// Define a RabbitMQ sink task configuration
var config = new QueueSinkConfiguration()
{
    // Sink name
    Name = "RabbitMqSinkTaskName",
    // The connection string to connect the broker with
    ConnectionStringName = "RabbitMqConStr",
    // What queue broker is this task using
    BrokerType = QueueBrokerType.RabbitMq,
    // The list of scripts to run
    Scripts = { queueSinkScript }
};

AddQueueSinkOperationResult addQueueSinkOperationResult =
    store.Maintenance.Send(new AddQueueSinkOperation<QueueConnectionString>(config));

For the QueueSinkConfiguration and QueueSinkScript property reference, see the Sink task classes in the Syntax section.

Configuration Options​

Use these configuration options to gain more control over queue sink tasks.

Syntax​

Classes​

Connection string​

The connection string a sink task uses to reach a RabbitMQ broker.

class QueueConnectionString
{
    string Name
    QueueBrokerType BrokerType
    RabbitMqConnectionSettings RabbitMqConnectionSettings
}

PropertyTypeDescription
NamestringThe connection string name
BrokerTypeQueueBrokerTypeSet to QueueBrokerType.RabbitMq for a RabbitMQ connection string
RabbitMqConnectionSettingsRabbitMqConnectionSettingsThe broker's connection details

QueueConnectionString is shared by all queue brokers, so it also defines settings for the other broker types (KafkaConnectionSettings, AzureQueueStorageConnectionSettings, AmazonSqsConnectionSettings, AzureServiceBusConnectionSettings).
Set only the one matching BrokerType.

Sink task​

The configuration of a RabbitMQ sink task.

class QueueSinkConfiguration
{
    string Name
    string ConnectionStringName
    QueueBrokerType BrokerType
    List<QueueSinkScript> Scripts
    bool Disabled
    string MentorNode
    bool PinToMentorNode
    long TaskId
}

PropertyTypeDescription
NamestringThe sink task name
ConnectionStringNamestringThe name of the RabbitMQ connection string the task uses
BrokerTypeQueueBrokerTypeSet to QueueBrokerType.RabbitMq (must match the connection string's broker type)
ScriptsList<QueueSinkScript>The scripts the task runs
DisabledboolWhether the task is created in a disabled state
MentorNodestringThe preferred responsible node for the task, if any
PinToMentorNodeboolWhether to pin the task to its mentor node
TaskIdlongThe task's identifier, assigned by the server

Enums​

Identifies the message broker that a connection string and sink task use.

enum QueueBrokerType
{
    None,
    Kafka,
    RabbitMq,
    AzureQueueStorage,
    AmazonSqs,
    AzureServiceBus
}

ValueDescription
RabbitMqSelects RabbitMQ. Use this value for a RabbitMQ connection string and sink task.
None, Kafka, AzureQueueStorage, AmazonSqs, AzureServiceBusThe other broker types, used by the remaining Queue Sink and Queue ETL brokers.

In this article