face: Stop using shared_ptr to manage acceptors
This commit also includes a major cleanup of all channels.
Change-Id: I10db9709e0cba6a0691a86482c60b5dbb2956f68
Refs: #2613
diff --git a/daemon/face/tcp-channel.cpp b/daemon/face/tcp-channel.cpp
index 0a6ce78..637cfa3 100644
--- a/daemon/face/tcp-channel.cpp
+++ b/daemon/face/tcp-channel.cpp
@@ -24,6 +24,7 @@
*/
#include "tcp-channel.hpp"
+#include "tcp-face.hpp"
#include "core/global-io.hpp"
namespace nfd {
@@ -33,14 +34,10 @@
using namespace boost::asio;
TcpChannel::TcpChannel(const tcp::Endpoint& localEndpoint)
- : m_localEndpoint(localEndpoint)
- , m_isListening(false)
+ : m_acceptor(getGlobalIoService())
+ , m_localEndpoint(localEndpoint)
{
- this->setUri(FaceUri(localEndpoint));
-}
-
-TcpChannel::~TcpChannel()
-{
+ setUri(FaceUri(m_localEndpoint));
}
void
@@ -48,24 +45,21 @@
const ConnectFailedCallback& onAcceptFailed,
int backlog/* = tcp::acceptor::max_connections*/)
{
- m_acceptor = make_shared<ip::tcp::acceptor>(ref(getGlobalIoService()));
- m_acceptor->open(m_localEndpoint.protocol());
- m_acceptor->set_option(ip::tcp::acceptor::reuse_address(true));
+ if (isListening()) {
+ NFD_LOG_WARN("[" << m_localEndpoint << "] Already listening");
+ return;
+ }
+
+ m_acceptor.open(m_localEndpoint.protocol());
+ m_acceptor.set_option(ip::tcp::acceptor::reuse_address(true));
if (m_localEndpoint.address().is_v6())
- {
- m_acceptor->set_option(ip::v6_only(true));
- }
- m_acceptor->bind(m_localEndpoint);
- m_acceptor->listen(backlog);
+ m_acceptor.set_option(ip::v6_only(true));
- shared_ptr<ip::tcp::socket> clientSocket =
- make_shared<ip::tcp::socket>(ref(getGlobalIoService()));
- m_acceptor->async_accept(*clientSocket,
- bind(&TcpChannel::handleSuccessfulAccept, this, _1,
- clientSocket,
- onFaceCreated, onAcceptFailed));
+ m_acceptor.bind(m_localEndpoint);
+ m_acceptor.listen(backlog);
- m_isListening = true;
+ // start accepting connections
+ accept(onFaceCreated, onAcceptFailed);
}
void
@@ -74,9 +68,9 @@
const TcpChannel::ConnectFailedCallback& onConnectFailed,
const time::seconds& timeout/* = time::seconds(4)*/)
{
- ChannelFaceMap::iterator i = m_channelFaces.find(remoteEndpoint);
- if (i != m_channelFaces.end()) {
- onFaceCreated(i->second);
+ auto it = m_channelFaces.find(remoteEndpoint);
+ if (it != m_channelFaces.end()) {
+ onFaceCreated(it->second);
return;
}
@@ -84,10 +78,11 @@
make_shared<ip::tcp::socket>(ref(getGlobalIoService()));
scheduler::EventId connectTimeoutEvent = scheduler::schedule(timeout,
- bind(&TcpChannel::handleFailedConnect, this, clientSocket, onConnectFailed));
+ bind(&TcpChannel::handleConnectTimeout, this, clientSocket, onConnectFailed));
clientSocket->async_connect(remoteEndpoint,
- bind(&TcpChannel::handleSuccessfulConnect, this, _1,
+ bind(&TcpChannel::handleConnect, this,
+ boost::asio::placeholders::error,
clientSocket, connectTimeoutEvent,
onFaceCreated, onConnectFailed));
}
@@ -107,22 +102,25 @@
shared_ptr<Face> face;
- ChannelFaceMap::iterator faceMapPos = m_channelFaces.find(remoteEndpoint);
- if (faceMapPos == m_channelFaces.end())
+ auto it = m_channelFaces.find(remoteEndpoint);
+ if (it == m_channelFaces.end())
{
if (socket->local_endpoint().address().is_loopback())
face = make_shared<TcpLocalFace>(socket, isOnDemand);
else
face = make_shared<TcpFace>(socket, isOnDemand);
- face->onFail.connectSingleShot(bind(&TcpChannel::afterFaceFailed, this, remoteEndpoint));
+ face->onFail.connectSingleShot([this, remoteEndpoint] (const std::string&) {
+ NFD_LOG_TRACE("Erasing " << remoteEndpoint << " from channel face map");
+ m_channelFaces.erase(remoteEndpoint);
+ });
m_channelFaces[remoteEndpoint] = face;
}
else
{
// we've already created a a face for this endpoint, just reuse it
- face = faceMapPos->second;
+ face = it->second;
boost::system::error_code error;
socket->shutdown(ip::tcp::socket::shutdown_both, error);
@@ -135,50 +133,47 @@
}
void
-TcpChannel::afterFaceFailed(tcp::Endpoint &remoteEndpoint)
+TcpChannel::accept(const FaceCreatedCallback& onFaceCreated,
+ const ConnectFailedCallback& onAcceptFailed)
{
- NFD_LOG_DEBUG("afterFaceFailed: " << remoteEndpoint);
- m_channelFaces.erase(remoteEndpoint);
+ auto socket = make_shared<ip::tcp::socket>(ref(getGlobalIoService()));
+
+ m_acceptor.async_accept(*socket,
+ bind(&TcpChannel::handleAccept, this,
+ boost::asio::placeholders::error,
+ socket, onFaceCreated, onAcceptFailed));
}
void
-TcpChannel::handleSuccessfulAccept(const boost::system::error_code& error,
- const shared_ptr<boost::asio::ip::tcp::socket>& socket,
- const FaceCreatedCallback& onFaceCreated,
- const ConnectFailedCallback& onAcceptFailed)
+TcpChannel::handleAccept(const boost::system::error_code& error,
+ const shared_ptr<boost::asio::ip::tcp::socket>& socket,
+ const FaceCreatedCallback& onFaceCreated,
+ const ConnectFailedCallback& onAcceptFailed)
{
if (error) {
- if (error == boost::system::errc::operation_canceled) // when socket is closed by someone
+ if (error == boost::asio::error::operation_aborted) // when the socket is closed by someone
return;
- NFD_LOG_DEBUG("Connect to remote endpoint failed: "
- << error.category().message(error.value()));
-
- if (static_cast<bool>(onAcceptFailed))
- onAcceptFailed("Connect to remote endpoint failed: " +
- error.category().message(error.value()));
+ NFD_LOG_DEBUG("[" << m_localEndpoint << "] Accept failed: " << error.message());
+ if (onAcceptFailed)
+ onAcceptFailed(error.message());
return;
}
- // prepare accepting the next connection
- shared_ptr<ip::tcp::socket> clientSocket =
- make_shared<ip::tcp::socket>(ref(getGlobalIoService()));
- m_acceptor->async_accept(*clientSocket,
- bind(&TcpChannel::handleSuccessfulAccept, this, _1,
- clientSocket,
- onFaceCreated, onAcceptFailed));
+ NFD_LOG_DEBUG("[" << m_localEndpoint << "] Connection from " << socket->remote_endpoint());
- NFD_LOG_DEBUG("[" << m_localEndpoint << "] "
- "<< Connection from " << socket->remote_endpoint());
+ // prepare accepting the next connection
+ accept(onFaceCreated, onAcceptFailed);
+
createFace(socket, onFaceCreated, true);
}
void
-TcpChannel::handleSuccessfulConnect(const boost::system::error_code& error,
- const shared_ptr<ip::tcp::socket>& socket,
- const scheduler::EventId& connectTimeoutEvent,
- const FaceCreatedCallback& onFaceCreated,
- const ConnectFailedCallback& onConnectFailed)
+TcpChannel::handleConnect(const boost::system::error_code& error,
+ const shared_ptr<ip::tcp::socket>& socket,
+ const scheduler::EventId& connectTimeoutEvent,
+ const FaceCreatedCallback& onFaceCreated,
+ const ConnectFailedCallback& onConnectFailed)
{
scheduler::cancel(connectTimeoutEvent);
@@ -191,33 +186,34 @@
#else
if (error) {
#endif
-
- if (error == boost::system::errc::operation_canceled) // when socket is closed by someone
+ if (error == boost::asio::error::operation_aborted) // when the socket is closed by someone
return;
socket->close();
- NFD_LOG_DEBUG("Connect to remote endpoint failed: "
- << error.category().message(error.value()));
-
- onConnectFailed("Connect to remote endpoint failed: " +
- error.category().message(error.value()));
+ NFD_LOG_DEBUG("[" << m_localEndpoint << "] Connect failed: " << error.message());
+ if (onConnectFailed)
+ onConnectFailed(error.message());
return;
}
- NFD_LOG_DEBUG("[" << m_localEndpoint << "] "
- ">> Connection to " << socket->remote_endpoint());
+ NFD_LOG_DEBUG("[" << m_localEndpoint << "] Connected to " << socket->remote_endpoint());
createFace(socket, onFaceCreated, false);
}
void
-TcpChannel::handleFailedConnect(const shared_ptr<ip::tcp::socket>& socket,
- const ConnectFailedCallback& onConnectFailed)
+TcpChannel::handleConnectTimeout(const shared_ptr<ip::tcp::socket>& socket,
+ const ConnectFailedCallback& onConnectFailed)
{
NFD_LOG_DEBUG("Connect to remote endpoint timed out");
- onConnectFailed("Connect to remote endpoint timed out");
- socket->close(); // abort the connection
+
+ // abort the connection attempt
+ boost::system::error_code error;
+ socket->close(error);
+
+ if (onConnectFailed)
+ onConnectFailed("Connect to remote endpoint timed out");
}
} // namespace nfd
diff --git a/daemon/face/tcp-channel.hpp b/daemon/face/tcp-channel.hpp
index 25c5fb0..ba59dbe 100644
--- a/daemon/face/tcp-channel.hpp
+++ b/daemon/face/tcp-channel.hpp
@@ -27,7 +27,6 @@
#define NFD_DAEMON_FACE_TCP_CHANNEL_HPP
#include "channel.hpp"
-#include "tcp-face.hpp"
#include "core/scheduler.hpp"
namespace nfd {
@@ -55,9 +54,6 @@
explicit
TcpChannel(const tcp::Endpoint& localEndpoint);
- virtual
- ~TcpChannel();
-
/**
* \brief Enable listening on the local endpoint, accept connections,
* and create faces when remote host makes a connection
@@ -97,39 +93,37 @@
bool isOnDemand);
void
- afterFaceFailed(tcp::Endpoint &endpoint);
+ accept(const FaceCreatedCallback& onFaceCreated,
+ const ConnectFailedCallback& onConnectFailed);
void
- handleSuccessfulAccept(const boost::system::error_code& error,
- const shared_ptr<boost::asio::ip::tcp::socket>& socket,
- const FaceCreatedCallback& onFaceCreated,
- const ConnectFailedCallback& onConnectFailed);
+ handleAccept(const boost::system::error_code& error,
+ const shared_ptr<boost::asio::ip::tcp::socket>& socket,
+ const FaceCreatedCallback& onFaceCreated,
+ const ConnectFailedCallback& onConnectFailed);
void
- handleSuccessfulConnect(const boost::system::error_code& error,
- const shared_ptr<boost::asio::ip::tcp::socket>& socket,
- const scheduler::EventId& connectTimeoutEvent,
- const FaceCreatedCallback& onFaceCreated,
- const ConnectFailedCallback& onConnectFailed);
+ handleConnect(const boost::system::error_code& error,
+ const shared_ptr<boost::asio::ip::tcp::socket>& socket,
+ const scheduler::EventId& connectTimeoutEvent,
+ const FaceCreatedCallback& onFaceCreated,
+ const ConnectFailedCallback& onConnectFailed);
void
- handleFailedConnect(const shared_ptr<boost::asio::ip::tcp::socket>& socket,
- const ConnectFailedCallback& onConnectFailed);
+ handleConnectTimeout(const shared_ptr<boost::asio::ip::tcp::socket>& socket,
+ const ConnectFailedCallback& onConnectFailed);
private:
+ std::map<tcp::Endpoint, shared_ptr<Face>> m_channelFaces;
+
+ boost::asio::ip::tcp::acceptor m_acceptor;
tcp::Endpoint m_localEndpoint;
-
- typedef std::map< tcp::Endpoint, shared_ptr<Face> > ChannelFaceMap;
- ChannelFaceMap m_channelFaces;
-
- bool m_isListening;
- shared_ptr<boost::asio::ip::tcp::acceptor> m_acceptor;
};
inline bool
TcpChannel::isListening() const
{
- return m_isListening;
+ return m_acceptor.is_open();
}
} // namespace nfd
diff --git a/daemon/face/udp-channel.cpp b/daemon/face/udp-channel.cpp
index fd79d9d..bd860cc 100644
--- a/daemon/face/udp-channel.cpp
+++ b/daemon/face/udp-channel.cpp
@@ -24,6 +24,7 @@
*/
#include "udp-channel.hpp"
+#include "udp-face.hpp"
#include "core/global-io.hpp"
namespace nfd {
@@ -38,64 +39,55 @@
, m_isListening(false)
, m_idleFaceTimeout(timeout)
{
- /// \todo the reuse_address works as we want in Linux, but in other system could be different.
- /// We need to check this
- /// (SO_REUSEADDR doesn't behave uniformly in different OS)
+ setUri(FaceUri(m_localEndpoint));
m_socket = make_shared<ip::udp::socket>(ref(getGlobalIoService()));
m_socket->open(m_localEndpoint.protocol());
- m_socket->set_option(boost::asio::ip::udp::socket::reuse_address(true));
+ m_socket->set_option(ip::udp::socket::reuse_address(true));
if (m_localEndpoint.address().is_v6())
- {
- m_socket->set_option(ip::v6_only(true));
- }
+ m_socket->set_option(ip::v6_only(true));
try {
m_socket->bind(m_localEndpoint);
}
- catch (boost::system::system_error& e) {
- //The bind failed, so the socket is useless now
+ catch (const boost::system::system_error& e) {
+ // bind failed, so the socket is useless now
m_socket->close();
- throw Error("Failed to properly configure the socket. "
- "UdpChannel creation aborted, check the address (" + std::string(e.what()) + ")");
+ throw Error("bind failed: " + std::string(e.what()));
}
-
- this->setUri(FaceUri(localEndpoint));
}
void
UdpChannel::listen(const FaceCreatedCallback& onFaceCreated,
- const ConnectFailedCallback& onListenFailed)
+ const ConnectFailedCallback& onReceiveFailed)
{
- if (m_isListening) {
- throw Error("Listen already called on this channel");
+ if (isListening()) {
+ NFD_LOG_WARN("[" << m_localEndpoint << "] Already listening");
+ return;
}
m_isListening = true;
- onFaceCreatedNewPeerCallback = onFaceCreated;
- onConnectFailedNewPeerCallback = onListenFailed;
-
m_socket->async_receive_from(boost::asio::buffer(m_inputBuffer, ndn::MAX_NDN_PACKET_SIZE),
m_newRemoteEndpoint,
- bind(&UdpChannel::newPeer, this,
+ bind(&UdpChannel::handleNewPeer, this,
boost::asio::placeholders::error,
- boost::asio::placeholders::bytes_transferred));
+ boost::asio::placeholders::bytes_transferred,
+ onFaceCreated, onReceiveFailed));
}
-
void
UdpChannel::connect(const udp::Endpoint& remoteEndpoint,
const FaceCreatedCallback& onFaceCreated,
const ConnectFailedCallback& onConnectFailed)
{
- ChannelFaceMap::iterator i = m_channelFaces.find(remoteEndpoint);
- if (i != m_channelFaces.end()) {
- i->second->setOnDemand(false);
- onFaceCreated(i->second);
+ auto it = m_channelFaces.find(remoteEndpoint);
+ if (it != m_channelFaces.end()) {
+ it->second->setOnDemand(false);
+ onFaceCreated(it->second);
return;
}
- //creating a new socket for the face that will be created soon
+ // creating a new socket for the face that will be created soon
shared_ptr<ip::udp::socket> clientSocket =
make_shared<ip::udp::socket>(ref(getGlobalIoService()));
@@ -109,11 +101,12 @@
//async_connect, make sure that if in the meantime we receive a UDP pkt from
//that endpoint nothing bad happen (it's difficult, but it could happen)
}
- catch (boost::system::system_error& e) {
+ catch (const boost::system::system_error& e) {
clientSocket->close();
onConnectFailed("Failed to configure socket (" + std::string(e.what()) + ")");
return;
}
+
createFace(clientSocket, onFaceCreated, false);
}
@@ -123,7 +116,6 @@
return m_channelFaces.size();
}
-
shared_ptr<UdpFace>
UdpChannel::createFace(const shared_ptr<ip::udp::socket>& socket,
const FaceCreatedCallback& onFaceCreated,
@@ -133,19 +125,22 @@
shared_ptr<UdpFace> face;
- ChannelFaceMap::iterator faceMapPos = m_channelFaces.find(remoteEndpoint);
- if (faceMapPos == m_channelFaces.end())
+ auto it = m_channelFaces.find(remoteEndpoint);
+ if (it == m_channelFaces.end())
{
face = make_shared<UdpFace>(socket, isOnDemand, m_idleFaceTimeout);
- face->onFail.connectSingleShot(bind(&UdpChannel::afterFaceFailed, this, remoteEndpoint));
+ face->onFail.connectSingleShot([this, remoteEndpoint] (const std::string&) {
+ NFD_LOG_TRACE("Erasing " << remoteEndpoint << " from channel face map");
+ m_channelFaces.erase(remoteEndpoint);
+ });
m_channelFaces[remoteEndpoint] = face;
}
else
{
// we've already created a a face for this endpoint, just reuse it
- face = faceMapPos->second;
+ face = it->second;
boost::system::error_code error;
socket->shutdown(ip::udp::socket::shutdown_both, error);
@@ -160,23 +155,27 @@
}
void
-UdpChannel::newPeer(const boost::system::error_code& error,
- size_t nBytesReceived)
+UdpChannel::handleNewPeer(const boost::system::error_code& error,
+ size_t nBytesReceived,
+ const FaceCreatedCallback& onFaceCreated,
+ const ConnectFailedCallback& onReceiveFailed)
{
if (error) {
- if (error == boost::asio::error::operation_aborted)
+ if (error == boost::asio::error::operation_aborted) // when the socket is closed by someone
return;
- NFD_LOG_ERROR("newPeer: " << error.message());
+ NFD_LOG_DEBUG("[" << m_localEndpoint << "] Receive failed: " << error.message());
+ if (onReceiveFailed)
+ onReceiveFailed(error.message());
return;
}
- NFD_LOG_DEBUG("newPeer from " << m_newRemoteEndpoint);
+ NFD_LOG_DEBUG("[" << m_localEndpoint << "] New peer " << m_newRemoteEndpoint);
shared_ptr<UdpFace> face;
- ChannelFaceMap::iterator i = m_channelFaces.find(m_newRemoteEndpoint);
- if (i != m_channelFaces.end()) {
+ auto it = m_channelFaces.find(m_newRemoteEndpoint);
+ if (it != m_channelFaces.end()) {
//The face already exists.
//Usually this shouldn't happen, because the channel creates a Udpface
//as soon as it receives a pkt from a new endpoint and then the
@@ -187,10 +186,8 @@
//ready. In this case, the channel has to pass the pkt to the face
NFD_LOG_DEBUG("The creation of the face for the remote endpoint "
- << m_newRemoteEndpoint
- << " is in progress");
-
- face = i->second;
+ << m_newRemoteEndpoint << " is already in progress");
+ face = it->second;
}
else {
shared_ptr<ip::udp::socket> clientSocket =
@@ -198,6 +195,7 @@
clientSocket->open(m_localEndpoint.protocol());
clientSocket->set_option(ip::udp::socket::reuse_address(true));
clientSocket->bind(m_localEndpoint);
+
boost::system::error_code ec;
clientSocket->connect(m_newRemoteEndpoint, ec);
if (ec) {
@@ -206,9 +204,7 @@
return;
}
- face = createFace(clientSocket,
- onFaceCreatedNewPeerCallback,
- true);
+ face = createFace(clientSocket, onFaceCreated, true);
}
// dispatch the datagram to the face for processing
@@ -216,17 +212,10 @@
m_socket->async_receive_from(boost::asio::buffer(m_inputBuffer, ndn::MAX_NDN_PACKET_SIZE),
m_newRemoteEndpoint,
- bind(&UdpChannel::newPeer, this,
+ bind(&UdpChannel::handleNewPeer, this,
boost::asio::placeholders::error,
- boost::asio::placeholders::bytes_transferred));
-}
-
-
-void
-UdpChannel::afterFaceFailed(udp::Endpoint &endpoint)
-{
- NFD_LOG_DEBUG("afterFaceFailed: " << endpoint);
- m_channelFaces.erase(endpoint);
+ boost::asio::placeholders::bytes_transferred,
+ onFaceCreated, onReceiveFailed));
}
} // namespace nfd
diff --git a/daemon/face/udp-channel.hpp b/daemon/face/udp-channel.hpp
index b51d50d..1451713 100644
--- a/daemon/face/udp-channel.hpp
+++ b/daemon/face/udp-channel.hpp
@@ -27,8 +27,6 @@
#define NFD_DAEMON_FACE_UDP_CHANNEL_HPP
#include "channel.hpp"
-#include "core/global-io.hpp"
-#include "udp-face.hpp"
namespace nfd {
@@ -36,10 +34,10 @@
typedef boost::asio::ip::udp::endpoint Endpoint;
} // namespace udp
+class UdpFace;
+
/**
* \brief Class implementing UDP-based channel to create faces
- *
- *
*/
class UdpChannel : public Channel
{
@@ -79,7 +77,7 @@
*/
void
listen(const FaceCreatedCallback& onFaceCreated,
- const ConnectFailedCallback& onAcceptFailed);
+ const ConnectFailedCallback& onReceiveFailed);
/**
* \brief Create a face by establishing connection to remote endpoint
@@ -97,22 +95,28 @@
size_t
size() const;
+ bool
+ isListening() const;
+
private:
shared_ptr<UdpFace>
createFace(const shared_ptr<boost::asio::ip::udp::socket>& socket,
const FaceCreatedCallback& onFaceCreated,
bool isOnDemand);
- void
- afterFaceFailed(udp::Endpoint& endpoint);
/**
- * \brief The UdpChannel has received a new pkt from a remote endpoint not yet
- * associated with any UdpFace
+ * \brief The channel has received a new packet from a remote
+ * endpoint that is not associated with any UdpFace yet
*/
void
- newPeer(const boost::system::error_code& error, size_t nBytesReceived);
+ handleNewPeer(const boost::system::error_code& error,
+ size_t nBytesReceived,
+ const FaceCreatedCallback& onFaceCreated,
+ const ConnectFailedCallback& onReceiveFailed);
private:
+ std::map<udp::Endpoint, shared_ptr<UdpFace>> m_channelFaces;
+
udp::Endpoint m_localEndpoint;
/**
@@ -121,25 +125,12 @@
udp::Endpoint m_newRemoteEndpoint;
/**
- * Callbacks for face creation.
- * New communications are detected using async_receive_from.
- * Its handler has a fixed signature. No space for the face callback
- */
- FaceCreatedCallback onFaceCreatedNewPeerCallback;
-
- // @todo remove the onConnectFailedNewPeerCallback if it remains unused
- ConnectFailedCallback onConnectFailedNewPeerCallback;
-
- /**
* \brief Socket used to "accept" new communication
**/
shared_ptr<boost::asio::ip::udp::socket> m_socket;
uint8_t m_inputBuffer[ndn::MAX_NDN_PACKET_SIZE];
- typedef std::map< udp::Endpoint, shared_ptr<UdpFace> > ChannelFaceMap;
- ChannelFaceMap m_channelFaces;
-
/**
* \brief If true, it means the function listen has already been called
*/
@@ -152,6 +143,12 @@
time::seconds m_idleFaceTimeout;
};
+inline bool
+UdpChannel::isListening() const
+{
+ return m_isListening;
+}
+
} // namespace nfd
#endif // NFD_DAEMON_FACE_UDP_CHANNEL_HPP
diff --git a/daemon/face/unix-stream-channel.cpp b/daemon/face/unix-stream-channel.cpp
index d64f333..52df192 100644
--- a/daemon/face/unix-stream-channel.cpp
+++ b/daemon/face/unix-stream-channel.cpp
@@ -1,11 +1,12 @@
/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
/**
- * Copyright (c) 2014 Regents of the University of California,
- * Arizona Board of Regents,
- * Colorado State University,
- * University Pierre & Marie Curie, Sorbonne University,
- * Washington University in St. Louis,
- * Beijing Institute of Technology
+ * Copyright (c) 2014-2015, Regents of the University of California,
+ * Arizona Board of Regents,
+ * Colorado State University,
+ * University Pierre & Marie Curie, Sorbonne University,
+ * Washington University in St. Louis,
+ * Beijing Institute of Technology,
+ * The University of Memphis.
*
* This file is part of NFD (Named Data Networking Forwarding Daemon).
* See AUTHORS.md for complete list of NFD authors and contributors.
@@ -20,9 +21,10 @@
*
* You should have received a copy of the GNU General Public License along with
* NFD, e.g., in COPYING.md file. If not, see <http://www.gnu.org/licenses/>.
- **/
+ */
#include "unix-stream-channel.hpp"
+#include "unix-stream-face.hpp"
#include "core/global-io.hpp"
#include <boost/filesystem.hpp>
@@ -35,23 +37,22 @@
using namespace boost::asio::local;
UnixStreamChannel::UnixStreamChannel(const unix_stream::Endpoint& endpoint)
- : m_endpoint(endpoint)
- , m_isListening(false)
+ : m_acceptor(getGlobalIoService())
+ , m_endpoint(endpoint)
{
- setUri(FaceUri(endpoint));
+ setUri(FaceUri(m_endpoint));
}
UnixStreamChannel::~UnixStreamChannel()
{
- if (m_isListening)
- {
- // use the non-throwing variants during destruction
- // and ignore any errors
- boost::system::error_code error;
- m_acceptor->close(error);
- NFD_LOG_TRACE("[" << m_endpoint << "] Removing socket file");
- boost::filesystem::remove(m_endpoint.path(), error);
- }
+ if (isListening()) {
+ // use the non-throwing variants during destruction
+ // and ignore any errors
+ boost::system::error_code error;
+ m_acceptor.close(error);
+ NFD_LOG_DEBUG("[" << m_endpoint << "] Removing socket file");
+ boost::filesystem::remove(m_endpoint.path(), error);
+ }
}
void
@@ -59,7 +60,7 @@
const ConnectFailedCallback& onAcceptFailed,
int backlog/* = acceptor::max_connections*/)
{
- if (m_isListening) {
+ if (isListening()) {
NFD_LOG_WARN("[" << m_endpoint << "] Already listening");
return;
}
@@ -69,78 +70,73 @@
fs::path socketPath(m_endpoint.path());
fs::file_type type = fs::symlink_status(socketPath).type();
- if (type == fs::socket_file)
- {
- boost::system::error_code error;
- stream_protocol::socket socket(getGlobalIoService());
- socket.connect(m_endpoint, error);
- NFD_LOG_TRACE("[" << m_endpoint << "] connect() on existing socket file returned: "
- + error.message());
- if (!error)
- {
- // someone answered, leave the socket alone
- throw Error("Socket file at " + m_endpoint.path()
- + " belongs to another NFD process");
- }
- else if (error == boost::system::errc::connection_refused ||
- error == boost::system::errc::timed_out)
- {
- // no one is listening on the remote side,
- // we can safely remove the socket file
- NFD_LOG_INFO("[" << m_endpoint << "] Removing stale socket file");
- fs::remove(socketPath);
- }
+ if (type == fs::socket_file) {
+ boost::system::error_code error;
+ stream_protocol::socket socket(getGlobalIoService());
+ socket.connect(m_endpoint, error);
+ NFD_LOG_TRACE("[" << m_endpoint << "] connect() on existing socket file returned: "
+ + error.message());
+ if (!error) {
+ // someone answered, leave the socket alone
+ throw Error("Socket file at " + m_endpoint.path()
+ + " belongs to another NFD process");
}
- else if (type != fs::file_not_found)
- {
- throw Error(m_endpoint.path() + " already exists and is not a socket file");
+ else if (error == boost::asio::error::connection_refused ||
+ error == boost::asio::error::timed_out) {
+ // no one is listening on the remote side,
+ // we can safely remove the stale socket
+ NFD_LOG_DEBUG("[" << m_endpoint << "] Removing stale socket file");
+ fs::remove(socketPath);
}
+ }
+ else if (type != fs::file_not_found) {
+ throw Error(m_endpoint.path() + " already exists and is not a socket file");
+ }
- m_acceptor = make_shared<stream_protocol::acceptor>(ref(getGlobalIoService()));
- m_acceptor->open();
- m_acceptor->bind(m_endpoint);
- m_acceptor->listen(backlog);
- m_isListening = true;
+ m_acceptor.open();
+ m_acceptor.bind(m_endpoint);
+ m_acceptor.listen(backlog);
- if (::chmod(m_endpoint.path().c_str(), 0666) < 0)
- {
- throw Error("Failed to chmod() socket file at " + m_endpoint.path());
- }
+ if (::chmod(m_endpoint.path().c_str(), 0666) < 0) {
+ throw Error("chmod(" + m_endpoint.path() + ") failed: " + std::strerror(errno));
+ }
- shared_ptr<stream_protocol::socket> clientSocket =
- make_shared<stream_protocol::socket>(ref(getGlobalIoService()));
-
- m_acceptor->async_accept(*clientSocket,
- bind(&UnixStreamChannel::handleSuccessfulAccept, this,
- boost::asio::placeholders::error, clientSocket,
- onFaceCreated, onAcceptFailed));
+ // start accepting connections
+ accept(onFaceCreated, onAcceptFailed);
}
void
-UnixStreamChannel::handleSuccessfulAccept(const boost::system::error_code& error,
- const shared_ptr<stream_protocol::socket>& socket,
- const FaceCreatedCallback& onFaceCreated,
- const ConnectFailedCallback& onAcceptFailed)
+UnixStreamChannel::accept(const FaceCreatedCallback &onFaceCreated,
+ const ConnectFailedCallback &onAcceptFailed)
+{
+ auto socket = make_shared<stream_protocol::socket>(ref(getGlobalIoService()));
+
+ m_acceptor.async_accept(*socket,
+ bind(&UnixStreamChannel::handleAccept, this,
+ boost::asio::placeholders::error,
+ socket, onFaceCreated, onAcceptFailed));
+}
+
+void
+UnixStreamChannel::handleAccept(const boost::system::error_code& error,
+ const shared_ptr<stream_protocol::socket>& socket,
+ const FaceCreatedCallback& onFaceCreated,
+ const ConnectFailedCallback& onAcceptFailed)
{
if (error) {
- if (error == boost::system::errc::operation_canceled) // when socket is closed by someone
+ if (error == boost::asio::error::operation_aborted) // when the socket is closed by someone
return;
- NFD_LOG_DEBUG("[" << m_endpoint << "] Connection failed: " << error.message());
- onAcceptFailed("Connection failed: " + error.message());
+ NFD_LOG_DEBUG("[" << m_endpoint << "] Accept failed: " << error.message());
+ if (onAcceptFailed)
+ onAcceptFailed(error.message());
return;
}
- NFD_LOG_DEBUG("[" << m_endpoint << "] << Incoming connection");
-
- shared_ptr<stream_protocol::socket> clientSocket =
- make_shared<stream_protocol::socket>(ref(getGlobalIoService()));
+ NFD_LOG_DEBUG("[" << m_endpoint << "] Incoming connection");
// prepare accepting the next connection
- m_acceptor->async_accept(*clientSocket,
- bind(&UnixStreamChannel::handleSuccessfulAccept, this,
- boost::asio::placeholders::error, clientSocket,
- onFaceCreated, onAcceptFailed));
+ accept(onFaceCreated, onAcceptFailed);
shared_ptr<UnixStreamFace> face = make_shared<UnixStreamFace>(socket);
onFaceCreated(face);
diff --git a/daemon/face/unix-stream-channel.hpp b/daemon/face/unix-stream-channel.hpp
index c923c39..c90dcb0 100644
--- a/daemon/face/unix-stream-channel.hpp
+++ b/daemon/face/unix-stream-channel.hpp
@@ -26,7 +26,6 @@
#define NFD_DAEMON_FACE_UNIX_STREAM_CHANNEL_HPP
#include "channel.hpp"
-#include "unix-stream-face.hpp"
namespace nfd {
@@ -60,8 +59,7 @@
explicit
UnixStreamChannel(const unix_stream::Endpoint& endpoint);
- virtual
- ~UnixStreamChannel();
+ ~UnixStreamChannel() DECL_OVERRIDE;
/**
* \brief Enable listening on the local endpoint, accept connections,
@@ -77,19 +75,31 @@
const ConnectFailedCallback& onAcceptFailed,
int backlog = boost::asio::local::stream_protocol::acceptor::max_connections);
-private:
- void
- handleSuccessfulAccept(const boost::system::error_code& error,
- const shared_ptr<boost::asio::local::stream_protocol::socket>& socket,
- const FaceCreatedCallback& onFaceCreated,
- const ConnectFailedCallback& onConnectFailed);
+ bool
+ isListening() const;
private:
+ void
+ accept(const FaceCreatedCallback& onFaceCreated,
+ const ConnectFailedCallback& onAcceptFailed);
+
+ void
+ handleAccept(const boost::system::error_code& error,
+ const shared_ptr<boost::asio::local::stream_protocol::socket>& socket,
+ const FaceCreatedCallback& onFaceCreated,
+ const ConnectFailedCallback& onAcceptFailed);
+
+private:
+ boost::asio::local::stream_protocol::acceptor m_acceptor;
unix_stream::Endpoint m_endpoint;
- shared_ptr<boost::asio::local::stream_protocol::acceptor> m_acceptor;
- bool m_isListening;
};
+inline bool
+UnixStreamChannel::isListening() const
+{
+ return m_acceptor.is_open();
+}
+
} // namespace nfd
#endif // NFD_DAEMON_FACE_UNIX_STREAM_CHANNEL_HPP
diff --git a/daemon/face/websocket-channel.cpp b/daemon/face/websocket-channel.cpp
index 016dcac..855dd99 100644
--- a/daemon/face/websocket-channel.cpp
+++ b/daemon/face/websocket-channel.cpp
@@ -24,6 +24,8 @@
*/
#include "websocket-channel.hpp"
+#include "core/global-io.hpp"
+#include "core/scheduler.hpp"
#include <boost/date_time/posix_time/posix_time.hpp>
@@ -31,13 +33,13 @@
NFD_LOG_INIT("WebSocketChannel");
-using namespace boost::asio;
-
WebSocketChannel::WebSocketChannel(const websocket::Endpoint& localEndpoint)
: m_localEndpoint(localEndpoint)
, m_isListening(false)
, m_pingInterval(10000)
{
+ setUri(FaceUri(m_localEndpoint, "ws"));
+
// Setup WebSocket server
m_server.clear_access_channels(websocketpp::log::alevel::all);
m_server.clear_error_channels(websocketpp::log::alevel::all);
@@ -46,19 +48,19 @@
m_server.set_open_handler(bind(&WebSocketChannel::handleOpen, this, _1));
m_server.set_close_handler(bind(&WebSocketChannel::handleClose, this, _1));
m_server.init_asio(&getGlobalIoService());
+
+ // Detect disconnections using ping-pong messages
+ m_server.set_pong_handler(bind(&WebSocketChannel::handlePong, this, _1, _2));
+ m_server.set_pong_timeout_handler(bind(&WebSocketChannel::handlePongTimeout, this, _1, _2));
+
// Always set SO_REUSEADDR flag
m_server.set_reuse_addr(true);
-
- // Detect disconnection using PONG message
- m_server.set_pong_handler(bind(&WebSocketChannel::handlePong, this, _1, _2));
- m_server.set_pong_timeout_handler(bind(&WebSocketChannel::handlePongTimeout,
- this, _1, _2));
-
- this->setUri(FaceUri(localEndpoint, "ws"));
}
-WebSocketChannel::~WebSocketChannel()
+void
+WebSocketChannel::setPingInterval(time::milliseconds interval)
{
+ m_pingInterval = interval;
}
void
@@ -70,56 +72,53 @@
void
WebSocketChannel::handlePongTimeout(websocketpp::connection_hdl hdl, std::string msg)
{
- ChannelFaceMap::iterator it = m_channelFaces.find(hdl);
- if (it != m_channelFaces.end())
- {
- it->second->close();
- NFD_LOG_DEBUG("handlePongTimeout: remove " << it->second->getRemoteUri());
- m_channelFaces.erase(it);
- }
+ auto it = m_channelFaces.find(hdl);
+ if (it != m_channelFaces.end()) {
+ NFD_LOG_TRACE(__func__ << ": " << it->second->getRemoteUri());
+ it->second->close();
+ m_channelFaces.erase(it);
+ }
}
void
WebSocketChannel::handlePong(websocketpp::connection_hdl hdl, std::string msg)
{
- ChannelFaceMap::iterator it = m_channelFaces.find(hdl);
- if (it != m_channelFaces.end())
- {
- NFD_LOG_TRACE("handlePong: from " << it->second->getRemoteUri());
- }
+ auto it = m_channelFaces.find(hdl);
+ if (it != m_channelFaces.end()) {
+ NFD_LOG_TRACE("Pong from " << it->second->getRemoteUri());
+ }
}
void
WebSocketChannel::handleMessage(websocketpp::connection_hdl hdl,
websocket::Server::message_ptr msg)
{
- ChannelFaceMap::iterator it = m_channelFaces.find(hdl);
- if (it != m_channelFaces.end())
- {
- it->second->handleReceive(msg->get_payload());
- }
+ auto it = m_channelFaces.find(hdl);
+ if (it != m_channelFaces.end()) {
+ it->second->handleReceive(msg->get_payload());
+ }
}
void
WebSocketChannel::handleOpen(websocketpp::connection_hdl hdl)
{
std::string remote;
- try
- {
- remote = "wsclient://" + m_server.get_con_from_hdl(hdl)->get_remote_endpoint();
- }
- catch (websocketpp::lib::error_code&)
- {
- NFD_LOG_DEBUG("handleOpen: cannot get remote uri");
- websocketpp::lib::error_code ecode;
- m_server.close(hdl, websocketpp::close::status::normal, "closed by channel", ecode);
- }
- shared_ptr<WebSocketFace> face = ndn::make_shared<WebSocketFace>(FaceUri(remote), this->getUri(),
- hdl, ref(m_server));
+ try {
+ remote = "wsclient://" + m_server.get_con_from_hdl(hdl)->get_remote_endpoint();
+ }
+ catch (const websocketpp::lib::error_code& e) {
+ NFD_LOG_WARN("Cannot get remote URI");
+ websocketpp::lib::error_code error;
+ m_server.close(hdl, websocketpp::close::status::normal, "closed by channel", error);
+ return;
+ }
+
+ auto face = make_shared<WebSocketFace>(FaceUri(remote), this->getUri(),
+ hdl, ref(m_server));
m_onFaceCreatedCallback(face);
m_channelFaces[hdl] = face;
- // Schedule PING message
+ // Schedule ping message
scheduler::EventId pingEvent = scheduler::schedule(m_pingInterval,
bind(&WebSocketChannel::sendPing, this, hdl));
face->setPingEventId(pingEvent);
@@ -128,62 +127,49 @@
void
WebSocketChannel::sendPing(websocketpp::connection_hdl hdl)
{
- ChannelFaceMap::iterator it = m_channelFaces.find(hdl);
- if (it != m_channelFaces.end())
- {
- try
- {
- m_server.ping(hdl, "NFD-WebSocket");
- }
- catch (websocketpp::lib::error_code&)
- {
- it->second->close();
- NFD_LOG_DEBUG("sendPing: failed to ping " << it->second->getRemoteUri());
- m_channelFaces.erase(it);
- }
+ auto it = m_channelFaces.find(hdl);
+ if (it != m_channelFaces.end()) {
+ NFD_LOG_TRACE("Sending ping to " << it->second->getRemoteUri());
- NFD_LOG_TRACE("sendPing: to " << it->second->getRemoteUri());
-
- // Schedule next PING message
- scheduler::EventId pingEvent = scheduler::schedule(m_pingInterval,
- bind(&WebSocketChannel::sendPing, this, hdl));
- it->second->setPingEventId(pingEvent);
+ try {
+ m_server.ping(hdl, "NFD-WebSocket");
}
+ catch (const websocketpp::lib::error_code& e) {
+ NFD_LOG_WARN("Failed to ping " << it->second->getRemoteUri());
+ it->second->close();
+ m_channelFaces.erase(it);
+ return;
+ }
+
+ // Schedule next ping message
+ scheduler::EventId pingEvent = scheduler::schedule(m_pingInterval,
+ bind(&WebSocketChannel::sendPing, this, hdl));
+ it->second->setPingEventId(pingEvent);
+ }
}
void
WebSocketChannel::handleClose(websocketpp::connection_hdl hdl)
{
- ChannelFaceMap::iterator it = m_channelFaces.find(hdl);
- if (it != m_channelFaces.end())
- {
- it->second->close();
- NFD_LOG_DEBUG("handleClose: remove " << it->second->getRemoteUri());
- m_channelFaces.erase(it);
- }
+ auto it = m_channelFaces.find(hdl);
+ if (it != m_channelFaces.end()) {
+ NFD_LOG_TRACE(__func__ << ": " << it->second->getRemoteUri());
+ it->second->close();
+ m_channelFaces.erase(it);
+ }
}
-
void
WebSocketChannel::listen(const FaceCreatedCallback& onFaceCreated)
{
- if (m_isListening)
- {
- throw Error("Listen already called on this channel");
- }
+ if (isListening()) {
+ NFD_LOG_WARN("[" << m_localEndpoint << "] Already listening");
+ return;
+ }
m_isListening = true;
m_onFaceCreatedCallback = onFaceCreated;
-
- try
- {
- m_server.listen(m_localEndpoint);
- }
- catch (websocketpp::lib::error_code ec)
- {
- throw Error("Failed to listen on local endpoint");
- }
-
+ m_server.listen(m_localEndpoint);
m_server.start_accept();
}
diff --git a/daemon/face/websocket-channel.hpp b/daemon/face/websocket-channel.hpp
index f7c6ae2..f560ab3 100644
--- a/daemon/face/websocket-channel.hpp
+++ b/daemon/face/websocket-channel.hpp
@@ -27,34 +27,17 @@
#define NFD_DAEMON_FACE_WEBSOCKET_CHANNEL_HPP
#include "channel.hpp"
-#include "core/global-io.hpp"
-#include "core/scheduler.hpp"
#include "websocket-face.hpp"
namespace nfd {
/**
* \brief Class implementing WebSocket-based channel to create faces
- *
- *
*/
class WebSocketChannel : public Channel
{
public:
/**
- * \brief Exception of WebSocketChannel
- */
- class Error : public std::runtime_error
- {
- public:
- explicit
- Error(const std::string& what)
- : runtime_error(what)
- {
- }
- };
-
- /**
* \brief Create WebSocket channel for the local endpoint
*
* To enable creation of faces upon incoming connections,
@@ -66,9 +49,6 @@
explicit
WebSocketChannel(const websocket::Endpoint& localEndpoint);
- virtual
- ~WebSocketChannel();
-
/**
* \brief Enable listening on the local endpoint, accept connections,
* and create faces when remote host makes a connection
@@ -88,11 +68,9 @@
bool
isListening() const;
+PUBLIC_WITH_TESTS_ELSE_PRIVATE:
void
- setPingInterval(time::milliseconds interval)
- {
- m_pingInterval = interval;
- }
+ setPingInterval(time::milliseconds interval);
void
setPongTimeout(time::milliseconds timeout);
@@ -118,20 +96,16 @@
private:
websocket::Endpoint m_localEndpoint;
-
websocket::Server m_server;
+ std::map<websocketpp::connection_hdl, shared_ptr<WebSocketFace>,
+ std::owner_less<websocketpp::connection_hdl>> m_channelFaces;
+
/**
- * Callbacks for face creation.
- * New communications are detected using async_receive_from.
- * Its handler has a fixed signature. No space for the face callback
+ * Callback for face creation
*/
FaceCreatedCallback m_onFaceCreatedCallback;
- typedef std::map< websocketpp::connection_hdl, shared_ptr<WebSocketFace>,
- std::owner_less<websocketpp::connection_hdl> > ChannelFaceMap;
- ChannelFaceMap m_channelFaces;
-
/**
* \brief If true, it means the function listen has already been called
*/
diff --git a/tests/daemon/face/tcp.t.cpp b/tests/daemon/face/tcp.t.cpp
index ee2eb41..9ee9881 100644
--- a/tests/daemon/face/tcp.t.cpp
+++ b/tests/daemon/face/tcp.t.cpp
@@ -23,16 +23,19 @@
* NFD, e.g., in COPYING.md file. If not, see <http://www.gnu.org/licenses/>.
*/
+#include "face/tcp-channel.hpp"
+#include "face/tcp-face.hpp"
#include "face/tcp-factory.hpp"
-#include <ndn-cxx/util/dns.hpp>
-#include "core/network-interface.hpp"
-#include <ndn-cxx/security/key-chain.hpp>
+#include "core/network-interface.hpp"
#include "tests/test-common.hpp"
#include "tests/limited-io.hpp"
#include "dummy-stream-sender.hpp"
#include "packet-datasets.hpp"
+#include <ndn-cxx/security/key-chain.hpp>
+#include <ndn-cxx/util/dns.hpp>
+
namespace nfd {
namespace tests {
diff --git a/tests/daemon/face/udp.t.cpp b/tests/daemon/face/udp.t.cpp
index 44772ec..ea62922 100644
--- a/tests/daemon/face/udp.t.cpp
+++ b/tests/daemon/face/udp.t.cpp
@@ -23,9 +23,11 @@
* NFD, e.g., in COPYING.md file. If not, see <http://www.gnu.org/licenses/>.
*/
+#include "face/udp-channel.hpp"
+#include "face/udp-face.hpp"
#include "face/udp-factory.hpp"
-#include "core/network-interface.hpp"
+#include "core/network-interface.hpp"
#include "tests/test-common.hpp"
#include "tests/limited-io.hpp"
#include "face-history.hpp"
diff --git a/tests/daemon/face/unix-stream.t.cpp b/tests/daemon/face/unix-stream.t.cpp
index a3a5488..c1ee2cb 100644
--- a/tests/daemon/face/unix-stream.t.cpp
+++ b/tests/daemon/face/unix-stream.t.cpp
@@ -23,6 +23,8 @@
* NFD, e.g., in COPYING.md file. If not, see <http://www.gnu.org/licenses/>.
*/
+#include "face/unix-stream-channel.hpp"
+#include "face/unix-stream-face.hpp"
#include "face/unix-stream-factory.hpp"
#include "tests/test-common.hpp"
diff --git a/tests/daemon/face/websocket.t.cpp b/tests/daemon/face/websocket.t.cpp
index 8e30517..e0472bc 100644
--- a/tests/daemon/face/websocket.t.cpp
+++ b/tests/daemon/face/websocket.t.cpp
@@ -23,7 +23,10 @@
* NFD, e.g., in COPYING.md file. If not, see <http://www.gnu.org/licenses/>.
*/
+#include "face/websocket-channel.hpp"
+#include "face/websocket-face.hpp"
#include "face/websocket-factory.hpp"
+
#include "tests/test-common.hpp"
#include "tests/limited-io.hpp"
diff --git a/tests/daemon/mgmt/face-query-status-publisher.t.cpp b/tests/daemon/mgmt/face-query-status-publisher.t.cpp
index 076c384..217a863 100644
--- a/tests/daemon/mgmt/face-query-status-publisher.t.cpp
+++ b/tests/daemon/mgmt/face-query-status-publisher.t.cpp
@@ -28,8 +28,6 @@
namespace nfd {
namespace tests {
-NFD_LOG_INIT("FaceQueryStatusPublisherTest");
-
BOOST_FIXTURE_TEST_SUITE(MgmtFaceQueryStatusPublisher, FaceQueryStatusPublisherFixture)
BOOST_AUTO_TEST_CASE(NoConditionFilter)