server.hh

#ifndef VTESTBED_SERVER_SERVER_HH
#define VTESTBED_SERVER_SERVER_HH

#include <chrono>
#include <mutex>
#include <unordered_map>

#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 Server {

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

        private:
            sys::socket_address _address;
            sys::socket _socket;
            sys::event_poller _poller;
            std::unordered_map<sys::fd_type,client_type> _clients;
            testbed_type _testbed;
            std::mutex _mutex;
            bool _verbose = false;
            duration _timeout = std::chrono::seconds(7);
            duration _period = std::chrono::seconds(1);
            time_point _nextpush = time_point(duration::zero());
            time_point _lastpush = time_point(duration::zero());

        public:

            inline explicit
            Server(const sys::socket_address& address):
            _address(address) {}

            void
            run() {
                listen();
                _nextpush = clock_type::now();
                _lastpush = _nextpush;
                while (true) {
                    _poller.wait_until(_mutex, _nextpush);
                    time_point now = clock_type::now();
                    if (_nextpush <= now) {
                        push_updates(now);
                        _nextpush = now + _period;
                    }
                    for (const sys::epoll_event& ev : _poller) {
                        if (ev.fd() == _socket.fd()) {
                            accept_connection();
                        } else {
                            handle_client(ev);
                        }
                    }
                    for (auto& pair : _clients) {
                        auto& client = pair.second;
                        client.handshake();
                        client.flush();
                    }
                }
            }

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

            void
            listen() {
                this->info("add server _", _address);
                _socket.bind(_address);
                #if defined(UNISTDX_HAVE_TCP_USER_TIMEOUT)
                VTB_WARN(_socket.set_user_timeout(_timeout));
                #endif
                _socket.listen();
                _poller.emplace(_socket.fd(), sys::event::in);
            }

            void
            accept_connection() {
                sys::socket sock;
                sys::socket_address addr;
                try {
                    _socket.accept(sock, addr);
                } catch (const sys::bad_call& err) {
                    if (err.errc() == std::errc::resource_unavailable_try_again) {
                        return;
                    }
                    throw;
                }
                this->info("add client _", addr);
                #if defined(UNISTDX_HAVE_TCP_USER_TIMEOUT)
                VTB_WARN(sock.set_user_timeout(_timeout));
                #endif
                sys::fd_type fd = sock.fd();
                _poller.emplace(fd, sys::event::in);
                client_type client{std::move(sock), addr, true};
                client.verbose(_verbose);
                _clients.emplace(fd, std::move(client));
            }

            void
            handle_client(const sys::epoll_event& ev) {
                auto result = _clients.find(ev.fd());
                if (result == _clients.end()) {
                    this->error("unknown client, ev _", ev);
                    for (auto& pair : _clients) {
                        this->error(
                            "client _, fd _",
                            pair.second.address(),
                            pair.first
                        );
                    }
                    return;
                }
                auto& client = result->second;
                client.process(ev, _testbed);
                if (ev.hup()) {
                    this->info("remove client _", client.address());
                    _poller.erase(ev);
                    _clients.erase(result);
                }
            }

            void
            push_updates(time_point now) {
                float_duration delta = now - _lastpush;
                if (!(delta < float_duration::zero())) {
                    const T dt = delta.count();
                    _testbed.reset();
                    //_testbed.ship_motion_solver()->min_step(dt);
                    //_testbed.ship_motion_solver()->verbose(false);
                    _testbed.step(dt);
                }
                for (auto& pair : _clients) {
                    auto& client = pair.second;
                    client.push(_testbed);
                }
                _lastpush = now;
            }

        private:

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

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

        };

    }

}

#endif // vim:filetype=cpp