ioasync_client.h

Go to the documentation of this file.

// Evo C++ Library
/* Copyright 2019 Justin Crowell
Distributed under the BSD 2-Clause License -- see included file LICENSE.txt for details.
*/

#pragma once
#ifndef INCL_evo_ioasync_client_h
#define INCL_evo_ioasync_client_h

#include "ioasync_base.h"

namespace evo {


template<class T, class Q>
class AsyncClient : public AsyncBase {
public:
    typedef T ProtocolHandler;      
    typedef Q QueueItem;            
    typedef AsyncClient<T,Q> This;  

    struct OnConnect {
        virtual ~OnConnect() { }

        virtual void on_connect() { }
    };

    struct OnError {
        virtual ~OnError() { }

        virtual void on_error(AsyncError error) {
            EVO_PARAM_UNUSED(error);
        }
    };

    enum State {
        sNONE = 0,      
        sCONNECTING,    
        sCONNECTED      
    };

    AsyncClient(SizeT max_queue_size, SizeT max_read_size) : queue_(max_queue_size), id_(get_next_id()), state_(sNONE), bev_(NULL), on_connect_(NULL), on_error_(NULL), max_read_size_(max_read_size) {
    }

    ~AsyncClient() {
        close();
    }
    
    ulong get_id() const {
        return id_;
    }

    State get_state() const {
        return state_;
    }

    This& attach_to(AsyncBase& parent) {
        init_attach(parent);
        return *this;
    }

    void close() {
        if (state_ > sNONE) {
            bufs_.reset();
            ::bufferevent_free(bev_);
            bev_   = NULL;
            state_ = sNONE;
            queue_.clear();
            ((ProtocolHandler&)*this).on_close();
            if (logger.check(LOG_LEVEL_DEBUG_LOW))
                logger.log(LOG_LEVEL_DEBUG_LOW, String().reserve(32) << "AsyncClient " << id_ << " closed");
        }
    }

    This& set_on_connect(OnConnect* cb) {
        on_connect_ = cb;
        return *this;
    }

    This& set_on_error(OnError* cb) {
        on_error_ = cb;
        return *this;
    }

    bool connect_ip(const char* host, ushort port, int family=AF_INET) {
        close();
        SocketAddressInfo address_info(family);
        Error err = address_info.convert(host, port);
        if (err == ENone) {
            assert( address_info.ptr->ai_addrlen > 0 && (size_t)address_info.ptr->ai_addrlen <= (size_t)Int::MAX );
            if (connect_new(address_info.ptr->ai_addr, (int)address_info.ptr->ai_addrlen)) {
                if (logger.check(LOG_LEVEL_DEBUG)) {
                    const SubString host_str(host);
                    logger.log_direct(LOG_LEVEL_DEBUG, String().reserve(50 + host_str.size()) << "AsyncClient " << id_ << " connect_ip: '" << host_str << "' port " << port);
                }
                return true;
            }
            if (logger.check(LOG_LEVEL_ERROR))
                logger.log_direct(LOG_LEVEL_ERROR, String().reserve(44) << "AsyncClient " << id_ << " connect_new() failed");
        } else if (logger.check(LOG_LEVEL_ERROR)) {
            const SubString host_str(host);
            logger.log_direct(LOG_LEVEL_ERROR, String().reserve(58 + host_str.size()) << "AsyncClient " << id_ << " connect_ip() failed on bad host: '" << host_str << "'");
        }
        return false;
    }

protected:
    struct PreQueueItem {
        String buf;         
        QueueItem item;     

        PreQueueItem() {
        }

        PreQueueItem(const PreQueueItem& src) : buf(src.buf), item(src.item) {
        }

        PreQueueItem& operator=(const PreQueueItem& src) {
            buf = src.buf;
            item = src.item;
            return *this;
        }
    };

    struct RequestWriter : public AsyncBuffers::BulkWrite {
        typedef This Parent;    

        Parent& parent;         
        PreQueueItem pq;        

        RequestWriter(Parent& parent, size_t buf_size) : parent(parent) {
            init(parent.bufs_, buf_size);
        }

        ~RequestWriter() {
            if (!pq.item.null())
                parent.queue_.add(pq.item);
        }
    };

    AtomicBufferQueue<QueueItem> queue_;    

private:
    // Disable copy constructor
    AsyncClient(const This&) EVO_ONCPP11(= delete);

    AsyncBuffers bufs_;

    ulong id_;
    State state_;
    struct bufferevent* bev_;

    OnConnect* on_connect_;
    OnError*   on_error_;
    SizeT read_fixed_size_;
    SizeT max_read_size_;

    bool check_client_active() {
        return (!queue_.empty() || bufs_.write_size() > 0);
    }

    bool connect_new(struct sockaddr* addr, int addr_len) {
        init();
        bev_ = ::bufferevent_socket_new(evloop_->handle(), -1, BEV_OPT_CLOSE_ON_FREE);
        ::bufferevent_setcb(bev_, on_read, NULL, on_event, this);
        ::bufferevent_setwatermark(bev_, EV_READ, ProtocolHandler::MIN_INITIAL_READ, max_read_size_);
        if (read_timeout_ms_ > 0 || write_timeout_ms_ > 0) {
            struct timeval read_timeout, write_timeout;
            ::bufferevent_set_timeouts(bev_,
                get_timeout_ptr(read_timeout, read_timeout_ms_),
                get_timeout_ptr(write_timeout, write_timeout_ms_)
            );
        }

        read_fixed_size_ = 0;
        if (::bufferevent_enable(bev_, EV_READ | EV_WRITE) == 0) {
            bufs_.attach_write(bev_);
            if (::bufferevent_socket_connect(bev_, addr, addr_len) == 0) {
                state_ = sCONNECTING;
                return true;
            } else
                logger.log(LOG_LEVEL_ERROR, "AsyncClient libevent error: bufferevent_socket_connect() failed");
        } else
            logger.log(LOG_LEVEL_ERROR, "AsyncClient libevent error: bufferevent_enable() failed");
        close();
        return false;
    }

    static void on_read(struct bufferevent* bev, void* self_handler) {
        EVO_PARAM_UNUSED(bev);
        String logstr;
        ProtocolHandler& self = *(ProtocolHandler*)self_handler;
        AsyncBuffers& bufs = self.bufs_;
        bufs.attach_read();
        if (self.read_fixed_size_ > 0) {
            if (self.logger.check(LOG_LEVEL_DEBUG_LOW))
                self.logger.log_direct(LOG_LEVEL_DEBUG_LOW, logstr.set().reserve(64) << "AsyncClient " << self.id_ << " fixed read: " << self.read_fixed_size_);
            for (;;) {
                SubString data;
                if (!bufs.read_fixed(data, self.read_fixed_size_))
                    return; // wait for more data
                self.read_fixed_size_ = 0;
                if (!self.on_read_fixed(self.read_fixed_size_, data, NULL)) {
                    if (self.logger.check(LOG_LEVEL_DEBUG_LOW))
                        self.logger.log_direct(LOG_LEVEL_DEBUG_LOW, logstr.set().reserve(64) << "AsyncClient " << self.id_ << " on_read_fixed() returned false to close");
                    self.close();
                    return;
                }
                bufs.read_flush();
                if (self.read_fixed_size_ <= 0)
                    break;
            }
            bufs.read_reset(ProtocolHandler::MIN_INITIAL_READ, self.max_read_size_);
            if (bufs.read_size() == 0)
                return;
        }
        if (self.logger.check(LOG_LEVEL_DEBUG_LOW))
            self.logger.log_direct(LOG_LEVEL_DEBUG_LOW, logstr.set().reserve(64) << "AsyncClient " << self.id_ << " read: " << bufs.read_size());
        if (!self.on_read(self.read_fixed_size_, bufs, NULL)) {
            if (self.logger.check(LOG_LEVEL_DEBUG_LOW))
                self.logger.log_direct(LOG_LEVEL_DEBUG_LOW, logstr.set().reserve(64) << "AsyncClient " << self.id_ << " on_read() returned false to close");
            self.close();
            return;
        }
    }

    static void on_event(struct bufferevent* bev, short events, void* self_ptr) {
        EVO_PARAM_UNUSED(bev);
        String logstr;
        This& self = *(This*)self_ptr;
        if (events & BEV_EVENT_CONNECTED && self.state_ == sCONNECTING) {
            self.logger.log(LOG_LEVEL_DEBUG_LOW, logstr.set().reserve(34) << "AsyncClient " << self.id_ << " connected");
            self.state_ = sCONNECTED;
            ((ProtocolHandler&)self).on_connect();
            if (self.on_connect_ != NULL)
                self.on_connect_->on_connect();
        } else {
            AsyncError err;
            if (events & BEV_EVENT_EOF)
                err = aeCLOSED;
            else if (events & BEV_EVENT_TIMEOUT)
                err = aeTIMEOUT;
            else if (events & BEV_EVENT_READING)
                err = (self.state_ == sCONNECTING ? aeCONNECT : aeIO_READ);
            else if (events & BEV_EVENT_WRITING)
                err = (self.state_ == sCONNECTING ? aeCONNECT : aeIO_WRITE);
            else if (events & BEV_EVENT_CONNECTED) {
                self.logger.log(LOG_LEVEL_ERROR, logstr.set().reserve(64) << "AsyncClient " << self.id_ << " error: Unexpected 'connected' event");
                err = aeIO;
            } else
                err = aeIO;
            if (self.logger.check(LOG_LEVEL_ERROR)) {
                const SubString errmsg(async_error_msg(err));
                self.logger.log_direct(LOG_LEVEL_ERROR, logstr.set().reserve(42 + errmsg.size()) << "AsyncClient " << self.id_ << " error: " << errmsg << " (code: " << (int)err << ')');
            }

            self.close();
            ((ProtocolHandler&)self).on_error(err);
            if (self.on_error_ != NULL)
                self.on_error_->on_error(err);
        }
    }

    static ulong get_next_id() {
        static AtomicULong id;
        return ++id;
    }
};


}
#endif