src/transport/stream-transport.hpp - ndn-cxx - Gitiles

 /* -*- 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_inputBufferSize(0)
     , m_connectionInProgress(false)
     , m_connectTimer(ioService)
   {
   }

   void
   connectHandler(const boost::system::error_code& error)
   {
     m_connectionInProgress = false;
     m_connectTimer.cancel();

     if (!error)
       {
         resume();
         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_transport.m_isExpectingData = 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 time::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_transport.m_isExpectingData = false;
     m_sendQueue.clear();
     m_sendPairQueue.clear();
   }

   void
   pause()
   {
     if (m_transport.m_isExpectingData)
       {
         m_transport.m_isExpectingData = false;
         m_socket.cancel();
       }
   }

   void
   resume()
   {
     if (!m_transport.m_isExpectingData)
       {
         m_transport.m_isExpectingData = true;
         m_inputBufferSize = 0;
         m_socket.async_receive(boost::asio::buffer(m_inputBuffer, MAX_LENGTH), 0,
                                bind(&impl::handle_async_receive, this, _1, _2));
       }
   }

   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 bool
   processAll(uint8_t* buffer, size_t& offset, size_t availableSize)
   {
     Block element;
     while(offset < availableSize)
       {
         bool ok = Block::fromBuffer(buffer + offset, availableSize - offset, element);
         if (!ok)
           return false;

         m_transport.receive(element);
         offset += element.size();
       }
     return true;
   }

   void
   handle_async_receive(const boost::system::error_code& error, std::size_t bytes_recvd)
   {
     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");
       }

     m_inputBufferSize += bytes_recvd;
     // do magic

     std::size_t offset = 0;
     bool ok = processAll(m_inputBuffer, offset, m_inputBufferSize);
     if (!ok && m_inputBufferSize == MAX_LENGTH && offset == 0)
       {
         // very bad... should close connection
         m_socket.close();
         m_transport.m_isConnected = false;
         m_transport.m_isExpectingData = false;
         throw Transport::Error(boost::system::error_code(),
                                "input buffer full, but a valid TLV cannot be decoded");
       }

     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_LENGTH - m_inputBufferSize), 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];
   size_t m_inputBufferSize;

   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_transport.m_isExpectingData = 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 time::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
	/* -- 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_inputBufferSize(0)
	, m_connectionInProgress(false)
	, m_connectTimer(ioService)
	{
	}

	void
	connectHandler(const boost::system::error_code& error)
	{
	m_connectionInProgress = false;
	m_connectTimer.cancel();

	if (!error)
	{
	resume();
	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_transport.m_isExpectingData = 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 time::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_transport.m_isExpectingData = false;
	m_sendQueue.clear();
	m_sendPairQueue.clear();
	}

	void
	pause()
	{
	if (m_transport.m_isExpectingData)
	{
	m_transport.m_isExpectingData = false;
	m_socket.cancel();
	}
	}

	void
	resume()
	{
	if (!m_transport.m_isExpectingData)
	{
	m_transport.m_isExpectingData = true;
	m_inputBufferSize = 0;
	m_socket.async_receive(boost::asio::buffer(m_inputBuffer, MAX_LENGTH), 0,
	bind(&impl::handle_async_receive, this, _1, _2));
	}
	}

	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 bool
	processAll(uint8_t* buffer, size_t& offset, size_t availableSize)
	{
	Block element;
	while(offset < availableSize)
	{
	bool ok = Block::fromBuffer(buffer + offset, availableSize - offset, element);
	if (!ok)
	return false;

	m_transport.receive(element);
	offset += element.size();
	}
	return true;
	}

	void
	handle_async_receive(const boost::system::error_code& error, std::size_t bytes_recvd)
	{
	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");
	}

	m_inputBufferSize += bytes_recvd;
	// do magic

	std::size_t offset = 0;
	bool ok = processAll(m_inputBuffer, offset, m_inputBufferSize);
	if (!ok && m_inputBufferSize == MAX_LENGTH && offset == 0)
	{
	// very bad... should close connection
	m_socket.close();
	m_transport.m_isConnected = false;
	m_transport.m_isExpectingData = false;
	throw Transport::Error(boost::system::error_code(),
	"input buffer full, but a valid TLV cannot be decoded");
	}

	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_LENGTH - m_inputBufferSize), 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];
	size_t m_inputBufferSize;

	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_transport.m_isExpectingData = 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 time::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