Examples

1-1: sampler and sink

The 1-1 example has two devices: a sampler that sends a text message and a sink that receives it. It shows the minimum code of a device: a class that derives from fair::mq::Device, its hooks, one channel, and the main() function of fairmq/runDevice.h.

The files of the example are in examples/1-1 on GitHub.

Topology

The devices connect through one channel with the name data. The sampler binds a push socket, and the sink connects a pull socket to it.

Device	Channel	Type	Method	Address
`sampler1`	`data`	`push`	`bind`	`tcp://*:5555`
`sink1`	`data`	`pull`	`connect`	`tcp://localhost:5555`

The start script sets rateLogging=0 on both channels, so the devices do not log the transfer rates. It also starts the sampler with --rate 1, so the sampler sends one message per second.

Build the example

The examples build together with FairMQ when the CMake option BUILD_EXAMPLES is ON. ON is the default (see Install). The build puts these files into the build directory of the example:

Directorybuild/examples/1-1/
- fairmq-ex-1-1-sampler the sampler, from sampler.cxx
- fairmq-ex-1-1-sink the sink, from sink.cxx
- fairmq-start-ex-1-1.sh the start script
- test-ex-1-1.sh the test script

The install step copies the two executables and the start script to <prefix>/bin/. The installed start script calls the installed executables.

Run the example

Go to the directory of the example in the build tree:
Terminal window
```
cd build/examples/1-1
```
Run the start script:
Terminal window
```
./fairmq-start-ex-1-1.sh
```

The script opens one xterm window for each device. The windows stay open after a device stops (xterm -hold). The script also sets the environment variable FAIRMQ_PATH to <build>/fairmq. A device with --transport shmem and --shm-monitor true uses this path to find fairmq-shmmonitor.

If you do not use xterm, open two terminals in build/examples/1-1. Run one command of the script in each terminal:

./fairmq-ex-1-1-sampler --id sampler1 --rate 1 \
  --channel-config "name=data,type=push,method=bind,address=tcp://*:5555,rateLogging=0"

./fairmq-ex-1-1-sink --id sink1 \
  --channel-config "name=data,type=pull,method=connect,address=tcp://localhost:5555,rateLogging=0"

--id sets the ID of the device.
--channel-config configures one channel with a list of key=value pairs. The channel properties page lists all keys.
--rate sets the maximum number of ConditionalRun() calls per second. The default 0 sets no limit.

You can start the two devices in any order. Each device goes through its states to RUNNING without input. Then the default controller (--control dynamic) reads keys in the terminal. Push q to stop the device and end the process. Push h to see all keys. Ctrl-C also stops the device.

The sampler

The sampler is a struct that derives from fair::mq::Device. It overrides two hooks and has three members: fText, fMaxIterations and fNumIterations (sampler.cxx).

void InitTask() override
{
    // Get the fText and fMaxIterations values from the command line options (via fConfig)
    fText = fConfig->GetProperty<std::string>("text");
    fMaxIterations = fConfig->GetProperty<uint64_t>("max-iterations");
}

examples/1-1/sampler.cxx:18–23

The device calls InitTask() in the state INITIALIZING TASK. The sampler reads its two custom options there. fConfig points to the configuration of the device. GetProperty<T>() returns the value of an option as the type T.

bool ConditionalRun() override
{
    // create a copy of the data with new(), that will be deleted after the transfer is complete
    std::string* text = new std::string(fText);

    // create message object with a pointer to the data buffer, its size,
    // custom deletion function (called when transfer is done),
    // and pointer to the object managing the data buffer
    fair::mq::MessagePtr msg(NewMessage(
        const_cast<char*>(text->c_str()),
        text->length(),
        [](void* /*data*/, void* object) { delete static_cast<std::string*>(object); },
        text));

    LOG(info) << "Sending \"" << fText << "\"";

    // in case of error or transfer interruption, return false to go to the Ready state
    // successfull transfer will return number of bytes transfered (can be 0 if sending an empty message).
    if (Send(msg, "data") < 0) {
        return false;
    } else if (fMaxIterations > 0 && ++fNumIterations >= fMaxIterations) {
        LOG(info) << "Configured maximum number of iterations reached. Leaving RUNNING state.";
        return false;
    }

    return true;
}

examples/1-1/sampler.cxx:25–51

In the state RUNNING, the device calls ConditionalRun() again while the function returns true (Device.cxx):

NewMessage() with a buffer, a size, a free function and a hint creates a message for an existing buffer. Here the buffer is a new std::string, and the free function deletes it.
The zeromq transport sends the buffer without a copy and calls the free function after the transfer. The shmem transport copies the buffer into shared memory and calls the free function at once.
Send() sends the message on the sub-channel 0 of the channel data. It returns the number of bytes, or a negative transfer code.
LOG(info) writes a log line with the severity info (see Logging).
The return value false ends the state RUNNING, and the device goes to READY. The sampler returns false if Send() fails or if it reaches --max-iterations.

void addCustomOptions(bpo::options_description& options)
{
    options.add_options()
        ("text", bpo::value<std::string>()->default_value("Hello"), "Text to send out")
        ("max-iterations", bpo::value<uint64_t>()->default_value(0), "Maximum number of iterations of Run/ConditionalRun/OnData (0 - infinite)");
}

examples/1-1/sampler.cxx:59–64

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

examples/1-1/sampler.cxx:66–69

The header fairmq/runDevice.h contains the main() function of the executable. main() calls two functions that your source file must define (runDevice.h):

addCustomOptions() adds the options of the device to the command line. The sampler adds --text (default Hello) and --max-iterations (default 0, no limit). bpo is an alias of the namespace boost::program_options.
getDevice() creates the device.

The sink

Sink()
{
    // register a handler for data arriving on "data" channel
    OnData("data", &Sink::HandleData);
}

examples/1-1/sink.cxx:18–22

The sink has no ConditionalRun(). Its constructor registers the member function HandleData() for the channel data with OnData(). In the state RUNNING, the device receives the messages of the channel and calls the handler for each message. A device with data handlers does not call ConditionalRun() or Run(), and --rate has no effect on it (Device.cxx).

bool HandleData(fair::mq::MessagePtr& msg, int)
{
    LOG(info) << "Received: \"" << std::string(static_cast<char*>(msg->GetData()), msg->GetSize()) << "\"";

    if (fMaxIterations > 0 && ++fNumIterations >= fMaxIterations) {
        LOG(info) << "Configured maximum number of iterations reached. Leaving RUNNING state.";
        return false;
    }

    // return true if you want the handler to be called again (otherwise return false go to the
    // Ready state)
    return true;
}

examples/1-1/sink.cxx:30–42

The handler gets the message and the index of the sub-channel. GetData() returns a pointer to the data, and GetSize() returns the size in bytes. The handler returns true to get the next message, or false to end the state RUNNING. The sink also has the option --max-iterations.

Output

The listings show the severity and the message of each line. The real lines also have a time at the start; the option --verbosity sets this prefix. Before these lines, each device logs the FairMQ logo and its states from IDLE to READY. The interactive controller also prints its list of keys.

The sampler sends one message per second:

[STATE] READY ---> RUNNING
[INFO] fair::mq::Device running...
[INFO] Sending "Hello"
[INFO] Sending "Hello"
[INFO] Sending "Hello"

The sink logs each message that it receives:

[STATE] READY ---> RUNNING
[INFO] fair::mq::Device running...
[INFO] Received: "Hello"
[INFO] Received: "Hello"
[INFO] Received: "Hello"

With --max-iterations 3, each device logs these lines after the third message:

[INFO] Configured maximum number of iterations reached. Leaving RUNNING state.
[STATE] RUNNING ---> READY

Test the example

The test script test-ex-1-1.sh starts the two devices in the background, in one terminal. Among others, it adds these options to the commands of the start script:

Option	Value in the test	Effect
`--transport`	the first argument of the script, default `zeromq`	The transport of the devices.
`--session`	the output of `fairmq-uuid-gen -h`	A new session for each run.
`--control`	`static`	The process ends when the device leaves `RUNNING`.
`--max-iterations`	`1`	Each device stops after one message.
`--shm-segment-size`	`100000000`	A 100 MB shared memory segment.
`--shm-monitor`	`true`	With `shmem`, starts `fairmq-shmmonitor` as a daemon.

The test uses an ipc:// address instead of TCP: ipc:///tmp/fmq_<session>_data_<transport>. At the end, the script deletes this file.

CTest runs the script once for each transport. The tests are Example.1-1.zeromq and Example.1-1.shmem, each with a timeout of 30 s. CTest has these tests only if you configure with -DBUILD_TESTING=ON. The default is OFF.

cmake -S FairMQ -B build -DBUILD_TESTING=ON
cmake --build build
cd build
ctest -R '^Example\.1-1\.' --output-on-failure

You can also run the script directly, for example ./test-ex-1-1.sh shmem in build/examples/1-1. To write a test of this type for your own devices, see Test your devices. To add an example with its test to FairMQ, see Add an example.

Things to try

Use shared memory: add --transport shmem to both commands. Both devices must use the same session. If you do not set --session, both use the session default.
Change the rate: start the sampler with --rate 10. Without --rate, the sampler sends as fast as it can.
Send another text: start the sampler with --text "Hello FairMQ".
Stop after five messages: add --max-iterations 5 to both commands. Also add --control static to end the processes when the devices leave RUNNING. To check that no message is lost, see Check that no message is lost.
Show the transfer rates: set rateLogging=1 in both channel configurations. Each second, the device logs the messages per second and the MB per second of the channel.
Swap bind and connect: the sink binds tcp://*:5555, and the sampler connects to tcp://localhost:5555.
Show all options of a device, with the custom options: run it with --help.

Your first device: write and build a device like this one.
Device states: the states that the devices go through.
Channels and sockets: socket types and methods.
Receive data with callbacks or loops: OnData() compared to ConditionalRun().
1-n-1: sampler, processors and sink: the next example.