Redis and RabbitMQ

This section illustrates typical HOTROD usage patterns for Redis and RabbitMQ.

Lists and Queues

A Redis list is equivalent to a RabbitMQ queue — in both cases you push items on one end and pop items off the other end.

Push a value to a Redis list:

name: redis-write-queue

input:
  text: '{"msg":"hello"}'

output:
  redis:
    set:
      list: q

Pop a value from a Redis list:

name: redis-read-queue

input:
  redis:
    get:
        list: q
    raw: true        

output:
    write: console

Assuming you have a Redis server listening on 127.0.0.1:6379:

$> hotrod pipes run --file redis-write-queue.yml 

$> hotrod pipes run --file redis-read-queue.yml

{"msg":"hello"}

The input will now wait for anything new to be pushed to the queue. As with the other inputs, raw indicates that the data will not be quoted, e.g., {"_raw":"{\"msg\":\"hello\"}"}.

In cases where there are multiple consumers of the queue, run redis-read-queue in two separate terminals, and send new messages as seen below:

$> cat redis-input.json

{"msg":"dolly"}
{"msg":"so nice"}
{"msg":"to see you back"}
{"msg":"where you belong"}

Now write to the queue:

$> hotrod pipes run --file redis-write-queue.yml --json --input @redis-input.json

First terminal output:

{"msg":"dolly"}
{"msg":"to see you back"}

Second terminal output:

{"msg":"so nice"}
{"msg":"where you belong"}

Each consumer gets the next item available in turn as round-robin behaviour.

The RabbitMQ equivalent would be this pair of Pipes:

name: rabbit-write-queue

input:
  text: '{"msg":"hello"}'

output:
  amqp:
    queue:
      name: some-name

name: rabbit-read-queue

input:
  amqp:
    raw: true
    queue:
      name: some-name

output:
  write: console

Here again, multiple consumers will get the items in round-robin order.

RabbitMQ provides greater control, which is why it may be preferable to use instead of Redis. This does, however, mean that there are more fields to consider, such as the important passive field:

queue:
  name: some-name
  passive: true

This result indicates that we should not attempt to redeclare this queue. You will need this after defining the queue properties in the RabbitMQ server.

The amqp model involves both queues and exchanges, which we need to change the default round-robin behaviour. If an exchange is specified, then a queue will be defined implicitly.

To receive a fan-out — where each consumer gets a copy of the messages — we must explicitly define an exchange:

name: rabbit-write-fanout

input:
  text: '{"msg":"hello"}'

output:
  amqp:
    exchange:
      name: some-name
      type: fanout

name: rabbit-read-fanout

input:
  amqp:
    raw: true
    exchange: 
      name: some-name
      type: fanout

output:
  write: console

note

passive also applies to exchanges as default behaviour attempts to declare them if they do not exist.

Publish and Subscribe

Much like the queue consumers above, subscribers listen to an exchange. However, they are only interested in certain topics. If a publisher puts out a message with a routing key that matches the topic specified by the subscriber, then the subscriber gets the message.

The subscription topic is a set of words separated by .. * means "exactly one word" and # means "zero or many words".

In the pair of Pipes found in the example below, we have used Context Expansion for variety. In real-world cases, this will allow you to customize the Pipe parameters for each Agent. Here we can experiment with different routing keys without editing the Pipes:

name: rabbit-publish

context:
  TOPIC: animal.dog

input:
  text: '{"msg":"hello","my_topic":"{{TOPIC}}"}'

output:
  amqp:
    exchange:
      name: topic-exchange
      type: topic
    routing-key-field: my_topic

name: rabbit-subscribe

context:
  TOPIC: animal.*

input:
  amqp:
    raw: true
    exchange:
      name: topic-exchange
      type: topic
      passive: true
    routing-key: '{{TOPIC}}'

output:
  write: console

The first Pipe publishes the message which contains the topic name. We can exploit the fact that the routing key can be used as the value of a field:

$> hotrod pipes run --file rabbit-publish.yml

$> hotrod pipes run --file rabbit-publish.yml TOPIC='animal.cat'

In another terminal:

$> hotrod pipes run --file rabbit-subscribe.yml

{"msg":"hello","my_topic":"animal.dog"}
{"msg":"hello","my_topic":"animal.cat"}

By default, the subscribing Pipe is interested in everything related to animals.

Redis also performs publish-subscribe. Note that the topic is matched with glob patterns (does not use #) and there can be subscriptions to multiple topics.

name: redis-publish

context:
  TOPIC: animal.dog

input:
  exec:
    raw: true
    command: |
      echo '{"msg":"hello","my_topic":"{{TOPIC}}"}'

output:
  redis:
    set:
      publish: '${my_topic}'

name: redis-subscribe

input:
  redis:
    get:
      subscribe:
      - animal.*
    raw: true        

output:
    write: console

Redis Hash Operations

Originally, Redis was designed as a pure in-memory, key-value store where values could be set and retrieved by key. However, these values have types, such as the previously detailed list type, which operates like a queue. Let’s explore the hash type, which is a map of fields to values.

output:
  redis:
    set:
      hash-value: [key, field]

This hash may have the key employees, and the field bob. The value is a string, which is usually interpreted as JSON.

The getting process is similar:

input:
  redis:
    get:
      hash-value: [key, field]

Flat JSON events map well to Redis hashes, a feature that can be used to merge JSON from different sources:

# Input: {"one":1}, {"two":"hello"}

output:
  redis:
    set:
      hash: keyname

input:
  redis:
    interval: 30s
    get:
      hash: keyname

# Output: {"one":1,"two":"hello"}

There are several small Pipes that individually report the specialized knowledge of a system and all write to the same hash. A coordinating Pipe reads this hash before sending the combined event to its desired destination. This scheme is especially powerful because the Pipes do not have to coordinate their scheduling — Redis acts as the shared cache.

We can also ask for the values inside the hash:

input:
  redis:
    interval: 1m
    get:
      hash-values: keyname
    raw: true

# Output: 1, hello

This is useful for inter-Pipe communication. On a server, events from different Agents can come in and be written to a hash keyed by the Agent ID. A collector Pipe can then access all these events at chosen intervals with Redis as the event cache.