TSAPlatformSocket.h

/*  ===================================================================
*
*                 T R A D I N G   S Y S T E M   A P I ™
*    Copyright © 1999 - 2014 by Peter Ritter ( TradingSystemAPI.com )
*                A L L   R I G H T S   R E S E R V E D
*
*     Consult your license regarding permissions and restrictions.
*             You may obtain a copy of the License at:
*     http://www.TradingSystemAPI.com/licenses/LICENSE-TSAPI-3.0.html
*
*  ====================================================================
*/

#ifndef TSA_PLATFORM_SOCKET__INCLUDED
#define TSA_PLATFORM_SOCKET__INCLUDED

#include <map>
#include <vector>
#include <chrono>
#include <thread>
#include <atomic>

#include "TSATypeDef.h"
#include "TSAStreams.h"
#include "TSAOrder.h"
#include "TSAIntraday.h"
#include "TSAPlatformMsg.h"
#include "TSAJSON.h"

namespace tsa {

    class transaction;
    class strategy;
    class platform_proxy;

    namespace net {

        /*
        ** ====================================================
        ** >>>             class socket                     <<<
        ** ====================================================
        */

        class dll_export socket_base : public object {
            tsa_declare_testable;
          protected:
            bool test__no_dispatch_to_strategy = false;
            bool test__no_thread = false;
          public:
              static std::atomic<int> s_signal_value;
          protected:
            struct header {
                int process_id;
                int strategy_id;
                message::type message_type;
            };
            std::vector<strategy*> m_strategies;
            mutable std::mutex m_mutex;
            mutable std::mutex m_mutex_handle_exception;
          public:
            const std::string& id(void)const;
            void id(const std::string& id);
          private:
            std::string m_idendity;
            bool m_has_already_run = false;
            bool m_message_pump_is_running = false;
            bool m_is_connected = false;
            bool m_strategies_are_running = false;
          private:
            mutable std::mutex m_shutdown_request_mutex;
            bool m_shutdown_requested = false;
            std::string m_strategy_exception_what;
            enum class shutdown_type {
                normal_shutdown,
                shutdown_with_exception
            };
            shutdown_type m_shutdown_type = shutdown_type::normal_shutdown;
            void request_shutdown(shutdown_type type, const std::string& exception_what = "");

          public:
            bool is_running(void)const;
            bool is_connected(void)const;
            void set_connected(bool);
          public:
            socket_base(void);
            virtual ~socket_base(void);
            void run(void);
            void on_strategy_exception(const strategy*, const std::exception&);
          public:
            //==============
            //  Strategies
            //==============

            // Calls platform initialize.
            void init_strategies(void);
            // Number of strategies
            size_t strategy_count(void)const;
            // Registers a new strategy
            identifier_t register_strategy(strategy*);
            void remove_strategy_ref(strategy*);
            bool has_valid_strategy__no_lock(void)const;
            //void exit(void);
            // Removes a strategy with given identifier
            void remove_strategy(identifier_t);
            // Returns the strategy pointer for given pos
            strategy* strategy_ptr(int pos)const;
            // Returns the strategy for given id.
            //tsa::strategy& strategy_ref(identifier_t id)const;
          public:
            // sends message
            virtual void send(net::message&) = 0;
            // respond to incoming message
            virtual void on_message(net::message&);
            // respond to error
            virtual void on_error(const std::string& error);
            //
            virtual void on_connection_lost(void);
          public:
            // waits for a message to arrive within given timeout interval
            virtual bool wait_message(net::message&, int usec_timeout) = 0;
          public:
            // dispatch message to one of the strategies
            void dispatch_message(net::message&);
          private:
            void send_ctrl_c_exit_command_to_all_strategies(void);
          protected:
            //========
            //  RPC
            //========
            std::mutex m_mutex__RPC_values;
            std::map<int, std::string> m_RPC_return_vals;
            int m_RPC_token_count = 1;
            int RPC_next_token(void);
          public:
            bool RPC(const json& call_definition, json& response, int timeout_millisec = 6000);
            void RPC__except(const json& call_definition, json& response, int timeout_millisec = 6000);

            int send__RPC_request(const json& call_def);// , identifier_t strat_id);
            bool fetch_RPC_result(int token, std::string& return_value, int timeout_millisec = 6000);

            void RPC__set_return_value(int token, const std::string& value);
            bool RPC__return_value_exists(int token);
            bool RPC__return_value_exists__no_lock(int token);

            void notify_strategy_stopped(identifier_t strat_id, const std::string& exit_return_string);
          protected:
            //========
            // THREAD
            //========
            std::thread m_message_pump_thread;
            void thread_fn__message_pump(void);
            bool m_stop_msg_pump_thread = false;

            std::thread m_command_line_thread;
            void thread_fn__command_line(void);
            bool m_stop_command_line_thread = false;

            void start_message_pump_thread(void);
            //void start_command_line_thread(void);
            void start_threads(void);
          public:
            void stop_threads(void);
          public:
            //========
            // LOG
            //========
            void write_log(const std::string& error);
        };


    }//namespace net

}//namespace tsa

#endif