Server_Side_RPC.h

Go to the documentation of this file.

#ifndef Server_Side_RPC_h
#define Server_Side_RPC_h
 
// Local includes.
#include "RPC_Callback.h"
#include "IAPI_Implementation.h"
 
 
// server-side RPC topics.
char constexpr RPC_SUBSCRIBE_TOPIC[] = "v1/devices/me/rpc/request/+";
char constexpr RPC_REQUEST_TOPIC[] = "v1/devices/me/rpc/request/";
char constexpr RPC_SEND_RESPONSE_TOPIC[] = "v1/devices/me/rpc/response/%u";
// Log messages.
char constexpr RPC_RESPONSE_OVERFLOWED[] = "Server-side RPC response overflowed, increase MaxRPC (%u)";
#if !THINGSBOARD_ENABLE_DYNAMIC
char constexpr SERVER_SIDE_RPC_SUBSCRIPTIONS[] = "server-side RPC";
#endif // !THINGSBOARD_ENABLE_DYNAMIC
#if THINGSBOARD_ENABLE_DEBUG
char constexpr SERVER_RPC_METHOD_NULL[] = "Server-side RPC method name is NULL";
char constexpr RPC_RESPONSE_NULL[] = "Response JsonDocument is NULL, skipping sending";
char constexpr NO_RPC_PARAMS_PASSED[] = "No parameters passed with RPC, passing null JSON";
char constexpr CALLING_RPC_CB[] = "Calling subscribed callback for rpc with methodname (%s)";
#endif // THINGSBOARD_ENABLE_DEBUG
 
 
#if THINGSBOARD_ENABLE_DYNAMIC
template <typename Logger = DefaultLogger>
#else
template<size_t MaxSubscriptions = DEFAULT_SUBSCRIPTION_AMOUNT, size_t MaxRPC = DEFAULT_RPC_AMOUNT, typename Logger = DefaultLogger>
#endif // THINGSBOARD_ENABLE_DYNAMIC
class Server_Side_RPC : public IAPI_Implementation {
  public:
    Server_Side_RPC() = default;
 
    ~Server_Side_RPC() override = default;
 
    template<typename InputIterator>
    bool RPC_Subscribe(InputIterator const & first, InputIterator const & last) {
#if !THINGSBOARD_ENABLE_DYNAMIC
        auto const size = Helper::distance(first, last);
        if (m_rpc_callbacks.size() + size > m_rpc_callbacks.capacity()) {
            Logger::printfln(MAX_SUBSCRIPTIONS_EXCEEDED, SERVER_SIDE_RPC_SUBSCRIPTIONS, MAX_SUBSCRIPTIONS_TEMPLATE_NAME);
            return false;
        }
#endif // !THINGSBOARD_ENABLE_DYNAMIC
        (void)m_subscribe_topic_callback.Call_Callback(RPC_SUBSCRIBE_TOPIC);
        // Push back complete vector into our local m_rpc_callbacks vector.
        m_rpc_callbacks.insert(m_rpc_callbacks.end(), first, last);
        return true;
    }
 
    bool RPC_Subscribe(RPC_Callback const & callback) {
#if !THINGSBOARD_ENABLE_DYNAMIC
        if (m_rpc_callbacks.size() + 1 > m_rpc_callbacks.capacity()) {
            Logger::printfln(MAX_SUBSCRIPTIONS_EXCEEDED, SERVER_SIDE_RPC_SUBSCRIPTIONS, MAX_SUBSCRIPTIONS_TEMPLATE_NAME);
            return false;
        }
#endif // !THINGSBOARD_ENABLE_DYNAMIC
        (void)m_subscribe_topic_callback.Call_Callback(RPC_SUBSCRIBE_TOPIC);
        m_rpc_callbacks.push_back(callback);
        return true;
    }
 
    bool RPC_Unsubscribe() {
        m_rpc_callbacks.clear();
        return m_unsubscribe_topic_callback.Call_Callback(RPC_SUBSCRIBE_TOPIC);
    }
 
    API_Process_Type Get_Process_Type() const override {
        return API_Process_Type::JSON;
    }
 
    void Process_Response(char const * topic, uint8_t * payload, uint32_t length) override {
        // Nothing to do
    }
 
    void Process_Json_Response(char const * topic, JsonDocument const & data) override {
        if (!data.containsKey(RPC_METHOD_KEY)) {
#if THINGSBOARD_ENABLE_DEBUG
            Logger::printfln(SERVER_RPC_METHOD_NULL);
#endif // THINGSBOARD_ENABLE_DEBUG
            return;
        }
        char const * method_name = data[RPC_METHOD_KEY];
 
#if THINGSBOARD_ENABLE_STL
        auto it = std::find_if(m_rpc_callbacks.begin(), m_rpc_callbacks.end(), [&method_name](RPC_Callback const & rpc) {
            char const * subscribedMethodName = rpc.Get_Name();
            return (!Helper::String_IsNull_Or_Empty(subscribedMethodName) && strncmp(subscribedMethodName, method_name, strlen(subscribedMethodName)) == 0);
        });
        if (it != m_rpc_callbacks.end()) {
            auto & rpc = *it;
#else
        for (auto const & rpc : m_rpc_callbacks) {
            char const * subscribedMethodName = rpc.Get_Name();
            if (Helper::String_IsNull_Or_Empty(subscribedMethodName) || strncmp(subscribedMethodName, method_name, strlen(subscribedMethodName)) != 0) {
              continue;
            }
#endif // THINGSBOARD_ENABLE_STL
#if THINGSBOARD_ENABLE_DEBUG
            if (!data.containsKey(RPC_PARAMS_KEY)) {
                Logger::printfln(NO_RPC_PARAMS_PASSED);
            }
#endif // THINGSBOARD_ENABLE_DEBUG
 
#if THINGSBOARD_ENABLE_DEBUG
            Logger::printfln(CALLING_RPC_CB, method_name);
#endif // THINGSBOARD_ENABLE_DEBUG
 
            JsonVariantConst const param = data[RPC_PARAMS_KEY];
#if THINGSBOARD_ENABLE_DYNAMIC
            auto const & rpc_response_size = rpc.Get_Response_Size();
            TBJsonDocument json_buffer(rpc_response_size);
#else
            size_t constexpr rpc_response_size = MaxRPC;
            StaticJsonDocument<JSON_OBJECT_SIZE(MaxRPC)> json_buffer;
#endif // THINGSBOARD_ENABLE_DYNAMIC
            rpc.Call_Callback(param, json_buffer);
 
            if (json_buffer.isNull()) {
#if THINGSBOARD_ENABLE_DEBUG
                Logger::printfln(RPC_RESPONSE_NULL);
#endif // THINGSBOARD_ENABLE_DEBUG
                return;
            }
            else if (json_buffer.overflowed()) {
                Logger::printfln(RPC_RESPONSE_OVERFLOWED, rpc_response_size);
                return;
            }
 
            auto const request_id = Helper::Split_Topic_Into_Request_ID(topic, strlen(RPC_REQUEST_TOPIC));
            char responseTopic[Helper::Calculate_Print_Size(RPC_SEND_RESPONSE_TOPIC, request_id)] = {};
            (void)snprintf(responseTopic, sizeof(responseTopic), RPC_SEND_RESPONSE_TOPIC, request_id);
            (void)m_send_json_callback.Call_Callback(responseTopic, json_buffer);
            return;
        }
    }
 
    bool Is_Response_Topic_Matching(char const * topic) const override {
        return strncmp(RPC_REQUEST_TOPIC, topic, strlen(RPC_REQUEST_TOPIC)) == 0;
    }
 
    bool Unsubscribe() override {
        return RPC_Unsubscribe();
    }
 
    bool Resubscribe_Permanent_Subscriptions() override {
        if (!m_rpc_callbacks.empty() && !m_subscribe_topic_callback.Call_Callback(RPC_SUBSCRIBE_TOPIC)) {
            Logger::printfln(SUBSCRIBE_TOPIC_FAILED, RPC_SUBSCRIBE_TOPIC);
            return false;
        }
        return true;
    }
 
#if !THINGSBOARD_USE_ESP_TIMER
    void loop() override {
        // Nothing to do
    }
#endif // !THINGSBOARD_USE_ESP_TIMER
 
    void Initialize() override {
        // Nothing to do
    }
 
    void Set_Client_Callbacks(Callback<void, IAPI_Implementation &>::function subscribe_api_callback, Callback<bool, char const * const, JsonDocument const &>::function send_json_callback, Callback<bool, char const * const, char const * const>::function send_json_string_callback, Callback<bool, char const * const>::function subscribe_topic_callback, Callback<bool, char const * const>::function unsubscribe_topic_callback, Callback<uint16_t>::function get_receive_size_callback, Callback<uint16_t>::function get_send_size_callback, Callback<bool, uint16_t, uint16_t>::function set_buffer_size_callback, Callback<size_t *>::function get_request_id_callback) override {
        m_send_json_callback.Set_Callback(send_json_callback);
        m_subscribe_topic_callback.Set_Callback(subscribe_topic_callback);
        m_unsubscribe_topic_callback.Set_Callback(unsubscribe_topic_callback);
    }
 
  private:
#if THINGSBOARD_ENABLE_DYNAMIC
    using Callback_Container = Container<RPC_Callback>;
#else
    using Callback_Container = Container<RPC_Callback, MaxSubscriptions>;
#endif // THINGSBOARD_ENABLE_DYNAMIC
 
    Callback<bool, char const * const, JsonDocument const &> m_send_json_callback = {};         // Send json document callback
    Callback<bool, char const * const>                       m_subscribe_topic_callback = {};   // Subscribe mqtt topic client callback
    Callback<bool, char const * const>                       m_unsubscribe_topic_callback = {}; // Unubscribe mqtt topic client callback
    Callback_Container                                       m_rpc_callbacks = {};              // server-side RPC callbacks array
};
 
#endif // Server_Side_RPC_h