Index: trunk/FACT++/src/Connection.cc =================================================================== --- trunk/FACT++/src/Connection.cc (revision 10267) +++ trunk/FACT++/src/Connection.cc (revision 10268) @@ -31,9 +31,9 @@ } -void Connection::AsyncRead(ba::mutable_buffers_1 buffers) +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)); + dummy::error, dummy::bytes_transferred, type)); } @@ -89,12 +89,16 @@ } - // Close the connection + // Stop any pending connection attempt + fConnectionTimer.cancel(); + + // Close possible open connections + close(); + + // Reset the connection status fConnectionStatus = 0; - close(); // Stop deadline counters fInTimeout.cancel(); fOutTimeout.cancel(); - //fConnectTimeout.cancel(); if (!restart || IsConnecting()) @@ -106,5 +110,6 @@ // Start trying to reconnect - AsyncConnect(); + fMsgConnect = ""; + StartConnect(); } @@ -201,6 +206,27 @@ } +void Connection::PostMessage(vector 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 msg((char*)&inp.front(), (char*)&inp.last()+1); +// get_io_service().post(boost::bind(&Connection::SendMessageImp, this, msg)); +} + void Connection::PostMessage(const vector &msg) { + get_io_service().post(boost::bind(&Connection::SendMessageImp, this, msg)); +} + +void Connection::PostMessage(const void *ptr, size_t max) +{ + const vector msg(reinterpret_cast(ptr), + reinterpret_cast(ptr)+max); + get_io_service().post(boost::bind(&Connection::SendMessageImp, this, msg)); } @@ -244,5 +270,5 @@ // to run. if (fConnectionTimer.expires_at() < ba::deadline_timer::traits_type::now()) - AsyncConnect(); + StartConnect(); } @@ -272,4 +298,6 @@ // fSocket.get_io_service()/*fEventQueue*/.stop(); + + ConnectionEstablished(); return; } @@ -287,5 +315,6 @@ { stringstream msg; - msg << "Connecting to " << URL() << ": " << error.message() << " (" << error << ")"; + if (URL()!=":") + msg << "Connecting to " << URL() << ": " << error.message() << " (" << error << ")"; if (fErrConnect!=msg.str()) @@ -315,5 +344,5 @@ // FIXME: Async connect should get address and port as an argument -void Connection::AsyncConnect() +void Connection::StartConnect() { fConnectionStatus = 1; @@ -321,21 +350,31 @@ tcp::resolver resolver(get_io_service()); + boost::system::error_code ec; + tcp::resolver::query query(fAddress, fPort); - tcp::resolver::iterator iterator = resolver.resolve(query); + tcp::resolver::iterator iterator = resolver.resolve(query, ec); stringstream msg; - msg << "Trying to connect to " << fAddress << ":" << fPort << "..."; - + 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(); - Message(msg); - } - - // Start connection attempts (will also reset deadline counter) - AsyncConnect(iterator); -} - -void Connection::SetEndpoint(const char *addr, int port) + 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) @@ -346,13 +385,37 @@ } +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"), +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), @@ -369,3 +432,3 @@ { } - +*/ Index: trunk/FACT++/src/Connection.h =================================================================== --- trunk/FACT++/src/Connection.h (revision 10267) +++ trunk/FACT++/src/Connection.h (revision 10268) @@ -5,4 +5,5 @@ #include #include +#include #include @@ -40,8 +41,6 @@ int Write(const Time &t, const char *txt, int qos=kInfo); - void AsyncRead(boost::asio::mutable_buffers_1 buffers/*, - void (Connection::*handler)(const boost::system::error_code&, size_t)*/); - void AsyncWrite(boost::asio::mutable_buffers_1 buffers/*, - void (Connection::*handler)(const boost::system::error_code&)*/); + void AsyncRead(boost::asio::mutable_buffers_1 buffers, int type=0); + void AsyncWrite(boost::asio::mutable_buffers_1 buffers); void AsyncWait(boost::asio::deadline_timer &timer, int millisec, void (Connection::*handler)(const boost::system::error_code&)); @@ -50,9 +49,13 @@ std::string URL() const { return fAddress + ":" + fPort; } + void CloseImp(bool restart=true); + + void ConnectImp(const boost::system::error_code& error, + boost::asio::ip::tcp::resolver::iterator endpoint_iterator); + public: // ------------------------ close -------------------------- // close from another thread - void CloseImp(bool restart=true); void PostClose(bool restart=true); @@ -62,23 +65,34 @@ void HandleSentData(const boost::system::error_code& error, size_t); void SendMessageImp(const std::vector &msg); + void PostMessage(std::vector msg); void PostMessage(const std::vector &msg); + void PostMessage(const void *msg, size_t s=0); void PostMessage(const std::string &cmd, size_t s=-1); + + template + void PostMessage(boost::array arr) + { + // Convert to network byte order + for_each(arr.begin(), arr.end(), htons); + PostMessage(std::vector(arr.begin(), arr.end())); + } // ------------------------ connect -------------------------- - virtual void ConnectImp(const boost::system::error_code& error, - boost::asio::ip::tcp::resolver::iterator endpoint_iterator); + virtual void ConnectionEstablished() { } - void AsyncConnect(); + void StartConnect(); Connection(boost::asio::io_service& io_service, std::ostream &out); - Connection(boost::asio::io_service& io_service, const std::string &addr="localhost", int port=5000); - Connection(boost::asio::io_service& io_service, const std::string &addr, const std::string &port); +// Connection(boost::asio::io_service& io_service, const std::string &addr="localhost", int port=5000); +// Connection(boost::asio::io_service& io_service, const std::string &addr, const std::string &port); - void SetEndpoint(const char *addr, int port); + void SetEndpoint(const std::string &addr, int port); + void SetEndpoint(const std::string &addr, const std::string &port); + void SetEndpoint(const std::string &addr); // ------------------------ others -------------------------- - virtual void HandleReceivedData(const boost::system::error_code&, size_t) { } + virtual void HandleReceivedData(const boost::system::error_code&, size_t, int = 0) { } virtual void HandleReadTimeout(const boost::system::error_code&) { }