async-socket-transport.hpp

/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil -*- */
#ifndef NDN_ASYNC_SOCKET_TRANSPORT_HPP
#define NDN_ASYNC_SOCKET_TRANSPORT_HPP

#include <stdexcept>
#include <boost/bind.hpp>
#include <boost/asio.hpp>
#include <boost/enable_shared_from_this.hpp>
#include <ndn-cpp/transport/transport.hpp>
#include "../c/encoding/element-reader.h"
#include "../encoding/element-listener.hpp"
#include "../util/dynamic-uint8-vector.hpp"

namespace ndn {

template<class AsioProtocol> class AsyncSocketTransport {
public:
  AsyncSocketTransport(boost::asio::io_service& ioService)
  : impl_(new Impl(ioService))
  {
  }

  void
  connect
    (const typename AsioProtocol::endpoint& endPoint,
     ElementListener& elementListener, const Transport::OnConnected& onConnected)
  {
    impl_->connect(endPoint, elementListener, onConnected);
  }

  void
  send(const uint8_t *data, size_t dataLength)
  {
    impl_->send(data, dataLength);
  }

  bool
  getIsConnected()
  {
    return impl_->getIsConnected();
  }

  void
  close()
  {
    impl_->close();
  }

private:
  class Impl : public boost::enable_shared_from_this<Impl> {
  public:
    Impl(boost::asio::io_service& ioService)
    : ioService_(ioService), socket_(new typename AsioProtocol::socket(ioService)),
      elementBuffer_(new DynamicUInt8Vector(1000)), isConnected_(false)
    {
      ndn_ElementReader_initialize(&elementReader_, 0, elementBuffer_.get());
    }

    void
    connect
      (const typename AsioProtocol::endpoint& endPoint,
       ElementListener& elementListener, const Transport::OnConnected& onConnected)
    {
      close();

      ndn_ElementReader_reset(&elementReader_, &elementListener);

      socket_->async_connect
        (endPoint,
         boost::bind(&AsyncSocketTransport::Impl::connectHandler,
                     this->shared_from_this(), _1, onConnected));
    }

    void
    send(const uint8_t *data, size_t dataLength)
    {
      if (!isConnected_)
        throw std::runtime_error
          ("AsyncSocketTransport.send: The socket is not connected");

      // Assume that this is called from a dispatch so that we are already in the
      // ioService_ thread. Just do a blocking write.
      boost::system::error_code errorCode;
      boost::asio::write
        (*socket_, boost::asio::buffer(data, dataLength), errorCode);
      if (errorCode != boost::system::errc::success)
        throw std::runtime_error("AsyncSocketTransport.send: Error in write");
    }

    bool
    getIsConnected()
    {
      return isConnected_;
    }

    void
    close()
    {
      try {
        socket_->close();
      }
      catch (...) {
        // Ignore any exceptions.
      }

      isConnected_ = false;
    }

  private:
    void
    connectHandler
      (const boost::system::error_code& errorCode,
       const Transport::OnConnected& onConnected)
    {
      if (errorCode != boost::system::errc::success)
        // TODO: How to report errors to the application?
        throw std::runtime_error("AsyncSocketTransport: Error in async_connect");

      isConnected_ = true;
      onConnected();

      socket_->async_receive
        (boost::asio::buffer(receiveBuffer_, sizeof(receiveBuffer_)), 0,
         boost::bind(&AsyncSocketTransport::Impl::readHandler,
                     this->shared_from_this(), _1, _2));
    }

    void
    readHandler(const boost::system::error_code& errorCode, size_t nBytesReceived)
    {
      if (errorCode != boost::system::errc::success) {
        if (errorCode == boost::system::errc::operation_canceled)
          // Assume the socket has been closed. Do nothing.
          return;

        close();
        // TODO: How to report errors to the application?
        throw std::runtime_error("AsyncSocketTransport: Error in async_receive");
      }

      ndn_Error error;
      if ((error = ndn_ElementReader_onReceivedData
           (&elementReader_, receiveBuffer_, nBytesReceived)))
        throw std::runtime_error(ndn_getErrorString(error));

      // Request another async receive to loop back to here.
      if (socket_->is_open())
        socket_->async_receive
          (boost::asio::buffer(receiveBuffer_, sizeof(receiveBuffer_)), 0,
           boost::bind(&AsyncSocketTransport::Impl::readHandler,
                       this->shared_from_this(), _1, _2));
    }

    boost::asio::io_service& ioService_;
    boost::shared_ptr<typename AsioProtocol::socket> socket_;
    uint8_t receiveBuffer_[MAX_NDN_PACKET_SIZE];
    boost::shared_ptr<DynamicUInt8Vector> elementBuffer_;
    ndn_ElementReader elementReader_;
    bool isConnected_;
  };

  boost::shared_ptr<Impl> impl_;
};

}

#endif