SynchronousTimeout.hpp

Go to the documentation of this file.

#ifndef SYNCHRONOUS_TIMEOUT_HPP
#define SYNCHRONOUS_TIMEOUT_HPP

#include "CLogger.hpp"

#include <stdexcept>

#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/optional.hpp>
#include <boost/shared_ptr.hpp>

namespace freedm {
namespace broker {
namespace device {

namespace {
CLocalLogger SyncTimeoutLogger(__FILE__);
}

typedef boost::optional<boost::system::error_code> OptionalError;

void SetResult(boost::shared_ptr<OptionalError> status,
        const boost::system::error_code & error);

template <typename ReadStream, typename BufferSequence>
void TimedRead(ReadStream & stream, const BufferSequence & buffer,
        unsigned int duration)
{
    SyncTimeoutLogger.Trace << __PRETTY_FUNCTION__ << std::endl;

    boost::asio::io_service & ios = stream.get_io_service();
    boost::shared_ptr<OptionalError> result(new OptionalError);
    boost::shared_ptr<OptionalError> timeout(new OptionalError);
    boost::asio::deadline_timer timer(ios);

    SyncTimeoutLogger.Info << "Blocking for synchronous read." << std::endl;

    boost::asio::async_read(stream, buffer,
            boost::bind(SetResult, result, boost::asio::placeholders::error));
    timer.expires_from_now(boost::posix_time::milliseconds(duration));
    timer.async_wait(boost::bind(SetResult, timeout,
            boost::asio::placeholders::error));

    while( !(*result) && !(*timeout) )
    {
        ios.poll();
    }

    if( *result && (*result)->value() == boost::system::errc::success )
    {
        SyncTimeoutLogger.Info << "Synchronous read complete." << std::endl;
    }
    else if( *timeout && (*timeout)->value() == boost::system::errc::success )
    {
        throw std::runtime_error("Synchronous Read Timeout");
    }
    else
    {
        throw std::runtime_error("Synchronous Read Failed");
    }
}

template <typename ReadStream, typename Allocator>
void TimedReadUntil(ReadStream & stream,
        boost::asio::basic_streambuf<Allocator> & buffer,
        const std::string & delim,
        unsigned int duration)
{
    SyncTimeoutLogger.Trace << __PRETTY_FUNCTION__ << std::endl;

    boost::asio::io_service & ios = stream.get_io_service();
    boost::shared_ptr<OptionalError> result(new OptionalError);
    boost::shared_ptr<OptionalError> timeout(new OptionalError);
    boost::asio::deadline_timer timer(ios);

    SyncTimeoutLogger.Info << "Blocking for synchronous read." << std::endl;

    boost::asio::async_read_until(stream, buffer, delim,
            boost::bind(SetResult, result, boost::asio::placeholders::error));
    timer.expires_from_now(boost::posix_time::milliseconds(duration));
    timer.async_wait(boost::bind(SetResult, timeout,
            boost::asio::placeholders::error));

    while( !(*result) && !(*timeout) )
    {
        ios.poll();
    }

    if( *result && (*result)->value() == boost::system::errc::success )
    {
        SyncTimeoutLogger.Info << "Synchronous read complete." << std::endl;
    }
    else if( *timeout && (*timeout)->value() == boost::system::errc::success )
    {
        throw std::runtime_error("Synchronous Read Until Timeout");
    }
    else
    {
        throw std::runtime_error("Synchronous Read Until Failed");
    }
}

template <typename WriteStream, typename BufferSequence>
void TimedWrite(WriteStream & stream, const BufferSequence & buffer,
        unsigned int duration)
{
    SyncTimeoutLogger.Trace << __PRETTY_FUNCTION__ << std::endl;

    boost::asio::io_service & ios = stream.get_io_service();
    boost::shared_ptr<OptionalError> result(new OptionalError);
    boost::shared_ptr<OptionalError> timeout(new OptionalError);
    boost::asio::deadline_timer timer(ios);

    SyncTimeoutLogger.Info << "Blocking for synchronous write." << std::endl;

    boost::asio::async_write(stream, buffer,
            boost::bind(SetResult, result, boost::asio::placeholders::error));
    timer.expires_from_now(boost::posix_time::milliseconds(duration));
    timer.async_wait(boost::bind(SetResult, timeout,
            boost::asio::placeholders::error));

    while( !(*result) && !(*timeout) )
    {
        ios.poll();
    }

    if( *result && (*result)->value() == boost::system::errc::success )
    {
        SyncTimeoutLogger.Info << "Synchronous write complete." << std::endl;
    }
    else if( *timeout && (*timeout)->value() == boost::system::errc::success )
    {
        throw std::runtime_error("Synchronous Write Timeout");
    }
    else
    {
        throw std::runtime_error("Synchronous Write Failed");
    }
}

template <typename WriteStream, typename Allocator>
void TimedWrite(WriteStream & stream,
        boost::asio::basic_streambuf<Allocator> & buffer,
        unsigned int duration)
{
    SyncTimeoutLogger.Trace << __PRETTY_FUNCTION__ << std::endl;

    boost::asio::io_service & ios = stream.get_io_service();
    boost::shared_ptr<OptionalError> result(new OptionalError);
    boost::shared_ptr<OptionalError> timeout(new OptionalError);
    boost::asio::deadline_timer timer(ios);

    SyncTimeoutLogger.Info << "Blocking for synchronous write." << std::endl;

    boost::asio::async_write(stream, buffer,
            boost::bind(SetResult, result, boost::asio::placeholders::error));
    timer.expires_from_now(boost::posix_time::milliseconds(duration));
    timer.async_wait(boost::bind(SetResult, timeout,
            boost::asio::placeholders::error));

    while( !(*result) && !(*timeout) )
    {
        ios.poll();
    }

    if( *result && (*result)->value() == boost::system::errc::success )
    {
        SyncTimeoutLogger.Info << "Synchronous write complete." << std::endl;
    }
    else if( *timeout && (*timeout)->value() == boost::system::errc::success )
    {
        throw std::runtime_error("Synchronous Write Timeout");
    }
    else
    {
        throw std::runtime_error("Synchronous Write Failed");
    }
}

} // namespace device
} // namespace broker
} // namespace freedm

#endif // SYNCHRONOUS_TIMEOUT_HPP