face/tcp: Implementing accepting and establishing connections with TCP Channel
refs: #1132, #1133, #1134, #1135
Change-Id: I8878784059fd962187bfcdd23b5adcaf9135cf4e
diff --git a/daemon/face/channel-factory.hpp b/daemon/face/channel-factory.hpp
index 7f4a5d4..a2d7ff9 100644
--- a/daemon/face/channel-factory.hpp
+++ b/daemon/face/channel-factory.hpp
@@ -30,7 +30,7 @@
};
protected:
- typedef std::map<Endpoint, Channel> ChannelMap;
+ typedef std::map< Endpoint, shared_ptr<Channel> > ChannelMap;
ChannelMap m_channels;
};
diff --git a/daemon/face/tcp-channel-factory.cpp b/daemon/face/tcp-channel-factory.cpp
index 8ec1c3f..4c6df87 100644
--- a/daemon/face/tcp-channel-factory.cpp
+++ b/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
diff --git a/daemon/face/tcp-channel-factory.hpp b/daemon/face/tcp-channel-factory.hpp
index 04932ca..1baeb58 100644
--- a/daemon/face/tcp-channel-factory.hpp
+++ b/daemon/face/tcp-channel-factory.hpp
@@ -23,6 +23,8 @@
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 @@
*/
shared_ptr<TcpChannel>
find(const tcp::Endpoint& localEndpoint);
+
+private:
+ boost::asio::io_service& m_ioService;
};
} // namespace ndn
diff --git a/daemon/face/tcp-channel.cpp b/daemon/face/tcp-channel.cpp
index 8aacbd6..4018161 100644
--- a/daemon/face/tcp-channel.cpp
+++ b/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
diff --git a/daemon/face/tcp-channel.hpp b/daemon/face/tcp-channel.hpp
index 50846a9..c410769 100644
--- a/daemon/face/tcp-channel.hpp
+++ b/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 @@
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
diff --git a/daemon/face/tcp-face.cpp b/daemon/face/tcp-face.cpp
index f198877..42d4fc7 100644
--- a/daemon/face/tcp-face.cpp
+++ b/daemon/face/tcp-face.cpp
@@ -14,4 +14,14 @@
{
}
+void
+TcpFace::sendInterest(const Interest& interest)
+{
+}
+
+void
+TcpFace::sendData(const Data& data)
+{
+}
+
} // namespace ndn
diff --git a/daemon/face/tcp-face.hpp b/daemon/face/tcp-face.hpp
index f010495..86ec086 100644
--- a/daemon/face/tcp-face.hpp
+++ b/daemon/face/tcp-face.hpp
@@ -21,6 +21,13 @@
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;
};