blob: 95738131d2cfff92bd54af63adbc9221638f5790 [file] [log] [blame]
/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
/**
* Copyright (C) 2014 Named Data Networking Project
* See COPYING for copyright and distribution information.
*/
#ifndef NFD_FACE_STREAM_FACE_HPP
#define NFD_FACE_STREAM_FACE_HPP
#include "face.hpp"
namespace nfd {
template <class T>
class StreamFace : public Face
{
public:
typedef T protocol;
/**
* \brief Create instance of StreamFace
*/
StreamFace(const shared_ptr<typename protocol::socket>& socket);
virtual
~StreamFace();
// from Face
virtual void
sendInterest(const Interest& interest);
virtual void
sendData(const Data& data);
virtual void
close();
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);
void
keepFaceAliveUntilAllHandlersExecuted(const shared_ptr<Face>& face);
void
closeSocket();
protected:
shared_ptr<typename protocol::socket> m_socket;
private:
uint8_t m_inputBuffer[MAX_NDN_PACKET_SIZE];
std::size_t m_inputBufferSize;
NFD_LOG_INCLASS_DECLARE();
};
// All inherited classes must use
// NFD_LOG_INCLASS_TEMPLATE_SPECIALIZATION_DEFINE(StreamFace, <specialization-parameter>, "Name");
template <class T>
inline
StreamFace<T>::StreamFace(const shared_ptr<typename StreamFace::protocol::socket>& socket)
: m_socket(socket)
, m_inputBufferSize(0)
{
m_socket->async_receive(boost::asio::buffer(m_inputBuffer, MAX_NDN_PACKET_SIZE), 0,
bind(&StreamFace<T>::handleReceive, this, _1, _2));
}
template <class T>
inline
StreamFace<T>::~StreamFace()
{
}
template <class T>
inline void
StreamFace<T>::sendInterest(const Interest& interest)
{
m_socket->async_send(boost::asio::buffer(interest.wireEncode().wire(),
interest.wireEncode().size()),
bind(&StreamFace<T>::handleSend, this, _1, interest.wireEncode()));
// anything else should be done here?
}
template <class T>
inline void
StreamFace<T>::sendData(const Data& data)
{
m_socket->async_send(boost::asio::buffer(data.wireEncode().wire(),
data.wireEncode().size()),
bind(&StreamFace<T>::handleSend, this, _1, data.wireEncode()));
// anything else should be done here?
}
template <class T>
inline void
StreamFace<T>::close()
{
if (!m_socket->is_open())
return;
NFD_LOG_INFO("[id:" << this->getId()
<< ",endpoint:" << m_socket->local_endpoint()
<< "] Close connection");
closeSocket();
onFail("Close connection");
}
template <class T>
inline void
StreamFace<T>::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;
if (!m_socket->is_open())
{
onFail("Connection closed");
return;
}
if (error == boost::asio::error::eof)
{
NFD_LOG_INFO("[id:" << this->getId()
<< ",endpoint:" << m_socket->local_endpoint()
<< "] Connection closed");
}
else
{
NFD_LOG_WARN("[id:" << this->getId()
<< ",endpoint:" << m_socket->local_endpoint()
<< "] Send operation failed, closing socket: "
<< error.category().message(error.value()));
}
closeSocket();
if (error == boost::asio::error::eof)
{
onFail("Connection closed");
}
else
{
onFail("Send operation failed, closing socket: " +
error.category().message(error.value()));
}
return;
}
NFD_LOG_TRACE("[id:" << this->getId()
<< ",endpoint:" << m_socket->local_endpoint()
<< "] Successfully sent: " << wire.size() << " bytes");
// do nothing (needed to retain validity of wire memory block
}
template <class T>
inline void
StreamFace<T>::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;
// this should be unnecessary, but just in case
if (!m_socket->is_open())
{
onFail("Connection closed");
return;
}
if (error == boost::asio::error::eof)
{
NFD_LOG_INFO("[id:" << this->getId()
<< ",endpoint:" << m_socket->local_endpoint()
<< "] Connection closed");
}
else
{
NFD_LOG_WARN("[id:" << this->getId()
<< ",endpoint:" << m_socket->local_endpoint()
<< "] Receive operation failed: "
<< error.category().message(error.value()));
}
closeSocket();
if (error == boost::asio::error::eof)
{
onFail("Connection closed");
}
else
{
onFail("Receive operation failed, closing socket: " +
error.category().message(error.value()));
}
return;
}
NFD_LOG_TRACE("[id:" << this->getId()
<< ",endpoint:" << m_socket->local_endpoint()
<< "] Received: " << bytes_recvd << " bytes");
m_inputBufferSize += bytes_recvd;
// do magic
std::size_t offset = 0;
/// @todo Eliminate reliance on exceptions in this path
try {
while(m_inputBufferSize - offset > 0)
{
Block element(m_inputBuffer + offset, m_inputBufferSize - offset);
offset += element.size();
BOOST_ASSERT(offset <= m_inputBufferSize);
/// @todo Ensure lazy field decoding process
if (element.type() == tlv::Interest)
{
shared_ptr<Interest> i = make_shared<Interest>();
i->wireDecode(element);
onReceiveInterest(*i);
}
else if (element.type() == tlv::Data)
{
shared_ptr<Data> d = make_shared<Data>();
d->wireDecode(element);
onReceiveData(*d);
}
// @todo Add local header support
// else if (element.type() == tlv::LocalHeader)
// {
// shared_ptr<Interest> i = make_shared<Interest>();
// i->wireDecode(element);
// }
else
{
NFD_LOG_WARN("[id:" << this->getId()
<< ",endpoint:" << m_socket->local_endpoint()
<< "] Received unrecognized block of type ["
<< element.type() << "]");
// ignore unknown packet and proceed
}
}
}
catch(const tlv::Error& e) {
if (m_inputBufferSize == MAX_NDN_PACKET_SIZE && offset == 0)
{
NFD_LOG_WARN("[id:" << this->getId()
<< ",endpoint:" << m_socket->local_endpoint()
<< "] Received input is invalid or too large to process, "
<< "closing down the face");
closeSocket();
onFail("Received input is invalid or too large to process, closing down the face");
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<T>::handleReceive, this, _1, _2));
}
template <class T>
inline void
StreamFace<T>::keepFaceAliveUntilAllHandlersExecuted(const shared_ptr<Face>& face)
{
}
template <class T>
inline void
StreamFace<T>::closeSocket()
{
boost::asio::io_service& io = m_socket->get_io_service();
// use the non-throwing variants and ignore errors, if any
boost::system::error_code error;
m_socket->shutdown(protocol::socket::shutdown_both, error);
m_socket->close(error);
// after this, handlers will be called with an error code
// ensure that the Face object is alive at least until all pending
// handlers are dispatched
io.post(bind(&StreamFace<T>::keepFaceAliveUntilAllHandlersExecuted,
this, this->shared_from_this()));
}
} // namespace nfd
#endif // NFD_FACE_STREAM_FACE_HPP