rpc_api.h

Go to the documentation of this file.

/*
-------------------------------------------------------------------------------
                            PLEBBLE
-------------------------------------------------------------------------------
 Copyright (C) 2019-2020 KATLAS Technology. (http://katlastechnology.com)
 Copyright (C) 2017-2020 Marcos Mayorga. (mm@mm-studios.com)
 
 This file is part of our Plebble(R) Platform.
 
 The code below cannot be copied, used for any purpose other than the one
 agreed and/or distributed without the express permission of
 KATLAS Technology.
-------------------------------------------------------------------------------
 
 
*/
#ifndef USGOV_be7dc84108784c4af35dada40b3f9b325e2d9f0782781c75529a2c286e0cfa5c
#define USGOV_be7dc84108784c4af35dada40b3f9b325e2d9f0782781c75529a2c286e0cfa5c
 
#include <us/gov/auth/peer_t.h>
#include <us/gov/socket/client.h>
#include <us/gov/socket/busyled.h>
#include <us/gov/socket/datagram.h>
#include <chrono>
#include <thread>
#include <condition_variable>
#include <mutex>
#include <atomic>
#include <stack>
#include <us/gov/bgtask.h>
#include <us/gov/config.h>
 
namespace us { namespace gov {
namespace cli {
 
using namespace std;
 
struct rpc_api: gov::auth::peer_t, bgtask {
    typedef gov::auth::peer_t b;
    typedef bgtask recv;
    typedef us::gov::socket::client::queue_t queue_t;
    typedef us::gov::socket::datagram::dispatcher_t dispatcher_t;
    typedef us::gov::socket::busyled_t busyled_t;
    using b::pub_t;
 
    struct rendezvous_t final:unordered_map<uint16_t,datagram*> {
        static const char* KO_76103;
        rendezvous_t();
        ~rendezvous_t();
        bool arrived(datagram *);
        std::pair<ko,datagram*> sendrecv(rpc_api& endpoint, datagram*d);
 
        void flush();
 
        uint16_t next_seq{0};
        mutex mx;
        condition_variable cv;
        bool finished{false};
    };
 
    static const char* KO_40928;
    static const char* KO_10210;
 
 
    rpc_api(const b::keys&, const string& host, uint16_t port, role_t role, uint16_t pport_pin, dispatcher_t*);
    rpc_api(const b::keys&, const string& host, uint16_t port, role_t role, uint16_t pport_pin, dispatcher_t*, queue_t* sendq);
    ~rpc_api() override;
    inline const keys& get_keys() const override { return id; }
    inline bool authorize(const pub_t& p, uint16_t pin) const override { return true; }
 
    //static const char* KO_30928;
/*
    void ask_void(uint16_t service, ostream&os);
    void ask_string(uint16_t service, const string& args, ostream&os);
    void ask_void(uint16_t service);
    void ask_string(uint16_t service, const string& args);
*/
 //   ko ask_void(uint16_t service);
 //   ko ask_string(uint16_t service, const string& args);
 
 //   template<typename... args>
 //   datagram* mk_datagram(uint16_t service, args... a) {
 //       return new datagram(service,0,a...);
 //   }
/*
    ko ans_string(datagram* d, ostream&os);
    ko ask_hsh_ans_string(uint16_t service, const hash_t&, const string&, const hash_t&, ostream&os);
    ko ask_ss_ans_string(uint16_t service, const string&s1, const string&s2, ostream&os);
    ko ask_sss_ans_string(uint16_t service, const string&s1, const string&s2, const string&s3, ostream&os);
    ko ask_hs_ans_string(uint16_t service, const hash_t&h, const string&s, ostream&os);
    ko ask_void_ans_string(uint16_t service, ostream&os);
    ko ask_string_ans_string(uint16_t service, const string& args, ostream&os);
*/
 
    ko connect(uint16_t pport_pin, ostream*);
    std::pair<ko,datagram*> sendrecv(datagram*);
    ko send1(datagram*) override;
//    void wait(int ms);
    void wait(const chrono::system_clock::time_point& deadline);
    virtual queue_t* create_sendq();
    void on_peer_disconnected(const string& reason) override;
    void on_connect(ko) override;
 
    ko start() override;
    ko wait_ready(const chrono::system_clock::time_point& deadline) const override;
    void stop() override;
    void join() override;
    bool is_active() const;
    void run_recv();
    void onwakeup();
    ko wait_connected();
    ko wait_connection(bool initiate);
    bool is_connected() const;
    void string_dgram(const datagram&);
    void process_work(datagram*);
 
    void set_busy_handlers(busyled_t::handler_t* hsend, busyled_t::handler_t* hrecv);
 
    static const char* KO_3029;
   //static const char* KO_30112;
 
    condition_variable cv_recv;
    condition_variable cv_connected;
    string host;
    uint16_t port;
    keys id;
    chrono::system_clock::time_point connected_since;
    bool ownsendq;
    role_t role;
    uint16_t pport_pin;
    bool connect_for_recv{true};
    bool stop_on_disconnection{true};
    mutable mutex mx_connect;
    rendezvous_t rendezvous;
    stack<ostream*> cinfo;
    dispatcher_t* dispatcher;
    busyled_t busyled_recv;
    int ms{10000}; //wait after failed connection
    chrono::system_clock::time_point last_try_connect{chrono::system_clock::now()-chrono::milliseconds(ms)};
 
    #if CFG_LOGS==1
        string logdir;
    #endif
};
 
}
}}
 
#endif