Examples

multiple-channels: data and broadcast

The multiple-channels example shows devices with more than one channel. The sampler sends data to the sink on the channel data. Both devices also receive the messages of a broadcaster on the channel broadcast. The sampler checks its two channels with a poller, and the sink registers one data handler for each channel.

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

Topology

Device	Channel	Type	Method	Address
`sampler1`	`data`	`push`	`bind`	`tcp://*:5555`
`sampler1`	`broadcast`	`sub`	`connect`	`tcp://localhost:5005`
`sink1`	`data`	`pull`	`connect`	`tcp://localhost:5555`
`sink1`	`broadcast`	`sub`	`connect`	`tcp://localhost:5005`
`broadcaster1`	`broadcast`	`pub`	`bind`	`tcp://*:5005`

The pub socket of the broadcaster sends each message to all connected sub sockets. FairMQ subscribes each sub socket to all messages (zeromq/Socket.h). All channels use rateLogging=0.

Build the example

The examples build together with FairMQ when the CMake option BUILD_EXAMPLES is ON (the default):

Directorybuild/examples/multiple-channels/
- fairmq-ex-multiple-channels-sampler
- fairmq-ex-multiple-channels-sink
- fairmq-ex-multiple-channels-broadcaster
- fairmq-start-ex-multiple-channels.sh
- test-ex-multiple-channels.sh

Run the example

Go to the directory of the example in the build tree:
Terminal window
```
cd build/examples/multiple-channels
```
Run the start script. It opens three xterm windows, one for each device:
Terminal window
```
./fairmq-start-ex-multiple-channels.sh
```

To run the example without xterm, run each command in its own terminal in build/examples/multiple-channels. One --channel-config option takes one value for each channel:

./fairmq-ex-multiple-channels-sampler --id sampler1 \
  --channel-config "name=data,type=push,method=bind,rateLogging=0,address=tcp://*:5555" \
                   "name=broadcast,type=sub,method=connect,rateLogging=0,address=tcp://localhost:5005"

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

./fairmq-ex-multiple-channels-broadcaster --id broadcaster1 \
  --channel-config "name=broadcast,type=pub,method=bind,rateLogging=0,address=tcp://*:5005"

The sampler

void Run() override
{
    fair::mq::PollerPtr poller(NewPoller("data", "broadcast"));

    while (!NewStatePending()) {
        poller->Poll(100);

        if (poller->CheckInput("broadcast", 0)) {
            fair::mq::MessagePtr msg(NewMessage());

            if (Receive(msg, "broadcast") > 0) {
                LOG(info) << "Received broadcast: \"" << std::string(static_cast<char*>(msg->GetData()), msg->GetSize()) << "\"";
            }
        }

        if (poller->CheckOutput("data", 0)) {
            std::this_thread::sleep_for(std::chrono::seconds(1));

            fair::mq::MessagePtr msg(NewSimpleMessage(fText));

            if (Send(msg, "data") > 0) {
                LOG(info) << "Sent \"" << fText << "\"";

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

examples/multiple-channels/sampler.cxx:27–57

The sampler overrides Run() instead of ConditionalRun(). The device calls Run() once in the state RUNNING. Thus Run() has its own loop, and it must return when NewStatePending() is true.

NewPoller() creates a poller for the channels data and broadcast, with all their sub-channels.
Poll(100) waits for max. 100 ms for an event on one of the channels.
CheckInput("broadcast", 0) is true if the sub-channel broadcast[0] has a message to receive.
CheckOutput("data", 0) is true if the sub-channel data[0] can send. The zeromq poller checks push sockets for output and sub sockets for input (zeromq/Poller.h).
Before each send, the sampler waits 1 s. NewSimpleMessage() copies the text into a new message.
At --max-iterations, the loop ends. When Run() returns, the device goes to READY.

All channels of one poller must use the same transport. Otherwise NewPoller() throws an exception (Device.h).

The sink

Sink()
{
    OnData("broadcast", &Sink::HandleBroadcast);
    OnData("data", &Sink::HandleData);
}

examples/multiple-channels/sink.cxx:19–23

The sink calls OnData() once for each channel. With more than one input channel, the device polls all input channels and calls the handler of the channel that has a message (Device.cxx).

bool CheckIterations()
{
    if (fMaxIterations > 0) {
        if (fReceivedData && fReceivedBroadcast && ++fNumIterations >= fMaxIterations) {
            LOG(info) << "Configured maximum number of iterations reached & Received messages from both sources. Leaving RUNNING state.";
            return false;
        }
    }

    return true;
}

examples/multiple-channels/sink.cxx:46–56

Both handlers return the result of CheckIterations(). With --max-iterations, the sink leaves RUNNING only after it got messages on both channels.

The broadcaster

bool ConditionalRun() override
{
    std::this_thread::sleep_for(std::chrono::seconds(1));

    // NewSimpleMessage creates a copy of the data and takes care of its destruction (after the transfer takes place).
    // Should only be used for small data because of the cost of an additional copy
    fair::mq::MessagePtr msg(NewSimpleMessage("OK"));

    LOG(info) << "Sending OK";

    Send(msg, "broadcast");

    return true;
}

examples/multiple-channels/broadcaster.cxx:19–32

Once per second, the broadcaster sends the text OK on its pub channel. It has no custom options, so it runs until you stop it.

Output

The listings show the severity and the message of each line, without the time. The lines of the two channels mix, because the messages arrive in any order.

The sampler:

[INFO] Received broadcast: "OK"
[INFO] Sent "Hello"

The sink:

[INFO] Received broadcast: "OK"
[INFO] Received message: "Hello"

The broadcaster:

[INFO] Sending OK

Test the example

The test script test-ex-multiple-channels.sh uses ipc:// addresses in /tmp, a new session and --control static. It starts the sink first, waits 1 s, and then starts the sampler and the broadcaster. The sampler and the sink stop after one iteration (--max-iterations 1). Then the script stops the broadcaster with SIGINT.

The CMake file of the example sets TRANSPORT zeromq. Thus CTest has only one test, Example.multiple-channels.zeromq. The test exists if you configure with -DBUILD_TESTING=ON.

cd build
ctest -R '^Example\.multiple-channels\.' --output-on-failure

Things to try

Start the broadcaster some seconds after the other devices. The sampler and the sink exchange data before the first broadcast arrives.
Stop the broadcaster. The data on the channel data continues.
Send another text: start the sampler with --text "Hi".
Stop after three iterations: add --max-iterations 3 to the commands of the sampler and the sink.

Poll several channels: the Poller interface in detail.
Receive data with callbacks or loops: OnData(), ConditionalRun() and Run().
Channels and sockets: the socket types pub and sub.
multiple-transports: zeromq and shmem: input channels with two transports.
Design a topology: a data link and a broadcast link in one topology.