Client_Side_RPC.h

Go to the documentation of this file.

#ifndef Client_Side_RPC_h
#define Client_Side_RPC_h
 
// Local includes.
#include "RPC_Request_Callback.h"
#include "IAPI_Implementation.h"
 
 
// client-side RPC topics.
char constexpr RPC_RESPONSE_SUBSCRIBE_TOPIC[] = "v1/devices/me/rpc/response/+";
char constexpr RPC_RESPONSE_TOPIC[] = "v1/devices/me/rpc/response/";
char constexpr RPC_SEND_REQUEST_TOPIC[] = "v1/devices/me/rpc/request/%u";
// Log messages.
char constexpr CLIENT_RPC_METHOD_NULL[] = "Client-side RPC method name is NULL";
#if !THINGSBOARD_ENABLE_DYNAMIC
char constexpr RPC_REQUEST_OVERFLOWED[] = "Client-side RPC request overflowed, increase MaxRequestRPC (%u)";
char constexpr CLIENT_SIDE_RPC_SUBSCRIPTIONS[] = "client-side RPC";
#endif // !THINGSBOARD_ENABLE_DYNAMIC
char constexpr RPC_EMPTY_PARAMS_VALUE[] = "{}";
 
 
#if THINGSBOARD_ENABLE_DYNAMIC
template <typename Logger = DefaultLogger>
#else
template<size_t MaxSubscriptions = DEFAULT_SUBSCRIPTION_AMOUNT, size_t MaxRequestRPC = DEFAULT_REQUEST_RPC_AMOUNT, typename Logger = DefaultLogger>
#endif // THINGSBOARD_ENABLE_DYNAMIC
class Client_Side_RPC : public IAPI_Implementation {
  public:
    Client_Side_RPC() = default;
 
    ~Client_Side_RPC() override = default;
 
    bool RPC_Request(RPC_Request_Callback const & callback) {
        char const * method_name = callback.Get_Name();
 
        if (Helper::String_IsNull_Or_Empty(method_name)) {
            Logger::printfln(CLIENT_RPC_METHOD_NULL);
            return false;
        }
        RPC_Request_Callback * registered_callback = nullptr;
        if (!RPC_Request_Subscribe(callback, registered_callback)) {
            return false;
        }
        else if (registered_callback == nullptr) {
            return false;
        }
 
        JsonArray const * parameters = callback.Get_Parameters();
 
#if THINGSBOARD_ENABLE_DYNAMIC
        // String are const char* and therefore stored as a pointer --> zero copy, meaning the size for the strings is 0 bytes,
        // Data structure size depends on the amount of key value pairs passed + the default method name and params key needed for the request.
        // See https://arduinojson.org/v6/assistant/ for more information on the needed size for the JsonDocument
        TBJsonDocument request_buffer(JSON_OBJECT_SIZE(parameters != nullptr ? parameters->size() + 2U : 2U));
#else
        // Ensure to have enough size for the infinite amount of possible parameters that could be sent to the cloud
        StaticJsonDocument<JSON_OBJECT_SIZE(MaxRequestRPC)> request_buffer;
#endif // THINGSBOARD_ENABLE_DYNAMIC
 
        request_buffer[RPC_METHOD_KEY] = method_name;
 
        if (parameters != nullptr && !parameters->isNull()) {
            request_buffer[RPC_PARAMS_KEY] = *parameters;
        }
        else {
            request_buffer[RPC_PARAMS_KEY] = RPC_EMPTY_PARAMS_VALUE;
        }
 
#if !THINGSBOARD_ENABLE_DYNAMIC
        if (request_buffer.overflowed()) {
            Logger::printfln(RPC_REQUEST_OVERFLOWED, MaxRequestRPC);
            return false;
        }
#endif // !THINGSBOARD_ENABLE_DYNAMIC
 
        size_t * p_request_id = m_get_request_id_callback.Call_Callback();
        if (p_request_id == nullptr) {
            Logger::printfln(REQUEST_ID_NULL);
            return false;
        }
        auto & request_id = *p_request_id;
 
        registered_callback->Set_Request_ID(++request_id);
        auto & request_callback = registered_callback->Get_Request_Timeout();
        request_callback.Start_Timeout_Timer();
 
        char topic[Helper::Calculate_Print_Size(RPC_SEND_REQUEST_TOPIC, request_id)] = {};
        (void)snprintf(topic, sizeof(topic), RPC_SEND_REQUEST_TOPIC, request_id);
        return m_send_json_callback.Call_Callback(topic, request_buffer);
    }
 
    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 {
        auto const request_id = Helper::Split_Topic_Into_Request_ID(topic, strlen(RPC_RESPONSE_TOPIC));
 
#if THINGSBOARD_ENABLE_STL
        auto it = std::find_if(m_rpc_request_callbacks.begin(), m_rpc_request_callbacks.end(), [&request_id](RPC_Request_Callback & rpc_request) {
            return rpc_request.Get_Request_ID() == request_id;
        });
        if (it != m_rpc_request_callbacks.end()) {
            auto & rpc_request = *it;
#else
        for (auto it = m_rpc_request_callbacks.begin(); it != m_rpc_request_callbacks.end(); ++it) {
            auto & rpc_request = *it;
 
            if (rpc_request.Get_Request_ID() != request_id) {
                continue;
            }
#endif // THINGSBOARD_ENABLE_STL
            auto & request_timeout = rpc_request.Get_Request_Timeout();
            request_timeout.Stop_Timeout_Timer();
            rpc_request.Call_Callback(data);
 
            // Delete callback because the changes have been requested and the callback is no longer needed
            m_rpc_request_callbacks.erase(it);
#if !THINGSBOARD_ENABLE_STL
            break;
#endif // !THINGSBOARD_ENABLE_STL
        }
 
        // Attempt to unsubscribe from the shared attribute request topic,
        // if we are not waiting for any further responses with shared attributes from the server.
        // Will be resubscribed if another request is sent anyway
        if (m_rpc_request_callbacks.empty()) {
            (void)RPC_Request_Unsubscribe();
        }
    }
 
    bool Is_Response_Topic_Matching(char const * topic) const override {
        return strncmp(RPC_RESPONSE_TOPIC, topic, strlen(RPC_RESPONSE_TOPIC)) == 0;
    }
 
    bool Unsubscribe() override {
        return RPC_Request_Unsubscribe();
    }
 
    bool Resubscribe_Permanent_Subscriptions() override {
        m_rpc_request_callbacks.clear();
        return Unsubscribe();
    }
 
#if !THINGSBOARD_USE_ESP_TIMER
    void loop() override {
        for (auto & rpc_request : m_rpc_request_callbacks) {
            auto & request_callback = rpc_request.Get_Request_Timeout();
            request_callback.Update_Timeout_Timer();
        }
    }
#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);
        m_get_request_id_callback.Set_Callback(get_request_id_callback);
    }
 
  private:
#if THINGSBOARD_ENABLE_DYNAMIC
    using Callback_Container = Container<RPC_Request_Callback>;
#else
    using Callback_Container = Container<RPC_Request_Callback, MaxSubscriptions>;
#endif // THINGSBOARD_ENABLE_DYNAMIC
 
    bool RPC_Request_Subscribe(RPC_Request_Callback const & callback, RPC_Request_Callback * & registered_callback) {
#if !THINGSBOARD_ENABLE_DYNAMIC
        if (m_rpc_request_callbacks.size() + 1 > m_rpc_request_callbacks.capacity()) {
            Logger::printfln(MAX_SUBSCRIPTIONS_EXCEEDED, CLIENT_SIDE_RPC_SUBSCRIPTIONS, MAX_SUBSCRIPTIONS_TEMPLATE_NAME);
            return false;
        }
#endif // !THINGSBOARD_ENABLE_DYNAMIC
        if (!m_subscribe_topic_callback.Call_Callback(RPC_RESPONSE_SUBSCRIBE_TOPIC)) {
            Logger::printfln(SUBSCRIBE_TOPIC_FAILED, RPC_RESPONSE_SUBSCRIBE_TOPIC);
            return false;
        }
        m_rpc_request_callbacks.push_back(callback);
        registered_callback = &m_rpc_request_callbacks.back();
        return true;
    }
 
    bool RPC_Request_Unsubscribe() {
        (void)Resubscribe_Permanent_Subscriptions();
        return m_unsubscribe_topic_callback.Call_Callback(RPC_RESPONSE_SUBSCRIBE_TOPIC);
    }
 
    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<size_t *>                                       m_get_request_id_callback = {};     // Get internal request id callback
    Callback_Container                                       m_rpc_request_callbacks = {};       // Server side RPC callbacks vector
};
 
#endif // Client_Side_RPC_h