ClientCoreState.hpp

/*
 * Copyright 2010-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License").
 * You may not use this file except in compliance with the License.
 * A copy of the License is located at
 *
 *  http://aws.amazon.com/apache2.0
 *
 * or in the "license" file accompanying this file. This file is distributed
 * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
 * express or implied. See the License for the specific language governing
 * permissions and limitations under the License.
 */

/**
 * @file ClientCoreState.hpp
 * @brief
 *
 */

#pragma once

#include <condition_variable>
#include <chrono>

#include "util/Core_EXPORTS.hpp"

#include "util/Utf8String.hpp"
#include "util/memory/stl/Map.hpp"
#include "util/memory/stl/Queue.hpp"

#include "Action.hpp"
#include "ResponseCode.hpp"
#include "NetworkConnection.hpp"

/**
 * Default sleep duration between each execution of Client Core thread operations
 */
#define DEFAULT_CORE_THREAD_SLEEP_DURATION_MS 100

/**
 * Max size of the queue
 */
#define DEFAULT_MAX_QUEUE_SIZE 16

namespace awsiotsdk {

    /**
     * @brief MQTT Disconnect Callback Context Data
     *
     * This class can be used to provide customer context data to be provided with each disconnect callback.
     * Uses a pure virtual destructor to allow for polymorphism
     */
    class DisconnectCallbackContextData {
    public:
        virtual ~DisconnectCallbackContextData() = 0;
    };

    /**
     * @brief MQTT Reconnect Callback Context Data
     *
     * This class can be used to provide customer context data to be provided with each reconnect callback.
     * Uses a pure virtual destructor to allow for polymorphism
     */
    class ReconnectCallbackContextData {
    public:
        virtual ~ReconnectCallbackContextData() = 0;
    };

    /**
     * @brief MQTT Resubscribe Callback Context Data
     *
     * This class can be used to provide customer context data to be provided with each resubscribe callback.
     * Uses a pure virtual destructor to allow for polymorphism
     */
    class ResubscribeCallbackContextData {
    public:
        virtual ~ResubscribeCallbackContextData() = 0;
    };

    /**
     * @brief Client Core State Class
     *
     * Defining a class for the Core Client State.
     * This class is responsible for maintaing the state information for the core client
     * It can also be extended to provide state information to Actions.
     * It contains the action queue, an action registry and also keeps track of
     * the Common Action State as well as the Network connection
     *
     */
    class ClientCoreState : public ActionState {
    protected:

        /**
         * @brief Pending Ack Data Class
         *
         * Defining an internal class for storing information about Pending Acks.
         *
         */
        class PendingAckData {
        public:
            std::chrono::system_clock::time_point time_of_request_;           ///< Time at which the request was sent
            ActionData::AsyncAckNotificationHandlerPtr p_async_ack_handler_;  ///< Handler to which response must be sent
        };

        std::atomic<uint16_t> next_action_id_;                                                   ///< Atomic, ID of the next Action that will be enqueued
        std::atomic_int cur_core_threads_;                                                       ///< Atomic, Count of currently running core threads
        std::atomic_int max_hardware_threads_;                                                   ///< Atomic, Count of the maximum allowed hardware threads
        std::atomic_size_t max_queue_size_;                                                      ///< Atomic, Current configured max queue size
        std::chrono::seconds ack_timeout_;                                                       ///< Timeout for pending Acks, older Acks are deleted with a failed response

        std::mutex register_action_lock_;                                                        ///< Mutex for Register Action Request flow
        std::mutex ack_map_lock_;                                                                ///< Mutex for Ack Map operations

        // Used to perform blocking sync actions
        std::mutex sync_action_request_lock_;                                                    ///< Mutex for Sync Action Request flow
        std::mutex sync_action_response_lock_;                                                   ///< Mutex for Sync Action Response flow
        std::condition_variable sync_action_response_wait_;                                      ///< Condition variable used to wake up calling thread on Sync Action response
        ResponseCode sync_action_response_;                                                      ///< Variable to store received Sync Action response

        std::atomic_bool process_queued_actions_;                                                ///< Atomic, indicates whether currently queued Actions should be processed or not
        std::shared_ptr<std::atomic_bool> continue_execution_;                                   ///< Atomic, Used to synchronize running threads, false value causes running threads to stop

        util::Map<ActionType, std::unique_ptr<Action>> action_map_;                              ///< Map containing currently initialized Action Instances
        util::Map<uint16_t, std::unique_ptr<PendingAckData>> pending_ack_map_;                   ///< Map containing currently pending Acks
        util::Map<ActionType, Action::CreateHandlerPtr> action_create_handler_map_;              ///< Map containing currently registered Action Types and corrosponding Factories

        util::Queue<std::pair<ActionType, std::shared_ptr<ActionData>>> outbound_action_queue_;  ///< Queue of outbound actions

        /**
         * @brief Internal Action Handler for Sync Action responses
         *
         * @param action_id - ID of the Action that response was received for
         * @param rc - Received response
         */
        void SyncActionHandler(uint16_t action_id, ResponseCode rc);

    public:
        /**
         * @brief Define Handler for Disconnect Callbacks
         *
         * This handler is used to provide notification to the application when a disconnect occurs
         * NOTE: This handler should be NON-BLOCKING
         */
        typedef std::function<ResponseCode(util::String mqtt_client_id,
                                           std::shared_ptr<DisconnectCallbackContextData> p_app_handler_data)> ApplicationDisconnectCallbackPtr;

        ApplicationDisconnectCallbackPtr disconnect_handler_ptr_;                         ///< Pointer to the Application Disconnect Callback
        std::shared_ptr<DisconnectCallbackContextData> p_disconnect_app_handler_data_;  ///< Data to be passed to the Application Handler

        /**
         * @brief Define Handler for Reconnect Callbacks
         *
         * This handler is used to provide notification to the application when a reconnect occurs
         * NOTE: This handler should be NON-BLOCKING
         */
        typedef std::function<ResponseCode(util::String mqtt_client_id,
                                           std::shared_ptr<ReconnectCallbackContextData> p_app_handler_data,
                                           ResponseCode reconnect_result)> ApplicationReconnectCallbackPtr;

        ApplicationReconnectCallbackPtr reconnect_handler_ptr_;                           ///< Pointer to the Application Reconnect Callback
        std::shared_ptr<ReconnectCallbackContextData> p_reconnect_app_handler_data_;   ///< Data to be passed to the Application Handler

        /**
         * @brief Define Handler for Resubscribe Callbacks
         *
         * This handler is used to provide notification to the application when a resubscribe occurs.
         * NOTE: This handler should be NON-BLOCKING
         */
        typedef std::function<ResponseCode(util::String mqtt_client_id,
                                           std::shared_ptr<ResubscribeCallbackContextData> p_app_handler_data,
                                           ResponseCode resubscribe_result)> ApplicationResubscribeCallbackPtr;

        ApplicationResubscribeCallbackPtr resubscribe_handler_ptr_;                       ///< Pointer to the Application Resubscribe Callback
        std::shared_ptr<ResubscribeCallbackContextData> p_resubscribe_app_handler_data_; ///< Data to be passed to the Application Handler

        /**
         * @brief Network connection instance to use for this instance of the Client
         * This is shared between Actions, public to avoid multiple shared pointer operations while passing as argument by ClientCore
         */
        std::shared_ptr<NetworkConnection> p_network_connection_;

        /**
         * @brief Overload for Get next Action ID
         * @return uint16_t Action ID
         */
        virtual uint16_t GetNextActionId() {
            if (UINT16_MAX == next_action_id_) {
                next_action_id_ = 1;
                return UINT16_MAX;
            }
            return next_action_id_++;
        };

        /**
         * @brief Get current value of maximum action queue size
         * @return size_t max_queue_size_
         */
        size_t GetMaxActionQueueSize() { return max_queue_size_; }

        /**
         * @brief Set max size for action queue
         * @param size_t max_queue_size
         */
        void SetMaxActionQueueSize(size_t max_queue_size) { max_queue_size_ = max_queue_size; }

        /**
         * @brief Get pointer to sync point used for execution status of the Core instance
         *
         * This sync point is used to indicate SDK is still continuing execution. Set to false when exiting
         *
         * @return std::shared_ptr<std::atomic_bool> pointer to the sync point
         */
        std::shared_ptr<std::atomic_bool> GetCoreExecutionSyncPoint() { return continue_execution_; }

        /**
         * @brief Sets whether the Client is allowed to process queue actions
         * @param process_queued_actions value to set it to
         */
        void SetProcessQueuedActions(bool process_queued_actions) { process_queued_actions_ = process_queued_actions; }

        /**
         * @brief Get whether the Client can process queued actions
         * @return boolean value indicating status
         */
        bool CanProcessQueuedActions() { return process_queued_actions_; }

        /**
         * @brief Process the outbound action queue
         *
         * This function processes the actions queued up in the Outbound action queue.
         * The function accepts a Sync point that can be used to control execution in a separate thread.
         * If the value is set to false for the sync point, the function will perform one action from the queue.
         * This puts the running thread to sleep if there are no queued up actions.
         * DO NOT call from main thread unless you have a separate thread to queue up actions
         *
         * @param thread_task_out_sync
         */
        void ProcessOutboundActionQueue(std::shared_ptr<std::atomic_bool> thread_task_out_sync);

        /**
         * @brief Perform Action in Blocking Mode
         *
         * This API will perform the Action in Blocking mode. The timeout for the action to give a valid response
         * is provided as an argument. This API stops processing of all outbound actions until Response is received
         * for the requested Action Type
         *
         * @param action_type - Type of the Action to be executed. Must be registered
         * @param action_data - Action Data to be passed as argument to the Action instance
         * @param action_reponse_timeout - Timeout for this API call
         * @return ResponseCode indicating result of the API call
         */
        ResponseCode PerformAction(ActionType action_type, std::shared_ptr<ActionData> action_data,
                                   std::chrono::milliseconds action_reponse_timeout);

        /**
         * @brief Register Action for execution by Client Core
         *
         * This function allows Actions to be registered to be executed at a later stage by Client Core.
         * Actions must be registered before PerformAction can be called using the Action Type.
         * This also applies to Creating Action runners which allow running Actions in dedicated Thread Tasks.
         * Only one Action can be registered to each Action Type. If a second call is made with the same Action Type,
         * the previous registration will be overwritten
         *
         * @param action_type - Type of the Action that will be created using the provided handler
         * @param p_action_create_handler - Factory method pointer which returns an Action instance
         * @param p_action_state - Shared_ptr to use as argument for Action create
         * @return ResponseCode indicating result of the API call
         */
        ResponseCode RegisterAction(ActionType action_type, Action::CreateHandlerPtr p_action_create_handler,
                                    std::shared_ptr<ActionState> p_action_state);

        /**
         * @brief Get the Create Factory Method for the specified action type
         *
         * @param action_type - Type of the Action for which the handler is required
         * @param p_action_create_handler[out] - Create factory method for this Action
         * @return ResponseCode indicating result of the API call
         */
        ResponseCode GetActionCreateHandler(ActionType action_type, Action::CreateHandlerPtr *p_action_create_handler);

        /**
         * @brief Enqueue Action for processing in Outbound Queue
         *
         * @param action_type - Type of the Action
         * @param action_data - Data to be passed to perform Action
         * @param action_id_out[out] - Action ID that was assigned to this action by the Client
         * @return ResponseCode indicating result of the API call
         */
        ResponseCode EnqueueOutboundAction(ActionType action_type, std::shared_ptr<ActionData> action_data,
                                           uint16_t &action_id_out);

        /**
         * @brief Register Ack Handler for provided action id
         * @param action_id - Action ID
         * @param p_async_ack_handler - Handler to call on response
         * @return ResponseCode indicating result of the API call
         */
        ResponseCode RegisterPendingAck(uint16_t action_id,
                                        ActionData::AsyncAckNotificationHandlerPtr p_async_ack_handler);

        /**
         * @brief Delete Ack Handler for specified Action ID
         * @param action_id - Action ID
         */
        void DeletePendingAck(uint16_t action_id);

        /**
         * @brief Call registered Ack handler if it exists for specified Packet id
         * @param action_id - Action ID
         * @param rc - Response Code to pass to the Handler if found
         */
        void ForwardReceivedAck(uint16_t action_id, ResponseCode rc);

        /**
         * @brief Delete all expired Acks
         *
         * Deletes all Acks where the timeouts have expired. Responds with Code indicating request timeout
         */
        void DeleteExpiredAcks();

        /**
         * @brief Clears all registered Actions
         *
         * Utility method to remove all registered actions by the client.
         * Also helps in breaking out of cyclic reference introduced when ::RegisterAction is called.
         */
        void ClearRegisteredActions();

        /**
         * @brief Clears all pending outbound Actions.
         *
         * Utility method to remove all pending outbound actions registered by the client.
         * Also helps in breaking out of cyclic reference introduced when ::EnqueueOutboundAction is called.
         */
        void ClearOutboundActionQueue();

        /**
         * @brief Default Constructor
         */
        ClientCoreState();

        /**
         * @brief Destructor
         */
        virtual ~ClientCoreState();

        // Rule of 5 stuff
        // Contains data for running thread tasks, should not be moved or copied
        ClientCoreState(const ClientCoreState &) = delete;               // Copy constructor
        ClientCoreState(ClientCoreState &&) = delete;                    // Move constructor
        ClientCoreState &operator=(const ClientCoreState &) & = delete;  // Copy assignment operator
        ClientCoreState &operator=(ClientCoreState &&) & = delete;       // Move assignment operator
    };
}