Examples

multipart: several parts in one message

The multipart example sends one message that has several parts. The sampler puts a header and six data parts into a fair::mq::Parts object and sends all parts with one Send() call. The sink gets all seven parts in one call of its data handler. The Messages and multipart page explains the concept.

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

Topology

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

Here the sink binds and the sampler connects. Both devices use rateLogging=0.

Build the example

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

Directorybuild/examples/multipart/
- fairmq-ex-multipart-sampler
- fairmq-ex-multipart-sink
- fairmq-start-ex-multipart.sh
- test-ex-multipart.sh

Run the example

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

The first argument of the script sets the transport, for example ./fairmq-start-ex-multipart.sh shmem. The default is zeromq. The script gives both devices the same session name, from fairmq-uuid-gen -h.

To run the example without xterm, run each command in its own terminal in build/examples/multipart. Any session name works if both devices use the same one:

./fairmq-ex-multipart-sampler --id sampler1 --transport zeromq --session ex-multipart \
  --channel-config "name=data,type=push,method=connect,rateLogging=0,address=tcp://127.0.0.1:5555"

./fairmq-ex-multipart-sink --id sink1 --transport zeromq --session ex-multipart \
  --channel-config "name=data,type=pull,method=bind,rateLogging=0,address=tcp://127.0.0.1:5555"

The header

struct Header
{
    int32_t stopFlag;
};

examples/multipart/Header.h:17–20

The sampler and the sink include Header.h. The first part of each message is a Header. The sink stops when stopFlag is 1.

The sampler

bool ConditionalRun() override
{
    example_multipart::Header header;
    header.stopFlag = 0;

    // Set stopFlag to 1 for last message.
    if (fMaxIterations > 0 && fNumIterations == fMaxIterations - 1) {
        header.stopFlag = 1;
    }
    LOG(info) << "Sending header with stopFlag: " << header.stopFlag;

    fair::mq::Parts parts;

    // 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
    parts.AddPart(NewSimpleMessage(header));
    parts.AddPart(NewMessage(1000));

    // create more data parts, testing the fair::mq::Parts in-place constructor
    fair::mq::Parts auxData{ NewMessage(500), NewMessage(600), NewMessage(700) };
    assert(auxData.Size() == 3);
    parts.AddPart(std::move(auxData));
    assert(auxData.Size() == 0);
    assert(parts.Size() == 5);

    parts.AddPart(NewMessage());

    assert(parts.Size() == 6);

    parts.AddPart(NewMessage(100));

    assert(parts.Size() == 7);

    LOG(info) << "Sending body of size: " << parts.At(1)->GetSize();

    Send(parts, "data");

    // Go out of the sending loop if the stopFlag was sent.
    if (fMaxIterations > 0 && ++fNumIterations >= fMaxIterations) {
        LOG(info) << "Configured maximum number of iterations reached. Leaving RUNNING state.";
        return false;
    }

    // Wait a second to keep the output readable.
    std::this_thread::sleep_for(std::chrono::seconds(1));

    return true;
}

examples/multipart/sampler.cxx:27–74

fair::mq::Parts parts; creates an empty multipart message. Parts holds a vector of MessagePtr.
NewSimpleMessage() copies the header into a new message. The transport deletes this copy when it does not need it any more. Use it for small data only.
AddPart() adds one message at the end.
NewMessage(1000) creates a message of 1000 bytes. The example does not write data into it. NewMessage() without arguments creates an empty part of 0 bytes.
The constructor of Parts takes several messages. AddPart() with a Parts object moves all its parts to the end.
At() returns the part at an index. Size() returns the number of parts.
Send() sends all seven parts as one message.

The sampler sets stopFlag to 1 in its last message. The option --max-iterations of this sampler has the default 5, so the sampler stops after five messages. The value 0 sets no limit.

The sink

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

examples/multipart/sink.cxx:18–21

bool HandleData(fair::mq::Parts& parts, int)
{
    LOG(info) << "Received message with " << parts.Size() << " parts";

    if (parts.Size() != 7) {
        throw std::runtime_error("Number of received parts != 7");
    }

    example_multipart::Header header;
    header.stopFlag = (static_cast<example_multipart::Header*>(parts.At(0)->GetData()))->stopFlag;

    LOG(info) << "Received part 1 (header) with stopFlag: " << header.stopFlag;
    LOG(info) << "Received part 2 of size: " << parts.At(1)->GetSize();
    assert(parts.At(1)->GetSize() == 1000);
    LOG(info) << "Received part 3 of size: " << parts.At(2)->GetSize();
    assert(parts.At(2)->GetSize() == 500);
    LOG(info) << "Received part 4 of size: " << parts.At(3)->GetSize();
    assert(parts.At(3)->GetSize() == 600);
    LOG(info) << "Received part 5 of size: " << parts.At(4)->GetSize();
    assert(parts.At(4)->GetSize() == 700);
    LOG(info) << "Received part 6 of size: " << parts.At(5)->GetSize();
    assert(parts.At(5)->GetSize() == 0);
    LOG(info) << "Received part 7 of size: " << parts.At(6)->GetSize();
    assert(parts.At(6)->GetSize() == 100);

    if (header.stopFlag == 1) {
        LOG(info) << "stopFlag is 1, going IDLE";
        return false;
    }

    return true;
}

examples/multipart/sink.cxx:23–54

The handler takes a fair::mq::Parts& instead of a message. With this handler type, OnData() receives multipart messages on the channel. The sink checks the number of parts and reads the header from the data of part 0. To also check the size of each part, see Read the data safely.

If the message does not have seven parts, the handler throws an exception. The device then goes to the state ERROR.
If stopFlag is 1, the handler returns false. The log line says going IDLE, but the device goes to the state READY.

Output

The listings show the severity and the message of each line, without the time. The sampler logs two lines for each message, once per second. The listing shows the first and the last message:

[INFO] Sending header with stopFlag: 0
[INFO] Sending body of size: 1000
[INFO] Sending header with stopFlag: 1
[INFO] Sending body of size: 1000
[INFO] Configured maximum number of iterations reached. Leaving RUNNING state.

The sink logs the parts of each message:

[INFO] Received message with 7 parts
[INFO] Received part 1 (header) with stopFlag: 0
[INFO] Received part 2 of size: 1000
[INFO] Received part 3 of size: 500
[INFO] Received part 4 of size: 600
[INFO] Received part 5 of size: 700
[INFO] Received part 6 of size: 0
[INFO] Received part 7 of size: 100

After the message with stopFlag 1, the sink also logs stopFlag is 1, going IDLE.

Test the example

The test script test-ex-multipart.sh uses a pair channel instead of push and pull. The sampler connects with linger=1000, and the sink binds. The channel uses an ipc:// address in /tmp, and the devices use --control static. The sampler has --max-iterations 1, so its first message has the stopFlag 1. Thus both devices stop after one message.

CTest runs the script once for each transport: Example.multipart.zeromq and Example.multipart.shmem. The tests exist if you configure with -DBUILD_TESTING=ON.

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

Things to try

Use shared memory: run ./fairmq-start-ex-multipart.sh shmem.
Send without a limit: start the sampler with --max-iterations 0. stopFlag then stays 0, and the sink does not stop.
Send to two sinks: use the JSON file of the next section.

Send to two sinks

The source folder of the example has the file ex-multipart-multiple-endpoints.json. The build does not copy it. In this file, one sampler sends to two sinks.

The address of the sampler has two endpoints, tcp://localhost:5555,tcp://localhost:5556. The socket of the sampler connects to both. The entry of sink2 has the method connect, but the address starts with @:

"id": "sink2",
"channels": [
    {
        "name": "data",
        "sockets": [
            {
                "type": "pull",
                "method": "connect",
                "address": "@tcp://*:5556",
                "sndBufSize": 1000,
                "rcvBufSize": 1000,
                "rateLogging": 0
            }
        ]
    }
]

examples/multipart/ex-multipart-multiple-endpoints.json:41–56

@ in front of an endpoint makes the socket bind there, and + or > makes it connect (Device.cxx). Run each device in its own terminal, with the path of the file in your source tree:

./fairmq-ex-multipart-sink --id sink1 --mq-config /path/to/FairMQ/examples/multipart/ex-multipart-multiple-endpoints.json
./fairmq-ex-multipart-sink --id sink2 --mq-config /path/to/FairMQ/examples/multipart/ex-multipart-multiple-endpoints.json
./fairmq-ex-multipart-sampler --id sampler1 --mq-config /path/to/FairMQ/examples/multipart/ex-multipart-multiple-endpoints.json

The push socket of the sampler distributes the messages over the two sinks. Only the sink that gets the message with stopFlag 1 leaves RUNNING.

Messages and multipart: messages, parts and their memory.
Receive data with callbacks or loops: handlers for messages and for parts.
Channel properties: the address with several endpoints.
1-1: sampler and sink: a sampler and a sink with single-part messages.
Stop a topology without losing messages: end messages, linger and the stop order.