source: trunk/FACT++/src/Connection.cc@ 10278

// **************************************************************************
/** @class Connection

@brief Maintains an ansynchronous TCP/IP client connection

*/
// **************************************************************************
#include "Connection.h"

#include <boost/bind.hpp>
#include <boost/lexical_cast.hpp>

using namespace std;

namespace ba    = boost::asio;
namespace bs    = boost::system;
namespace dummy = ba::placeholders;

using boost::lexical_cast;
using ba::ip::tcp;


    // -------- Abbreviations for starting async tasks ---------

int Connection::Write(const Time &t, const char *txt, int qos)
{
    if (fLog)
        return fLog->Write(t, txt, qos);

    return MessageImp::Write(t, txt, qos);
}

void Connection::AsyncRead(ba::mutable_buffers_1 buffers, int type)
{
    ba::async_read(*this, buffers,
                   boost::bind(&Connection::HandleReceivedData, this,
                               dummy::error, dummy::bytes_transferred, type));
}

void Connection::AsyncWrite(ba::mutable_buffers_1 buffers)
{
    ba::async_write(*this, buffers,
                    boost::bind(&Connection::HandleSentData, this,
                                dummy::error, dummy::bytes_transferred));
}

void Connection::AsyncWait(ba::deadline_timer &timer, int millisec,
                           void (Connection::*handler)(const bs::error_code&))
{
    // - The boost::asio::basic_deadline_timer::expires_from_now()
    //   function cancels any pending asynchronous waits, and returns
    //   the number of asynchronous waits that were cancelled. If it
    //   returns 0 then you were too late and the wait handler has
    //   already been executed, or will soon be executed. If it
    //   returns 1 then the wait handler was successfully cancelled.
    // - If a wait handler is cancelled, the bs::error_code passed to
    //   it contains the value bs::error::operation_aborted.
    timer.expires_from_now(boost::posix_time::milliseconds(millisec));

    timer.async_wait(boost::bind(handler, this, dummy::error));
}

void Connection::AsyncConnect(tcp::resolver::iterator iterator)
{
    //cout << "Start async connect(...)" << endl;
    fConnectionStatus = 1;

    tcp::endpoint endpoint = *iterator;

    // AsyncConnect + Deadline
    async_connect(endpoint,
                  boost::bind(&Connection::ConnectImp,
                              this, ba::placeholders::error,
                              ++iterator));

    // We will get a "Connection timeout anyway"
    //AsyncWait(fConnectTimeout, 5, &Connection::HandleConnectTimeout);
}

// ------------------------ close --------------------------
// close from another thread
void Connection::CloseImp(bool restart)
{
    if (IsConnected())
    {
        stringstream str;
        str << "Connection closed to " << URL() << ".";// << endl;
        Message(str);
    }

    // Stop any pending connection attempt
    fConnectionTimer.cancel();

    // Close possible open connections
    close();

    // Reset the connection status
    fConnectionStatus = 0;

    // Stop deadline counters
    fInTimeout.cancel();
    fOutTimeout.cancel();

    if (!restart || IsConnecting())
        return;

    // We need some timeout before reconnecting!
    // And we have to check if we are alreayd trying to connect
    // We shoudl wait until all operations in progress were canceled

    // Start trying to reconnect
    fMsgConnect = "";
    StartConnect();
}

void Connection::PostClose(bool restart)
{
    get_io_service().post(boost::bind(&Connection::CloseImp, this, restart));
}

// ------------------------ write --------------------------
void Connection::HandleWriteTimeout(const bs::error_code &error)
{
    // 125: Operation canceled
    if (error && error!=bs::error_code(125, bs::system_category))
    {
        stringstream str;
        str << "Write timeout of " << URL() << ": " << error.message() << " (" << error << ")";// << endl;
        Error(str);
    }

    if (!is_open())
    {
        // For example: Here we could schedule a new accept if we
        // would not want to allow two connections at the same time.
        return;
    }

    // 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 (fOutTimeout.expires_at() > ba::deadline_timer::traits_type::now())
        return;

    Error("fOutTimeout has expired, writing data to "+URL());

    CloseImp();
}

void Connection::HandleSentData(const bs::error_code& error, size_t)
{
    if (error)
    {
        stringstream str;
        str << "Writing to " << URL() << ": " << error.message() << " (" << error << ")";// << endl;
        Error(str);

        CloseImp();
        return;
    }

    Message("Data successfully sent to "+URL());

    // This is "thread" safe because SendMessage and HandleSentMessage
    // are serialized in the EventQueue. Note: Do not call these
    // functions directly from any other place then Handlers, use
    // PostMessage instead
    fOutQueue.pop_front();

    if (fOutQueue.empty())
    {
        // Queue went empty, remove deadline
        fOutTimeout.cancel();
        return;
    }

    // AsyncWrite + Deadline
    AsyncWrite(ba::buffer(fOutQueue.front())/*, &Connection::HandleSentData*/);
    AsyncWait(fOutTimeout, 5000, &Connection::HandleWriteTimeout);
}

void Connection::SendMessageImp(const vector<char> &msg)
{
    /*
    if (!fConnectionEstablished)
    {
        UpdateWarn("SendMessageImp, but no connection to "+fAddress+":"+fPort+".");
        return;
    }*/

    const bool first_message_in_queue = fOutQueue.empty();

    // This is "thread" safe because SendMessage and HandleSentMessage
    // are serialized in the EventQueue. Note: Do not call these
    // functions directly from any other place then Handlers, use
    // PostMessage instead
    fOutQueue.push_back(msg);

    if (!first_message_in_queue)
        return;

    // AsyncWrite + Deadline
    AsyncWrite(ba::buffer(fOutQueue.front())/*, &Connection::HandleSentData*/);
    AsyncWait(fOutTimeout, 5000, &Connection::HandleWriteTimeout);
}

void Connection::PostMessage(vector<uint16_t> inp)
{
    // Convert to network byte order
    for_each(inp.begin(), inp.end(), htons);

    // FIXME FIXME

    PostMessage((char*)&inp.front(), sizeof(uint16_t)*inp.size());

//    const vector<char> msg((char*)&inp.front(), (char*)&inp.last()+1);
//    get_io_service().post(boost::bind(&Connection::SendMessageImp, this, msg));
}

void Connection::PostMessage(const vector<char> &msg)
{
    get_io_service().post(boost::bind(&Connection::SendMessageImp, this, msg));
}

void Connection::PostMessage(const void *ptr, size_t max)
{
    const vector<char> msg(reinterpret_cast<const char*>(ptr),
                           reinterpret_cast<const char*>(ptr)+max);

    get_io_service().post(boost::bind(&Connection::SendMessageImp, this, msg));
}

void Connection::PostMessage(const string &cmd, size_t max)
{
    if (max==size_t(-1))
        max = cmd.length()+1;

    vector <char>msg(max);

    for (unsigned int i=0; i<max; i++)
        msg[i] = 0;

    for (unsigned int i=0; i<min(cmd.length()+1, max); i++)
        msg[i] = cmd[i];

    get_io_service().post(boost::bind(&Connection::SendMessageImp, this, msg));
}

void Connection::HandleConnectionTimer(const bs::error_code &error)
{
    // 125: Operation canceled
    if (error && error!=bs::error_code(125, bs::system_category))
    {
        stringstream str;
        str << "Connetion timer of " << URL() << ": " << error.message() << " (" << error << ")";// << endl;
        Error(str);
    }

    if (is_open())
    {
        // For example: Here we could schedule a new accept if we
        // would not want to allow two connections at the same time.
        return;
    }

    // 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 (fConnectionTimer.expires_at() < ba::deadline_timer::traits_type::now())
        StartConnect();
}

void Connection::ConnectImp(const bs::error_code& error,
                            tcp::resolver::iterator endpoint_iterator)
{
    // Connection established
    if (!error)
    {
        Message("Connection established to "+URL()+"...");

        // Initialize hardware...
        // Set state to : connected/undefined until we get the first message
        //                (send request for configuration) ?
        //                (send configuration?)
        fMsgConnect = "";
        fErrConnect = "";

        fConnectionStatus = 2;
        //StartAsyncRead();

        // We can create the buffer here and delete it in
        // Handle Read. However, the read buffer is not filled again
        // before it was not read within HandleReceivedData -- so a single
        // buffer should be good enough. Before HandleReceivedData
        // returns the data must be copied to a "safe" place.

        // fSocket.get_io_service()/*fEventQueue*/.stop();

        ConnectionEstablished();
        return;
    }

    // If returning from run will lead to deletion of this
    // instance, close() is not needed (maybe implicitly called).
    // If run is called again, close() is needed here. Otherwise:
    // Software caused connection abort when we try to resolve
    // the endpoint again.
    CloseImp(false);
    //fSocket.close();

    // 111: Connection refused
    if (1/*error!=bs::error_code(111, bs::system_category)*/)
    {
        stringstream msg;
        if (URL()!=":")
            msg << "Connecting to " << URL() << ": " << error.message() << " (" << error << ")";

        if (fErrConnect!=msg.str())
        {
            fMsgConnect = "";
            fErrConnect = msg.str();
            Warn(fErrConnect);
        }
    }

    // Go on with the next
    if (endpoint_iterator != tcp::resolver::iterator())
    {
        AsyncConnect(endpoint_iterator);
        return;
    }

    // No more entries to try, if we would not put anything else
    // into the queue anymore it would now return (run() would return)

    // Since we don't want to block the main loop, we wait using an
    // asnychronous timer

    // FIXME: Should we move this before AsyncConnect() ?
    AsyncWait(fConnectionTimer, 250, &Connection::HandleConnectionTimer);
}

// FIXME: Async connect should get address and port as an argument
void Connection::StartConnect()
{
    fConnectionStatus = 1;

    tcp::resolver resolver(get_io_service());

    boost::system::error_code ec;

    tcp::resolver::query query(fAddress, fPort);
    tcp::resolver::iterator iterator = resolver.resolve(query, ec);

    stringstream msg;
    if (!fAddress.empty() || !fPort.empty() || ec)
        msg << "Trying to connect to " << URL() << "...";

    if (ec)
        msg << " " << ec.message() << " (" << ec << ")";

    // Only output message if it has changed
    if (fMsgConnect!=msg.str())
    {
        fMsgConnect = msg.str();
        ec ? Error(msg) : Message(msg);
    }

    if (ec)
        AsyncWait(fConnectionTimer, 250, &Connection::HandleConnectionTimer);
    else
        // Start connection attempts (will also reset deadline counter)
        AsyncConnect(iterator);
}

void Connection::SetEndpoint(const string &addr, int port)
{
    if (fConnectionStatus>=1)
        Warn("Connection or connection attempt in progress. New endpoint only valid for next connection.");

    fAddress = addr;
    fPort    = lexical_cast<string>(port);
}

void Connection::SetEndpoint(const string &addr, const string &port)
{
    if (fConnectionStatus>=1 && URL()!=":")
        Warn("Connection or connection attempt in progress. New endpoint only valid for next connection.");

    fAddress = addr;
    fPort    = port;
}

void Connection::SetEndpoint(const string &addr)
{
    const size_t p0 = addr.find_first_of(':');
    const size_t p1 = addr.find_last_of(':');

    if (p0==string::npos || p0!=p1)
    {
        Error("Connection::SetEndPoint - Wrong format of argument.");
        return;
    }

    SetEndpoint(addr.substr(0, p0), addr.substr(p0+1));
}



Connection::Connection(ba::io_service& ioservice, ostream &out) :
MessageImp(out), tcp::socket(ioservice),
fLog(0), //fAddress("localhost"), fPort("5000"),
fInTimeout(ioservice), fOutTimeout(ioservice), fConnectionTimer(ioservice),
fConnectionStatus(0)
{
}
/*
Connection::Connection(ba::io_service& ioservice, const string &addr, int port) :
tcp::socket(ioservice),
fLog(0), fAddress(addr), fPort(lexical_cast<string>(port)),
fInTimeout(ioservice), fOutTimeout(ioservice), fConnectionTimer(ioservice),
fConnectionStatus(0)
{
}

Connection::Connection(ba::io_service& ioservice, const string &addr, const string &port) :
tcp::socket(ioservice), fLog(0), fAddress(addr), fPort(port),
fInTimeout(ioservice), fOutTimeout(ioservice), fConnectionTimer(ioservice),
fConnectionStatus(0)
{
}
*/