client.hh

#ifndef VTESTBED_SERVER_CLIENT_HH
#define VTESTBED_SERVER_CLIENT_HH

#include <chrono>
#include <mutex>

#include <unistdx/base/log_message>
#include <unistdx/io/poller>
#include <unistdx/net/socket>

#include <vtestbed/core/testbed.hh>
#include <vtestbed/server/macros.hh>
#include <vtestbed/server/remote_client.hh>

namespace vtb {

    namespace srv {

        template <class T>
        class Client {

        private:
            typedef vtb::core::Testbed<T,3> testbed_type;
            typedef std::chrono::system_clock clock_type;
            typedef clock_type::duration duration;
            typedef clock_type::time_point time_point;
            typedef std::recursive_mutex mutex_type;
            typedef std::lock_guard<mutex_type> lock_type;
            typedef Remote_client<T> client_type;

        private:
            client_type _client;
            sys::event_poller _poller;
            testbed_type _testbed;
            mutex_type _mutex;
            bool _verbose = false;
            duration _timeout = std::chrono::seconds(7);

        public:

            inline explicit
            Client(const sys::socket_address& address):
            _client(sys::socket(), address, false)
            {}

            template <class Handler>
            void
            run(Handler handler) {
                connect();
                while (true) {
                    _poller.wait(_mutex);
                    for (const sys::epoll_event& ev : _poller) {
                        if (ev.fd() == _client.socket().fd()) {
                            if (ev.in()) {
                                _client.process(ev, _testbed);
                                handler.on_receive(_testbed);
                            }
                            if (ev.hup()) {
                                this->info("disconnect from _", _client.address());
                                _poller.erase(ev);
                                return;
                            }
                        }
                    }
                    if (_client.handshake()) {
                        handler.on_handshake(_client);
                    }
                    _client.flush();
                }
            }

            void
            connect() {
                this->info("connect to _", _client.address());
                try {
                    _client.socket().connect(_client.address());
                    #if defined(UNISTDX_HAVE_TCP_USER_TIMEOUT)
                    VTB_WARN(_client.socket().set_user_timeout(_timeout));
                    #endif
                } catch (const sys::bad_call& err) {
                    if (err.errc() != std::errc::operation_in_progress) {
                        throw;
                    }
                }
                _poller.emplace(_client.socket().fd(), sys::event::in);
                _poller.notify_one();
            }

            inline void
            verbose(bool rhs) {
                _verbose = rhs;
            }

            inline bool
            verbose() const noexcept {
                return _verbose;
            }

            inline testbed_type&
            testbed() noexcept {
                return _testbed;
            }

            inline const testbed_type&
            testbed() const noexcept {
                return _testbed;
            }

            inline void
            push(const testbed_type& testbed) {
                lock_type lock(_mutex);
                _client.push(testbed);
                _poller.notify_one();
            }

            inline const client_type&
            client() const noexcept {
                return _client;
            }

            inline client_type&
            client() noexcept {
                return _client;
            }

        private:

            template <class ... Args>
            inline void
            info(const Args& ... args) {
                sys::log_message("client", args ...);
            }

        };

    }

}

#endif // vim:filetype=cpp