C++17 message queuing · LGPL-3.0

Message queuing for large-scale data processing

FairMQ connects processes through channels. Each process is a device with a state machine. On one host, the shared memory transport moves data without a copy.

Get started Read the concepts

spack install fairmq

1 sampler · 3 processors · 1 sink

zeromq: the payload goes through each socket.

A complete device

Derive from fair::mq::Device and override one hook. runDevice.h adds the main() function. It parses the command line, loads the plugins and runs the state machine.

In the state RUNNING, the device calls ConditionalRun() in a loop, at the rate of --rate. Each call sends one message. When the function returns false, the device leaves the state RUNNING.

Your first device →

#include <fairmq/Device.h>
#include <fairmq/runDevice.h>

struct Sampler : fair::mq::Device
{
    bool ConditionalRun() override
    {
        auto msg = NewSimpleMessage("Hello");
        LOG(info) << "Sending \"Hello\"";
        return Send(msg, "data") >= 0; // false ends RUNNING
    }
};

void addCustomOptions(boost::program_options::options_description&) {}

std::unique_ptr<fair::mq::Device> getDevice(fair::mq::ProgOptions&)
{
    return std::make_unique<Sampler>();
}

$ fairmq-ex-1-1-sampler --id sampler1 --rate 1 \
    --channel-config "name=data,type=push,method=bind,address=tcp://*:5555,rateLogging=0"
[10:42:07][STATE] IDLE ---> INITIALIZING DEVICE
...
[10:42:07][STATE] READY ---> RUNNING
[10:42:07][INFO] fair::mq::Device running...
[10:42:07][INFO] Sending "Hello"
[10:42:08][INFO] Sending "Hello"