diff --git a/daemon/face/channel-factory.hpp b/daemon/face/channel-factory.hpp index 7f4a5d49..a2d7ff94 100644 --- a/daemon/face/channel-factory.hpp +++ b/daemon/face/channel-factory.hpp @@ -30,7 +30,7 @@ public: }; protected: - typedef std::map ChannelMap; + typedef std::map< Endpoint, shared_ptr > ChannelMap; ChannelMap m_channels; }; diff --git a/daemon/face/tcp-channel-factory.cpp b/daemon/face/tcp-channel-factory.cpp index 8ec1c3fb..4c6df871 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 TcpChannelFactory::create(const tcp::Endpoint& endpoint) { - return shared_ptr(); + shared_ptr channel = find(endpoint); + if(static_cast(channel)) + return channel; + + channel = make_shared(boost::ref(m_ioService), boost::cref(endpoint)); + m_channels[endpoint] = channel; + return channel; } shared_ptr TcpChannelFactory::create(const std::string& localHost, const std::string& localPort) { - return shared_ptr(); + using boost::asio::ip::tcp; + + tcp::resolver::query query(localHost, localPort); + // shared_ptr resolver = + // make_shared(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(); + + return create(*i); } shared_ptr TcpChannelFactory::find(const tcp::Endpoint& localEndpoint) { - return shared_ptr(); + ChannelMap::iterator i = m_channels.find(localEndpoint); + if (i != m_channels.end()) + return i->second; + else + return shared_ptr(); } } // namespace ndn diff --git a/daemon/face/tcp-channel-factory.hpp b/daemon/face/tcp-channel-factory.hpp index 04932ca6..1baeb582 100644 --- a/daemon/face/tcp-channel-factory.hpp +++ b/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 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 8aacbd6c..40181619 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(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 clientSocket = + make_shared(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 clientSocket = + make_shared(boost::ref(m_ioService)); + + shared_ptr connectTimeoutTimer = + make_shared(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 clientSocket = + make_shared(boost::ref(m_ioService)); + + shared_ptr connectTimeoutTimer = + make_shared(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 resolver = + make_shared(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& 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 face = make_shared(1, boost::cref(socket)); + + // what's next? +} + +void +TcpChannel::handleSuccessfulConnect(const boost::system::error_code& error, + const shared_ptr& socket, + const shared_ptr& 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& socket, + const shared_ptr& 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& socket, + const shared_ptr& 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 50846a96..c410769f 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 +#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& newFace)> FaceCreatedCallback; + /** + * \brief Prototype for the callback that is called when face is failed to + * get created + */ + typedef function 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& socket); + + void + handleSuccessfulConnect(const boost::system::error_code& error, + const shared_ptr& socket, + const shared_ptr& timer); + + void + handleFailedConnect(const boost::system::error_code& error, + const shared_ptr& socket, + const shared_ptr& timer); + + void + handleEndpointResoution(const boost::system::error_code& error, + boost::asio::ip::tcp::resolver::iterator remoteEndpoint, + const shared_ptr& socket, + const shared_ptr& timer); + +private: + boost::asio::io_service& m_ioService; tcp::Endpoint m_localEndpoint; + + bool isListening; + shared_ptr m_acceptor; }; } // namespace ndn diff --git a/daemon/face/tcp-face.cpp b/daemon/face/tcp-face.cpp index f1988774..42d4fc77 100644 --- a/daemon/face/tcp-face.cpp +++ b/daemon/face/tcp-face.cpp @@ -14,4 +14,14 @@ TcpFace::TcpFace(FaceId id, const shared_ptr& socket) { } +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 f0104956..86ec0862 100644 --- a/daemon/face/tcp-face.hpp +++ b/daemon/face/tcp-face.hpp @@ -21,6 +21,13 @@ class TcpFace : public StreamFace public: TcpFace(FaceId id, const shared_ptr& socket); + // from Face + virtual void + sendInterest(const Interest& interest); + + virtual void + sendData(const Data& data); + private: shared_ptr m_socket; };