API

Component API

Message Handlers

Message handlers play central role in RIAPS applications. Message handlers are C++ functions and they called when a given type of message arrived to the ComponentBase. The name of the message handler is derived from the name of the port (defined in the RIAPS model file ❗ LINK ❗) where the message arrived. The naming convention for the message handler is: On{port_name}(...). For example, if a component defines a subscriber port with tha name value and type ValueMsgType:

component C {
    sub value : ValueMsgType;
}

The C++ message handler will be:

virtual void OnValue(messages::ValueMsgType::Reader& msg, riaps::ports::PortBase* port) override;

RIAPS uses Cap'n Proto for serializing/deserializing messages. The msg argument (ValueMsgType::Reader) is a Cap'n Proto object and contains the received message. The PortBase pointer encapsulates the RIAPS port, where the message arrived.

Timers

Timers are defined in the application model file with name and period values. Any number of timers can be defined and they belong to the component. Timers run in the background and sending a message in every period. The following code defines a timer, where the clock is the name of the timer and it sends a message in every 5th second to the component:

component C {
    timer clock 5000; // millisecond
}

Handling a timer event is very similar to the regular message handlers, the arguments are different, but the philosophy is the same: a handler function must be implemented implemented in the component with a particular name. The naming convention for timers is: On{timer_name(...)}.

The corresponding C++ timer handler is:

virtual void OnClock(riaps::ports::PortBase *port);

Request/Response

Behind the scenes RIAPS uses ZeroMQ for messaging. No surprise the same messaging patterns can be used in the RIAPS components (REQUEST-RESPONSE, PUBLISH-SUBSCRIBE, ROUTER-DEALER).

In ZeroMQ the REQ-REP pattern requires strict communication flow. Namely, every send() call must be followed by a recv() call on the same port. (In other words: if you call send() two times without calling the recv() between them, it is an error). In this section we show how to handle REQ-REP messaging in RIAPS.

Let suppose two components are defined in the application model file with a REQ and a REP port:

component Server {
    rep resp  : (QueryMsg, ResponseMsg);
}

component ClientBlock {
    timer clock 1000;
    req query : (QueryMsg, ResponseMsg);
}

ClientBlock component defines a request port (query is the name of the port), and in the parenthesis the message type of the request (QueryMsg) and the response message type (ResponseMsg) are. Server defines the response port, which accepts messages from the request ports.

The following code snippet sends a message in every second and waits for the answer (:exclamation: link_to_the _example_source :exclamation:):

// Timer handler, called in every sec
void ClientBlock::OnClock(riaps::ports::PortBase *port) {
    // Create a capnp message to be sent
    capnp::MallocMessageBuilder message_builder;
    auto query_message = message_builder.initRoot<messages::QueryMsg>();
    // The value is a counter, increased in every `OnClock()`.
    query_message.setMsg(counter_);
    
    // Send the message. Returns true if the message was sent.
    if (SendQuery(message_builder, query_message)){
        component_logger()->info("To server: {}", counter_);

        // Waits for the answer.
        // Blocks the component thread until the response.
        reqreptutorial::messages::ResponseMsg::Reader msg;
        if (RecvQuery(&msg)) {
            component_logger()->info("From server: {}", msg.getMsg());
        } else {
            component_logger()->error("No response from the server");
        }
    } else {
        component_logger()->error("Couldn't send message");
    }
    counter_++;
}

Note, that the above implementation blocks the execution until a RecvQuery() is waiting for the response. After the successful Send the Recv is called, so the strict send-recv flow is kept.

Another way to deal with the request-response pattern is to implement a handler for the incoming message. In this case there is no blocking between the send and the recv, but you have to keep track of the communication flow. The implementation below shows how to implement this use case.

❗ link_to_the _example_source ❗

Add a new component to the existing model file:

component ClientHandler {
    timer clock 1000;
    req query : (QueryMsg, ResponseMsg);
}

ClientHandler component defines the same ports using the same names, the difference is in the implementation:

void ClientHandler::OnClock(riaps::ports::PortBase *port) {
    capnp::MallocMessageBuilder message_builder;
    auto query_message = message_builder.initRoot<messages::QueryMsg>();
    query_message.setMsg(counter_);
    
    // No Recv() call after the Send(), the component thread is not blocked
    if (SendQuery(message_builder, query_message)){
        component_logger()->info("To server: {}", counter_);
    } else {
        component_logger()->error("Couldn't send message");
    }
    counter_--;
}

// Message handler when a message arrives on the `query` port
void ClientHandler::OnQuery(const reqreptutorial::messages::ResponseMsg::Reader &message,
                            riaps::ports::PortBase *port) {
    component_logger()->info("From server: {}", message.getMsg());
}

The OnQuery() is called when a message arrives on the query port (the arrived message is passed in the message argument). However, there is an error in this implementation. The execution of the handlers is synchronous, concurrent execution of the handlers cannot happen, this is not an issue here. The issue in this implementation is that the arrival time of the response messages cannot be guaranteed, thus if the OnClock() is ticking faster then the incoming response messages, then the strict send-recv flow is violated. One option is to call the send() only when the response for the previous request has already arrived:

Publish/Subscribe

Distributed Coordination

Group API

Leader Election

Consensus

Time-Synchronized Coordinated Action

Fault Management

Resource Management