Skip to content

API

Jump to bottom
Istvan Madari edited this page Sep 19, 2019 · 21 revisions

Component API

Message Handlers

Message handlers play central role in RIAPS components. 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) 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 the name value and incoming message type is ValueMsgType:

component C {
    sub value : ValueMsgType;
}

The C++ message handler will be:

virtual void OnValue();

The incoming message is not read from the port buffer, the read must be called explicitly in the handler function:

virtual void OnValue() {
    auto [msg, err] = RecvValue();
}

The naming convention of the Recv() is deducted from the name of the port: Recv{port_name}(). The return type is std::tuple<MessageReader, PortError>. In the example above, the tuple elements are mapped to msg and err variables (C++17).

The err object contains the error code if any error (e.g.: read timeout) occurred during the receive. The error codes are defined in the ZMQ API (ZMQ errno). The PortError class overloads the operator bool(), thus the appropriate way to check if there is any error:

virtual void OnValue() {
    auto [exmpl_msg, err] = RecvValue();
    if (!err) {
        // No error, the message was read successfully
    } else {
        // Error, log the error code
        component_logger()->warn("Recv() error in {}, errorcode: ", __func__, err.error_code());
    }
}

Receive Message

RIAPS uses Cap'n Proto to serialize/deserialize messages. However, the exmpl_msg object in the above example is a NOT a Cap'n Proto object, but a wrapper around the Cap'n Proto buffer. The type of the exmpl_msg is riaps::MessageReader<T>. The MessageReader<T> holds the capnp buffer, automatically releases the allocated memory and provides functions to access the message in type-safe way. The fields of the wrapped capnp message are accessible by the -> operator. For example, print the field msg from the received message:

virtual void OnValue() {
    auto [msg, err] = RecvValue();
    auto logger = component_logger();
    if (!err) {
        // No error, the message was read successfully
        logger()->info("{}", msg->getMsg().cStr());
    } else {
        // Error, log the error code
        logger->warn("Recv() error in {}, errorcode: ", __func__, err.error_code());
    }
}
struct ExampleMessage {

};

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);

Send Message

Strongly-typed Send() functions are generated for each message structure. The naming convention of the Send() is deducted from the name of the port: Send{port_name}(). The return type is PortError. The following model defines a component with a publisher port, the code below shows how to send a message and handle the errors.

component C {
    timer clock 1000;
    pub value : ValueMessage;
}
void C:OnClock() {
    auto time = RecvClock();
    MessageBuilder<messages::ValueMessage> builder;
    builder->setValue(42);
    auto error = SendValue(builder);
    if (error) {
        component_logger()->warn("Error sending message: {}, errorcode: {}", __func__, error.error_code());
    }
}

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:

// Timer handler, called in every sec
void ClientBlock::OnClock() {
    // Read the timestamp, sent by the 'clock'
    auto time = RecvClock();

    // Create a capnp message to be sent
    riaps::MessageBuilder<QueryMsg> query_message;
    // The value is a counter, increased in every `OnClock()`.
    query_message->setMsg(counter_);

    // Send the message. Returns true if the message was sent.
    auto error = SendQuery(query_message);

    // Message was sent succesfully
    if (!error) {
        component_logger()->info("To server: {}", counter_);

        // Waits for the answer.
        // Blocks the component thread until the response.
        auto [answer_message, answer_error] = RecvQuery;
        if (!answer_error) {
            component_logger()->info("From server: {}", answer_message->getMsg().cStr());
        } 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.

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() {
    // Create a capnp message to be sent
    riaps::MessageBuilder<QueryMsg> query_message;
    // The value is a counter, increased in every `OnClock()`.
    query_message->setMsg(counter_);

    // Send the message. Returns true if the message was sent.
    auto error = SendQuery(query_message);

    if (!error){
        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() {
    auto [msg, err] = RecvQuery();
    if (!err)
        component_logger()->info("From server: {}", msg->getMsg().cStr());
}

The OnQuery() is called when a message arrives on the query port (the arrived message is read by the RecQuery()). 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 a problem 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 than the incoming response messages are arriving, then the strict send-recv flow might be violated. One option is to call the send() only when the response for the previous request has already arrived:

void ClientHandler::OnClock(riaps::ports::PortBase *port) {
    // Create a capnp message to be sent
    riaps::MessageBuilder<QueryMsg> query_message;
    // The value is a counter, increased in every `OnClock()`.
    query_message->setMsg(counter_);

    // If no unanswered messages (request sent, but no reply so far).
    if (!has_pending_){
        // Sends the message.
        auto error = SendQuery(query_message);
        if (!error) {
            // Set the flag, component will not send any message until the response.
            has_pending_ = true;
            component_logger()->info("To server: {}", counter_);
        }
        else {
            component_logger()->error("Couldn't send message");
        }
    }
    counter_--;
}

void ClientHandler::OnQuery() {
    auto [msg, err] = RecvQuery();
    if (!err) {
        has_pending_ = false;
        component_logger()->info("From server: {}", message->getMsg().cStr());
    }
}

You may find the presented REQ-REP pattern is not enough flexible in several situations due to the strict send-recv flow. If your application needs more versatile request-response like communication pattern please check the Query/Answer (ROUTER-DEALER in ZeroMQ).

Publish/Subscribe

In contrast to the REQ-REP the PUB-SUB pattern is more flexible, since there is no strict communication flow. The publisher can send() a message at anytime even if the previous message wasn't even received by the subscribers. Actually, the publisher doesn't know whether the sent message reached the subscribers or not, that extra communication would contain overwhelming amount of ACK receipts, which would greatly decrease the publisher's performance.

Let's see how the PUB-SUB pattern can be implemented in RIAPS. The following model defines three components: one publisher, two subscribers.

component PubComponent {
    timer clock 1000;
    pub value: PubMsg;
}

component SubComponentOne {
    sub portone : PubMsg;
}

component SubComponentTwo {
    sub porttwo : PubMsg;
}

The PubComponent publishes a message in every second. Intentionally the name of the port in the subscribers is different to show that RIAPS matches the subscribers with the publishers based on the message type only, the name of the port doesn't matter at all. The implementation of the publisher (PubComponent) is shown below:

void PubComponent::OnClock() {
    // Create a capnp message to be sent
    riaps::MessageBuilder<PubMsg> value_message;
    // The value is a counter, increased in every `OnClock()`.
    value_message->setMsg(counter_);

    // Send the message. Returns true if the message was sent.
    auto error = SendValue(value_message);

    if (!error){
        component_logger()->info("Published: {}", counter_);
    } else {
        component_logger()->error("Couldn't send message");
    }
    counter_--;
}

Note, that even if there is no error during the publish, we cannot be sure that all the subscribers received the message. If you need guaranteed publish-subscribe message delivery, then you have to add extra logic to your application (e.g.: using counters in the messages, implement resend feature in the component). The subscriber side looks very similar to implementation in the REQ-REP example:

void SubComponentOne::OnPortone() {
    auto [msg, err] = RecvPubMsg();
    if (!err) {
        component_logger()->info("From publisher: {}", message->getMsg().cStr());
    }
}

void SubComponentTwo::OnPorttwo() {
    auto [msg, err] = RecvPubMsg();
    if (!err) {
        component_logger()->info("From publisher: {}", message->getMsg().cStr());
    }
}

Query/Answer

Distributed Coordination

Group API

Leader Election

Consensus

Time-Synchronized Coordinated Action

Fault Management

Resource Management

Clone this wiki locally