msgbus.hpp

Go to the documentation of this file.

// Copyright (c) 2021 Intel Corporation.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to
// deal in the Software without restriction, including without limitation the
// rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
// sell copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
// FROM,OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
// IN THE SOFTWARE.
 
#ifndef _EII_MSGBUS_HPP
#define _EII_MSGBUS_HPP
 
#include <atomic>
#include <string>
#include <chrono>
#include <thread>
#include <condition_variable>
#include <eii/utils/profiling.h>
#include <eii/utils/thread_safe_queue.h>
#include "eii/msgbus/msgbus.h"
#include "eii/msgbus/msg_envelope.hpp"
 
namespace eii {
namespace msgbus {
 
// Forward declarations
class ReceiveContext;
class Publisher;
class Service;
class ServiceRequester;
 
typedef ReceiveContext Subscriber;
 
class MsgbusContext {
private:
    // Underlying message bus context
    void* m_msgbus_ctx;
 
    MsgbusContext(const MsgbusContext& src);
 
    MsgbusContext& operator=(const MsgbusContext& src);
 
public:
    MsgbusContext(config_t* config);
 
    ~MsgbusContext();
 
    Publisher* new_publisher(const std::string topic);
 
    Subscriber* new_subscriber(
            const std::string topic, user_data_t* user_data=NULL);
 
    Service* new_service(std::string service_name, void* user_data=NULL);
 
    ServiceRequester* get_service(
            std::string service_name, void* user_data=NULL);
};
 
class ReceiveContext {
protected:
    // Message bus context which the receive context belongs to
    void* m_msgbus_ctx;
 
    // Internal EII Message Bus receive context
    recv_ctx_t* m_recv_ctx;
 
    ReceiveContext(void* msgbus_ctx, recv_ctx_t* recv_ctx);
 
public:
    ~ReceiveContext();
 
    MsgEnvelope* recv_wait();
 
    template<class Rep, class Period>
    MsgEnvelope* recv_timedwait(
            const std::chrono::duration<Rep, Period>& timeout) {
        int timeout_ms = std::chrono::milliseconds(timeout).count();
        msg_envelope_t* msg = NULL;
 
        msgbus_ret_t ret = msgbus_recv_timedwait(
                m_msgbus_ctx, m_recv_ctx, timeout_ms, &msg);
        if (ret == MSG_RECV_NO_MESSAGE) {
            return NULL;
        } else if (ret != MSG_SUCCESS) {
            throw MsgbusException(ret, "Failed to receive message");
        }
 
        return new MsgEnvelope(msg);
    }
 
    MsgEnvelope* recv_nowait();
 
    user_data_t* get_user_data();
 
 
    // MsgbusContext class friend so it can call the private constructor
    friend class MsgbusContext;
};
 
class Service : public ReceiveContext {
private:
    Service(void* msgbus_Ctx, recv_ctx_t* recv_ctx);
 
public:
    ~Service();
 
    void response(MsgEnvelope* response);
 
    // MsgbusContext class friend so it can call the private constructor
    friend class MsgbusContext;
};
 
class ServiceRequester : public ReceiveContext {
private:
    ServiceRequester(void* msgbus_ctx, recv_ctx_t* recv_ctx);
 
public:
    ~ServiceRequester();
 
    void request(MsgEnvelope* request);
 
    // MsgbusContext class friend so it can call the private constructor
    friend class MsgbusContext;
};
 
class Publisher {
private:
    // Publisher's message bus context
    void* m_msgbus_ctx;
 
    // Internal publisher context
    publisher_ctx_t* m_pub_ctx;
 
protected:
    Publisher(void* msgbus_ctx, publisher_ctx_t* pub_ctx);
 
public:
    ~Publisher();
 
    void publish(MsgEnvelope* msg);
 
    // MsgbusContext class friend so it can call the private constructor
    friend class MsgbusContext;
};
 
typedef eii::utils::ThreadSafeQueue<eii::msgbus::Serializable*> MessageQueue;
 
class BaseMsgbusThread {
private:
    // Publisher thread handler
    std::thread* m_th;
 
protected:
    // Message bus context
    MsgbusContext* m_ctx;
 
    // Stop flag
    std::atomic<bool> m_stop;
 
    // Error condition variable to notify users of an error
    std::condition_variable& m_err_cv;
 
    virtual void run() = 0;
 
public:
    BaseMsgbusThread(config_t* msgbus_config, std::condition_variable& err_cv);
 
    virtual ~BaseMsgbusThread();
 
    virtual void start();
 
    virtual void join();
 
    virtual void stop();
};
 
class PublisherThread : public BaseMsgbusThread {
private:
    // Publisher context
    Publisher* m_pub;
 
    // Input message queue
    MessageQueue* m_input_queue;
 
    // AppName variable
    std::string m_service_name;
 
    // Profiling variable
    eii::utils::Profiling* m_profile;
 
    PublisherThread(const PublisherThread& src);
 
    PublisherThread& operator=(const PublisherThread& src);
 
protected:
    void run() override;
 
public:
    PublisherThread(
            config_t* msgbus_config, std::condition_variable& err_cv,
            std::string topic, MessageQueue* input_queue,
            std::string service_name);
 
    ~PublisherThread();
};
 
template<class T>
class SubscriberThread : public BaseMsgbusThread {
private:
    // Subscriber context
    ReceiveContext* m_recv_ctx;
 
    // Output message queue
    MessageQueue* m_output_queue;
 
    // Timeout for time to wait between receiving and checking if the stop
    // signal has been sent
    std::chrono::milliseconds m_timeout;
 
    // AppName variable
    std::string m_service_name;
 
    // Profiling variable
    eii::utils::Profiling* m_profile;
 
    SubscriberThread& operator=(const SubscriberThread& src) { return *this; };
 
protected:
    void run() override {
        LOG_DEBUG_0("Subscriber thread started");
 
        msg_envelope_t* msg = NULL;
        MsgEnvelope* env = NULL;
        T* received = NULL;
        utils::QueueRetCode ret_queue = utils::QueueRetCode::SUCCESS;
 
        // Profiling related variables
        std::string subscriber_ts = m_service_name + "_subscriber_ts";
        std::string subscriber_exit_ts =
            m_service_name + "_subscriber_exit_ts";
        std::string subscriber_blocked_ts =
            m_service_name + "_subscriber_blocked_ts";
 
        try {
            while (!m_stop.load()) {
                env = m_recv_ctx->recv_timedwait(m_timeout);
                if (env == NULL) {
                    // Timeout...
                    continue;
                }
 
                // Received message
                msg = env->get_msg_envelope();
 
                // Add timestamp after message is received if profiling is
                // enabled
                DO_PROFILING(this->m_profile, msg, subscriber_ts.c_str());
 
                // Parse the message into user's provided object
                received = new T(msg);
 
                // env is no longer needed after this point, delete it
                delete env;
                env = NULL;
 
                ret_queue = m_output_queue->push(received);
                if (ret_queue == utils::QueueRetCode::QUEUE_FULL) {
                    ret_queue = m_output_queue->push_wait(received);
                    if(ret_queue != utils::QueueRetCode::SUCCESS) {
                        LOG_ERROR_0("Failed to enqueue received message, "
                                    "message dropped");
                        m_err_cv.notify_all();
                        break;
                    } else {
                        // Dropping pointer to message here because the memory for
                        // the envelope is now owned by the received variable
                        msg = NULL;
                        received = NULL;
                    }
                } else {
                        msg = NULL;
                        received = NULL;
                }
            }
        } catch (const std::exception& ex) {
            LOG_ERROR("Error in subscriber thread: %s", ex.what());
        }
 
        // When the while loop exits, if received is not NULL then it has been
        // initialized with the underlying msg_envelope_t and now owns the
        // deletion of its memory. If msg is not NULL and received is NULL,
        // then this thread owns the msg_envelope_t* structure and must free
        // it.
        if (received != NULL) {
            delete received;
        } else if (msg != NULL) {
            msgbus_msg_envelope_destroy(msg);
        }
 
        // Even if the above are true, if the MsgEnvelope (env) is not NULL it
        // must be deleted. If received is NULL, msg is NULL, and env is not
        // NULL then it currently owns a msg_envelope_t and will approprietly
        // dispose of it.
        if (env != NULL) {
            delete env;
        }
 
        LOG_DEBUG_0("Subscriber thread stopped");
    };
 
public:
    SubscriberThread(config_t* msgbus_config, std::condition_variable& err_cv,
                     std::string topic, MessageQueue* output_queue,
                     std::string service_name,
                     std::chrono::milliseconds timeout=
                         std::chrono::milliseconds(250)) :
        BaseMsgbusThread(msgbus_config, err_cv) {
 
        static_assert(std::is_base_of<Serializable, T>::value,
                  "Template must be subclass of eii::msgbus::Serializable");
 
        m_recv_ctx = m_ctx->new_subscriber(topic);
        m_output_queue = output_queue;
        m_timeout = timeout;
        m_service_name = service_name;
        m_profile = new eii::utils::Profiling();
    };
 
    ~SubscriberThread() {
        this->stop();
        delete m_recv_ctx;
        delete m_ctx;
        delete m_profile;
    };
};
 
}  // namespace msgbus
}  // namespace eii
 
#endif  // _EII_MSGBUS_HPP