connector.h

Go to the documentation of this file.

 
#ifndef QB_IO_ASYNC_TCP_CONNECTOR_H
#define QB_IO_ASYNC_TCP_CONNECTOR_H
 
#include <qb/io.h>
#include <qb/io/system/sys__socket.h>
#include "../../uri.h"
#include "../event/io.h"
#include "../listener.h"
 
namespace qb::io::async::tcp {

template <typename Socket_, typename Func_>
class connector {
    Func_        _func;    
    const double _timeout; 
    Socket_      _socket;  
    uri          _remote;  
 
private:
    void _establish_connection() {
        LOG_DEBUG("Started async connect to " << _remote.source());
        auto ret = _socket.n_connect(_remote);
        if (!ret) {
            LOG_DEBUG("Connected directly to " << _remote.source());
            _func(std::move(_socket));
        } else if (socket_no_error(qb::io::socket::get_last_errno())) {
            listener::current
                .registerEvent<event::io>(*this, _socket.native_handle(), EV_WRITE)
                .start();
            return;
        } else {
            _socket.disconnect();
            LOG_DEBUG("Failed to connect to "
                      << _remote.source() << " err=" << qb::io::socket::get_last_errno());
            _func(Socket_{});
        }
        delete this;
    }
 
public:
    connector(uri const &remote, Func_ &&func, double timeout = 0.)
        : _func(std::forward<Func_>(func))
        , _timeout(timeout > 0. ? ev_time() + timeout : 0.)
        , _remote{remote} {
        LOG_DEBUG("Connector: Initializing for " << remote.source());
        // _socket is default-initialized here
        _establish_connection();
    }

    connector(Socket_&& existing_socket, uri const &remote, Func_ &&func, double timeout = 0.)
        : _func(std::forward<Func_>(func))
        , _timeout(timeout > 0. ? ev_time() + timeout : 0.)
        , _socket(std::move(existing_socket)) // Move the existing socket
        , _remote{remote} {
        LOG_DEBUG("Connector: Initializing with existing socket for " << remote.source());
        _establish_connection();
    }

    void
    on(event::io const &event) {
        int err = 0;
        if (!(event._revents & EV_WRITE) ||
            _socket.template get_optval<int>(SOL_SOCKET, SO_ERROR, err)) {
            _socket.disconnect();
            err = 1;
        } else if (err && (err != EISCONN) && (!_timeout || ev_time() < _timeout))
            return;
        listener::current.unregisterEvent(event._interface);
        if (!err || err == EISCONN) {
            LOG_DEBUG("Connected async to " << _remote.source());
            _socket.connected();
            _func(std::move(_socket));
        } else {
            LOG_DEBUG("Failed to connect to " << _remote.source() << " err="
                                              << qb::io::socket::get_last_errno());
            _func(Socket_{});
        }
        delete this;
    }
};

template <typename Socket_, typename Func_>
void
connect(uri const &remote, Func_ &&func, double timeout = 0.) {
    new connector<Socket_, Func_>(remote, std::forward<Func_>(func), timeout);
}

template <typename Socket_, typename Func_>
void
connect(Socket_&& existing_socket, uri const &remote, Func_ &&func, double timeout = 0.) {
    new connector<Socket_, Func_>(std::move(existing_socket), remote, std::forward<Func_>(func), timeout);
}
 
} // namespace qb::io::async::tcp
 
#endif // QB_IO_ASYNC_TCP_CONNECTOR_H