UdpConnection.h and UdpConnection.cpp

/*
 * UdpConnection.h
 *
 *  Created on: 2011-02-25
 *      Author: cdunphy
 */

// Portions Copyright (c) 2003-2011 Christopher M. Kohlhoff (chris at kohlhoff dot com)
//
// Distributed under the Boost Software License, Version 1.0. (See accompanying
// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
//
#ifndef DATAGRAMCONNECTION_H_
#define DATAGRAMCONNECTION_H_

#include "AsioConnection.h"

#include <boost/lexical_cast.hpp>

namespace shaw_rsc
{

struct UdpException: public std::runtime_error
{
    UdpException(const std::string& msg) : std::runtime_error(msg) { }
};

/*
 * This is the concrete class that manages UDP connections.
 */
class UdpConnection: public AsioConnection
{
public:
    /*
     * Use this constructor for clients (connecting to a remote socket).
     */
    UdpConnection(
        const std::string& _host,
        const std::string& _port,
        int r_timeout,
        Mode mode
    ) : AsioConnection(0, r_timeout),
            socket(io_service),
            remote_endpoint(),
            host(_host),
            port(_port)
    {
        check_deadline();
        connect(mode);
    }

    std::size_t write(const DataBuffer& data);
    std::size_t read(DataBuffer& data);
    const std::string str() const;

private:

    void connect(Mode mode);
    void check_deadline();

    boost::asio::ip::udp::socket socket;
    boost::asio::ip::udp::endpoint remote_endpoint;
    std::string host;
    std::string port;
};

}

#endif /* DATAGRAMCONNECTION_H_ */


/*
 * UdpConnection.cpp
 *
 *  Created on: 2011-02-25
 *      Author: cdunphy
 */

#include "UdpConnection.h"

using std::string;
using std::endl;
using std::stringstream;
using std::exception;

using boost::asio::buffer;
using boost::asio::ip::udp;
using boost::system::error_code;
using boost::system::system_error;
using boost::asio::deadline_timer;
using boost::bind;
using boost::lexical_cast;

namespace shaw_rsc
{

size_t UdpConnection::write(const DataBuffer& data)
{
    size_t bytes_written = 0;

    /*
     * Check to see if the socket is bad before writing
     */
    if (error &&
            error.value() != boost::asio::error::operation_aborted &&
            error.value() != boost::asio::error::timed_out &&
            error != boost::asio::error::try_again)
        throw UdpException(error.message());

    socket.async_send_to(buffer(data), remote_endpoint,
                         strand.wrap(bind(&AsioConnection::handle_send, this, _1, _2,
                                          &error, &bytes_written)));

    do
    {
        strand.get_io_service().run_one();
    }
    while (error == boost::asio::error::would_block
            || error == boost::asio::error::try_again);

    if (error)
    {
        if (error.value() == boost::asio::error::operation_aborted
                || error.value() == boost::asio::error::timed_out)
            throw SocketTimeoutException(error.message());
        else
            throw UdpException(error.message());
    }

    reset_deadline();
    return bytes_written;
}

size_t UdpConnection::read(DataBuffer& data)
{
    /*
    * Check to see if the socket is bad before writing
    */
    if (error &&
            error.value() != boost::asio::error::operation_aborted &&
            error.value() != boost::asio::error::timed_out &&
            error != boost::asio::error::try_again)
        throw UdpException(error.message());

    data.clear();

    /*
     * Reset the deadline timer to expire according
     * to the configured read timeout.
     */
    deadline.expires_from_now(read_timeout);

    size_t bytes_read = 0;
    boost::array<char, BUF_SIZE> buff;

    error = boost::asio::error::would_block;
    socket.async_receive_from(buffer(buff), remote_endpoint,
                              strand.wrap(boost::bind(&AsioConnection::handle_receive, this, _1, _2, &error,
                                                      &bytes_read)));

    do
    {
        strand.get_io_service().run_one();
    }
    while (error == boost::asio::error::would_block ||
            error == boost::asio::error::try_again);

    // Reset the deadline timer so we can leave this socket open as long
    // as we want.
    reset_deadline();

    /*
     * Check for errors after the read.
     */
    if (error)
    {
        if (error.value() == boost::asio::error::operation_aborted
                || error.value() == boost::asio::error::timed_out)
            throw SocketTimeoutException(error.message());
        else
            throw UdpException(error.message());
    }
    else
        data.insert(data.end(), buff.begin(), buff.begin() + bytes_read);

    return bytes_read;
}

void UdpConnection::connect(Mode mode)
{
    if (!socket.is_open())
        socket.open(boost::asio::ip::udp::v4());

    if (mode == SERVER)
    {
        socket.bind(
            udp::endpoint(udp::v4(),
                          lexical_cast<int>(port)), error);
    }
    else if (mode == CLIENT)
    {
        udp::resolver resolver(io_service);
        udp::resolver::query query(udp::v4(), host, port);
        remote_endpoint = *resolver.resolve(query, error);
    }
}

void UdpConnection::check_deadline()
{
    // Check whether the deadline has passed. We compare the deadline against
    // the current time since a new asynchronous operation may have moved the
    // deadline before this actor had a chance to run.
    if (deadline.expires_at() <= deadline_timer::traits_type::now())
    {
        // The deadline has passed. The outstanding asynchronous operation needs
        // to be cancelled so that the blocked receive() function will return.
        //
        // Please note that cancel() has portability issues on some versions of
        // Microsoft Windows, and it may be necessary to use close() instead.
        // Consult the documentation for cancel() for further information.
        socket.cancel();

        // There is no longer an active deadline. The expiry is set to positive
        // infinity so that the actor takes no action until a new deadline is set.
        reset_deadline();
    }

    // Put the actor back to sleep.
    deadline.async_wait(strand.wrap(boost::bind(&UdpConnection::check_deadline, this)));
}

/*
 * This member function is good for diagnostic purposes
 */
const string UdpConnection::str() const
{
    stringstream sstr;
    sstr << "Host: " << host << endl;
    sstr << "Port: " << port << endl;
    sstr << "Read timeout: " << read_timeout_int << endl;
    sstr << "Remote Endpoint Address: " << remote_endpoint.address().to_string()
    << endl;
    sstr << "Remote Endpoint Port: " << remote_endpoint.port() << endl;

    try
    {
        sstr << "Socket Remote Endpoint Address: "
        << socket.remote_endpoint().address().to_string() << endl;
        sstr << "Socket Remote Endpoint Port: "
        << socket.remote_endpoint().port() << endl;
    }
    catch (exception& e) { }

    try
    {
        sstr << "Socket Local Endpoint Address: "
        << socket.local_endpoint().address().to_string() << endl;
        sstr << "Socket Local Endpoint Port: " << socket.local_endpoint().port()
        << endl;
    }
    catch (exception& e) { }

    return sstr.str();
}

}