chained sending for connections, ping/pong, namespaced

app/main.cpp

@@ -10,6 +10,7 @@
 #include "libf2f/protocol.h"
 
 using namespace std;
+using namespace libf2f;
 
 void iorun( boost::asio::io_service * ios )
 {
@@ -62,6 +63,12 @@ int main(int argc, char **argv)
             boost::asio::ip::tcp::endpoint ep(ipaddr, rp);
             r.connect_to_remote( ep );
         }
+
+        if(parts[0] == "pingall")
+        {
+            message_ptr ping = PingMessage::factory();
+            r.foreach_conns( boost::bind(&Connection::async_write, _1, ping) );
+        }
     }
 
     ios.stop();

include/libf2f/connection.h

@@ -15,6 +15,8 @@
 
 #include "libf2f/message.h"
 
+namespace libf2f {
+
 class Connection;
 typedef boost::shared_ptr<Connection> connection_ptr;
 
@@ -27,7 +29,8 @@ class Connection
 public:
 
     Connection( boost::asio::io_service& io_service, 
-                boost::function< void(message_ptr, connection_ptr) > cb );
+                boost::function< void(message_ptr, connection_ptr) > msg_cb,
+                boost::function< void(connection_ptr) > fin_cb );
 
     ~Connection();
 
@@ -39,9 +42,10 @@ class Connection
     boost::asio::ip::tcp::socket& socket();
 
     /// Asynchronously write a data structure to the socket.
-    /// This just enqueues the data, the router's flow-control algo
-    /// decides when/if to send it.
+    /// This just enqueues the data and returns immediately
     void async_write(message_ptr msg);
+    /// this is called internally to do actual sending:
+    void do_async_write(const boost::system::error_code& e, message_ptr finished_msg);
 
     /// Setup a call to read the next msg_header
     void async_read();
@@ -71,11 +75,14 @@ class Connection
     /// Stateful stuff the protocol handler/servent will set:
     std::string m_username; // username of user at end of Connection
     bool m_authed;
+    bool m_sending;
+
 
-    //Router * m_router;
     boost::function< void(message_ptr, connection_ptr) > m_message_received_cb;
+    boost::function< void(connection_ptr) > m_fin_cb; // call when we die.
 
 };
 
+} //ns
 
 #endif 

include/libf2f/message.h

@@ -20,6 +20,8 @@
 #define SIDCANCEL        8
 #define BYE              9
 
+namespace libf2f {
+
 class Message;
 typedef boost::shared_ptr<Message> message_ptr;
 
@@ -70,23 +72,27 @@ class Message
     {}
 
     Message(const message_header& header)
+        : m_payload(0)
     {
         m_header = header;
         m_guid = std::string(header.guid, 36);
-        std::cout << "CTOR Msg(" << m_guid << ")" << std::endl;
+        //std::cout << "CTOR Msg(" << m_guid << ")" << std::endl;
     }
 
     virtual ~Message()
     {
-        std::cout << "DTOR Msg(" << m_guid << ")" << std::endl;
-        free(m_payload);
+        //std::cout << "DTOR Msg(" << m_guid << ")" << std::endl;
+        if(m_payload) free(m_payload);
     }
 
     virtual const std::string str() const
     {
         std::ostringstream os;
-        os  << "[Msg type:" << type() << " ttl:" << ttl() << " hops:" << hops()
-            << " length:" << length() << " guid:" << guid() << "]";
+        os  << "[Msg type:" << (int)type() 
+            << " ttl:" << (int)ttl() 
+            << " hops:" << (int)hops()
+            << " length:" << (int)length() 
+            << " guid:" << guid() << "]";
         return os.str();
     }
 
@@ -190,5 +196,5 @@ class DataMessage : public Message
     mutable std::string m_sid;
 };
 
-
+}
 #endif

include/libf2f/protocol.h

@@ -4,9 +4,15 @@
 #include "libf2f/router.h"
 #include "libf2f/connection.h"
 
+namespace libf2f {
+
 class Protocol
 {
 public:
+    Protocol();
+
+    virtual void set_router(Router * r) { m_router = r; }
+
     /// called when a client connects to us
     virtual bool new_incoming_connection( connection_ptr conn );
 
@@ -16,6 +22,10 @@ class Protocol
     /// we received a msg from this connection
     virtual void message_received( message_ptr msgp, connection_ptr conn );
 
-
+private:
+    Router * m_router;
 };
+
+} //ns
+
 #endif

include/libf2f/router.h

@@ -11,6 +11,8 @@
 #include "libf2f/message.h"
 #include "libf2f/connection.h"
 
+namespace libf2f {
+
 class Protocol;
 
 
@@ -43,13 +45,18 @@ class Router
     /// Default message recvd callback
     void message_received( message_ptr msgp, connection_ptr conn );
 
-    /// Run in a thread, does the flow-control and sends messages
-    void message_dispatch_runner();
-
+    /// apply function to all registered connections
+    void foreach_conns( boost::function<void(connection_ptr)> );
 
 private:
+    /// Router keeps track of connections:
+    void register_connection( connection_ptr conn );
+    void unregister_connection( connection_ptr conn );
+    void connections_str();
+
     /// all connections:
     std::vector< connection_ptr > m_connections;
+    boost::mutex m_connections_mutex; // protects connections
 
     /// The acceptor object used to accept incoming socket connections.
     boost::asio::ip::tcp::acceptor & m_acceptor;
@@ -62,6 +69,6 @@ class Router
     unsigned int seen_connections; // num incoming connections accepted
 };
 
-
+} //ns
 
 #endif

src/connection.cpp

@@ -1,10 +1,16 @@
 #include "libf2f/connection.h"
 #include <boost/foreach.hpp>
 
+namespace libf2f {
+
+using namespace std;
+
 Connection::Connection( boost::asio::io_service& io_service, 
-            boost::function< void(message_ptr, connection_ptr) > cb )
-    : m_socket(io_service), m_authed(false),
-        m_message_received_cb(cb)
+            boost::function< void(message_ptr, connection_ptr) > msg_cb,
+            boost::function< void(connection_ptr) > fin_cb )
+    : m_socket(io_service), m_authed(false), m_sending(false),
+      m_message_received_cb(msg_cb),
+      m_fin_cb(fin_cb)
 {
     std::cout << "CTOR connection" << std::endl;
     max_writeq_size = 20*1024; // 20kb
@@ -25,6 +31,7 @@ void
 Connection::fin()
 {
     std::cout << "FIN connection " << str() << std::endl;
+    m_fin_cb( shared_from_this() );
     close();
 }
 
@@ -43,8 +50,13 @@ Connection::async_write(message_ptr msg)
         m_writeq.push_back(msg);
         m_writeq_size += sizeof(message_header) + msg->length();
     }
+    // make sure our sending loop is running:
+    boost::system::error_code e;
+    do_async_write( e, message_ptr() );
 }
 
+/// Reading incoming messages is a chain of 3 async methods:
+/// async_read() -> handle_read_headers() -> handle_read_data()
 void 
 Connection::async_read()
 {
@@ -59,7 +71,7 @@ Connection::async_read()
                                         msgp
                                         ));
 }
-
+/// called when we've read the header off the wire
 void 
 Connection::handle_read_header(const boost::system::error_code& e, message_ptr msgp)
 {
@@ -91,7 +103,7 @@ Connection::handle_read_header(const boost::system::error_code& e, message_ptr m
                                         ));
 
 }
-
+/// called when we've read header and payload off the wire
 void 
 Connection::handle_read_data(const boost::system::error_code& e, message_ptr msgp)
 {
@@ -112,7 +124,7 @@ Connection::set_message_received_cb( boost::function< void(message_ptr, connecti
 {
     m_message_received_cb = cb;
 }
-
+/// unused atm.. todo flow control?
 size_t
 Connection::drain_writeq( std::deque< message_ptr > & out )
 {
@@ -137,3 +149,39 @@ Connection::str() const
     return os.str();
 }
 
+/// Calls to do_async_write are chained - it will call itself when a write
+/// completes, to send the next msg in the queue. Bails when none left.
+void 
+Connection::do_async_write(const boost::system::error_code& e, message_ptr finished_msg)
+{
+    message_ptr msgp;
+    { // mutex scope
+        boost::mutex::scoped_lock lk(m_mutex);
+        if(m_sending && !finished_msg)
+        {
+            // this call is telling us to send, but we already are.
+            //cout << "bailing from do_async_write - already sending (sending=true)" << endl;
+            return;
+        }
+
+        if(m_writeq.empty())
+        {
+            //cout << "bailing from do_async_write, q empty (sending=false)" << endl;
+            m_sending = false;
+            return;
+        }
+
+        msgp = m_writeq.front();
+        m_writeq.pop_front();
+        m_writeq_size -= (sizeof(message_header) + msgp->length());
+        m_sending = true;
+    } // mutex scope
+
+    boost::asio::async_write( socket(), msgp->to_buffers(),
+                              boost::bind( &Connection::do_async_write, this,
+                                           boost::asio::placeholders::error,
+                                           msgp ) );
+}
+
+} //ns
+

src/protocol.cpp

@@ -1,7 +1,14 @@
 #include "libf2f/protocol.h"
 
+namespace libf2f {
+
 using namespace std;
 
+Protocol::Protocol()
+{
+    cout << "CTOR Protocol" << endl;
+}
+
 bool
 Protocol::new_incoming_connection( connection_ptr conn )
 {
@@ -25,4 +32,6 @@ Protocol::message_received( message_ptr msgp, connection_ptr conn )
 {
     cout << "Protocol::message_received " << conn->str() << endl 
          << msgp->str() << endl;
-}
+}
+
+} //ns