stream_socket_poll.C

// -*- C++ -*-
 
//  Copyright (c) 2005-2010, Isode Limited, London, England.
//  All rights reserved.
//                                                                       
//  Acquisition and use of this software and related materials for any      
//  purpose requires a written licence agreement from Isode Limited,
//  or a written licence from an organisation licenced by Isode Limited
//  to grant such a licence.
 
// 
//
// stream_socket_poll.C
//
// Stream provider using sockets and poll.
//
// @VERSION@
 
#include <openssl/err.h>
 
#include "Event_p.h"
#include "Proxy.h"
 
#include "../include/Syncmsg.h"
 
#include <isode/messages/base.h>
#include <isode/base/crypto.h>
#include <isode/compat/Common.h>
 
#include <boost/optional.hpp>
 
// Should be set to the size of the longest sockaddr
// There is no easy and portable way to define this.
// The size here is probably generous.
#define MAXSOCKADDRLEN 1024
 
namespace Stream {
 
    struct ProxyBuf {
        enum class Version { VERS_1, VERS_2 } version;
        enum class State   { BEGIN, HEADER, HEADER_BODY, HEADER_DONE } state;
 
        ProxyBuf() : version(Version::VERS_2), state(State::BEGIN), header_read(0), header_len(0), remaining(0), local(false)
        {
            memset(&raddr, 0, sizeof raddr);
        }
 
        uint16_t header_read;
        uint16_t header_len;
        uint16_t remaining;
        std::string buf;
        bool             local;
        sockaddr_storage raddr;
    };
 
    static inline bool starts_with(const std::string& str, const std::string& prefix)
    {
        return str.size() >= prefix.size() && str.substr(0, prefix.size()) == prefix;
    }
 
 
    // This is the core interface which we wrap to get the thread
    // synchronization. In a non-thread environment, this could be
    // used directly, which is more efficient.
 
    class SocketPoll : public Provider, public ::Poll::User {
    private:
        ::Poll::Provider  *_poller; 
        enum State {
            State_noconnect,
            State_listening,
            State_connecting,
            State_accepting,
            State_need_proxy,
            State_connected,
            State_tls
        } _state;  
 
        // Reader stuff
        char       * _rbuf;   
        char       * _rptr;   
        size_t       _rlen;   
        bool         _rfill;  
        bool         _rfull;  
 
        // Writer stuff
        const char * _wbuf;   
        const char * _wptr;   
        size_t       _wlen;   
 
        // Addresses for ConnectionIndication
        socklen_t               _clen;   
        struct sockaddr_storage _caddr;  
        socklen_t               _llen;   
        struct sockaddr_storage _laddr;  
 
        SSL        * _ssl;          
        const char * _hostname;     
 
        SSLTLS::tls_verify_client_choice _choice; 
 
        SyncEvent<time_t, SocketPoll>    _timer;  
        unsigned                         _emfile; 
 
        enum TLSstate {
            TLS_idle,
            TLS_handshake,
            TLS_write,
            TLS_close
        } _tlsstate;
 
        bool _noshut;  
 
        // PROXY buffer
        boost::optional<ProxyBuf> proxybuf_;
 
        inline bool checkUser () {
            if ( GetUser() == 0 ) {
                MSG_IOevent_MissingUser_LOG ();
                return true;
            }
            return false;
        }
 
        inline bool wouldBlock () {
            return ( errno == EWOULDBLOCK || errno == EAGAIN );
        }
 
        inline void waitEvent (unsigned events);
 
        // Send status indication and tidy up
        inline void handleProxyError();
 
        //  Common error handling code for doRead / doProxy
        inline bool handleReadError (ssize_t nbytes);
 
        // This routine is responsible for signalling the User
        inline bool doProxy ();
 
        // This routine is responsible for signalling the User
        inline bool doRead ();
 
        // This routine is responsible for signalling the User
        inline bool doWrite ();
 
        inline void doAccept ();
 
        void doTLS ();
 
        inline int doTLSread();
 
        // returns true if worked, false on SSL error
        inline int doTLSwrite();
 
        // returns true if worked, false on SSL error
        inline int doTLShandshake();
 
        // returns true if worked, false on SSL error
        inline int doTLSshutdown();
 
        inline bool setTrafficClass (
            endpoint_t fd,
            int family,
            unsigned traffic_class);
 
    protected:
        ~SocketPoll() override {
            tidy ();
        }
 
    public:
        SocketPoll () :
            _poller (0),
            _state  (State_noconnect),
            _rlen  (0),
            _rfill (false),
            _rfull (false),
            _wlen  (0),
            _ssl   (0),
            _hostname (0),
            _emfile (0),
            _noshut (false)
        {
            _event.Init (this);
            _timer.Init (this);
        }
 
        void Die () override { 
            delete this;
        }
 
        inline void tidy () {
            SetUser (0);
 
            if ( _ssl != 0 ) {
                if ( _poller != 0 ) {
                    _poller->Deregister (this);
                    _poller = 0;
                }
 
                // This closes the socket
                int err = SSL_get_error(_ssl, 0);
                if (err != SSL_ERROR_NONE) {
                    // Report but clear and continue
                    LOG_DEBUG (("SocketPoll::tidy() - SSL error (%d)", err));
                    ERR_clear_error();
                }
                SSL_free (_ssl);
                _ssl = 0;
 
                _fd = INVALID_ENDPOINT;
                _event.Cancel ();
 
            } else if ( _fd != INVALID_ENDPOINT ) {
                if ( _poller != 0 ) {
                    _poller->Deregister (this);
                    _poller = 0;
                }
 
                // This seems to be the way to close sockets cleanly 
                // on Windows. If you use SHUT_RDWR (aka SH_BOTH) the
                // remote end can be reset.
                if ( !_noshut )
                    shutdown (_fd, SHUT_WR);
 
                closesocket (_fd);
                _fd = INVALID_ENDPOINT;
 
                _event.Cancel ();
            }
 
            _rbuf = _rptr = 0;
            _rlen = 0;
            _rfull = false;
 
            _state = State_noconnect;
        }
 
        // If requested make sure endpoint can be inherited
        endpoint_t GetEndpoint (bool inheriting) override {
            if ( inheriting ) {
                _noshut = true;
 
#ifdef IC_WIN32
                if ( SetHandleInformation ((HANDLE)_fd,
                            HANDLE_FLAG_INHERIT,
                            HANDLE_FLAG_INHERIT) == 0 ) {
                    MSG_IOevent_Inherit_LOG();
                }
 
#elif defined (FD_CLOEXEC)
                int flags = fcntl (_fd, F_GETFD, 0);
 
                if ( flags >= 0 ) {
                    flags &= ~FD_CLOEXEC;
                    if ( fcntl (_fd, F_SETFD, flags) != 0 ) {
                    MSG_IOevent_Inherit_LOG ();
                    }
                }
#endif
            }
            return _fd;
        }
 
        unsigned Events () override {
            return _uevents;
        }
 
        int GetLocalAddress (struct sockaddr *laddr,
                        socklen_t *lalenp,
                        MSGstruct *msp) override {
 
            if ( _fd == INVALID_ENDPOINT ) {
                MSG_IOevent_NotConnected_SET (msp);
                return MSG_IOevent_NotConnected;
            }
 
#ifdef GETSOCKNAME_NEEDS_INTP
            int lalen;
#else
            socklen_t lalen;
#endif
            lalen = *lalenp;
            if ( getsockname (_fd, laddr, &lalen) != 0 ) {
                *lalenp = lalen;
                MSG_IOevent_IOerror_SET (msp, "getsockname");
                return MSG_IOevent_IOerror;
            }
 
            // In case encapsulated IPv4 address
            *lalenp = convert2ipv4 (laddr, lalen);
            return 0;
        }
 
        int GetPeerAddress (struct sockaddr *laddr,
                        socklen_t *lalenp,
                        MSGstruct *msp) override {
 
            if ( _fd == INVALID_ENDPOINT ) {
                MSG_IOevent_NotConnected_SET (msp);
                return MSG_IOevent_NotConnected;
            }
 
#ifdef GETPEERNAME_NEEDS_INTP
            int lalen;
#else
            socklen_t lalen;
#endif
            lalen = *lalenp;
            if ( getpeername (_fd, laddr, &lalen) != 0 ) {
                *lalenp = lalen;
                MSG_IOevent_IOerror_SET (msp, "getpeername");
                return MSG_IOevent_IOerror;
            }
 
            // In case encapsulated IPv4 address
            *lalenp = convert2ipv4 (laddr, lalen);
            return 0;
        }
 
        TLS_CipherSuite GetCipherSuite () override {
            if ( _state != State_tls || _ssl == 0 )
                return TLS_NULL_WITH_NULL_NULL;
 
            const SSL_CIPHER *cipher = SSL_get_current_cipher(_ssl);
            return SSLTLS::cipher_id(cipher);
        }
 
        X509* GetPeerCertificate () override {
            if ( _state != State_tls || _ssl == 0 )
                return 0;
 
            return SSL_get_peer_certificate(_ssl);
        }
 
        // This is virtual so that each subclass can override
        virtual SocketPoll *Clone () = 0;
 
        // Since the interface will be protected,
        // We need specific functions for each of the deliver cases
 
        void actualDeliver (time_t *msg);
        
        void actualDeliver (::Poll::pollmsg *msg) override;
    
        // Interface to stream user
        void actualDeliver (External *ext);
 
        void actualDeliver (ConnectRequest *req);
 
        void actualDeliver (ListenRequest *req);
 
        void actualDeliver (ConnectAccept *req);
 
        void actualDeliver (DisconnectRequest *req);
 
        void actualDeliver (DataRequest *req);
 
        void actualDeliver (ReadRequest *req);
 
        void actualDeliver (StartTLS *req);
 
        void actualDeliver (StreamControl *option);
 
        virtual void Deliver (time_t *msg) {
            actualDeliver (msg);
        }
 
        // Interface from poll provider
 
        void Deliver (::Poll::pollmsg *msg) override {
            actualDeliver (msg);
        }
 
        // Interface to stream user
        void Deliver (External *ext) override {
            actualDeliver (ext);
        }
 
        void Deliver (ConnectRequest *req) override {
            actualDeliver (req);
        }
 
        void Deliver (ListenRequest *req) override {
            actualDeliver (req);
        }
 
        void Deliver (ConnectAccept *req) override {
            actualDeliver (req);
        }
 
        void Deliver (DisconnectRequest *req) override {
            actualDeliver (req);
        }
 
        void Deliver (DataRequest *req) override {
            actualDeliver (req);
        }
 
        void Deliver (ReadRequest *req) override {
            actualDeliver (req);
        }
 
        void Deliver (StartTLS *req) override {
            actualDeliver (req);
        }
 
        void Deliver (StreamControl *option) override {
            actualDeliver (option);
        }
 
        void Deliver (LengthFnxRequest *) override {
            // Don't bother to synchronize this
        }
    };
 
    class SyncSocketPoll : public Syncmsgobj<SocketPoll> {
    protected:
        ~SyncSocketPoll () override {
            // Need to call tidy() here to ensure stuff is cleared up
            // while this actual object exists.
            tidy();
        }
 
    public:
        // for Deliver(Stream::LengthFnxRequest*)
        using Syncmsgobj<SocketPoll>::Deliver;
 
        SocketPoll *Clone () override {
            return new SyncSocketPoll ();
        }
 
        void Die () override {
            dieAux ();
        }
 
        void Deliver (time_t *msg) override {
            actualDeliver (msg);
        }
    
        void Deliver (::Poll::pollmsg *msg) override {
            msgDeliver (msg);
        }
 
        // Interface to stream user
        void Deliver (External *ext) override {
            msgDeliver (ext);
        }
 
        void Deliver (ConnectRequest *req) override {
            msgDeliver (req);
        }
 
        void Deliver (ListenRequest *req) override {
            msgDeliver (req);
        }
 
        void Deliver (ConnectAccept *req) override {
            msgDeliver (req);
        }
 
        void Deliver (DisconnectRequest *req) override {
            msgDeliver (req);
        }
 
        void Deliver (DataRequest *req) override {
            msgDeliver (req);
        }
 
        void Deliver (ReadRequest *req) override {
            msgDeliver (req);
        }
 
        void Deliver (StartTLS *req) override {
            msgDeliver (req);
        }
 
        void Deliver (StreamControl *option) override {
            msgDeliver (option);
        }
    };
}
 
// SocketPoll implementation
 
inline void Stream::SocketPoll::waitEvent (unsigned events)
{
    StatusIndication statusind (this);
 
    if ( _fd == INVALID_ENDPOINT )
        return;
 
    if ( _poller == 0 ) {
        _poller = ::Poll::Provider::GetPollService ();
 
        if ( _poller == 0 ) {
            MSG_IOevent_NoPollProvider_SET (&statusind.msg);
            GetUser()->Deliver (&statusind);
            tidy ();
            return;
        }
 
        int rc = _poller->Register (this);
 
        if ( rc != 0 ) {
            MSG_IOevent_PollRegister_SET (&statusind.msg, rc);
            GetUser()->Deliver (&statusind);
            tidy ();
            return;
        }
    }
 
    _uevents |= events;
 
    // If have events which the poller is not already listening for,
    // tell it to listen
    if ( _uevents & ~_pevents )
        _poller->Control (this);
}
 
bool Stream::SocketPoll::doWrite ()
{
    ssize_t remain = (ssize_t)_wlen - (_wptr - _wbuf);
    ssize_t tosend = remain;
    ssize_t nbytes;
    
    LOG_DEBUG (("SocketPoll::doWrite() - send %ld", (long)tosend));
 
    while (1) {
        nbytes = send (_fd, _wptr, tosend, 0);
 
        if ( nbytes <= 0 ) {
            StatusIndication statusind (this);
 
            if ( nbytes == 0 ) {
                // EOF
                MSG_IOevent_ConnectionClosed_SET (&statusind.msg);
                break;
            } else {
                isode_set_errno();
                // If error is that read would block, then just return
                // Caller should ensure that poller knows we want to
                // know when readable.
                if ( wouldBlock () )
                    return false;
#ifdef IC_WIN32
                // On Windows we may get WSAENOBUFS (converted to ENOBUFS in isode_set_errno)
                // if the data we are attempting to send is 'too big'
                if (errno == ENOBUFS) {
                    // Try and send half as much data....
                    tosend = tosend / 2;
                    LOG_DEBUG (("SocketPoll::doWrite() - retry send with %d", tosend));
                }
        
                if ((tosend == 0) || (errno != ENOBUFS)) {
                    // Error!
                    MSG_IOevent_IOerror_SET (&statusind.msg, "send");
                    break;
                }
                continue;
#else
                // Error
                MSG_IOevent_IOerror_SET (&statusind.msg, "send");
                break;
#endif
            }
 
            GetUser()->Deliver (&statusind);
 
            tidy ();
 
            return true;
        } else {
            break;
        }
    }
 
    _wptr   += nbytes;
    remain  -= nbytes;
 
    if ( remain == 0 ) {
        ReleaseBuf release;
 
        release.buf = _wbuf;
        release.len = _wptr - _wbuf;
 
        _wbuf = _wptr = 0;
        _wlen = 0;
 
        GetUser()->Deliver (&release);
 
        return true;
    }
 
    // Presume that the reason the write was incomplete is that
    // the next attempt would block
    return false;
}
 
void Stream::SocketPoll::handleProxyError() {
    Stream::StatusIndication statusind (this);
    MSG_IOevent_IOerror_SET (&statusind.msg, "PROXY");
    GetUser()->Deliver (&statusind);
    tidy ();
}
 
bool Stream::SocketPoll::handleReadError (ssize_t nbytes)
{
    StatusIndication statusind (this);
 
    if ( nbytes == 0 ) {
        // EOF
        MSG_IOevent_ConnectionClosed_SET (&statusind.msg);
    } else {
        isode_set_errno();
        // If error is that read would block, then just return
        // Caller should ensure that poller knows we want to
        // know when readable.
        if ( wouldBlock () )
            return false;
 
        // Error
        MSG_IOevent_IOerror_SET (&statusind.msg, "recv");
    }
 
    GetUser()->Deliver (&statusind);
 
    tidy ();
 
    return true;
}
 
bool Stream::SocketPoll::doProxy ()
{
    if (!proxybuf_) {
        proxybuf_ = ProxyBuf();
    }
 
    LOG_DEBUG (("doProxy: state=%d read=%d, len=%d remain=%d",
                static_cast<int>(proxybuf_->state), proxybuf_->header_read,
                proxybuf_->header_len, proxybuf_->remaining));
 
    ProxyBuf& pb = *proxybuf_;
    while (true) {
        char      b;
        ssize_t   len;
        switch (pb.state) {
        case ProxyBuf::State::BEGIN: {
            len = recv (_fd, &b, 1, 0);
            if (1 == len) {
                switch (b) {
                case 'P':
                    pb.version = ProxyBuf::Version::VERS_1;
                    pb.buf.reserve(108);
                    pb.buf.push_back(b);
                    pb.state = ProxyBuf::State::HEADER;
                    break;
                case '\x0D':
                    pb.version = ProxyBuf::Version::VERS_2;
                    pb.buf.resize(16);
                    pb.buf[0] = b;
                    pb.state = ProxyBuf::State::HEADER;
                    pb.header_read = 1;
                    LOG_DEBUG (("doProxy: Read V2 BEGIN state=%d read=%d, len=%d remain=%d",
                                static_cast<int>(proxybuf_->state), proxybuf_->header_read,
                                proxybuf_->header_len, proxybuf_->remaining));
                    break;
                default:
                    MSG_IOevent_ProxyError_LOG(b, _fd, "Unknown PROXY version");
                    handleProxyError();
                    return true;
                }
            } else {
                return handleReadError(len);
            }
        }
        FALLTHROUGH;
        case ProxyBuf::State::HEADER:
            switch (pb.version) {
            case ProxyBuf::Version::VERS_1:
                len = recv (_fd, &b, 1, 0);
                if (1 == len) {
                    if (b == '\r') {
                        pb.buf.push_back(b);
                        continue;
                    }
 
                    if (b == '\n') {
                        if (pb.buf[pb.buf.length() - 1] != '\r') {
                            MSG_IOevent_ProxyError_LOG(1, _fd, "read CR but previous character was not newline");
                            handleProxyError();
                            return true;
                        }
 
                        pb.buf.push_back(b);
                        pb.state = ProxyBuf::State::HEADER_DONE;
                        continue;
                    }
 
                    if (isascii(b)) {
                        pb.buf.push_back(b);
                        continue;
                    }
 
                    MSG_IOevent_ProxyError_LOG(1, _fd, "read a non-ascii character");
                    handleProxyError();
                    return true;
 
                } else {
                    return handleReadError(len);
                }
            case ProxyBuf::Version::VERS_2: {
                ssize_t to_read = 16 - pb.header_read;
                len             = recv (_fd, &(*pb.buf.begin())+pb.header_read, to_read, 0);
                if (len <= 0) {
                    return handleReadError(len);
                }
 
                pb.header_read += len;
                if (len == to_read) {
                    pb.state = ProxyBuf::State::HEADER_BODY;
                    auto ph = reinterpret_cast<proxy_hdr_v2*>(&pb.buf[0]);
                    if (0 != std::memcmp(ph->sig, v2sig, 12) || 0x20 != (ph->ver_cmd & 0x20)) {
                        MSG_IOevent_ProxyError_LOG(2, _fd, "signature did not match");
                        handleProxyError();
                        return true;
                    }
                    pb.remaining = ntohs(ph->len);
                    pb.header_len = 16 + pb.remaining;
                    if (pb.remaining > 520) {
                        // Way too big!
                        MSG_IOevent_ProxyError_LOG(2, _fd, "length is much too big");
                        handleProxyError();
                        return true;
                    }
                    pb.buf.resize(pb.header_len);
                    LOG_DEBUG (("doProxy: Read V2 header state=%d read=%d, len=%d remain=%d",
                                static_cast<int>(proxybuf_->state), proxybuf_->header_read,
                                proxybuf_->header_len, proxybuf_->remaining));
                    continue;
                }
                LOG_DEBUG (("doProxy: Read V2 partial header state=%d read=%d, len=%d remain=%d",
                            static_cast<int>(proxybuf_->state), proxybuf_->header_read,
                            proxybuf_->header_len, proxybuf_->remaining));
                return false;
            }
            }
        FALLTHROUGH;
        case ProxyBuf::State::HEADER_BODY:
            len = recv (_fd, &(*pb.buf.begin()) + pb.header_read, pb.remaining, 0);
            if (len <= 0) {
                return handleReadError(len);
            }
            pb.remaining   -= len;
            pb.header_read += len;
            if (0 == pb.remaining) {
                pb.state = ProxyBuf::State::HEADER_DONE;
                LOG_DEBUG (("doProxy: Read V2 body state=%d read=%d, len=%d remain=%d",
                            static_cast<int>(proxybuf_->state), proxybuf_->header_read,
                            proxybuf_->header_len, proxybuf_->remaining));
            } else {
                LOG_DEBUG (("doProxy: Read V2 partial body state=%d read=%d, len=%d remain=%d",
                            static_cast<int>(proxybuf_->state), proxybuf_->header_read,
                            proxybuf_->header_len, proxybuf_->remaining));
                return false;
            }
        FALLTHROUGH;
        case ProxyBuf::State::HEADER_DONE: {
            if (pb.version == ProxyBuf::Version::VERS_1) {
                // Chop off the CR LF
                pb.buf = pb.buf.substr(0, pb.buf.length()-2);
                if (!starts_with(pb.buf, "PROXY ")) {
                    MSG_IOevent_ProxyError_LOG(1, _fd, "header did not begin with PROXY");
                    handleProxyError();
                    return true;
                }
                pb.buf = pb.buf.substr(6);
                if (starts_with(pb.buf, "LOCAL ")) {
                    // We should probably parse the rest of the line, but
                    // assume it's OK. This is used by the PROXY server
                    // (for health checking the server)
                    pb.local = true;
 
                } else if (starts_with(pb.buf, "TCP4 ")) {
                    ipv4_addr pa;
                    if (!read_proxy_ipv4(pb.buf.substr(5), pa)) {
                        MSG_IOevent_ProxyError_LOG(1, _fd, "failed to decode TCP4 header");
                        handleProxyError();
                        return true;
                    }
                    auto raddr     = reinterpret_cast<sockaddr_in*>(&pb.raddr);
                    raddr->sin_family = AF_INET;
                    *reinterpret_cast<in_addr_t*>(&raddr->sin_addr) = pa.src_addr;
                    raddr->sin_port = htons(pa.src_port);
 
                } else if (starts_with(pb.buf, "TCP6 ")) {
                    ipv6_addr pa;
                    if (!read_proxy_ipv6(pb.buf.substr(5), pa)) {
                        MSG_IOevent_ProxyError_LOG(1, _fd, "failed to decode TCP6 header");
                        handleProxyError();
                        return true;
                    }
                    auto raddr     = reinterpret_cast<sockaddr_in6*>(&pb.raddr);
                    raddr->sin6_family = AF_INET6;
                    std::memcpy(&raddr->sin6_addr, &pa.src_addr, sizeof(raddr->sin6_addr));
                    raddr->sin6_port = htons(pa.src_port);
 
                } else {
                    MSG_IOevent_ProxyError_LOG(1, _fd, "second word was not LOCAL, TCP4, or TCP6");
                    handleProxyError();
                    return true;
                }
            } else if (pb.version == ProxyBuf::Version::VERS_2) {
                auto ph = reinterpret_cast<proxy_hdr_v2*>(&pb.buf[0]);
                std::uint8_t cmd = ph->ver_cmd & 0x0f;
                if (0x0 == cmd) {
                    // This is a local command, so just allow it
                    pb.local = true;
                } else {
                    if (0x1 != cmd) {
                        // Something other than PROXY, so we drop the connection
                        MSG_IOevent_ProxyError_LOG(2, _fd, "command was not LOCAL or PROXY");
                        handleProxyError();
                        return true;
                    }
 
                    if ((ph->fam & 0x0f) != 0x1) {
                        // we only support STREAM
                        MSG_IOevent_ProxyError_LOG(2, _fd, "non-STREAM transport protocol");
                        handleProxyError();
                        return true;
                    }
 
                    switch (ph->fam & 0xf0) {
                    case 0x10: {
                        if (pb.header_len < 16 + sizeof(ipv4_addr)) {
                            MSG_IOevent_ProxyError_LOG(2, _fd, "message length too small to hold TCP4 information");
                            handleProxyError();
                            return true;
                        }
                        auto pa    = reinterpret_cast<const ipv4_addr*>(pb.buf.c_str()+16);
                        auto raddr = reinterpret_cast<sockaddr_in*>(&pb.raddr);
                        raddr->sin_family = AF_INET;
                        *reinterpret_cast<in_addr_t*>(&raddr->sin_addr) = pa->src_addr;
                        raddr->sin_port                                 = pa->src_port;
                        break;
                    }
                    case 0x20: {
                        if (pb.header_len < 16 + sizeof(ipv6_addr)) {
                            MSG_IOevent_ProxyError_LOG(2, _fd, "message length too small to hold TCP6 information");
                            handleProxyError();
                            return true;
                        }
                        auto pa    = reinterpret_cast<const ipv6_addr*>(pb.buf.c_str()+16);
                        auto raddr = reinterpret_cast<sockaddr_in6*>(&pb.raddr);
                        raddr->sin6_family = AF_INET6;
                        std::memcpy(&raddr->sin6_addr, &pa->src_addr, sizeof(raddr->sin6_addr));
                        raddr->sin6_port = pa->src_port;
                        break;
                    }
                    default:
                        // Anything else we'll ignore (LOCAL is accepted earlier)
                        MSG_IOevent_ProxyError_LOG(2, _fd, "unknown family");
                        handleProxyError();
                        return true;
                    }
                }
            }
            
            if (!pb.local && pb.raddr.ss_family == AF_UNSPEC) {
                MSG_IOevent_ProxyError_LOG(-1, _fd, "unable to determine remote address family");
                handleProxyError();
            } else {
                _state = State_connected;
                LOG_DEBUG (("doProxy: Read proxy state=%d read=%d, len=%d remain=%d",
                            static_cast<int>(proxybuf_->state), proxybuf_->header_read,
                            proxybuf_->header_len, proxybuf_->remaining));
 
                // Reset read buffer
                _rbuf = _rptr = 0;
                _rlen = 0;
 
                ProxyIndication proxyind;
                proxyind.local = pb.local;
                proxyind.raddr = reinterpret_cast<sockaddr *>(&pb.raddr);
                GetUser()->DeliverProxy (&proxyind);
            }
            return true;
        }
        }
    }
    return true;
}
 
bool Stream::SocketPoll::doRead ()
{
    ssize_t remain = (ssize_t)_rlen - (_rptr - _rbuf);
 
    LOG_DEBUG (("doRead: _rlen=%d remain=%d", (int)_rlen, (int)remain));
 
    ssize_t nbytes = recv (_fd, _rptr, remain, 0);
 
    LOG_DEBUG (("doRead: recv -> %d", (int)nbytes));
 
    if ( nbytes <= 0 ) {
        return handleReadError (nbytes);
    }
 
    // The read filled the buffer - there may be more queued
    if ( nbytes == remain )
        _rfull = true;
 
    _rptr  += nbytes;
    remain -= nbytes;
 
    LOG_DEBUG (("doRread: remain=%d _rfill=%d", (int)remain, _rfill));
 
    // If not filling buffer or the buffer is full, send
    // data indication
    if ( !_rfill || remain == 0 ) {
        DataIndication dataind;
 
        dataind.buf = _rbuf;
        dataind.len = _rptr - _rbuf;
 
        _rbuf = _rptr = 0;
        _rlen = 0;
 
        GetUser()->Deliver (&dataind);
 
        return true;
    }
 
    // Need to read some more
    return false;
}
 
void Stream::SocketPoll::doAccept ()
{
    StatusIndication statusind (this);
    char abuf[MAXSOCKADDRLEN];
#ifdef ACCEPT_NEEDS_INTP
    int alen;
#else
    socklen_t alen;
#endif
    alen = sizeof abuf;
 
    endpoint_t newfd = accept (_fd, (struct sockaddr *)abuf, &alen);
    if (INVALID_ENDPOINT == newfd) {
        isode_set_errno();
        // Note: need to distinguish the serious cases where accept should
        // not have been called, and failures of the individual accept.
        switch ( errno ) {
        default:
            // Note: could get repeating errors here. E.g. if at limit of
            // number of open files for the process or system.
            MSG_IOevent_AcceptError_SET (&statusind.msg);
            MSGappendsyserror (&statusind.msg);
            GetUser()->Deliver (&statusind);
            break;
 
        case EMFILE:
            // No point in trying again immediately, as it would spin
            // Notify user and wait for a while before trying again
            MSG_IOevent_AcceptError_SET (&statusind.msg);
            MSGappendsyserror (&statusind.msg);
            GetUser()->Deliver (&statusind);
            {
                time_t dummy = 0;
                _emfile++;
                if ( _emfile > 100 ) _emfile = 100;
 
                _timer.Sendms (100*_emfile, dummy);  // 100 ms times # errors
            }
            return;
 
        case EBADF:
        case EINVAL:
#ifdef ENOTSOCK
        case ENOTSOCK:
#endif
#ifdef EOPNOTSUPP
        case EOPNOTSUPP:
#endif
#ifdef EFAULT
        case EFAULT:
#endif
        // A severe error, abandon listen
        MSG_IOevent_IOerror_SET (&statusind.msg, "accept");
        GetUser()->Deliver (&statusind);
        tidy ();
        return;
 
#ifdef EINTR
        case EINTR:
#endif
#ifdef EAGAIN
        case EAGAIN:
#endif
#if defined(EWOULDBLOCK)
#  if defined(EAGAIN) && EAGAIN != EWOULDBLOCK
        case EWOULDBLOCK:
#  endif
#endif
            // False alarm
            break;
        }
    } else {
        // Have new connection from peer
#ifdef IC_WIN32
        // Windows sockets are inherited by default
        if ( SetHandleInformation(
                (HANDLE)newfd,
                HANDLE_FLAG_INHERIT, // mask of bits to change
                0) == 0 ) {
            MSG_IOevent_Inherit_LOG();
        }
#elif defined (FD_CLOEXEC)
        (void) fcntl(newfd, F_SETFD, FD_CLOEXEC);
#endif
 
        bool do_setsockopt = true;
#ifdef O_NONBLOCK
        if ( fcntl(newfd, F_SETFL, O_NONBLOCK) != 0 ) {
            closesocket(newfd);
            MSG_IOevent_IoctlError_SET(&statusind.msg);
            MSGappendsyserror(&statusind.msg);
            GetUser()->Deliver(&statusind);
            do_setsockopt = false;
        }
#elif defined(FIONBIO)
        char onoff = 1;
        if ( ioctl(newfd, FIONBIO, &onoff) != 0 ) {
            closesocket(newfd);
            MSG_IOevent_IoctlError_SET(&statusind.msg);
            MSGappendsyserror(&statusind.msg);
            GetUser()->Deliver(&statusind);
            do_setsockopt = false;
        }
#else
#error Cannot set socket into non-blocking more
#endif
        if (do_setsockopt) {
#ifdef TCP_NODELAY
            // Always set this on by default
#ifdef IC_WIN32
            char nodelay = 1;
#else
            int nodelay = 1;
#endif
            // But ignore failure
            setsockopt(newfd, IPPROTO_TCP, TCP_NODELAY, &nodelay, sizeof nodelay);
#endif
            // Always set keepalives (and ignore failure)
#ifdef IC_WIN32
            char keepalive = 1;
#else
            int keepalive = 1;
#endif
            setsockopt(newfd, SOL_SOCKET, SO_KEEPALIVE, (char *)&keepalive, sizeof keepalive);
 
            Stream::ConnectIndication connind;
            Stream::SocketPoll *      newprov = Clone();
 
            connind.listener    = this;
            connind.newprovider = newprov;
 
            newprov->_fd    = newfd;
            newprov->_state = State_accepting;
 
            memcpy(&newprov->_caddr, abuf, alen);
            newprov->_clen = alen;
 
            connind.saddr = reinterpret_cast<struct sockaddr *>(&newprov->_caddr);
            connind.salen = convert2ipv4(reinterpret_cast<struct sockaddr *>(&newprov->_caddr), alen);
 
#ifdef GETSOCKNAME_NEEDS_INTP
            int llen;
#else
            socklen_t llen;
#endif
            llen           = sizeof newprov->_laddr;
            newprov->_llen = sizeof newprov->_laddr;
 
            getsockname(newfd, reinterpret_cast<struct sockaddr *>(&newprov->_laddr), &llen);
 
            newprov->_llen = llen;
 
            connind.laddr = reinterpret_cast<struct sockaddr *>(&newprov->_laddr);
            connind.lalen = newprov->_llen;
 
            GetUser()->Deliver(&connind);
        }
    }
 
    // Accept did not fail or have EMFILE
    _emfile = 0;
 
    // Go back for more
    waitEvent (::Poll::Event_In);
}
 
 
int Stream::SocketPoll::doTLSread ()
{
    int remain = (int)_rlen - (_rptr - _rbuf);
 
    LOG_DEBUG (("doTLSread: _rlen=%d remain=%d", (int)_rlen, (int)remain));
 
    int nbytes = SSL_read (_ssl, _rptr, remain);
 
    LOG_DEBUG (("doTLSread: SSL_read -> %d", nbytes));
 
    // SSL errors handled by caller
    if ( nbytes < 0 )
        return nbytes;
 
    if ( nbytes == 0 ) {
        StatusIndication statusind (this);
 
        MSG_IOevent_ConnectionClosed_SET (&statusind.msg);
 
        GetUser()->Deliver (&statusind);
 
        _tlsstate = TLS_idle;
 
        tidy ();
        return 1; // Fudge to stop error handling
    }
 
    // The read filled the buffer - there may be more queued
    if ( nbytes == remain )
        _rfull = true;
 
    _rptr  += nbytes;
    remain -= nbytes;
 
    LOG_DEBUG (("doTLSread: remain=%d _rfill=%d", (int)remain, _rfill));
 
    // If not filling buffer or the buffer is full, send
    // data indication
    if ( !_rfill || remain == 0 ) {
        DataIndication dataind;
 
        dataind.buf = _rbuf;
        dataind.len = _rptr - _rbuf;
 
        _rbuf = _rptr = 0;
        _rlen = 0;
 
        _tlsstate = TLS_idle;
 
        GetUser()->Deliver (&dataind);
    }
 
    return nbytes;
}
 
int Stream::SocketPoll::doTLSwrite ()
{
    int remain = (int)_wlen - (_wptr - _wbuf);
    LOG_DEBUG (("doTLSwrite: _wlen=%d remain=%d", (int)_wlen, (int)remain));
 
    int nbytes = SSL_write (_ssl, _wptr, remain);
    LOG_DEBUG (("doTLSwrite: SSL_write -> %d", nbytes));
 
    // Errors handled in caller
    if ( nbytes <= 0 )
        return nbytes;
 
    _wptr   += nbytes;
    remain  -= nbytes;
 
    if ( remain == 0 ) {
        ReleaseBuf release;
 
        release.buf = _wbuf;
        release.len = _wptr - _wbuf;
 
        _wbuf = _wptr = 0;
        _wlen = 0;
 
        GetUser()->Deliver (&release);
 
        _tlsstate = TLS_idle;
    }
 
    return nbytes;
}
 
int Stream::SocketPoll::doTLShandshake ()
{
    int rc = SSL_do_handshake (_ssl);
    LOG_DEBUG (("doTLShandshake: SSL_do_handshake -> %d", rc));
 
    if ( rc < 0 )
        return rc;
 
    StatusIndication statusind (this);
 
    if ( rc == 0 ) {
        _tlsstate = TLS_idle;
 
        return 0;
    }
 
    _tlsstate = TLS_idle;
 
    const char *version = SSL_get_version(_ssl);
 
    const SSL_CIPHER *cipher = SSL_get_current_cipher(_ssl);
    TLS_CipherSuite cs = SSLTLS::cipher_id(cipher);
    int strength = SSL_CIPHER_get_bits(cipher, NULL);
 
    X509 *peer = SSL_get_peer_certificate(_ssl);
 
    bool closing = true;
 
    MSGstruct submsg;
 
    if ( !peer ) {
        if ( _choice == SSLTLS::tls_verify_require ) {
            MSG_IOevent_SSLnotverified_SET (&statusind.msg);
            MSG_IOevent_SSLnocert_SET (&submsg);
            MSGappend (&statusind.msg, &submsg);
 
        } else {
            MSG_IOevent_SSLOK_SET (&statusind.msg,
                       version, cs, strength);
 
            closing = false;
        }
 
    } else if ( _choice == SSLTLS::tls_verify_none ) {
        MSG_IOevent_SSLOK_SET (&statusind.msg,
                               version, cs, strength);
 
        closing = false;
 
    } else {
 
        X509_NAME *subject_name = X509_get_subject_name(peer);
 
        long result = SSL_get_verify_result(_ssl);
        char *peername = X509_NAME_to_text (subject_name);
 
        switch (result) {
        case X509_V_OK:
            closing = false;
            MSG_IOevent_SSLverified_SET (&statusind.msg,
                         version, cs, strength,
                         peername);
            break;
 
        case X509_V_ERR_UNABLE_TO_GET_ISSUER_CERT:
            MSG_IOevent_SSLnoissuercert_SET (&submsg);
            break;
        case X509_V_ERR_UNABLE_TO_DECRYPT_CERT_SIGNATURE:
            MSG_IOevent_SSLcertdecryptfailed_SET (&submsg);
            break;
        case X509_V_ERR_CERT_SIGNATURE_FAILURE:
            MSG_IOevent_SSLcertsignature_SET (&submsg);
            break;
        case X509_V_ERR_CERT_NOT_YET_VALID:
            MSG_IOevent_SSLcertnotyetvalid_SET (&submsg);
            break;
        case X509_V_ERR_CERT_HAS_EXPIRED:
            MSG_IOevent_SSLcertexpired_SET (&submsg);
            break;
        case X509_V_ERR_ERROR_IN_CERT_NOT_BEFORE_FIELD:
            MSG_IOevent_SSLcertnotbefore_SET (&submsg);
            break;
        case X509_V_ERR_OUT_OF_MEM:
            MSG_Base_Nomem_SET (&submsg, 0);
            break;
        case X509_V_ERR_SELF_SIGNED_CERT_IN_CHAIN:
            MSG_IOevent_SSLselfsigned_SET (&submsg);
            break;
        case X509_V_ERR_UNABLE_TO_GET_ISSUER_CERT_LOCALLY:
            MSG_IOevent_SSLlocalissuer_SET (&submsg);
            break;
        case X509_V_ERR_UNABLE_TO_VERIFY_LEAF_SIGNATURE:
            MSG_IOevent_SSLleafsignature_SET (&submsg);
            break;
        case X509_V_ERR_CERT_CHAIN_TOO_LONG:
            MSG_IOevent_SSLchaintoolong_SET (&submsg);
            break;
        case X509_V_ERR_INVALID_CA:
            MSG_IOevent_SSLinvalidCA_SET (&submsg);
            break;
        case X509_V_ERR_PATH_LENGTH_EXCEEDED:
            MSG_IOevent_SSLpathlength_SET (&submsg);
            break;
        case X509_V_ERR_INVALID_PURPOSE:
            MSG_IOevent_SSLinvalidpurpose_SET (&submsg);
            break;
        case X509_V_ERR_CERT_UNTRUSTED:
            MSG_IOevent_SSLuntrusted_SET (&submsg);
            break;
        case X509_V_ERR_CERT_REJECTED:
            MSG_IOevent_SSLrejected_SET (&submsg);
            break;
        case X509_V_ERR_SUBJECT_ISSUER_MISMATCH:
            MSG_IOevent_SSLsubjectissuer_SET (&submsg);
            break;
        case X509_V_ERR_AKID_ISSUER_SERIAL_MISMATCH:
            MSG_IOevent_SSLserialmismatch_SET (&submsg);
            break;
        case X509_V_ERR_KEYUSAGE_NO_CERTSIGN:
            MSG_IOevent_SSLkeyusage_SET (&submsg);
            break;
        default:
            MSG_IOevent_SSLunknown_SET (&submsg, result);
            break;
        }
 
        if ( closing ) {
            MSG_IOevent_SSLnotverified_SET (&statusind.msg);
            MSGappend (&statusind.msg, &submsg);
        }
 
        if (x509_acceptable_name(subject_name) != OK) {
            closing = true;
            MSG_IOevent_SSLnul_in_subject_LOG(peername);
        }
 
        if ( peername != 0 )
            free (peername);
    }
 
    if ( peer ) {
        if (!closing && _hostname &&
            SSLTLS::CheckHostname (peer, _hostname, &statusind.msg) != OK ) {
            closing = true;
        }
 
        X509_free(peer);
    }
 
    _tlsstate = TLS_idle;
 
    GetUser()->Deliver (&statusind);
 
    if ( closing )
        tidy ();
 
    return rc;
}
 
int Stream::SocketPoll::doTLSshutdown ()
{
    int rc = SSL_shutdown (_ssl);
    LOG_DEBUG (("SocketPoll::doTLSshutdown() -> %d", rc));
 
    if ( rc <= 0 ) {
        if (rc == 0) {
            LOG_DEBUG (("SocketPoll::doTLSshutdown() - shutdown in progress"));
        } else {
            int err = SSL_get_error(_ssl, rc);
            LOG_DEBUG (("SocketPoll::doTLSshutdown() - SSL error (%d)", err));
        }
        return rc;
    }
 
    // Shutdown successful, tell user
 
    StatusIndication statusind (this);
    MSG_IOevent_ConnectionClosed_SET (&statusind.msg);
    GetUser()->Deliver (&statusind);
 
    // Close up connection, etc.
    _tlsstate = TLS_idle;
    tidy ();
 
    return rc;
}
 
void Stream::SocketPoll::doTLS ()
{
    LOG_DEBUG (("SocketPoll::doTLS()"));
    for (;;) {
        int rc = 0;
 
        // If there is no SSL, then the connection is being closed down
        if ( _ssl == 0 )
            return;
 
        switch ( _tlsstate ) {
        case TLS_idle:
            if ( _wlen != 0 ) {
                _tlsstate = TLS_write;
                rc = doTLSwrite();
 
            } else if ( _rlen != 0 ) {
                rc = doTLSread();
 
            } else {
                return;
            }
            break;
 
        case TLS_write:
            rc = doTLSwrite();
            break;
 
        case TLS_handshake:
            rc = doTLShandshake();
            break;
 
        case TLS_close:
            rc = doTLSshutdown();
            break;
        }
 
        if ( rc > 0 ) continue;
 
        StatusIndication statusind (this);
 
        switch (SSL_get_error(_ssl, rc)) {
        case SSL_ERROR_NONE:
            // No problem, carry on
            continue;
 
        case SSL_ERROR_ZERO_RETURN:
            // Connection has been closed
            MSG_IOevent_ConnectionClosed_SET (&statusind.msg);
            break;
 
        case SSL_ERROR_WANT_READ:
            waitEvent (::Poll::Event_In);
            return;
 
        case SSL_ERROR_WANT_WRITE:
            waitEvent (::Poll::Event_Out);
            return;
 
        case SSL_ERROR_SYSCALL:
            // Check the error queue
            if (ERR_get_error()==0) {
                if (rc == 0) {
                    // Connection has closed
                    MSG_IOevent_ConnectionClosed_SET (&statusind.msg);
                } else {
                    MSG_IOevent_IOerror_SET (&statusind.msg, "SSL");
                }
            } else {
                SSLTLS::SSLError (&statusind.msg, MSG_IOevent_SSLerror, "SSL_ERROR_SYSCALL");
            }
            break;
 
        case SSL_ERROR_SSL:
            SSLTLS::SSLError (&statusind.msg, MSG_IOevent_SSLerror, "SSL_ERROR_SSL");
            break;
 
        case SSL_ERROR_WANT_CONNECT:
            SSLTLS::SSLError (&statusind.msg, MSG_IOevent_SSLerror,"SSL_WANT_CONNECT");
            break;
 
        case SSL_ERROR_WANT_ACCEPT:
            SSLTLS::SSLError (&statusind.msg, MSG_IOevent_SSLerror,"SSL_WANT_ACCEPT");
            break;
 
        default:
            {
                char buf[32];
                snprintf (buf, sizeof buf, "err=%d", SSL_get_error(_ssl, rc));
                SSLTLS::SSLError (&statusind.msg, MSG_IOevent_SSLerror, buf);
                break;
            }
        }
 
        GetUser()->Deliver (&statusind);
        tidy();
        return;
    }
}
 
void Stream::SocketPoll::actualDeliver (time_t *)
{
    // Timer has expired after EMFILE, try again
    waitEvent (::Poll::Event_In);
}
 
void Stream::SocketPoll::actualDeliver (::Poll::pollmsg *msg)
{
    if ( checkUser() ) return;
 
    StatusIndication statusind (this);
 
    LOG_DEBUG (("SocketPoll::actualDeliver (pollmsg = %x) state=%d",
        msg->events, _state));
 
    // Poller is no longer listening for any events
    _pevents = 0;
 
    if ( (msg->events & ::Poll::Event_Terminate) ) {
        // System terminating
        _uevents = 0;
 
        // Close first, otherwise calls back to poller
        closesocket (_fd);
        _fd = INVALID_ENDPOINT;
        _poller = 0;
 
        // Tell User
        MSG_IOevent_Terminating_SET (&statusind.msg);
        GetUser()->Deliver (&statusind);
 
        return;
    }
 
    if ( msg->events & 
      (::Poll::Event_Err | ::Poll::Event_Hup | ::Poll::Event_Nval) ) {
 
        // Don't want any events
        _uevents = 0;
 
        // Get error from socket
        int eno;
#ifdef GETSOCKOPT_NEEDS_INTP
        int olen;
#else
        socklen_t olen;
#endif
        olen = sizeof eno;
 
        /* Casting to (char *) is required for Windows */
        int rc = getsockopt (_fd, SOL_SOCKET, SO_ERROR, (char *)&eno, &olen);
 
        const char *argv[1];
        argv[0] = "poll";
 
        MSGsetv (&statusind.msg, MSG_IOevent_IOerror, 1, argv);
 
        if ( rc == 0 ) {
            MSGstruct sockerr;
            MSGsetv (&sockerr,
                 MSGID_BUILD(MSGLEVEL_ERROR, MSGID_SYSFAC, eno),
                 0, 0);
            MSGappend (&statusind.msg, &sockerr);
        } else {
            MSGappendsyserror (&statusind.msg);
        }
 
        GetUser()->Deliver (&statusind);
 
        tidy ();
 
        return;
    }
 
    // Cancel desired events
    _uevents &= ~msg->events;
 
    if ( msg->events & ::Poll::Event_In ) {
 
        switch ( _state ) {
        case State_tls:
            doTLS();
            break;
 
        case State_need_proxy:
            if ( !doProxy() )
                waitEvent (::Poll::Event_In);
            break;
        case State_connected:
            LOG_DEBUG (("Read stuff: _rlen = %d", (int)_rlen));
            if ( _rlen != 0 ) {
                // Stuff to read. If does not complete, then wait again
                if ( !doRead () )
                    waitEvent (::Poll::Event_In);
            }
            break;
 
        case State_listening:
            doAccept ();
            break;
 
        default:
            break;
        }
    }
 
    if ( msg->events & ::Poll::Event_Out ) {
 
        switch ( _state ) {
        case State_tls:
            doTLS();
            break;
 
        case State_need_proxy:
            // Unexpected state at this point, fall through to connected
            LOG_DEBUG (("Need proxy state seen when writing"));
            FALLTHROUGH;
        case State_connected:
            if ( _wlen != 0 ) {
                // Stuff to write. If does not complete, then wait again
                if ( !doWrite () )
                    waitEvent (::Poll::Event_Out);
            }
            break;
 
        case State_connecting:
            _state = GetUser()->Proxy() ? State_need_proxy : State_connected;
            MSG_IOevent_Connected_SET (&statusind.msg);
            GetUser()->Deliver (&statusind);
            break;
 
        default:
            break;
        }
    }
 
    if ( (_uevents & ~_pevents) && _poller )
        _poller->Control (this);
}
 
void Stream::SocketPoll::actualDeliver (Stream::External *ext)
{
    if ( checkUser() ) return;
 
    StatusIndication statusind (this);
 
    LOG_DEBUG (("SocketPoll::actualDeliver (External)"));
 
    if ( _state != State_noconnect ) {
        MSG_IOevent_NotConnected_SET (&statusind.msg);
        GetUser()->Deliver (&statusind);
        return;
    }
 
    _fd = ext->fd;
 
    if ( ext->listen ) {
        _state = State_listening;
 
        waitEvent (::Poll::Event_In);
 
    } else {
        _state = GetUser()->Proxy() ? State_need_proxy : State_connected;
    }
}
 
void Stream::SocketPoll::actualDeliver (Stream::ConnectRequest *req)
{
    if ( checkUser() ) return;
 
    StatusIndication statusind (this);
 
    LOG_DEBUG (("SocketPoll::actualDeliver (ConnectRequest)"));
 
    if ( _state != State_noconnect ) {
        MSG_IOevent_NotConnected_SET (&statusind.msg);
        GetUser()->Deliver (&statusind);
        return;
    }
 
    int family;
#if defined(AF_INET6) && defined(PF_INET6)
    struct sockaddr_in6 ipv6addr;
#endif
 
    switch ( req->daddr->sa_family ) {
#if defined(AF_UNIX) && defined(PF_UNIX)
    case AF_UNIX:
        family = PF_UNIX;
        break;
#endif
 
    case AF_INET:
        family = PF_INET;
        break;
 
#if defined(AF_INET6) && defined(PF_INET6)
    case AF_INET6:
        family = PF_INET6;
        if ( req->dscp != 0 ) {
            // Set in the flowinfo in the socket address
            ipv6addr = * reinterpret_cast<const struct sockaddr_in6 *>
            (req->daddr);
 
            // Note: IPv6 traffic class is offset 20 bits
            // and the Codepoint is the top 6 bits.
            ipv6addr.sin6_flowinfo = htonl (req->dscp << 22);
 
            req->daddr = reinterpret_cast<struct sockaddr *>(&ipv6addr);
        }
        break;
#endif
 
    default:
        MSG_IOevent_AddrType_SET (&statusind.msg, req->daddr->sa_family);
        GetUser()->Deliver (&statusind);
        return;
    }
 
    endpoint_t fd = socket (family, SOCK_STREAM, 0);
 
    if ( fd == INVALID_ENDPOINT ) {
        MSG_IOevent_IOerror_SET (&statusind.msg, "socket");
        GetUser()->Deliver (&statusind);
        return;
    }
 
#ifdef IC_WIN32
    // Windows sockets are inherited by default
    if ( SetHandleInformation ((HANDLE)fd,
                   HANDLE_FLAG_INHERIT, // mask of bits to change
                   0) == 0 ) {
        MSG_IOevent_Inherit_LOG ();
    }
#elif defined (FD_CLOEXEC)
    (void) fcntl (fd, F_SETFD, FD_CLOEXEC);
#endif
 
#ifdef O_NONBLOCK
    if ( fcntl (fd, F_SETFL, O_NONBLOCK) != 0 ) {
        MSG_IOevent_IOerror_SET (&statusind.msg, "fcntl");
        GetUser()->Deliver (&statusind);
        closesocket (fd);
        return;
    }
#elif defined (FIONBIO)
    char onoff = 1;
    if ( ioctl (fd, FIONBIO, &onoff) != 0 ) {
        MSG_IOevent_IOerror_SET (&statusind.msg, "ioctl");
        GetUser()->Deliver (&statusind);
        closesocket (fd);
        return;
    }
#else
# error Cannot set socket into non-blocking more
#endif
 
#ifdef TCP_NODELAY
    // Always set this on by default
#ifdef IC_WIN32
    char nodelay = 1;
#else
    int nodelay = 1;
#endif
 
    // But ignore failure
    setsockopt (fd, IPPROTO_TCP, TCP_NODELAY, &nodelay, sizeof nodelay);
#endif
 
    // Always set keepalives (and ignore failure)
#ifdef IC_WIN32
    char keepalive = 1;
#else
    int keepalive = 1;
#endif
    setsockopt (fd, SOL_SOCKET, SO_KEEPALIVE,
        (char *)&keepalive, sizeof keepalive);
 
    // Set the traffic class
    if ( req->dscp != 0 && !setTrafficClass (fd, family, req->dscp) ) {
        MSG_IOevent_IOerror_SET (&statusind.msg, "traffic class");
        GetUser()->Deliver (&statusind);
        closesocket (fd);
        return;
    }
 
    // Bind to the local address here, if required */
    if ((req->saddr) && (bind(fd, req->saddr, req->salen) == -1)) {
        MSG_IOevent_IOerror_SET (&statusind.msg, "bind calling address");
        GetUser()->Deliver (&statusind);
        closesocket (fd);
        return;
    }
 
    if ( connect (fd, req->daddr, req->dalen) != 0 ) {
        isode_set_errno();
        if ( errno == EINPROGRESS || wouldBlock () ) {
            // Connect is async, need to wait
            _fd    = fd;
            _state = State_connecting;
 
            // Connected state indicated by socket becoming writable
            waitEvent (::Poll::Event_Out);
        } else {
            // Must be error in connect
            MSG_IOevent_IOerror_SET (&statusind.msg, "connect");
            GetUser()->Deliver (&statusind);
            closesocket (fd);
        }
 
    } else {
        // Immediate connection
 
        _fd    = fd;
        _state = GetUser()->Proxy() ? State_need_proxy : State_connected;
 
        MSG_IOevent_Connected_SET (&statusind.msg);
        GetUser()->Deliver (&statusind);
    }
}
 
bool Stream::SocketPoll::setTrafficClass (
    endpoint_t fd,
    int family,
    unsigned traffic_class)
{
    if ( family == PF_INET ) {
        // IPv4
        // traffic class is in TOS byte, low 2 bits are not used
        int tosbyte = traffic_class << 2;
#ifdef IP_TOS
#  ifdef SOL_IP
#    define TOSLEVEL SOL_IP
#  else
#    define TOSLEVEL IPPROTO_IP
#  endif
        return setsockopt (fd, TOSLEVEL, IP_TOS,
                   (char *)&tosbyte, sizeof tosbyte) == 0;
#else
        return false;
#endif
    }
 
    // If IPv6 then already actioned, otherwise ignore
    return true;
}
 
void Stream::SocketPoll::actualDeliver (Stream::ListenRequest *req)
{
    if ( checkUser() ) return;
 
    StatusIndication statusind (this);
 
    LOG_DEBUG (("SocketPoll::actualDeliver (ListenRequest)"));
 
    if ( _state != State_noconnect ) {
        MSG_IOevent_NotConnected_SET (&statusind.msg);
 
        GetUser()->Deliver (&statusind);
 
        return;
    }
 
    int family;
 
    switch ( req->laddr->sa_family ) {
#ifdef AF_UNIX
    case AF_UNIX:
        family = PF_UNIX;
        break;
#endif
 
    case AF_INET:
        family = PF_INET;
        break;
 
#if defined(AF_INET6) && defined(PF_INET6)
    case AF_INET6:
        family = PF_INET6;
        break;
#endif
 
    default:
        MSG_IOevent_AddrType_SET (&statusind.msg, req->laddr->sa_family);
        GetUser()->Deliver (&statusind);
        return;
    }
 
    endpoint_t fd = socket (family, SOCK_STREAM, 0);
 
    if ( fd == INVALID_ENDPOINT ) {
        MSG_IOevent_IOerror_SET (&statusind.msg, "socket");
        GetUser()->Deliver (&statusind);
        return;
    }
 
#if defined(PF_INET6) && defined(IPV6_V6ONLY) && defined(IPPROTO_IPV6)
    if ( family == PF_INET6 ) {
        int v6only = 1;
 
        if ( setsockopt (fd, IPPROTO_IPV6, IPV6_V6ONLY,
                 (char *)&v6only, sizeof v6only) != 0 ) {
            MSG_IOevent_IOerror_SET (&statusind.msg, "setsockopt");
            GetUser()->Deliver (&statusind);
            closesocket (fd);
            return;
        }
    }
#endif
 
#if !defined(IC_WIN32) && defined(SO_REUSEADDR)
    int onoff = 1;
 
    if ( setsockopt (fd, SOL_SOCKET, SO_REUSEADDR,
                    (char *)&onoff, sizeof onoff) != 0 ) {
        MSG_IOevent_IOerror_SET (&statusind.msg, "setsockopt");
        GetUser()->Deliver (&statusind);
        closesocket (fd);
        return;
    }
#endif
 
    if ( bind (fd, req->laddr, req->lalen ) != 0 ) {
        MSG_IOevent_IOerror_SET (&statusind.msg, "bind");
        GetUser()->Deliver (&statusind);
        closesocket (fd);
        return;
    }
 
#ifdef IC_WIN32
    // Windows sockets are inherited by default
    if ( SetHandleInformation ((HANDLE)fd,
                   HANDLE_FLAG_INHERIT, // mask of bits to change
                   0) == 0 ) {
        MSG_IOevent_Inherit_LOG ();
    }
#elif defined (FD_CLOEXEC)
    (void) fcntl (fd, F_SETFD, FD_CLOEXEC);
#endif
 
#ifdef O_NONBLOCK
    if ( fcntl (fd, F_SETFL, O_NONBLOCK) != 0 ) {
        MSG_IOevent_IOerror_SET (&statusind.msg, "fcntl");
        GetUser()->Deliver (&statusind);
        closesocket (fd);
        return;
    }
#elif defined (FIONBIO)
    char con = 1;
    if ( ioctl (fd, FIONBIO, &con) != 0 ) {
        MSG_IOevent_IOerror_SET (&statusind.msg, "ioctl");
        GetUser()->Deliver (&statusind);
        closesocket (fd);
        return;
    }
#else
# error Cannot set socket into non-blocking more
#endif
 
    if ( listen (fd, req->backlog ? req->backlog : SOMAXCONN) != 0 ) {
        MSG_IOevent_IOerror_SET(&statusind.msg, "listen");
        GetUser()->Deliver (&statusind);
        closesocket (fd);
        return;
    }
 
    // Listening
    _fd    = fd;
    _state = State_listening;
 
    // Listen events are 'In' events
    waitEvent (::Poll::Event_In);
}
 
void Stream::SocketPoll::actualDeliver (Stream::ConnectAccept *req ARGNOTUSED)
{
    if ( checkUser() ) return;
 
    StatusIndication statusind (this);
 
    LOG_DEBUG (("SocketPoll::actualDeliver (ConnectAccept)"));
 
    if ( _fd == INVALID_ENDPOINT || _state != State_accepting ) {
        MSG_IOevent_NotConnected_SET (&statusind.msg);
 
        GetUser()->Deliver (&statusind);
 
        return;
    }
 
    // For a socket interface there is nothing to do to accept the connection
    _state = GetUser()->Proxy() ? State_need_proxy : State_connected;
}
 
void Stream::SocketPoll::actualDeliver (Stream::DisconnectRequest *req ARGNOTUSED)
{
    if ( checkUser() ) return;
 
    StatusIndication statusind (this);
 
    LOG_DEBUG (("SocketPoll::actualDeliver (DisconnectRequest)"));
 
    switch ( _state ) {
    case State_need_proxy:
    case State_connected:
    case State_accepting:
    case State_tls:
        if ( _fd != INVALID_ENDPOINT )
            break;
        BOOST_FALLTHROUGH;
 
    default:
        MSG_IOevent_NotConnected_SET (&statusind.msg);
        GetUser()->Deliver (&statusind);
        return;
    }
 
    if ( _state == State_tls ) {
        _tlsstate = TLS_close;
 
        doTLS();
 
    } else {
        // Close up connection, etc.
 
        MSG_IOevent_ConnectionClosed_SET (&statusind.msg);
        GetUser()->Deliver (&statusind);
 
        tidy ();
    }
}
 
void Stream::SocketPoll::actualDeliver (Stream::DataRequest *req)
{
    if ( checkUser() ) return;
 
    StatusIndication statusind (this);
 
    LOG_DEBUG (("SocketPoll::actualDeliver (DataRequest) state=%d _wlen=%ld",
        _state, (long)_wlen));
 
    switch ( _state ) {
    case State_need_proxy:
    case State_connected:
    case State_tls:
        if ( _fd != INVALID_ENDPOINT )
            break;
        BOOST_FALLTHROUGH;
 
    default:
        MSG_IOevent_NotConnected_SET (&statusind.msg);
        GetUser()->Deliver (&statusind);
        return;
    }
 
    // Note: should not attempt to write zero bytes
    if ( req->buf == 0 || req->len == 0 ) {
        MSG_IOevent_InvalidParameter_SET (&statusind.msg,
                          req->buf == NULL ?
                          "buffer pointer" :
                          "zero length");
 
        GetUser()->Deliver (&statusind);
 
        return;
    }
 
    if ( _wlen != 0 ) {
        ReleaseBuf release;
 
        release.buf = req->buf;
        release.len = -1;
 
        GetUser()->Deliver (&release);
 
        MSG_IOevent_DupWrite_SET (&statusind.msg);
 
        GetUser()->Deliver (&statusind);
 
        return;
    }
 
    _wbuf = _wptr = req->buf;
    _wlen = req->len;
 
    if ( _state == State_connected || _state == State_need_proxy ) {
        if (_state == State_need_proxy) {
            LOG_DEBUG (("Need proxy state seen when writing"));
        }
        if ( !doWrite () ) {
            // Write blocked, tell poller we want to know when can write
            waitEvent (::Poll::Event_Out);
        }
 
    } else {
        doTLS();
    }
}
 
void Stream::SocketPoll::actualDeliver (Stream::ReadRequest *req)
{
    if ( checkUser() ) return;
 
    DataIndication dataind;
    StatusIndication statusind (this);
 
    LOG_DEBUG (("SocketPoll::actualDeliver (ReadRequest)"));
 
    switch ( _state ) {
    case State_need_proxy:
    case State_connected:
    case State_tls:
        if ( _fd != INVALID_ENDPOINT )
            break;
        BOOST_FALLTHROUGH;
 
    default:
        MSG_IOevent_NotConnected_SET (&statusind.msg);
        GetUser()->Deliver (&statusind);
        return;
    }
 
    if ( req->buf == 0 || req->len == 0 ) {
        MSG_IOevent_InvalidParameter_SET (&statusind.msg,
                          req->buf == NULL ?
                          "buffer pointer" :
                          "zero length");
 
        GetUser()->Deliver (&statusind);
 
        return;
    }
 
    if ( _rlen != 0 ) {
        dataind.buf = req->buf;
        dataind.len = -1;
 
        GetUser()->Deliver (&dataind);
 
        MSG_IOevent_DupRead_SET (&statusind.msg);
 
        GetUser()->Deliver (&statusind);
 
        return;
    }
 
    _rbuf = _rptr = req->buf;
    _rlen  = req->len;
    _rfill = req->fill;
 
    LOG_DEBUG (("ReadRequest: _rbuf=%p _rlen=%d _rfill=%d",
        _rbuf, (int)_rlen, _rfill));
 
    if ( _state == State_connected || _state == State_need_proxy ) {
        // If the previous read filled the provided buffer, then
        // we attempt a read at this point. There may be more data
        // already buffered in the OS
 
        if ( _rfull
            && ((_state == State_connected  && doRead () )
             || (_state == State_need_proxy && doProxy() )) )
            return;
 
        // tell poller we are interested in read events
        waitEvent (::Poll::Event_In);
 
    } else {
        doTLS();
    }
}
 
void Stream::SocketPoll::actualDeliver (Stream::StartTLS *req)
{
    if ( checkUser() ) return;
 
    LOG_DEBUG (("SocketPoll::actualDeliver (StartTLS)"));
 
    StatusIndication statusind (this);
 
    switch ( _state ) {
    case State_tls:
        MSG_IOevent_SSLalready_SET (&statusind.msg);
        GetUser()->Deliver (&statusind);
        return;
 
    case State_connected:
        if ( _fd != INVALID_ENDPOINT )
            break;
        BOOST_FALLTHROUGH;
 
    default:
        MSG_IOevent_NotConnected_SET (&statusind.msg);
        GetUser()->Deliver (&statusind);
        return;
    }
 
    // Only check hostname of server
    _hostname = req->client ? req->hostname : 0;
 
    SSLTLS::OpenSSLContext *sslctx =
    dynamic_cast<SSLTLS::OpenSSLContext *>(req->context);
 
    _ssl = sslctx ? sslctx->GetSSL(_choice) : NULL;
    if (!_ssl) {
        MSG_IOevent_SSLerror_SET (&statusind.msg, "SSL_new", "this returned NULL. No ID configured?");
        GetUser()->Deliver (&statusind);
        return;
    }
 
    // Check for any OpenSSL error set during previous use of this thread
    int err = SSL_get_error(_ssl, 0);
    if (err != SSL_ERROR_NONE) {
        // Report but clear and continue
        LOG_DEBUG (("SocketPoll::actualDeliver(StartTLS) - SSL error code %d", err));
        ERR_clear_error();
    }
 
    BIO *bio = BIO_new_socket (_fd, TRUE);
 
    SSL_set_bio(_ssl, bio, bio);
 
    if ( req->client )
        SSL_set_connect_state (_ssl);
    else
        SSL_set_accept_state (_ssl);
 
    _state = State_tls;
    _tlsstate = TLS_handshake;
 
    doTLS();
}
 
void Stream::SocketPoll::actualDeliver (Stream::StreamControl *option)
{
    if ( checkUser() ) return;
 
    LOG_DEBUG (("SocketPoll::actualDeliver (StreamControl)"));
 
    StatusIndication statusind (this);
 
    if ( _fd == INVALID_ENDPOINT ) {
        MSG_IOevent_NotConnected_SET (&statusind.msg);
        GetUser()->Deliver (&statusind);
        return;
    }
 
    int level;
    int name;
 
    switch ( option->option ) {
#if defined(TCP_NODELAY)
        case StreamControlNoDelay:
            level = IPPROTO_TCP;
            name  = TCP_NODELAY;
            break;
#endif
 
        case StreamControlKeepAlive:
            level = SOL_SOCKET;
            name  = SO_KEEPALIVE;
            break;
 
#if defined(IP_TOS)
    case StreamControlTOS:
#ifdef SOL_IP
        level = SOL_IP;
#else
        level = IPPROTO_IP;
#endif
        name  = IP_TOS;
        break;
#endif
 
    default:
        MSG_IOevent_UnknownStreamOpt_SET (&statusind.msg, option->option);
        GetUser()->Deliver (&statusind);
        return;
    }
 
    int rc = setsockopt (_fd, level, name, (char *)&option->value, sizeof option->value);
    if ( rc != 0 ) {
        MSG_IOevent_IOerror_SET (&statusind.msg, "setsockopt");
        GetUser()->Deliver (&statusind);
        tidy ();
    }
}
 
 
Stream::Provider *Stream::CreateSocketPollProvider ()
{
    // In a single-threaded environment COULD create a SocketPoll
    return new Stream::SyncSocketPoll;
}