26 #ifndef NFD_DAEMON_FACE_STREAM_TRANSPORT_HPP 27 #define NFD_DAEMON_FACE_STREAM_TRANSPORT_HPP 42 template<
class Protocol>
66 doSend(
const Block& packet)
override;
72 handleSend(
const boost::system::error_code& error,
80 size_t nBytesReceived);
86 handleError(
const boost::system::error_code& error);
103 uint8_t m_receiveBuffer[ndn::MAX_NDN_PACKET_SIZE];
104 size_t m_receiveBufferSize;
105 std::queue<Block> m_sendQueue;
106 size_t m_sendQueueBytes;
113 , m_receiveBufferSize(0)
114 , m_sendQueueBytes(0)
129 NFD_LOG_FACE_WARN(
"Failed to obtain send queue length from socket: " << std::strerror(errno));
131 return getSendQueueBytes() + std::max<ssize_t>(0, queueLength);
140 if (m_socket.is_open()) {
144 boost::system::error_code error;
145 m_socket.cancel(error);
146 m_socket.shutdown(protocol::socket::shutdown_both, error);
175 boost::system::error_code error;
176 m_socket.close(error);
178 this->setState(TransportState::CLOSED);
187 if (getState() != TransportState::UP)
190 bool wasQueueEmpty = m_sendQueue.empty();
191 m_sendQueue.push(packet);
192 m_sendQueueBytes += packet.size();
202 boost::asio::async_write(m_socket, boost::asio::buffer(m_sendQueue.front()),
203 [
this] (
auto&&... args) { this->handleSend(std::forward<decltype(args)>(args)...); });
212 return processErrorCode(error);
216 BOOST_ASSERT(!m_sendQueue.empty());
217 BOOST_ASSERT(m_sendQueue.front().size() == nBytesSent);
218 m_sendQueueBytes -= nBytesSent;
221 if (!m_sendQueue.empty())
229 BOOST_ASSERT(getState() == TransportState::UP);
231 m_socket.async_receive(boost::asio::buffer(m_receiveBuffer + m_receiveBufferSize,
232 ndn::MAX_NDN_PACKET_SIZE - m_receiveBufferSize),
233 [
this] (
auto&&... args) { this->handleReceive(std::forward<decltype(args)>(args)...); });
239 size_t nBytesReceived)
242 return processErrorCode(error);
246 m_receiveBufferSize += nBytesReceived;
249 while (m_receiveBufferSize - offset > 0) {
251 std::tie(isOk, element) = Block::fromBuffer(m_receiveBuffer + offset, m_receiveBufferSize - offset);
255 offset += element.size();
256 BOOST_ASSERT(offset <= m_receiveBufferSize);
258 this->receive(element);
261 if (!isOk && m_receiveBufferSize == ndn::MAX_NDN_PACKET_SIZE && offset == 0) {
263 this->setState(TransportState::FAILED);
269 if (offset != m_receiveBufferSize) {
270 std::copy(m_receiveBuffer + offset, m_receiveBuffer + m_receiveBufferSize, m_receiveBuffer);
271 m_receiveBufferSize -= offset;
274 m_receiveBufferSize = 0;
287 if (getState() == TransportState::CLOSING ||
288 getState() == TransportState::FAILED ||
289 getState() == TransportState::CLOSED ||
290 error == boost::asio::error::operation_aborted ||
291 error == boost::asio::error::shut_down)
302 if (error == boost::asio::error::eof) {
303 this->setState(TransportState::CLOSING);
307 this->setState(TransportState::FAILED);
316 m_receiveBufferSize = 0;
323 std::queue<Block> emptyQueue;
324 std::swap(emptyQueue, m_sendQueue);
325 m_sendQueueBytes = 0;
332 return m_sendQueueBytes;
338 #endif // NFD_DAEMON_FACE_STREAM_TRANSPORT_HPP void resetReceiveBuffer()
#define NFD_LOG_FACE_TRACE(msg)
Log a message at TRACE level.
const ssize_t QUEUE_ERROR
indicates that the transport was unable to retrieve the queue capacity/length
#define NFD_LOG_MEMBER_DECL()
The lower half of a Face.
Implements Transport for stream-based protocols.
void handleReceive(const boost::system::error_code &error, size_t nBytesReceived)
#define NFD_LOG_FACE_WARN(msg)
Log a message at WARN level.
Copyright (c) 2014-2015, Regents of the University of California, Arizona Board of Regents...
StreamTransport(typename protocol::socket &&socket)
Construct stream transport.
ssize_t getTxQueueLength(int fd)
obtain send queue length from a specified system socket
void processErrorCode(const boost::system::error_code &error)
void doClose() override
performs Transport specific operations to close the transport
size_t getSendQueueBytes() const
#define NFD_LOG_FACE_ERROR(msg)
Log a message at ERROR level.
void handleSend(const boost::system::error_code &error, size_t nBytesSent)
ssize_t getSendQueueLength() override
virtual void handleError(const boost::system::error_code &error)
protocol::socket m_socket
void doSend(const Block &packet) override
performs Transport specific operations to send a packet
boost::asio::io_service & getGlobalIoService()
Returns the global io_service instance for the calling thread.