From d32cb9657c36d975dcb4106fc9df00a8ddfd34be Mon Sep 17 00:00:00 2001 From: Alexander Afanasyev Date: Tue, 28 Jan 2014 12:43:47 -0800 Subject: [PATCH] face/tcp: Implementing mock version of TcpFace The real version should avoid all exceptions in the processing path, enable lazy decoding of Interest/Data packets, and implement LocalControlHeader processing and association with Interest/Data packets. Change-Id: Ied907b3d10b7110a1dcc05d0d6d3c0dd277df8e1 refs: #1132, #1133, #1134, #1135 --- daemon/common.hpp | 2 + daemon/face/face.hpp | 7 +- daemon/face/stream-face.cpp | 8 --- daemon/face/stream-face.hpp | 130 ++++++++++++++++++++++++++++++++++-- daemon/face/tcp-face.cpp | 16 ++++- daemon/face/tcp-face.hpp | 8 +-- 6 files changed, 150 insertions(+), 21 deletions(-) delete mode 100644 daemon/face/stream-face.cpp diff --git a/daemon/common.hpp b/daemon/common.hpp index 9b50fda7..a98719d7 100644 --- a/daemon/common.hpp +++ b/daemon/common.hpp @@ -14,6 +14,7 @@ #include #include +#include #include #include #include @@ -28,6 +29,7 @@ namespace ndn { using boost::noncopyable; using boost::shared_ptr; +using boost::enable_shared_from_this; using boost::make_shared; using boost::function; using boost::bind; diff --git a/daemon/face/face.hpp b/daemon/face/face.hpp index 8022e3dc..f014545d 100644 --- a/daemon/face/face.hpp +++ b/daemon/face/face.hpp @@ -17,10 +17,12 @@ namespace ndn { */ typedef int FaceId; +const std::size_t MAX_NDN_PACKET_SIZE = 8800; + /** \class Face * \brief represents a face */ -class Face : noncopyable +class Face : noncopyable, public enable_shared_from_this { public: Face(FaceId id); @@ -33,6 +35,9 @@ public: /// fires when a Data is received EventEmitter onReceiveData; + + /// fires when face disconnects or fails to perform properly + EventEmitter onFail; /// send an Interest virtual void diff --git a/daemon/face/stream-face.cpp b/daemon/face/stream-face.cpp deleted file mode 100644 index e84634db..00000000 --- a/daemon/face/stream-face.cpp +++ /dev/null @@ -1,8 +0,0 @@ -/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */ -/** - * Copyright (C) 2014 Named Data Networking Project - * See COPYING for copyright and distribution information. - */ - -#include "stream-face.hpp" - diff --git a/daemon/face/stream-face.hpp b/daemon/face/stream-face.hpp index 778eb710..8fd39a12 100644 --- a/daemon/face/stream-face.hpp +++ b/daemon/face/stream-face.hpp @@ -17,13 +17,133 @@ class StreamFace : public Face public: typedef T protocol; - StreamFace(FaceId id) - : Face(id) - { - } - + StreamFace(FaceId id, + const shared_ptr& socket); + +protected: + void + handleSend(const boost::system::error_code& error, + const Block& wire); + + void + handleReceive(const boost::system::error_code& error, + std::size_t bytes_recvd); + +protected: + shared_ptr m_socket; + +private: + uint8_t m_inputBuffer[MAX_NDN_PACKET_SIZE]; + std::size_t m_inputBufferSize; }; +template +inline +StreamFace::StreamFace(FaceId id, + const shared_ptr& socket) + : Face(id) + , m_socket(socket) +{ + m_socket->async_receive(boost::asio::buffer(m_inputBuffer, MAX_NDN_PACKET_SIZE), 0, + bind(&StreamFace::handleReceive, this, _1, _2)); +} + + +template +inline void +StreamFace::handleSend(const boost::system::error_code& error, + const Block& wire) +{ + if (error) { + if (error == boost::system::errc::operation_canceled) // when socket is closed by someone + return; + + onFail("Send operation failed: " + error.category().message(error.value())); + m_socket->close(); + return; + } + + // do nothing (needed to retain validity of wire memory block +} + +template +inline void +StreamFace::handleReceive(const boost::system::error_code& error, + std::size_t bytes_recvd) +{ + if (error || bytes_recvd == 0) { + if (error == boost::system::errc::operation_canceled) // when socket is closed by someone + return; + + onFail("Receive operation failed: " + error.category().message(error.value())); + m_socket->close(); + return; + } + + m_inputBufferSize += bytes_recvd; + // do magic + + std::size_t offset = 0; + /// @todo Eliminate reliance on exceptions in this path + try { + Block element(m_inputBuffer + offset, m_inputBufferSize - offset); + offset += element.size(); + + /// @todo Ensure lazy field decoding process + if (element.type() == Tlv::Interest) + { + shared_ptr i = make_shared(); + i->wireDecode(element); + onReceiveInterest(*i); + } + else if (element.type() == Tlv::Data) + { + shared_ptr d = make_shared(); + d->wireDecode(element); + onReceiveData(*d); + } + // @todo Add local header support + // else if (element.type() == Tlv::LocalHeader) + // { + // shared_ptr i = make_shared(); + // i->wireDecode(element); + // } + else + { + /// @todo Add loggin + + // ignore unknown packet and proceed + } + } + catch(const Tlv::Error&) { + if (m_inputBufferSize == MAX_NDN_PACKET_SIZE && offset == 0) + { + onFail("Received input is invalid or too large to process, closing down the face"); + m_socket->close(); + return; + } + } + + if (offset > 0) + { + if (offset != m_inputBufferSize) + { + std::copy(m_inputBuffer + offset, m_inputBuffer + m_inputBufferSize, + m_inputBuffer); + m_inputBufferSize -= offset; + } + else + { + m_inputBufferSize = 0; + } + } + + m_socket->async_receive(boost::asio::buffer(m_inputBuffer + m_inputBufferSize, + MAX_NDN_PACKET_SIZE - m_inputBufferSize), 0, + bind(&StreamFace::handleReceive, this, _1, _2)); +} + + } // namespace ndn #endif // NFD_FACE_STREAM_FACE_HPP diff --git a/daemon/face/tcp-face.cpp b/daemon/face/tcp-face.cpp index 42d4fc77..eb3ac94e 100644 --- a/daemon/face/tcp-face.cpp +++ b/daemon/face/tcp-face.cpp @@ -8,20 +8,30 @@ namespace ndn { -TcpFace::TcpFace(FaceId id, const shared_ptr& socket) - : StreamFace(id) - , m_socket(socket) +TcpFace::TcpFace(FaceId id, + const shared_ptr& socket) + : StreamFace(id, socket) { } void TcpFace::sendInterest(const Interest& interest) { + m_socket->async_send(boost::asio::buffer(interest.wireEncode().wire(), + interest.wireEncode().size()), + bind(&TcpFace::handleSend, this, _1, interest.wireEncode())); + + // anything else should be done here? } void TcpFace::sendData(const Data& data) { + m_socket->async_send(boost::asio::buffer(data.wireEncode().wire(), + data.wireEncode().size()), + bind(&TcpFace::handleSend, this, _1, data.wireEncode())); + + // anything else should be done here? } } // namespace ndn diff --git a/daemon/face/tcp-face.hpp b/daemon/face/tcp-face.hpp index 86ec0862..3d2bb53f 100644 --- a/daemon/face/tcp-face.hpp +++ b/daemon/face/tcp-face.hpp @@ -19,7 +19,10 @@ namespace ndn class TcpFace : public StreamFace { public: - TcpFace(FaceId id, const shared_ptr& socket); + typedef boost::asio::ip::tcp protocol; + + TcpFace(FaceId id, + const shared_ptr& socket); // from Face virtual void @@ -27,9 +30,6 @@ public: virtual void sendData(const Data& data); - -private: - shared_ptr m_socket; }; } // namespace ndn