eiennet_async.hpp

浏览该文件的文档.

#pragma once

namespace eiennet
{
class AsyncSocket;

// IO类型
enum IoType
{
    ioAccept,
    ioConnect,
    ioRecv,
    ioSend,
    ioRecvFrom,
    ioSendTo,
};

// IO场景基类
struct IoCtx
{
    winux::uint64 startTime; 
    winux::uint64 timeoutMs; 

    IoCtx() : startTime(winux::GetUtcTimeMs()), timeoutMs(-1)
    {
    }

    virtual ~IoCtx()
    {
    }
};

// 接受场景
struct IoAcceptCtx : IoCtx
{
    using OkFunction = std::function< bool ( winux::SharedPointer<AsyncSocket> servSock, winux::SharedPointer<AsyncSocket> clientSock, ip::EndPoint const & ep ) >;
    using TimeoutFunction = std::function< bool ( winux::SharedPointer<AsyncSocket> servSock, IoAcceptCtx * ctx ) >;

    OkFunction cbOk; 
    TimeoutFunction cbTimeout; 

    ip::EndPoint ep; 

    IoAcceptCtx() { }
};

// 连接场景
struct IoConnectCtx : IoCtx
{
    using OkFunction = std::function< void ( winux::SharedPointer<AsyncSocket> sock, winux::uint64 costTimeMs ) >;
    using TimeoutFunction = std::function< void ( winux::SharedPointer<AsyncSocket> sock, IoConnectCtx * ctx ) >;

    OkFunction cbOk; 
    TimeoutFunction cbTimeout; 

    winux::uint64 costTimeMs; 

    IoConnectCtx() : costTimeMs(0) { }
};

// 数据接收场景
struct IoRecvCtx : IoCtx
{
    using OkFunction = std::function< void ( winux::SharedPointer<AsyncSocket> sock, winux::Buffer & data, bool cnnAvail ) >;
    using TimeoutFunction = std::function< void ( winux::SharedPointer<AsyncSocket> sock, IoRecvCtx * ctx ) >;

    OkFunction cbOk; 
    TimeoutFunction cbTimeout; 

    size_t hadBytes;        
    size_t targetBytes;     
    winux::GrowBuffer data; 
    bool cnnAvail; 

    IoRecvCtx() : hadBytes(0), targetBytes(0), cnnAvail(false) { }
};

// 数据发送场景
struct IoSendCtx : IoCtx
{
    using OkFunction = std::function< void ( winux::SharedPointer<AsyncSocket> sock, winux::uint64 costTimeMs, bool cnnAvail ) >;
    using TimeoutFunction = std::function< void ( winux::SharedPointer<AsyncSocket> sock, IoSendCtx * ctx ) >;

    OkFunction cbOk; 
    TimeoutFunction cbTimeout; 

    size_t hadBytes;    
    winux::uint64 costTimeMs; 
    winux::Buffer data; 
    bool cnnAvail;

    IoSendCtx() : hadBytes(0), costTimeMs(0), cnnAvail(false) { }
};

// 无连接，数据接收场景
struct IoRecvFromCtx : IoCtx
{
    using OkFunction = std::function< void ( winux::SharedPointer<AsyncSocket> sock, winux::Buffer & data, EndPoint const & ep ) >;
    using TimeoutFunction = std::function< void ( winux::SharedPointer<AsyncSocket> sock, IoRecvFromCtx * ctx ) >;

    OkFunction cbOk; 
    TimeoutFunction cbTimeout; 

    size_t hadBytes;        
    size_t targetBytes;     
    winux::GrowBuffer data; 
    ip::EndPoint epFrom;    

    IoRecvFromCtx() : hadBytes(0), targetBytes(0) { }
};

// 无连接，数据发送场景
struct IoSendToCtx : IoCtx
{
    using OkFunction = std::function< void ( winux::SharedPointer<AsyncSocket> sock, winux::uint64 costTimeMs ) >;
    using TimeoutFunction = std::function< void ( winux::SharedPointer<AsyncSocket> sock, IoSendToCtx * ctx ) >;

    OkFunction cbOk; 
    TimeoutFunction cbTimeout; 

    size_t hadBytes;    
    winux::uint64 costTimeMs; 
    winux::Buffer data; 
    winux::SimplePointer<EndPoint> ep; // 目的地址

    IoSendToCtx() : hadBytes(0), costTimeMs(0) { }
};

class EIENNET_DLL IoService
{
public:
    using IoMapMap = std::map< winux::SharedPointer<AsyncSocket>, std::map< IoType, winux::SharedPointer<IoCtx> > >;
    using IoMap = IoMapMap::mapped_type;

    IoService( int threadCount = 4, double serverWait = 0.002 );

    virtual ~IoService();

    bool init( int threadCount = 4, double serverWait = 0.002 );

    void stop( bool b = true );

    void postAccept( winux::SharedPointer<AsyncSocket> sock, IoAcceptCtx::OkFunction cbOk, winux::uint64 timeoutMs = -1, IoAcceptCtx::TimeoutFunction cbTimeout = nullptr );
    void postConnect( winux::SharedPointer<AsyncSocket> sock, EndPoint const & ep, IoConnectCtx::OkFunction cbOk, winux::uint64 timeoutMs = -1, IoConnectCtx::TimeoutFunction cbTimeout = nullptr );
    void postRecv( winux::SharedPointer<AsyncSocket> sock, size_t targetSize, IoRecvCtx::OkFunction cbOk, winux::uint64 timeoutMs = -1, IoRecvCtx::TimeoutFunction cbTimeout = nullptr );
    void postSend( winux::SharedPointer<AsyncSocket> sock, void const * data, size_t size, IoSendCtx::OkFunction cbOk, winux::uint64 timeoutMs = -1, IoSendCtx::TimeoutFunction cbTimeout = nullptr );
    void postRecvFrom( winux::SharedPointer<AsyncSocket> sock, size_t targetSize, IoRecvFromCtx::OkFunction cbOk, winux::uint64 timeoutMs = -1, IoRecvFromCtx::TimeoutFunction cbTimeout = nullptr );
    void postSendTo( winux::SharedPointer<AsyncSocket> sock, EndPoint const & ep, void const * data, size_t size, IoSendToCtx::OkFunction cbOk, winux::uint64 timeoutMs = -1, IoSendToCtx::TimeoutFunction cbTimeout = nullptr );

    int run();

    void getSockIoCount( size_t * sockCount, size_t * ioCount ) const;

    void removeSock( winux::SharedPointer<AsyncSocket> sock );

    winux::ThreadPool & getPool() { return _pool; }
    IoMapMap & getIoMaps() { return _ioMaps; }
    winux::Mutex & getMutex() { return _mtx; }


protected:
    void _post( winux::SharedPointer<AsyncSocket> sock, IoType type, winux::SharedPointer<IoCtx> ctx );

    // IO请求加入Select轮询数组前
    DEFINE_CUSTOM_EVENT( RunBeforeJoin, (), () )
    // 完成Select.wait()后
    DEFINE_CUSTOM_EVENT( RunAfterWait, ( int rc ), (rc) )

private:
    winux::ThreadPool _pool;
    IoMapMap _ioMaps;
    winux::RecursiveMutex _mtx;
    winux::Condition _cdt;
    double _serverWait;         
    int _threadCount;
    bool _stop;

    friend class AsyncSocket;
    DISABLE_OBJECT_COPY(IoService)
};

class EIENNET_DLL AsyncSocket : public Socket, public winux::EnableSharedFromThis<AsyncSocket>
{
protected:
    explicit AsyncSocket( IoService & ioServ, int sock = -1, bool isNewSock = false ) : Socket( sock, isNewSock ), _ioServ(&ioServ), _data(nullptr)
    {
        this->setBlocking(false);
    }

    AsyncSocket( IoService & ioServ, AddrFamily af, SockType sockType, Protocol proto ) : Socket( af, sockType, proto ), _ioServ(&ioServ), _data(nullptr)
    {
        this->setBlocking(false);
    }

public:
    static winux::SharedPointer<AsyncSocket> New( IoService & ioServ, int sock = -1, bool isNewSock = false )
    {
        return winux::SharedPointer<AsyncSocket>( new AsyncSocket( ioServ, sock, isNewSock ) );
    }

    static winux::SharedPointer<AsyncSocket> New( IoService & ioServ, AddrFamily af, SockType sockType, Protocol proto )
    {
        return winux::SharedPointer<AsyncSocket>( new AsyncSocket( ioServ, af, sockType, proto ) );
    }

    winux::SharedPointer<AsyncSocket> accept( EndPoint * ep = nullptr )
    {
        int sock;
        return this->Socket::accept( &sock, ep ) ? winux::SharedPointer<AsyncSocket>( new AsyncSocket( *_ioServ, sock, true ) ) : winux::SharedPointer<AsyncSocket>();
    }

    void setDataPtr( void * data ) { _data = data; }
    void * getDataPtr() const { return _data; }
    template < typename _Ty >
    _Ty * getDataPtr() const { return reinterpret_cast<_Ty*>(_data); }

    IoService & getService() const { return *_ioServ; }
    void acceptAsync( IoAcceptCtx::OkFunction cbOk, winux::uint64 timeoutMs = -1, IoAcceptCtx::TimeoutFunction cbTimeout = nullptr );
    void connectAsync( EndPoint const & ep, IoConnectCtx::OkFunction cbOk, winux::uint64 timeoutMs = -1, IoConnectCtx::TimeoutFunction cbTimeout = nullptr );
    void recvUntilSizeAsync( size_t targetSize, IoRecvCtx::OkFunction cbOk, winux::uint64 timeoutMs = -1, IoRecvCtx::TimeoutFunction cbTimeout = nullptr );
    void recvAsync( IoRecvCtx::OkFunction cbOk, winux::uint64 timeoutMs = -1, IoRecvCtx::TimeoutFunction cbTimeout = nullptr )
    {
        this->recvUntilSizeAsync( 0, cbOk, timeoutMs, cbTimeout );
    }
    void sendAsync( void const * data, size_t size, IoSendCtx::OkFunction cbOk, winux::uint64 timeoutMs = -1, IoSendCtx::TimeoutFunction cbTimeout = nullptr );
    void sendAsync( winux::Buffer const & data, IoSendCtx::OkFunction cbOk, winux::uint64 timeoutMs = -1, IoSendCtx::TimeoutFunction cbTimeout = nullptr )
    {
        this->sendAsync( data.getBuf(), data.getSize(), cbOk, timeoutMs, cbTimeout );
    }
    void recvFromUntilSizeAsync( size_t targetSize, IoRecvFromCtx::OkFunction cbOk, winux::uint64 timeoutMs = -1, IoRecvFromCtx::TimeoutFunction cbTimeout = nullptr );
    void recvFromAsync( IoRecvFromCtx::OkFunction cbOk, winux::uint64 timeoutMs = -1, IoRecvFromCtx::TimeoutFunction cbTimeout = nullptr )
    {
        this->recvFromUntilSizeAsync( 0, cbOk, timeoutMs, cbTimeout );
    }
    void sendToAsync( EndPoint const & ep, void const * data, size_t size, IoSendToCtx::OkFunction cbOk, winux::uint64 timeoutMs = -1, IoSendToCtx::TimeoutFunction cbTimeout = nullptr );
    void sendToAsync( EndPoint const & ep, winux::Buffer const & data, IoSendToCtx::OkFunction cbOk, winux::uint64 timeoutMs = -1, IoSendToCtx::TimeoutFunction cbTimeout = nullptr )
    {
        this->sendToAsync( ep, data.getBuf(), data.getSize(), cbOk, timeoutMs, cbTimeout );
    }

    DEFINE_CUSTOM_EVENT( Error, ( winux::SharedPointer<AsyncSocket> sock ), (sock) )

private:
    IoService * _ioServ;
    void * _data;

    friend class IoService;
};

namespace ip
{
namespace tcp
{
class EIENNET_DLL AsyncSocket : public eiennet::AsyncSocket
{
public:
    typedef eiennet::AsyncSocket BaseClass;

protected:
    AsyncSocket( IoService & ioServ, int sock, bool isNewSock = false ) : BaseClass( ioServ, sock, isNewSock ) { }

    explicit AsyncSocket( IoService & ioServ ) : BaseClass( ioServ, BaseClass::afInet, BaseClass::sockStream, BaseClass::protoUnspec ) { }

public:
    static winux::SharedPointer<AsyncSocket> New( IoService & ioServ, int sock, bool isNewSock = false )
    {
        return winux::SharedPointer<AsyncSocket>( new AsyncSocket( ioServ, sock, isNewSock ) );
    }

    static winux::SharedPointer<AsyncSocket> New( IoService & ioServ )
    {
        return winux::SharedPointer<AsyncSocket>( new AsyncSocket(ioServ) );
    }
};

} // namespace tcp

namespace udp
{
class EIENNET_DLL AsyncSocket : public eiennet::AsyncSocket
{
public:
    typedef eiennet::AsyncSocket BaseClass;

protected:
    AsyncSocket( IoService & ioServ, int sock, bool isNewSock = false ) : BaseClass( ioServ, sock, isNewSock ) { }

    explicit AsyncSocket( IoService & ioServ ) : BaseClass( ioServ, BaseClass::afInet, BaseClass::sockDatagram, BaseClass::protoUnspec ) { }

public:
    static winux::SharedPointer<AsyncSocket> New( IoService & ioServ, int sock, bool isNewSock = false )
    {
        return winux::SharedPointer<AsyncSocket>( new AsyncSocket( ioServ, sock, isNewSock ) );
    }

    static winux::SharedPointer<AsyncSocket> New( IoService & ioServ )
    {
        return winux::SharedPointer<AsyncSocket>( new AsyncSocket(ioServ) );
    }
};

} // namespace udp

} // namespace ip

} // namespace eiennet