libplumbing/endpoint/client.cc

Go to the documentation of this file.

//
// plumbing - C++ wrapper facades for Berkeley sockets
// Copyright (C) 2009, 2010 Peter Miller
//
// This program is free software; you can redistribute it and/or modify it
// under the terms of the GNU General Public License, version 3, as published
// by the Free Software Foundation.
//
// This program is distributed in the hope that it will be useful, but WITHOUT
// ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
// FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
// more details.
//
// You should have received a copy of the GNU General Public License along with
// this program. If not, see <http://www.gnu.org/licenses/>.
//

#include <libplumbing/config.h>
#include <arpa/inet.h>
#include <cassert>
#include <cerrno>
#include <cstring>
#include <fcntl.h>
#include <netdb.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <libexplain/connect.h>
#include <libexplain/fcntl.h>
#include <libexplain/getsockopt.h>
#include <libexplain/read.h>
#include <libexplain/socket.h>
#include <libexplain/write.h>

#include <libplumbing/logging.h>
#include <libplumbing/endpoint/client.h>


plumbing::endpoint_client::~endpoint_client()
{
    delete (struct sockaddr_in *)sock_addr;
}


plumbing::endpoint_client::endpoint_client(const std::string &a_hostname,
        int a_port) :
    endpoint(-1),
    hostname(a_hostname),
    port(a_port),
    sock_addr(0),
    sock_addr_size(0),
    connecting(false),
    retry_after(0)
{
    char buf[100];
    snprintf(buf, sizeof(buf), "host \"%s\": port %d", hostname.c_str(), port);
    set_peer_name(buf);
    try_to_connect();
}


plumbing::endpoint_client::endpoint_client(const std::string &a_hostname,
        const std::string &a_port) :
    endpoint(-1),
    hostname(a_hostname),
    port(0),
    sock_addr(0),
    sock_addr_size(0),
    connecting(false),
    retry_after(0)
{
    char buf[100];
    snprintf
    (
        buf,
        sizeof(buf),
        "host \"%s\": port \"%s\"",
        hostname.c_str(),
        a_port.c_str()
    );
    set_peer_name(buf);

    port = parse_port_number_or_die(a_port);

    try_to_connect();
}


void
plumbing::endpoint_client::write(const std::string &text)
{
    write_queue.push_back(text + "\n");
}


int
plumbing::endpoint_client::get_write_file_descriptor()
{
    try_to_connect();
    if (connecting)
        return fd;
    return (write_queue.empty() ? -1 : fd);
}


void
plumbing::endpoint_client::process_write()
{
    if (connecting)
    {
        assert(fd >= 0);
        connecting = false;

        //
        // From the connect(2) man page:
        //
        // [If] the socket is non-blocking and the connection cannot
        // be completed immediately, it is possible to select(2) or
        // poll(2) for completion by selecting the socket for writing.
        // After select(2) indicates writability, use getsockopt(2) to
        // read the SO_ERROR option at level SOL_SOCKET to determine
        // whether connect() completed successfully (SO_ERROR is zero)
        // or unsuccessfully (SO_ERROR is one of the usual error codes
        // ... explaining the reason for the failure)."
        //
        // From the socket(7) man page:
        //
        // "SO_ERROR
        //     Get and clear the pending socket error.  Only valid as a
        //     getsockopt().  Expects an integer."
        //
        int error = 0;
        socklen_t len = sizeof(error);
        if (getsockopt(fd, SOL_SOCKET, SO_ERROR, &error, &len) < 0)
        {
            fatal_error
            (
                "%s",
                explain_getsockopt(fd, SOL_SOCKET, SO_ERROR, &error, &len)
            );
        }
        errno = error;
        switch (errno)
        {
        case 0:
            // No error.  Now successfully connected.
            break;

        case ECONNREFUSED:
        case ENETUNREACH:
            //
            // Definite negatives.
            // Wait a while and try again.
            //
            warning
            (
                "%s",
                explain_errno_connect
                (
                    error,
                    fd,
                    (const struct sockaddr *)sock_addr,
                    sock_addr_size
                )
            );
            close();
            kill_me_now();
            return;

        default:
            fatal_error
            (
                "%s",
                explain_errno_connect
                (
                    error,
                    fd,
                    (const struct sockaddr *)sock_addr,
                    sock_addr_size
                )
            );
            // NOTREACHED
            return;
        }

        //
        // Make the socket blocking again.
        //
        int flags = fcntl(fd, F_GETFL, 0);
        if (flags < 0)
            fatal_error("%s", explain_fcntl(fd, F_GETFL, 0));
        flags &= ~O_NONBLOCK;
        int err = fcntl(fd, F_SETFL, flags);
        if (err < 0)
            fatal_error("%s", explain_fcntl(fd, F_SETFL, flags));
    }

    if (fd < 0)
        return;
    if (write_queue.empty())
        return;
    std::string s = write_queue.front();
    ssize_t n = ::write(fd, s.c_str(), s.size());
    int err = errno;
    if (n != ssize_t(s.size()))
    {
        // Don't take it off the queue yet.
        warning("%s", explain_errno_write(err, fd, s.c_str(), s.size()));
        close();
        kill_me_now();
        return;
    }
    write_queue.pop_front();
}


int
plumbing::endpoint_client::get_read_file_descriptor()
{
    try_to_connect();
    if (connecting)
        return -1;
    return fd;
}


void
plumbing::endpoint_client::process_read()
{
    assert(fd >= 0);
    if (fd < 0)
        return;

    char buffer[1 << 12];
    size_t nbytes = sizeof(buffer);
    ssize_t n = ::read(fd, buffer, nbytes);
    if (n == -1)
    {
        int err = errno;
        fatal_error("%s", explain_errno_read(err, fd, buffer, nbytes));
    }
    if (n == 0)
    {
        //
        // This isn't a fatal error.  The connection was probably
        // closed by the server at the other end (maybe it crashed,
        // maybe it is restarting, maybe it closed the connection).
        //
        warning("connection closed by peer");
        data_received(0, 0);
        close();

        //
        // While you may expect kill_me_now to be called, that is for
        // the derived class to decide.  If it does not, the socket
        // connection will be retried.
        //
        return;
    }

    //
    // Do something with the data.
    //
    data_received(buffer, n);
}


void
plumbing::endpoint_client::try_to_connect()
{
    if (fd >= 0)
        return;
    if (connecting)
        return;

    //
    // Only try this every 10 seconds.
    //
    time_t now;
    time(&now);
    if (now < retry_after)
        return;
    retry_after = now + 10;

    if (!sock_addr)
    {
        struct hostent *he = gethostbyname(hostname.c_str());
        if (!he)
        {
            warning("gethostbyname: %s", hstrerror(h_errno));
            kill_me_now();
            return;
        }

        //
        // Build the appropriate network address to connect to.
        //
        struct sockaddr_in *sain_p = new struct sockaddr_in;
        memset(sain_p, 0, sizeof(*sain_p));
        sain_p->sin_addr.s_addr = *(in_addr_t *)he->h_addr;
        sain_p->sin_family = AF_INET;
        sain_p->sin_port = htons(port);

        sock_addr = sain_p;
        sock_addr_size = sizeof(*sain_p);
    }

    connecting = true;

    //
    // Create the socket.
    //
    fd = socket(AF_INET, SOCK_STREAM, 0);
    if (fd < 0)
        fatal_error("%s", explain_socket(AF_INET, SOCK_STREAM, 0));

    //
    // Make the socket non-blocking.
    //
    int flags = fcntl(fd, F_GETFL);
    if (flags < 0)
        fatal_error("%s", explain_fcntl(fd, F_GETFL, 0));
    flags |= O_NONBLOCK;
    int err = fcntl(fd, F_SETFL, flags);
    if (err < 0)
        fatal_error("%s", explain_fcntl(fd, F_SETFL, flags));

    //
    // From the connect(2) man page:
    //
    // "EINPROGRESS
    //     The socket is non-blocking and the connection cannot be
    //     completed immediately.  It is possible to select(2) or
    //     poll(2) for completion by selecting the socket for writing.
    //     After select(2) indicates writability, use getsockopt(2) to
    //     read the SO_ERROR option at level SOL_SOCKET to determine
    //     whether connect() completed successfully (SO_ERROR is zero)
    //     or unsuccessfully (SO_ERROR is one of the usual error codes
    //     ... explaining the reason for the failure)."
    //
    err = connect(fd, (const struct sockaddr *)sock_addr, sock_addr_size);

    //
    // We said non-blocking, but there is just a small chance that it
    // will be able to connect immediately.
    //
    if (err == 0)
    {
        connecting = false;
        return;
    }

    //
    // Depending on what errno says, we do different things.
    //
    int error_num = errno;
    switch (error_num)
    {
    case EAGAIN:
    case EALREADY:
    case EINPROGRESS:
        //
        // Cool our heels and wait.
        //
        return;

    case ECONNREFUSED:
    case ENETUNREACH:
        //
        // Definite negatives.
        // Wait a while and try again.
        //
        warning
        (
            "%s",
            explain_errno_connect
            (
                error_num,
                fd,
                (const struct sockaddr *)sock_addr,
                sock_addr_size
            )
        );
        connecting = false;
        close();
        kill_me_now();
        return;

    default:
        break;
    }

    fatal_error
    (
        "%s",
        explain_errno_connect
        (
            error_num,
            fd,
            (const struct sockaddr *)sock_addr,
            sock_addr_size
        )
    );
    // NOTREACHED
}


// vim: set ts=8 sw=4 et

---