face/tcp: Implementing accepting and establishing connections with TCP Channel

refs: #1132, #1133, #1134, #1135

Change-Id: I8878784059fd962187bfcdd23b5adcaf9135cf4e

daemon/face/channel-factory.hpp

@@ -30,7 +30,7 @@ public:
   };
 
 protected:
-  typedef std::map<Endpoint, Channel> ChannelMap;
+  typedef std::map< Endpoint, shared_ptr<Channel> > ChannelMap;
   ChannelMap m_channels;  
 };
 

daemon/face/tcp-channel-factory.cpp

@@ -8,22 +8,49 @@
 
 namespace ndn {
 
+TcpChannelFactory::TcpChannelFactory(boost::asio::io_service& ioService)
+  : m_ioService(ioService)
+{
+}
+
 shared_ptr<TcpChannel>
 TcpChannelFactory::create(const tcp::Endpoint& endpoint)
 {
-  return shared_ptr<ndn::TcpChannel>();
+  shared_ptr<TcpChannel> channel = find(endpoint);
+  if(static_cast<bool>(channel))
+    return channel;
+
+  channel = make_shared<TcpChannel>(boost::ref(m_ioService), boost::cref(endpoint));
+  m_channels[endpoint] = channel;
+  return channel;
 }
 
 shared_ptr<TcpChannel>
 TcpChannelFactory::create(const std::string& localHost, const std::string& localPort)
 {
-  return shared_ptr<ndn::TcpChannel>();
+  using boost::asio::ip::tcp;
+  
+  tcp::resolver::query query(localHost, localPort);
+  // shared_ptr<tcp::resolver> resolver =
+  //   make_shared<tcp::resolver>(boost::ref(m_ioService));
+  tcp::resolver resolver(m_ioService);
+
+  tcp::resolver::iterator end;
+  tcp::resolver::iterator i = resolver.resolve(query);
+  if (i == end)
+    return shared_ptr<ndn::TcpChannel>();
+
+  return create(*i);
 }
 
 shared_ptr<TcpChannel>
 TcpChannelFactory::find(const tcp::Endpoint& localEndpoint)
 {
-  return shared_ptr<ndn::TcpChannel>();
+  ChannelMap::iterator i = m_channels.find(localEndpoint);
+  if (i != m_channels.end())
+    return i->second;
+  else
+    return shared_ptr<ndn::TcpChannel>();
 }
 
 } // namespace ndn

daemon/face/tcp-channel-factory.hpp

@@ -23,6 +23,8 @@ public:
     Error(const std::string& what) : ChannelFactory::Error(what) {}
   };
 
+  TcpChannelFactory(boost::asio::io_service& ioService);
+
   /**
    * \brief Create TCP-based channel using tcp::Endpoint
    *
@@ -66,6 +68,9 @@ public:
    */
   shared_ptr<TcpChannel>
   find(const tcp::Endpoint& localEndpoint);
+  
+private:
+  boost::asio::io_service& m_ioService;
 };
 
 } // namespace ndn

daemon/face/tcp-channel.cpp

@@ -5,22 +5,166 @@
  */
 
 #include "tcp-channel.hpp"
+#include "tcp-face.hpp"
 
 namespace ndn {
 
-TcpChannel::TcpChannel(boost::asio::io_service& ioService,
-                       const tcp::Endpoint& endpoint)
+using namespace boost::asio;
+
+TcpChannel::TcpChannel(io_service& ioService,
+                       const tcp::Endpoint& localEndpoint)
+  : m_ioService(ioService)
+  , m_localEndpoint(localEndpoint)
 {
 }
 
 void
-TcpChannel::listen(const tcp::Endpoint& endpoint)
+TcpChannel::listen(int backlog/* = tcp::acceptor::max_connections*/)
 {
+  m_acceptor = make_shared<ip::tcp::acceptor>(boost::ref(m_ioService));
+  m_acceptor->open(m_localEndpoint.protocol());
+  m_acceptor->set_option(ip::tcp::acceptor::reuse_address(true));
+  m_acceptor->bind(m_localEndpoint);
+  m_acceptor->listen(backlog);
+
+  shared_ptr<ip::tcp::socket> clientSocket =
+    make_shared<ip::tcp::socket>(boost::ref(m_ioService));
+  m_acceptor->async_accept(*clientSocket,
+                           bind(&TcpChannel::handleConnection, this,
+                                _1, clientSocket));
 }
 
 void
-TcpChannel::connect(const tcp::Endpoint& endpoint)
+TcpChannel::connect(const tcp::Endpoint& remoteEndpoint,
+                    const time::Duration& timeout/* = time::seconds(4)*/)
 {
+  shared_ptr<ip::tcp::socket> clientSocket =
+    make_shared<ip::tcp::socket>(boost::ref(m_ioService));
+
+  shared_ptr<monotonic_deadline_timer> connectTimeoutTimer =
+    make_shared<monotonic_deadline_timer>(boost::ref(m_ioService));
+
+  // not sure if it works. This will bind to something...
+  // Do we need reuse here too?
+  clientSocket->bind(m_localEndpoint);
+
+  clientSocket->async_connect(remoteEndpoint,
+                              bind(&TcpChannel::handleSuccessfulConnect, this,
+                                   _1, clientSocket, connectTimeoutTimer));
+
+  connectTimeoutTimer->expires_from_now(timeout);
+  connectTimeoutTimer->async_wait(bind(&TcpChannel::handleFailedConnect, this,
+                                       _1, clientSocket, connectTimeoutTimer));
 }
 
+void
+TcpChannel::connect(const std::string& remoteHost, const std::string& remotePort,
+                    const time::Duration& timeout/* = time::seconds(4)*/)
+{
+  shared_ptr<ip::tcp::socket> clientSocket =
+    make_shared<ip::tcp::socket>(boost::ref(m_ioService));
+
+  shared_ptr<monotonic_deadline_timer> connectTimeoutTimer =
+    make_shared<monotonic_deadline_timer>(boost::ref(m_ioService));
+
+  // not sure if it works. This will bind to something...
+  // Do we need reuse here too?
+  clientSocket->bind(m_localEndpoint);
+
+  ip::tcp::resolver::query query(remoteHost, remotePort);
+  shared_ptr<ip::tcp::resolver> resolver =
+    make_shared<ip::tcp::resolver>(boost::ref(m_ioService));
+
+  resolver->async_resolve(query,
+                          bind(&TcpChannel::handleEndpointResoution, this, _1, _2,
+                               clientSocket, connectTimeoutTimer));
+
+  connectTimeoutTimer->expires_from_now(timeout);
+  connectTimeoutTimer->async_wait(bind(&TcpChannel::handleFailedConnect, this,
+                                       _1, clientSocket, connectTimeoutTimer));
+}
+
+
+void
+TcpChannel::handleConnection(const boost::system::error_code& error,
+                             const shared_ptr<ip::tcp::socket>& socket)
+{
+  if (error) {
+    if (error == boost::system::errc::operation_canceled) // when socket is closed by someone
+      return;
+
+    /// \todo Log the error
+    onConnectFailed("Connect to remote endpoint failed: " +
+                    error.category().message(error.value()));
+    return;
+  }
+
+  /**
+   * \todo Either remove FaceId from here or set it here to some real value
+   */
+  shared_ptr<TcpFace> face = make_shared<TcpFace>(1, boost::cref(socket));
+
+  // what's next?
+}
+
+void
+TcpChannel::handleSuccessfulConnect(const boost::system::error_code& error,
+                                    const shared_ptr<ip::tcp::socket>& socket,
+                                    const shared_ptr<monotonic_deadline_timer>& timer)
+{
+  timer->cancel();
+
+  if (error) {
+    if (error == boost::system::errc::operation_canceled) // when socket is closed by someone
+      return;
+
+    socket->close();
+    onConnectFailed("Connect to remote endpoint failed: " +
+                    error.category().message(error.value()));
+    return;
+  }
+
+  handleConnection(error, socket);
+}
+
+void
+TcpChannel::handleFailedConnect(const boost::system::error_code& error,
+                                const shared_ptr<ip::tcp::socket>& socket,
+                                const shared_ptr<monotonic_deadline_timer>& timer)
+{
+  if (error) { // e.g., cancelled
+    return;
+  }
+
+  onConnectFailed("Connect to remote endpoint timed out: " +
+                  error.category().message(error.value()));
+  socket->close(); // abort the connection
+}
+
+void
+TcpChannel::handleEndpointResoution(const boost::system::error_code& error,
+                                    ip::tcp::resolver::iterator remoteEndpoint,
+                                    const shared_ptr<boost::asio::ip::tcp::socket>& socket,
+                                    const shared_ptr<boost::asio::monotonic_deadline_timer>& timer)
+{
+  if (error ||
+      remoteEndpoint == ip::tcp::resolver::iterator())
+    {
+      if (error == boost::system::errc::operation_canceled) // when socket is closed by someone
+        return;
+
+      socket->close();
+      timer->cancel();
+      onConnectFailed("Remote endpoint hostname or port cannot be resolved: " +
+                      error.category().message(error.value()));
+      return;
+    }
+
+  // got endpoint, now trying to connect (only try the first resolution option)
+  socket->async_connect(*remoteEndpoint,
+                        bind(&TcpChannel::handleSuccessfulConnect, this,
+                             _1, socket, timer));
+}
+
+
 } // namespace ndn

daemon/face/tcp-channel.hpp

@@ -7,10 +7,8 @@
 #ifndef NFD_FACE_TCP_CHANNEL_HPP
 #define NFD_FACE_TCP_CHANNEL_HPP
 
-// #include "session-based-channel.hpp"
-
 #include "common.hpp"
-// #include <boost/asio/ip/tcp.hpp>
+#include "core/monotonic_deadline_timer.hpp"
 
 namespace tcp {
 typedef boost::asio::ip::tcp::endpoint Endpoint;
@@ -18,19 +16,96 @@ typedef boost::asio::ip::tcp::endpoint Endpoint;
 
 namespace ndn {
 
+/**
+ * \brief Class implementing TCP-based channel to create faces
+ *
+ * Channel can create faces as a response to incoming TCP
+ * connections (TcpChannel::listen needs to be called for that
+ * to work) or explicitly after using TcpChannel::connect method.
+ */
 class TcpChannel // : protected SessionBasedChannel
 {
 public:
-  TcpChannel(boost::asio::io_service& ioService, const tcp::Endpoint& endpoint);
+  /**
+   * \brief Prototype for the callback called when face is created
+   *        (as a response to incoming connection or after connection
+   *        is established)
+   */
+  typedef function<void(const shared_ptr<TcpFace>& newFace)> FaceCreatedCallback;
 
+  /**
+   * \brief Prototype for the callback that is called when face is failed to
+   *        get created
+   */
+  typedef function<void(const std::string& reason)> ConnectFailedCallback;
+  
+  /**
+   * \brief Create TCP channel for the local endpoint
+   *
+   * To enable creation faces upon incoming connections,
+   * one needs to explicitly call TcpChannel::listen method.
+   */
+  TcpChannel(boost::asio::io_service& ioService,
+             const tcp::Endpoint& localEndpoint);
+
+  /**
+   * \brief Enable listening on the local endpoint, accept connections,
+   *        and create faces when remote host makes a connection
+   * \param backlog The maximum length of the queue of pending incoming
+   *        connections
+   */
   void
-  listen(const tcp::Endpoint& endpoint);
+  listen(int backlog = boost::asio::ip::tcp::acceptor::max_connections);
 
+  /**
+   * \brief Create a face by establishing connection to remote endpoint
+   */
   void
-  connect(const tcp::Endpoint& endpoint);
+  connect(const tcp::Endpoint& remoteEndpoint,
+          const time::Duration& timeout = time::seconds(4));
 
+  /**
+   * \brief Create a face by establishing connection to the specified
+   *        remote host and remote port
+   *
+   * This method will never blocks and will return immediately. All
+   * necessary hostname and port resolution and connection will happen
+   * in asynchronous mode.
+   *
+   * If connection cannot be established within specified timeout, it
+   * will be aborted.
+   */
+  void
+  connect(const std::string& remoteHost, const std::string& remotePort,
+          const time::Duration& timeout = time::seconds(4));
+  
 private:
+  void
+  handleConnection(const boost::system::error_code& error,
+                   const shared_ptr<boost::asio::ip::tcp::socket>& socket);
+
+  void
+  handleSuccessfulConnect(const boost::system::error_code& error,
+                          const shared_ptr<boost::asio::ip::tcp::socket>& socket,
+                          const shared_ptr<boost::asio::monotonic_deadline_timer>& timer);
+  
+  void
+  handleFailedConnect(const boost::system::error_code& error,
+                      const shared_ptr<boost::asio::ip::tcp::socket>& socket,
+                      const shared_ptr<boost::asio::monotonic_deadline_timer>& timer);
+
+  void
+  handleEndpointResoution(const boost::system::error_code& error,
+                          boost::asio::ip::tcp::resolver::iterator remoteEndpoint,
+                          const shared_ptr<boost::asio::ip::tcp::socket>& socket,
+                          const shared_ptr<boost::asio::monotonic_deadline_timer>& timer);
+  
+private:
+  boost::asio::io_service& m_ioService;
   tcp::Endpoint m_localEndpoint;
+
+  bool isListening;
+  shared_ptr<boost::asio::ip::tcp::acceptor> m_acceptor;
 };
 
 } // namespace ndn

daemon/face/tcp-face.cpp

@@ -14,4 +14,14 @@ TcpFace::TcpFace(FaceId id, const shared_ptr<TcpFace::protocol::socket>& socket)
 {
 }
 
+void
+TcpFace::sendInterest(const Interest& interest)
+{
+}
+
+void
+TcpFace::sendData(const Data& data)
+{
+}
+
 } // namespace ndn

daemon/face/tcp-face.hpp

@@ -21,6 +21,13 @@ class TcpFace : public StreamFace<boost::asio::ip::tcp>
 public:
   TcpFace(FaceId id, const shared_ptr<protocol::socket>& socket);
 
+  // from Face
+  virtual void
+  sendInterest(const Interest& interest);
+
+  virtual void
+  sendData(const Data& data);
+  
 private:
   shared_ptr<protocol::socket> m_socket;
 };