Stream::SocketPoll Class Reference

Stream provider using sockets and using poll. More...

Inheritance diagram for Stream::SocketPoll:

Public Member Functions
void	Die () override
	Make this provider go away.
void	tidy ()
	Tidy up - close connections etc.
endpoint_t	GetEndpoint (bool inheriting) override
	Get the endpoint ID.
unsigned	Events () override
	Get the events.
int	GetLocalAddress (struct sockaddr *laddr, socklen_t *lalenp, MSGstruct *msp) override
	Get the local address.
int	GetPeerAddress (struct sockaddr *laddr, socklen_t *lalenp, MSGstruct *msp) override
	Get the peer address.
TLS_CipherSuite	GetCipherSuite () override
	Get the cipher suite.
X509 *	GetPeerCertificate () override
	Get the peer certificate.
virtual SocketPoll *	Clone ()=0
	Make another provider like this one.
void	actualDeliver (time_t *msg)
	Handle timer event.
void	actualDeliver (::Poll::pollmsg *msg) override
	Actual Deliver events to Poll provider.
void	actualDeliver (External *ext)
	Set external.
void	actualDeliver (ConnectRequest *req)
	connect
void	actualDeliver (ListenRequest *req)
	listen
void	actualDeliver (ConnectAccept *req)
	accept
void	actualDeliver (DisconnectRequest *req)
	disconnect
void	actualDeliver (DataRequest *req)
	send data
void	actualDeliver (ReadRequest *req)
	ReadRequest.
void	actualDeliver (StartTLS *req)
	Start SSL/TLS on stream.
void	actualDeliver (StreamControl *option)
	Control stream.
virtual void	Deliver (time_t *msg)
	Handle timer event.
void	Deliver (::Poll::pollmsg *msg) override
	Deliver events to Poll provider.
void	Deliver (External *ext) override
	Set external.
void	Deliver (ConnectRequest *req) override
	connect
void	Deliver (ListenRequest *req) override
	listen
void	Deliver (ConnectAccept *req) override
	accept
void	Deliver (DisconnectRequest *req) override
	disconnect
void	Deliver (DataRequest *req) override
	send data
void	Deliver (ReadRequest *req) override
	Read request.
void	Deliver (StartTLS *req) override
	Start SSL/TLS on stream.
void	Deliver (StreamControl *option) override
	Control stream.
void	Deliver (LengthFnxRequest *) override
	Set length function.
Public Member Functions inherited from Stream::Provider
	Provider ()
	The user of this provider.
virtual	~Provider ()
	Destructor should be virtual.
void	SetUser (User *u)
	set the user
User *	GetUser ()
	get the user
Public Member Functions inherited from Poll::User
	User ()
	User's view of events.
	~User () override
	Destructor is virtual.
virtual void	Deliver (pollmsg *msg)=0
	Deliver events to Poll user.
virtual void	actualDeliver (pollmsg *msg)=0
	Deliver events to Poll user.

Additional Inherited Members
Static Public Member Functions inherited from Stream::Provider
static EVENTSVC_DLL Provider *	ProviderFactory (const char *type, MSGstruct *msp)
	Factory method for Stream provider.
Data Fields inherited from Poll::User
UserEvent	_event
endpoint_t	_fd
	Event to User.
unsigned	_pevents
	FD for this user.
unsigned	_uevents
	Provider's view of events.

Detailed Description

Stream provider using sockets and using poll.

Definition at line 67 of file stream_socket_poll.C.

Constructor & Destructor Documentation

~SocketPoll()

Stream::SocketPoll::~SocketPoll ( )

inlineoverrideprotected

Definition at line 183 of file stream_socket_poll.C.

                               {
            tidy ();
        }

SocketPoll()

Stream::SocketPoll::SocketPoll ( )

inline

Definition at line 188 of file stream_socket_poll.C.

                      :
            _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);
        }

Member Function Documentation

Die()

void Stream::SocketPoll::Die ( )

inlineoverridevirtual

Make this provider go away.

Implements Stream::Provider.

Definition at line 205 of file stream_socket_poll.C.

                             { 
            delete this;
        }

tidy()

void Stream::SocketPoll::tidy ( )

inline

Tidy up - close connections etc.

Definition at line 210 of file stream_socket_poll.C.

                            {
            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;
        }

References Poll::User::_fd, Poll::Provider::Deregister(), and Stream::Provider::SetUser().

GetEndpoint()

endpoint_t Stream::SocketPoll::GetEndpoint ( bool  inheriting )

inlineoverridevirtual

Get the endpoint ID.

Implements Stream::Provider.

Definition at line 259 of file stream_socket_poll.C.

                                                          {
            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;
        }

References Poll::User::_fd.

Events()

unsigned Stream::SocketPoll::Events ( )

inlineoverridevirtual

Get the events.

Implements Stream::Provider.

Definition at line 285 of file stream_socket_poll.C.

                                    {
            return _uevents;
        }

References Poll::User::_uevents.

GetLocalAddress()

int Stream::SocketPoll::GetLocalAddress ( struct sockaddr *  laddr, socklen_t *  lalenp, MSGstruct *  msp  )

inlineoverridevirtual

Get the local address.

Implements Stream::Provider.

Definition at line 290 of file stream_socket_poll.C.

                                                 {
 
            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;
        }

References Poll::User::_fd.

GetPeerAddress()

int Stream::SocketPoll::GetPeerAddress ( struct sockaddr *  laddr, socklen_t *  lalenp, MSGstruct *  msp  )

inlineoverridevirtual

Get the peer address.

Implements Stream::Provider.

Definition at line 317 of file stream_socket_poll.C.

                                                 {
 
            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;
        }

References Poll::User::_fd.

GetCipherSuite()

TLS_CipherSuite Stream::SocketPoll::GetCipherSuite ( )

inlineoverridevirtual

Get the cipher suite.

Reimplemented from Stream::Provider.

Definition at line 344 of file stream_socket_poll.C.

                                                   {
            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);
        }

GetPeerCertificate()

X509 * Stream::SocketPoll::GetPeerCertificate ( )

inlineoverridevirtual

Get the peer certificate.

Reimplemented from Stream::Provider.

Definition at line 353 of file stream_socket_poll.C.

                                             {
            if ( _state != State_tls || _ssl == 0 )
                return 0;
 
            return SSL_get_peer_certificate(_ssl);
        }

actualDeliver() [1/9]

void SocketPoll::actualDeliver

(

time_t *

msg

)

Handle timer event.

Definition at line 1541 of file stream_socket_poll.C.

{
    // Timer has expired after EMFILE, try again
    waitEvent (::Poll::Event_In);
}

Referenced by Deliver(), Deliver(), Deliver(), Deliver(), Deliver(), Deliver(), Deliver(), Deliver(), Deliver(), Deliver(), and Deliver().

actualDeliver() [2/9]

void SocketPoll::actualDeliver ( ::Poll::pollmsg *  msg )

override

Actual Deliver events to Poll provider.

Definition at line 1547 of file stream_socket_poll.C.

{
    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);
}

actualDeliver() [3/9]

void SocketPoll::actualDeliver

(

Stream::External *

ext

)

Set external.

Definition at line 1681 of file stream_socket_poll.C.

{
    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;
    }
}

References Stream::External::listen.

actualDeliver() [4/9]

void SocketPoll::actualDeliver

(

Stream::ConnectRequest *

req

)

connect

Definition at line 1707 of file stream_socket_poll.C.

{
    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);
    }
}

References Stream::ConnectRequest::dalen, Stream::ConnectRequest::dscp, Stream::ConnectRequest::saddr, and Stream::ConnectRequest::salen.

actualDeliver() [5/9]

void SocketPoll::actualDeliver

(

Stream::ListenRequest *

req

)

listen

Definition at line 1888 of file stream_socket_poll.C.

{
    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);
}

References Stream::ListenRequest::backlog, and Stream::ListenRequest::lalen.

actualDeliver() [6/9]

void SocketPoll::actualDeliver

(

Stream::DataRequest *

req

)

send data

Definition at line 2073 of file stream_socket_poll.C.

{
    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();
    }
}

References Stream::DataRequest::len, and Stream::ReleaseBuf::len.

actualDeliver() [7/9]

void SocketPoll::actualDeliver

(

Stream::ReadRequest *

req

)

ReadRequest.

Definition at line 2140 of file stream_socket_poll.C.

{
    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();
    }
}

References Stream::ReadRequest::fill, Stream::ReadRequest::len, and Stream::DataIndication::len.

actualDeliver() [8/9]

void SocketPoll::actualDeliver

(

Stream::StartTLS *

req

)

Start SSL/TLS on stream.

Definition at line 2212 of file stream_socket_poll.C.

{
    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();
}

References Stream::StartTLS::client, and Stream::StartTLS::hostname.

actualDeliver() [9/9]

void SocketPoll::actualDeliver

(

Stream::StreamControl *

option

)

Control stream.

Definition at line 2273 of file stream_socket_poll.C.

{
    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 ();
    }
}

References Stream::StreamControlKeepAlive, Stream::StreamControlTOS, and Stream::StreamControl::value.

Deliver() [1/12]

virtual void Stream::SocketPoll::Deliver ( time_t *  msg )

inlinevirtual

Handle timer event.

Definition at line 402 of file stream_socket_poll.C.

                                           {
            actualDeliver (msg);
        }

References actualDeliver().

Deliver() [2/12]

void Stream::SocketPoll::Deliver ( ::Poll::pollmsg *  msg )

inlineoverride

Deliver events to Poll provider.

Definition at line 409 of file stream_socket_poll.C.

                                                 {
            actualDeliver (msg);
        }

References actualDeliver().

Deliver() [3/12]

void Stream::SocketPoll::Deliver ( External *  ext )

inlineoverridevirtual

Set external.

Implements Stream::Provider.

Definition at line 415 of file stream_socket_poll.C.

                                              {
            actualDeliver (ext);
        }

References actualDeliver().

Deliver() [4/12]

void Stream::SocketPoll::Deliver ( ConnectRequest *  req )

inlineoverridevirtual

connect

Implements Stream::Provider.

Definition at line 420 of file stream_socket_poll.C.

                                                    {
            actualDeliver (req);
        }

References actualDeliver().

Deliver() [5/12]

void Stream::SocketPoll::Deliver ( ListenRequest *  req )

inlineoverridevirtual

listen

Implements Stream::Provider.

Definition at line 425 of file stream_socket_poll.C.

                                                   {
            actualDeliver (req);
        }

References actualDeliver().

Deliver() [6/12]

void Stream::SocketPoll::Deliver ( ConnectAccept *  req )

inlineoverridevirtual

accept

Implements Stream::Provider.

Definition at line 430 of file stream_socket_poll.C.

                                                   {
            actualDeliver (req);
        }

References actualDeliver().

Deliver() [7/12]

void Stream::SocketPoll::Deliver ( DisconnectRequest *  req )

inlineoverridevirtual

disconnect

Implements Stream::Provider.

Definition at line 435 of file stream_socket_poll.C.

                                                       {
            actualDeliver (req);
        }

References actualDeliver().

Deliver() [8/12]

void Stream::SocketPoll::Deliver ( DataRequest *  req )

inlineoverridevirtual

send data

Implements Stream::Provider.

Definition at line 440 of file stream_socket_poll.C.

                                                 {
            actualDeliver (req);
        }

References actualDeliver().

Deliver() [9/12]

void Stream::SocketPoll::Deliver ( ReadRequest *  req )

inlineoverridevirtual

Read request.

Implements Stream::Provider.

Definition at line 445 of file stream_socket_poll.C.

                                                 {
            actualDeliver (req);
        }

References actualDeliver().

Deliver() [10/12]

void Stream::SocketPoll::Deliver ( StartTLS *  req )

inlineoverridevirtual

Start SSL/TLS on stream.

Implements Stream::Provider.

Definition at line 450 of file stream_socket_poll.C.

                                              {
            actualDeliver (req);
        }

References actualDeliver().

Deliver() [11/12]

void Stream::SocketPoll::Deliver ( StreamControl *  option )

inlineoverridevirtual

Control stream.

Implements Stream::Provider.

Definition at line 455 of file stream_socket_poll.C.

                                                      {
            actualDeliver (option);
        }

References actualDeliver().

Deliver() [12/12]

void Stream::SocketPoll::Deliver ( LengthFnxRequest *  )

inlineoverridevirtual

Set length function.

Implements Stream::Provider.

Definition at line 460 of file stream_socket_poll.C.

                                                   {
            // Don't bother to synchronize this
        }

---

The documentation for this class was generated from the following file:

• stream_socket_poll.C