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 "../include/Syncmsg.h"
 
#include <isode/messages/base.h>
#include <isode/base/crypto.h>
 
// 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 {
 
    // 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_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;  
 
    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);
 
    // 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:
    virtual ~SocketPoll() {
        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);
    }
 
    virtual void Die () { 
        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
    virtual endpoint_t GetEndpoint (bool inheriting) {
        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;
    }
 
        virtual unsigned Events () {
            return _uevents;
        }
 
    virtual int GetLocalAddress (struct sockaddr *laddr,
                     socklen_t *lalenp,
                     MSGstruct *msp) {
 
        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;
    }
 
    virtual int GetPeerAddress (struct sockaddr *laddr,
                    socklen_t *lalenp,
                    MSGstruct *msp) {
 
        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;
    }
 
    virtual TLS_CipherSuite GetCipherSuite () {
        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);
    }
 
    virtual X509* GetPeerCertificate () {
        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);
 
    // 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
 
    virtual void Deliver (::Poll::pollmsg *msg) {
        actualDeliver (msg);
    }
 
    // Interface to stream user
    virtual void Deliver (External *ext) {
        actualDeliver (ext);
    }
 
    virtual void Deliver (ConnectRequest *req) {
        actualDeliver (req);
    }
 
    virtual void Deliver (ListenRequest *req) {
        actualDeliver (req);
    }
 
    virtual void Deliver (ConnectAccept *req) {
        actualDeliver (req);
    }
 
    virtual void Deliver (DisconnectRequest *req) {
        actualDeliver (req);
    }
 
    virtual void Deliver (DataRequest *req) {
        actualDeliver (req);
    }
 
    virtual void Deliver (ReadRequest *req) {
        actualDeliver (req);
    }
 
    virtual void Deliver (StartTLS *req) {
        actualDeliver (req);
    }
 
    virtual void Deliver (StreamControl *option) {
        actualDeliver (option);
    }
 
    virtual void Deliver (LengthFnxRequest *) {
        // Don't bother to synchronize this
    }
    };
 
    class SyncSocketPoll : public Syncmsgobj<SocketPoll> {
    protected:
        virtual ~SyncSocketPoll () {
            // 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;
 
        virtual SocketPoll *Clone () {
            return new SyncSocketPoll ();
        }
 
        virtual void Die () {
            dieAux ();
        }
 
        virtual void Deliver (time_t *msg) {
            actualDeliver (msg);
        }
    
        virtual void Deliver (::Poll::pollmsg *msg) {
            msgDeliver (msg);
        }
 
        // Interface to stream user
        virtual void Deliver (External *ext) {
            msgDeliver (ext);
        }
 
        virtual void Deliver (ConnectRequest *req) {
            msgDeliver (req);
        }
 
        virtual void Deliver (ListenRequest *req) {
            msgDeliver (req);
        }
 
        virtual void Deliver (ConnectAccept *req) {
            msgDeliver (req);
        }
 
        virtual void Deliver (DisconnectRequest *req) {
            msgDeliver (req);
        }
 
        virtual void Deliver (DataRequest *req) {
            msgDeliver (req);
        }
 
        virtual void Deliver (ReadRequest *req) {
            msgDeliver (req);
        }
 
        virtual void Deliver (StartTLS *req) {
            msgDeliver (req);
        }
 
        virtual void Deliver (StreamControl *option) {
            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;
}
 
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 ) {
        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;
    }
 
    // 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 (("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;
 
        _tlsstate = TLS_idle;
 
        GetUser()->Deliver (&dataind);
    }
 
    return nbytes;
}
 
int Stream::SocketPoll::doTLSwrite ()
{
    int remain = (int)_wlen - (_wptr - _wbuf);
 
    int nbytes = SSL_write (_ssl, _wptr, remain);
 
    // 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);
 
    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);
 
    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 ()
{
    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_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_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 = 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 = 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 = 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 = 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_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_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 ) {
        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_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 ) {
        // 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 && doRead () )
            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;
}