management: Making LocalControlHeader encoding independent of Interest/Data wire
Boost.Asio support multi-buffer send operation, which is enabled in this
commit for prepending (potentially different) LocalControlHeader's to
Interest/Data wire.
Change-Id: I39b979f89f196d3e47d6466fb71f6d440bce74d4
refs: #1265
diff --git a/src/transport/stream-transport.hpp b/src/transport/stream-transport.hpp
new file mode 100644
index 0000000..f98b988
--- /dev/null
+++ b/src/transport/stream-transport.hpp
@@ -0,0 +1,337 @@
+/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil -*- */
+/**
+ * Copyright (C) 2013-2014 Regents of the University of California.
+ * See COPYING for copyright and distribution information.
+ */
+
+#ifndef NDN_TRANSPORT_STREAM_TRANSPORT_HPP
+#define NDN_TRANSPORT_STREAM_TRANSPORT_HPP
+
+#include "../common.hpp"
+
+namespace ndn {
+
+const size_t MAX_LENGTH = 9000;
+
+template<class BaseTransport, class Protocol>
+class StreamTransportImpl
+{
+public:
+ typedef BaseTransport base_transport;
+ typedef Protocol protocol;
+ typedef StreamTransportImpl<BaseTransport,Protocol> impl;
+
+ StreamTransportImpl(base_transport& transport, boost::asio::io_service& ioService)
+ : m_transport(transport)
+ , m_socket(ioService)
+ , m_partialDataSize(0)
+ , m_connectionInProgress(false)
+ , m_connectTimer(ioService)
+ {
+ }
+
+ void
+ connectHandler(const boost::system::error_code& error)
+ {
+ m_connectionInProgress = false;
+ m_connectTimer.cancel();
+
+ if (!error)
+ {
+ m_partialDataSize = 0;
+ m_socket.async_receive(boost::asio::buffer(m_inputBuffer, MAX_LENGTH), 0,
+ bind(&impl::handle_async_receive, this, _1, _2));
+
+ m_transport.m_isConnected = true;
+
+ for (std::list<Block>::iterator i = m_sendQueue.begin(); i != m_sendQueue.end(); ++i)
+ m_socket.async_send(boost::asio::buffer(i->wire(), i->size()),
+ bind(&impl::handle_async_send, this, _1, *i));
+
+ for (std::list< std::pair<Block,Block> >::iterator i = m_sendPairQueue.begin();
+ i != m_sendPairQueue.end(); ++i)
+ {
+ std::vector<boost::asio::const_buffer> buffer;
+ buffer.reserve(2);
+ buffer.push_back(boost::asio::buffer(i->first.wire(), i->first.size()));
+ buffer.push_back(boost::asio::buffer(i->second.wire(), i->second.size()));
+ m_socket.async_send(buffer,
+ bind(&impl::handle_async_send, this, _1, i->first, i->second));
+ }
+
+ m_sendQueue.clear();
+ m_sendPairQueue.clear();
+ }
+ else
+ {
+ // may need to throw exception
+ m_transport.m_isConnected = false;
+ m_transport.close();
+ throw Transport::Error(error, "error while connecting to the forwarder");
+ }
+ }
+
+ void
+ connectTimeoutHandler(const boost::system::error_code& error)
+ {
+ if (error) // e.g., cancelled timer
+ return;
+
+ m_connectionInProgress = false;
+ m_transport.m_isConnected = false;
+ m_socket.close();
+ throw Transport::Error(error, "error while connecting to the forwarder");
+ }
+
+ void
+ connect(const typename protocol::endpoint& endpoint)
+ {
+ if (!m_connectionInProgress) {
+ m_connectionInProgress = true;
+
+ // Wait at most 4 seconds to connect
+ /// @todo Decide whether this number should be configurable
+ m_connectTimer.expires_from_now(boost::posix_time::seconds(4));
+ m_connectTimer.async_wait(bind(&impl::connectTimeoutHandler, this, _1));
+
+ m_socket.open();
+ m_socket.async_connect(endpoint,
+ bind(&impl::connectHandler, this, _1));
+ }
+ }
+
+ void
+ close()
+ {
+ m_connectTimer.cancel();
+ m_socket.close();
+ m_transport.m_isConnected = false;
+ m_sendQueue.clear();
+ m_sendPairQueue.clear();
+ }
+
+ void
+ send(const Block& wire)
+ {
+ if (!m_transport.m_isConnected)
+ m_sendQueue.push_back(wire);
+ else
+ m_socket.async_send(boost::asio::buffer(wire.wire(), wire.size()),
+ bind(&impl::handle_async_send, this, _1, wire));
+ }
+
+ void
+ send(const Block& header, const Block& payload)
+ {
+ if (!m_transport.m_isConnected)
+ {
+ m_sendPairQueue.push_back(std::make_pair(header, payload));
+ }
+ else
+ {
+ std::vector<boost::asio::const_buffer> buffers;
+ buffers.reserve(2);
+ buffers.push_back(boost::asio::buffer(header.wire(), header.size()));
+ buffers.push_back(boost::asio::buffer(payload.wire(), payload.size()));
+
+ m_socket.async_send(buffers,
+ bind(&impl::handle_async_send, this, _1, header, payload));
+ }
+ }
+
+ inline void
+ processAll(uint8_t* buffer, size_t& offset, size_t availableSize)
+ {
+ while(offset < availableSize)
+ {
+ Block element(buffer + offset, availableSize - offset);
+ m_transport.receive(element);
+
+ offset += element.size();
+ }
+ }
+
+ void
+ handle_async_receive(const boost::system::error_code& error, std::size_t bytes_recvd)
+ {
+ /// @todo The socket is not datagram, so need to have internal buffer to handle partial data reception
+
+ if (error)
+ {
+ if (error == boost::system::errc::operation_canceled) {
+ // async receive has been explicitly cancelled (e.g., socket close)
+ return;
+ }
+
+ m_socket.close(); // closing at this point may not be that necessary
+ m_transport.m_isConnected = true;
+ throw Transport::Error(error, "error while receiving data from socket");
+ }
+
+ if (!error && bytes_recvd > 0)
+ {
+ // m_inputBuffer has bytes_recvd received bytes of data
+ if (m_partialDataSize > 0)
+ {
+ size_t newDataSize = std::min(bytes_recvd, MAX_LENGTH-m_partialDataSize);
+ std::copy(m_inputBuffer, m_inputBuffer + newDataSize, m_partialData + m_partialDataSize);
+
+ m_partialDataSize += newDataSize;
+
+ size_t offset = 0;
+ try
+ {
+ processAll(m_partialData, offset, m_partialDataSize);
+
+ // no exceptions => processed the whole thing
+ if (bytes_recvd - newDataSize > 0)
+ {
+ // there is a little bit more data available
+
+ offset = 0;
+ m_partialDataSize = bytes_recvd - newDataSize;
+ std::copy(m_inputBuffer + newDataSize, m_inputBuffer + newDataSize + m_partialDataSize, m_partialData);
+
+ processAll(m_partialData, offset, m_partialDataSize);
+
+ // no exceptions => processed the whole thing
+ m_partialDataSize = 0;
+ }
+ else
+ {
+ // done processing
+ m_partialDataSize = 0;
+ }
+ }
+ catch(Tlv::Error &)
+ {
+ if (offset > 0)
+ {
+ m_partialDataSize -= offset;
+ std::copy(m_partialData + offset, m_partialData + offset + m_partialDataSize, m_partialData);
+ }
+ else if (offset == 0 && m_partialDataSize == MAX_LENGTH)
+ {
+ // very bad... should close connection
+ m_socket.close();
+ m_transport.m_isConnected = true;
+ throw Transport::Error(boost::system::error_code(),
+ "input buffer full, but a valid TLV cannot be decoded");
+ }
+ }
+ }
+ else
+ {
+ size_t offset = 0;
+ try
+ {
+ processAll(m_inputBuffer, offset, bytes_recvd);
+ }
+ catch(Tlv::Error &error)
+ {
+ if (offset > 0)
+ {
+ m_partialDataSize = bytes_recvd - offset;
+ std::copy(m_inputBuffer + offset, m_inputBuffer + offset + m_partialDataSize, m_partialData);
+ }
+ }
+ }
+ }
+
+ m_socket.async_receive(boost::asio::buffer(m_inputBuffer, MAX_LENGTH), 0,
+ bind(&impl::handle_async_receive, this, _1, _2));
+ }
+
+ void
+ handle_async_send(const boost::system::error_code& error, const Block& wire)
+ {
+ // pass (needed to keep data block alive during the send)
+ }
+
+ void
+ handle_async_send(const boost::system::error_code& error,
+ const Block& header, const Block& payload)
+ {
+ // pass (needed to keep data blocks alive during the send)
+ }
+
+protected:
+ base_transport& m_transport;
+
+ typename protocol::socket m_socket;
+ uint8_t m_inputBuffer[MAX_LENGTH];
+
+ uint8_t m_partialData[MAX_LENGTH];
+ size_t m_partialDataSize;
+
+ std::list< Block > m_sendQueue;
+ std::list< std::pair<Block, Block> > m_sendPairQueue;
+ bool m_connectionInProgress;
+
+ boost::asio::deadline_timer m_connectTimer;
+};
+
+
+template<class BaseTransport, class Protocol>
+class StreamTransportWithResolverImpl : public StreamTransportImpl<BaseTransport, Protocol>
+{
+public:
+ typedef BaseTransport base_transport;
+ typedef Protocol protocol;
+ typedef StreamTransportWithResolverImpl<BaseTransport,Protocol> impl;
+
+ StreamTransportWithResolverImpl(base_transport& transport, boost::asio::io_service& ioService)
+ : StreamTransportImpl<base_transport, protocol>(transport, ioService)
+ {
+ }
+
+ void
+ resolveHandler(const boost::system::error_code& error,
+ typename protocol::resolver::iterator endpoint,
+ const shared_ptr<typename protocol::resolver>&)
+ {
+ if (error)
+ {
+ if (error == boost::system::errc::operation_canceled)
+ return;
+
+ throw Transport::Error(error, "Error during resolution of host or port");
+ }
+
+ typename protocol::resolver::iterator end;
+ if (endpoint == end)
+ {
+ this->m_connectionInProgress = false;
+ this->m_transport.m_isConnected = false;
+ this->m_socket.close();
+ throw Transport::Error(error, "Unable to resolve because host or port");
+ }
+
+ this->m_socket.async_connect(*endpoint,
+ bind(&impl::connectHandler, this, _1));
+ }
+
+ void
+ connect(const typename protocol::resolver::query& query)
+ {
+ if (!this->m_connectionInProgress) {
+ this->m_connectionInProgress = true;
+
+ // Wait at most 4 seconds to connect
+ /// @todo Decide whether this number should be configurable
+ this->m_connectTimer.expires_from_now(boost::posix_time::seconds(4));
+ this->m_connectTimer.async_wait(bind(&impl::connectTimeoutHandler, this, _1));
+
+ // typename boost::asio::ip::basic_resolver< protocol > resolver;
+ shared_ptr<typename protocol::resolver> resolver =
+ make_shared<typename protocol::resolver>(boost::ref(this->m_socket.get_io_service()));
+
+ resolver->async_resolve(query, bind(&impl::resolveHandler, this, _1, _2, resolver));
+ }
+ }
+};
+
+
+} // namespace ndn
+
+#endif // NDN_TRANSPORT_STREAM_TRANSPORT_HPP
diff --git a/src/transport/tcp-transport.cpp b/src/transport/tcp-transport.cpp
index 3c773a8..24ec85c 100644
--- a/src/transport/tcp-transport.cpp
+++ b/src/transport/tcp-transport.cpp
@@ -1,271 +1,19 @@
/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil -*- */
/**
- * Copyright (C) 2013 Regents of the University of California.
- * @author: Jeff Thompson <jefft0@remap.ucla.edu>
+ * Copyright (C) 2013-2014 Regents of the University of California.
* See COPYING for copyright and distribution information.
*/
#include "common.hpp"
#include "tcp-transport.hpp"
-
-#if NDN_CPP_HAVE_CXX11
-// In the std library, the placeholders are in a different namespace than boost.
-using namespace ndn::func_lib::placeholders;
-#endif
-
-using namespace std;
-typedef boost::asio::ip::tcp protocol;
+#include "stream-transport.hpp"
namespace ndn {
-const size_t MAX_LENGTH = 9000;
-
-class TcpTransport::Impl
-{
-public:
- Impl(TcpTransport &transport)
- : transport_(transport)
- , socket_(*transport_.ioService_)
- , partialDataSize_(0)
- , connectionInProgress_(false)
- , connectTimer_(*transport_.ioService_)
- {
- }
-
- void
- connectHandler(const boost::system::error_code& error)
- {
- connectionInProgress_ = false;
- connectTimer_.cancel();
-
- if (!error)
- {
- partialDataSize_ = 0;
- socket_.async_receive(boost::asio::buffer(inputBuffer_, MAX_LENGTH), 0,
- func_lib::bind(&Impl::handle_async_receive, this, _1, _2));
-
- transport_.isConnected_ = true;
-
- for (std::list<Block>::iterator i = sendQueue_.begin(); i != sendQueue_.end(); ++i)
- socket_.async_send(boost::asio::buffer(i->wire(), i->size()),
- func_lib::bind(&Impl::handle_async_send, this, _1, *i));
-
- sendQueue_.clear();
- }
- else
- {
- // may need to throw exception
- transport_.isConnected_ = false;
- transport_.close();
- throw Transport::Error(error, "error while connecting to the forwarder");
- }
- }
-
- void
- connectTimeoutHandler(const boost::system::error_code& error)
- {
- if (error) // e.g., cancelled timer
- return;
-
- connectionInProgress_ = false;
- transport_.isConnected_ = false;
- socket_.close();
- throw Transport::Error(error, "error while connecting to the forwarder");
- }
-
- void
- resolveHandler(const boost::system::error_code& error,
- boost::asio::ip::tcp::resolver::iterator endpoint,
- const ptr_lib::shared_ptr<boost::asio::ip::tcp::resolver>&)
- {
- if (error)
- {
- if (error == boost::system::errc::operation_canceled)
- return;
-
- throw Transport::Error(error, "Error during resolution of host or port [" + transport_.host_ + ":" + transport_.port_ + "]");
- }
-
- boost::asio::ip::tcp::resolver::iterator end;
- if (endpoint == end)
- {
- connectionInProgress_ = false;
- transport_.isConnected_ = false;
- socket_.close();
- throw Transport::Error(error, "Unable to connect because host or port [" + transport_.host_ + ":" + transport_.port_ + "] cannot be resolved");
- }
-
- socket_.async_connect(*endpoint,
- func_lib::bind(&Impl::connectHandler, this, _1));
- }
-
- void
- connect()
- {
- if (!connectionInProgress_) {
- connectionInProgress_ = true;
-
- // Wait at most 4 seconds to connect
- /// @todo Decide whether this number should be configurable
- connectTimer_.expires_from_now(boost::posix_time::seconds(4));
- connectTimer_.async_wait(func_lib::bind(&Impl::connectTimeoutHandler, this, _1));
-
- using boost::asio::ip::tcp;
-
- ptr_lib::shared_ptr<tcp::resolver> resolver =
- ptr_lib::make_shared<tcp::resolver>(boost::ref(*transport_.ioService_));
-
- tcp::resolver::query query(transport_.host_, transport_.port_);
-
- resolver->async_resolve(query, func_lib::bind(&Impl::resolveHandler, this, _1, _2, resolver));
- }
- }
-
- void
- close()
- {
- connectTimer_.cancel();
- socket_.close();
- transport_.isConnected_ = false;
- }
-
- void
- send(const Block &wire)
- {
- if (!transport_.isConnected_)
- sendQueue_.push_back(wire);
- else
- socket_.async_send(boost::asio::buffer(wire.wire(), wire.size()),
- func_lib::bind(&Impl::handle_async_send, this, _1, wire));
- }
-
- inline void
- processAll(uint8_t *buffer, size_t &offset, size_t availableSize)
- {
- while(offset < availableSize)
- {
- Block element(buffer + offset, availableSize - offset);
- transport_.receive(element);
-
- offset += element.size();
- }
- }
-
- void
- handle_async_receive(const boost::system::error_code& error, std::size_t bytes_recvd)
- {
- /// @todo The socket is not datagram, so need to have internal buffer to handle partial data reception
-
- if (error)
- {
- if (error == boost::system::errc::operation_canceled) {
- // async receive has been explicitly cancelled (e.g., socket close)
- return;
- }
-
- socket_.close(); // closing at this point may not be that necessary
- transport_.isConnected_ = true;
- throw Transport::Error(error, "error while receiving data from socket");
- }
-
- if (!error && bytes_recvd > 0)
- {
- // inputBuffer_ has bytes_recvd received bytes of data
- if (partialDataSize_ > 0)
- {
- size_t newDataSize = std::min(bytes_recvd, MAX_LENGTH-partialDataSize_);
- std::copy(inputBuffer_, inputBuffer_ + newDataSize, partialData_ + partialDataSize_);
-
- partialDataSize_ += newDataSize;
-
- size_t offset = 0;
- try
- {
- processAll(partialData_, offset, partialDataSize_);
-
- // no exceptions => processed the whole thing
- if (bytes_recvd - newDataSize > 0)
- {
- // there is a little bit more data available
-
- offset = 0;
- partialDataSize_ = bytes_recvd - newDataSize;
- std::copy(inputBuffer_ + newDataSize, inputBuffer_ + newDataSize + partialDataSize_, partialData_);
-
- processAll(partialData_, offset, partialDataSize_);
-
- // no exceptions => processed the whole thing
- partialDataSize_ = 0;
- }
- else
- {
- // done processing
- partialDataSize_ = 0;
- }
- }
- catch(Tlv::Error &)
- {
- if (offset > 0)
- {
- partialDataSize_ -= offset;
- std::copy(partialData_ + offset, partialData_ + offset + partialDataSize_, partialData_);
- }
- else if (offset == 0 && partialDataSize_ == MAX_LENGTH)
- {
- // very bad... should close connection
- socket_.close();
- transport_.isConnected_ = true;
- throw Transport::Error(boost::system::error_code(), "input buffer full, but a valid TLV cannot be decoded");
- }
- }
- }
- else
- {
- size_t offset = 0;
- try
- {
- processAll(inputBuffer_, offset, bytes_recvd);
- }
- catch(Tlv::Error &error)
- {
- if (offset > 0)
- {
- partialDataSize_ = bytes_recvd - offset;
- std::copy(inputBuffer_ + offset, inputBuffer_ + offset + partialDataSize_, partialData_);
- }
- }
- }
- }
-
- socket_.async_receive(boost::asio::buffer(inputBuffer_, MAX_LENGTH), 0,
- func_lib::bind(&Impl::handle_async_receive, this, _1, _2));
- }
-
- void
- handle_async_send(const boost::system::error_code& error, const Block &wire)
- {
- // pass (needed to keep data block alive during the send)
- }
-
-private:
- TcpTransport &transport_;
-
- protocol::socket socket_;
- uint8_t inputBuffer_[MAX_LENGTH];
-
- uint8_t partialData_[MAX_LENGTH];
- size_t partialDataSize_;
-
- std::list< Block > sendQueue_;
- bool connectionInProgress_;
-
- boost::asio::deadline_timer connectTimer_;
-};
-
TcpTransport::TcpTransport(const std::string& host, const std::string& port/* = "6363"*/)
- : host_(host)
- , port_(port)
+ : m_host(host)
+ , m_port(port)
{
}
@@ -274,27 +22,36 @@
}
void
-TcpTransport::connect(boost::asio::io_service &ioService,
- const ReceiveCallback &receiveCallback)
+TcpTransport::connect(boost::asio::io_service& ioService,
+ const ReceiveCallback& receiveCallback)
{
- if (!static_cast<bool>(impl_)) {
+ if (!static_cast<bool>(m_impl)) {
Transport::connect(ioService, receiveCallback);
- impl_ = ptr_lib::make_shared<TcpTransport::Impl> (ptr_lib::ref(*this));
+ m_impl = make_shared<Impl> (boost::ref(*this),
+ boost::ref(ioService));
}
- impl_->connect();
+
+ boost::asio::ip::tcp::resolver::query query(m_host, m_port);
+ m_impl->connect(query);
}
void
-TcpTransport::send(const Block &wire)
+TcpTransport::send(const Block& wire)
{
- impl_->send(wire);
+ m_impl->send(wire);
+}
+
+void
+TcpTransport::send(const Block& header, const Block& payload)
+{
+ m_impl->send(header, payload);
}
void
TcpTransport::close()
{
- impl_->close();
+ m_impl->close();
}
-}
+} // namespace ndn
diff --git a/src/transport/tcp-transport.hpp b/src/transport/tcp-transport.hpp
index c2ba56e..b97e2a1 100644
--- a/src/transport/tcp-transport.hpp
+++ b/src/transport/tcp-transport.hpp
@@ -1,18 +1,24 @@
/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil -*- */
/**
- * Copyright (C) 2013 Regents of the University of California.
- * @author: Jeff Thompson <jefft0@remap.ucla.edu>
+ * Copyright (C) 2013-2014 Regents of the University of California.
* See COPYING for copyright and distribution information.
*/
-#ifndef NDN_TCP_TRANSPORT_HPP
-#define NDN_TCP_TRANSPORT_HPP
+#ifndef NDN_TRANSPORT_TCP_TRANSPORT_HPP
+#define NDN_TRANSPORT_TCP_TRANSPORT_HPP
#include "../common.hpp"
#include "transport.hpp"
+// forward declaration
+namespace boost { namespace asio { namespace ip { class tcp; } } }
+
namespace ndn {
-
+
+// forward declaration
+template<class T, class U> class StreamTransportImpl;
+template<class T, class U> class StreamTransportWithResolverImpl;
+
class TcpTransport : public Transport
{
public:
@@ -21,23 +27,28 @@
// from Transport
virtual void
- connect(boost::asio::io_service &ioService,
- const ReceiveCallback &receiveCallback);
+ connect(boost::asio::io_service& ioService,
+ const ReceiveCallback& receiveCallback);
virtual void
close();
virtual void
- send(const Block &wire);
-
-private:
- std::string host_;
- std::string port_;
+ send(const Block& wire);
- class Impl;
- ptr_lib::shared_ptr<Impl> impl_;
+ virtual void
+ send(const Block& header, const Block& payload);
+
+private:
+ std::string m_host;
+ std::string m_port;
+
+ typedef StreamTransportWithResolverImpl<TcpTransport, boost::asio::ip::tcp> Impl;
+ friend class StreamTransportImpl<TcpTransport, boost::asio::ip::tcp>;
+ friend class StreamTransportWithResolverImpl<TcpTransport, boost::asio::ip::tcp>;
+ shared_ptr< Impl > m_impl;
};
-}
+} // namespace ndn
-#endif // NDN_TCP_TRANSPORT_HPP
+#endif // NDN_TRANSPORT_TCP_TRANSPORT_HPP
diff --git a/src/transport/transport.hpp b/src/transport/transport.hpp
index 3b62a21..c309288 100644
--- a/src/transport/transport.hpp
+++ b/src/transport/transport.hpp
@@ -1,12 +1,11 @@
/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil -*- */
/**
* Copyright (C) 2013 Regents of the University of California.
- * @author: Jeff Thompson <jefft0@remap.ucla.edu>
* See COPYING for copyright and distribution information.
*/
-#ifndef NDN_TRANSPORT_HPP
-#define NDN_TRANSPORT_HPP
+#ifndef NDN_TRANSPORT_TRANSPORT_HPP
+#define NDN_TRANSPORT_TRANSPORT_HPP
#include "../common.hpp"
#include "../encoding/block.hpp"
@@ -32,44 +31,54 @@
* @throws If connection cannot be established
*/
inline virtual void
- connect(boost::asio::io_service &io_service,
- const ReceiveCallback &receiveCallback);
+ connect(boost::asio::io_service& io_service,
+ const ReceiveCallback& receiveCallback);
/**
* Close the connection.
*/
virtual void
- close() = 0;
+ close() =0;
/**
- * Set data to the host
+ * @brief Set data to the host
+ *
* @param data A pointer to the buffer of data to send.
* @param dataLength The number of bytes in data.
*/
virtual void
- send(const Block &wire) = 0;
+ send(const Block& wire) =0;
+ /**
+ * @brief Alternative version of sending data, applying scatter/gather I/O concept
+ *
+ * Two non-consecutive memory blocks will be send out together, e.g., as part of the
+ * same message in datagram-oriented transports.
+ */
+ virtual void
+ send(const Block& header, const Block& payload) =0;
+
inline bool
isConnected();
protected:
inline void
- receive(const Block &wire);
+ receive(const Block& wire);
protected:
- boost::asio::io_service *ioService_;
- bool isConnected_;
- ReceiveCallback receiveCallback_;
+ boost::asio::io_service* m_ioService;
+ bool m_isConnected;
+ ReceiveCallback m_receiveCallback;
};
inline
Transport::Transport()
- : ioService_(0)
- , isConnected_(false)
+ : m_ioService(0)
+ , m_isConnected(false)
{
}
-inline Transport::Error::Error(const boost::system::error_code &code, const std::string &msg)
+inline Transport::Error::Error(const boost::system::error_code& code, const std::string& msg)
: std::runtime_error(msg + (code.value() ? " (" + code.category().message(code.value()) + ")" : ""))
{
}
@@ -80,25 +89,25 @@
}
inline void
-Transport::connect(boost::asio::io_service &ioService,
- const ReceiveCallback &receiveCallback)
+Transport::connect(boost::asio::io_service& ioService,
+ const ReceiveCallback& receiveCallback)
{
- ioService_ = &ioService;
- receiveCallback_ = receiveCallback;
+ m_ioService = &ioService;
+ m_receiveCallback = receiveCallback;
}
inline bool
Transport::isConnected()
{
- return isConnected_;
+ return m_isConnected;
}
inline void
-Transport::receive(const Block &wire)
+Transport::receive(const Block& wire)
{
- receiveCallback_(wire);
+ m_receiveCallback(wire);
}
-}
+} // namespace ndn
-#endif
+#endif // NDN_TRANSPORT_TRANSPORT_HPP
diff --git a/src/transport/unix-transport.cpp b/src/transport/unix-transport.cpp
index 84424f3..f1e26ec 100644
--- a/src/transport/unix-transport.cpp
+++ b/src/transport/unix-transport.cpp
@@ -1,246 +1,28 @@
/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil -*- */
/**
- * Copyright (C) 2013 Regents of the University of California.
- * @author: Jeff Thompson <jefft0@remap.ucla.edu>
+ * Copyright (C) 2013-2014 Regents of the University of California.
* See COPYING for copyright and distribution information.
*/
#include "common.hpp"
#include "unix-transport.hpp"
+#include "stream-transport.hpp"
#include "../face.hpp"
-#include <cstdlib>
-
-using namespace std;
-typedef boost::asio::local::stream_protocol protocol;
namespace ndn {
-const size_t MAX_LENGTH = 9000;
-
-class UnixTransport::Impl
-{
-public:
- Impl(UnixTransport &transport)
- : transport_(transport)
- , socket_(*transport_.ioService_)
- , partialDataSize_(0)
- , connectionInProgress_(false)
- , connectTimer_(*transport_.ioService_)
- {
- }
-
- void
- connectHandler(const boost::system::error_code& error)
- {
- connectionInProgress_ = false;
- connectTimer_.cancel();
-
- if (!error)
- {
- partialDataSize_ = 0;
- socket_.async_receive(boost::asio::buffer(inputBuffer_, MAX_LENGTH), 0,
- func_lib::bind(&Impl::handle_async_receive, this, _1, _2));
-
- transport_.isConnected_ = true;
-
- for (std::list<Block>::iterator i = sendQueue_.begin(); i != sendQueue_.end(); ++i)
- socket_.async_send(boost::asio::buffer(i->wire(), i->size()),
- func_lib::bind(&Impl::handle_async_send, this, _1, *i));
-
- sendQueue_.clear();
- }
- else
- {
- // may need to throw exception
- transport_.isConnected_ = false;
- transport_.close();
- throw Transport::Error(error, "error while connecting to the forwarder");
- }
- }
-
- void
- connectTimeoutHandler(const boost::system::error_code& error)
- {
- if (error) // e.g., cancelled timer
- return;
-
- connectionInProgress_ = false;
- transport_.isConnected_ = false;
- socket_.close();
- throw Transport::Error(error, "error while connecting to the forwarder");
- }
-
- void
- connect()
- {
- if (!connectionInProgress_) {
- connectionInProgress_ = true;
-
- // Wait at most 4 seconds to connect
- /// @todo Decide whether this number should be configurable
- connectTimer_.expires_from_now(boost::posix_time::seconds(4));
- connectTimer_.async_wait(func_lib::bind(&Impl::connectTimeoutHandler, this, _1));
-
- socket_.open();
- socket_.async_connect(protocol::endpoint(transport_.unixSocket_),
- func_lib::bind(&Impl::connectHandler, this, _1));
- }
- }
-
- void
- close()
- {
- connectTimer_.cancel();
- socket_.close();
- transport_.isConnected_ = false;
- }
-
- void
- send(const Block& wire)
- {
- if (!transport_.isConnected_)
- sendQueue_.push_back(wire);
- else
- socket_.async_send(boost::asio::buffer(wire.wire(), wire.size()),
- func_lib::bind(&Impl::handle_async_send, this, _1, wire));
- }
-
- inline void
- processAll(uint8_t* buffer, size_t& offset, size_t availableSize)
- {
- while(offset < availableSize)
- {
- Block element(buffer + offset, availableSize - offset);
- transport_.receive(element);
-
- offset += element.size();
- }
- }
-
- void
- handle_async_receive(const boost::system::error_code& error, std::size_t bytes_recvd)
- {
- /// @todo The socket is not datagram, so need to have internal buffer to handle partial data reception
-
- if (error)
- {
- if (error == boost::system::errc::operation_canceled) {
- // async receive has been explicitly cancelled (e.g., socket close)
- return;
- }
-
- socket_.close(); // closing at this point may not be that necessary
- transport_.isConnected_ = true;
- throw Transport::Error(error, "error while receiving data from socket");
- }
-
- if (!error && bytes_recvd > 0)
- {
- // inputBuffer_ has bytes_recvd received bytes of data
- if (partialDataSize_ > 0)
- {
- size_t newDataSize = std::min(bytes_recvd, MAX_LENGTH-partialDataSize_);
- std::copy(inputBuffer_, inputBuffer_ + newDataSize, partialData_ + partialDataSize_);
-
- partialDataSize_ += newDataSize;
-
- size_t offset = 0;
- try
- {
- processAll(partialData_, offset, partialDataSize_);
-
- // no exceptions => processed the whole thing
- if (bytes_recvd - newDataSize > 0)
- {
- // there is a little bit more data available
-
- offset = 0;
- partialDataSize_ = bytes_recvd - newDataSize;
- std::copy(inputBuffer_ + newDataSize, inputBuffer_ + newDataSize + partialDataSize_, partialData_);
-
- processAll(partialData_, offset, partialDataSize_);
-
- // no exceptions => processed the whole thing
- partialDataSize_ = 0;
- }
- else
- {
- // done processing
- partialDataSize_ = 0;
- }
- }
- catch(Tlv::Error &)
- {
- if (offset > 0)
- {
- partialDataSize_ -= offset;
- std::copy(partialData_ + offset, partialData_ + offset + partialDataSize_, partialData_);
- }
- else if (offset == 0 && partialDataSize_ == MAX_LENGTH)
- {
- // very bad... should close connection
- socket_.close();
- transport_.isConnected_ = true;
- throw Transport::Error(boost::system::error_code(), "input buffer full, but a valid TLV cannot be decoded");
- }
- }
- }
- else
- {
- size_t offset = 0;
- try
- {
- processAll(inputBuffer_, offset, bytes_recvd);
- }
- catch(Tlv::Error &error)
- {
- if (offset > 0)
- {
- partialDataSize_ = bytes_recvd - offset;
- std::copy(inputBuffer_ + offset, inputBuffer_ + offset + partialDataSize_, partialData_);
- }
- }
- }
- }
-
- socket_.async_receive(boost::asio::buffer(inputBuffer_, MAX_LENGTH), 0,
- func_lib::bind(&Impl::handle_async_receive, this, _1, _2));
- }
-
- void
- handle_async_send(const boost::system::error_code& error, const Block &wire)
- {
- // pass (needed to keep data block alive during the send)
- }
-
-private:
- UnixTransport &transport_;
-
- protocol::socket socket_;
- uint8_t inputBuffer_[MAX_LENGTH];
-
- uint8_t partialData_[MAX_LENGTH];
- size_t partialDataSize_;
-
- std::list< Block > sendQueue_;
- bool connectionInProgress_;
-
- boost::asio::deadline_timer connectTimer_;
-};
-
-
UnixTransport::UnixTransport()
{
if (std::getenv("NFD") != 0)
- unixSocket_ = "/var/run/nfd.sock";
+ m_unixSocket = "/var/run/nfd.sock";
else
- unixSocket_ = "/tmp/.ndnd.sock";
+ m_unixSocket = "/tmp/.ndnd.sock";
}
UnixTransport::UnixTransport(const std::string& unixSocket)
- : unixSocket_(unixSocket)
+ : m_unixSocket(unixSocket)
{
}
@@ -252,24 +34,32 @@
UnixTransport::connect(boost::asio::io_service& ioService,
const ReceiveCallback& receiveCallback)
{
- if (!static_cast<bool>(impl_)) {
+ if (!static_cast<bool>(m_impl)) {
Transport::connect(ioService, receiveCallback);
- impl_ = ptr_lib::make_shared<UnixTransport::Impl> (ptr_lib::ref(*this));
+ m_impl = make_shared<Impl> (boost::ref(*this),
+ boost::ref(ioService));
}
- impl_->connect();
+
+ m_impl->connect(boost::asio::local::stream_protocol::endpoint(m_unixSocket));
}
void
UnixTransport::send(const Block& wire)
{
- impl_->send(wire);
+ m_impl->send(wire);
+}
+
+void
+UnixTransport::send(const Block& header, const Block& payload)
+{
+ m_impl->send(header, payload);
}
void
UnixTransport::close()
{
- impl_->close();
+ m_impl->close();
}
}
diff --git a/src/transport/unix-transport.hpp b/src/transport/unix-transport.hpp
index dc85eac..4a89bc8 100644
--- a/src/transport/unix-transport.hpp
+++ b/src/transport/unix-transport.hpp
@@ -1,7 +1,6 @@
/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil -*- */
/**
- * Copyright (C) 2013 Regents of the University of California.
- * @author: Jeff Thompson <jefft0@remap.ucla.edu>
+ * Copyright (C) 2013-2014 Regents of the University of California.
* See COPYING for copyright and distribution information.
*/
@@ -11,8 +10,15 @@
#include "../common.hpp"
#include "transport.hpp"
+// forward declaration
+namespace boost { namespace asio { namespace local { class stream_protocol; } } }
+
namespace ndn {
+// forward declaration
+template<class T, class U>
+class StreamTransportImpl;
+
class UnixTransport : public Transport
{
public:
@@ -32,11 +38,15 @@
virtual void
send(const Block& wire);
+ virtual void
+ send(const Block& header, const Block& payload);
+
private:
- std::string unixSocket_;
+ std::string m_unixSocket;
- class Impl;
- ptr_lib::shared_ptr<Impl> impl_;
+ typedef StreamTransportImpl<UnixTransport, boost::asio::local::stream_protocol> Impl;
+ friend class StreamTransportImpl<UnixTransport, boost::asio::local::stream_protocol>;
+ ptr_lib::shared_ptr< Impl > m_impl;
};
}