stream-transport-impl.hpp
Go to the documentation of this file.
1 /* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
2 /*
3  * Copyright (c) 2013-2017 Regents of the University of California.
4  *
5  * This file is part of ndn-cxx library (NDN C++ library with eXperimental eXtensions).
6  *
7  * ndn-cxx library is free software: you can redistribute it and/or modify it under the
8  * terms of the GNU Lesser General Public License as published by the Free Software
9  * Foundation, either version 3 of the License, or (at your option) any later version.
10  *
11  * ndn-cxx library is distributed in the hope that it will be useful, but WITHOUT ANY
12  * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A
13  * PARTICULAR PURPOSE. See the GNU Lesser General Public License for more details.
14  *
15  * You should have received copies of the GNU General Public License and GNU Lesser
16  * General Public License along with ndn-cxx, e.g., in COPYING.md file. If not, see
17  * <http://www.gnu.org/licenses/>.
18  *
19  * See AUTHORS.md for complete list of ndn-cxx authors and contributors.
20  */
21 
22 #ifndef NDN_TRANSPORT_STREAM_TRANSPORT_IMPL_HPP
23 #define NDN_TRANSPORT_STREAM_TRANSPORT_IMPL_HPP
24 
25 #include "transport.hpp"
26 
27 #include <boost/asio/deadline_timer.hpp>
28 #include <boost/asio/write.hpp>
29 
30 #include <list>
31 
32 namespace ndn {
33 
39 template<typename BaseTransport, typename Protocol>
40 class StreamTransportImpl : public enable_shared_from_this<StreamTransportImpl<BaseTransport, Protocol>>
41 {
42 public:
44  typedef std::list<Block> BlockSequence;
45  typedef std::list<BlockSequence> TransmissionQueue;
46 
47  StreamTransportImpl(BaseTransport& transport, boost::asio::io_service& ioService)
48  : m_transport(transport)
49  , m_socket(ioService)
51  , m_isConnecting(false)
52  , m_connectTimer(ioService)
53  {
54  }
55 
56  void
57  connect(const typename Protocol::endpoint& endpoint)
58  {
59  if (!m_isConnecting) {
60  m_isConnecting = true;
61 
62  // Wait at most 4 seconds to connect
64  m_connectTimer.expires_from_now(boost::posix_time::seconds(4));
65  m_connectTimer.async_wait(bind(&Impl::connectTimeoutHandler, this->shared_from_this(), _1));
66 
67  m_socket.open();
68  m_socket.async_connect(endpoint, bind(&Impl::connectHandler, this->shared_from_this(), _1));
69  }
70  }
71 
72  void
74  {
75  m_isConnecting = false;
76 
77  boost::system::error_code error; // to silently ignore all errors
78  m_connectTimer.cancel(error);
79  m_socket.cancel(error);
80  m_socket.close(error);
81 
82  m_transport.m_isConnected = false;
83  m_transport.m_isReceiving = false;
84  m_transmissionQueue.clear();
85  }
86 
87  void
89  {
90  if (m_isConnecting)
91  return;
92 
93  if (m_transport.m_isReceiving) {
94  m_transport.m_isReceiving = false;
95  m_socket.cancel();
96  }
97  }
98 
99  void
101  {
102  if (m_isConnecting)
103  return;
104 
105  if (!m_transport.m_isReceiving) {
106  m_transport.m_isReceiving = true;
107  m_inputBufferSize = 0;
108  asyncReceive();
109  }
110  }
111 
112  void
113  send(const Block& wire)
114  {
115  BlockSequence sequence;
116  sequence.push_back(wire);
117  send(std::move(sequence));
118  }
119 
120  void
121  send(const Block& header, const Block& payload)
122  {
123  BlockSequence sequence;
124  sequence.push_back(header);
125  sequence.push_back(payload);
126  send(std::move(sequence));
127  }
128 
129 protected:
130  void
131  connectHandler(const boost::system::error_code& error)
132  {
133  m_isConnecting = false;
134  m_connectTimer.cancel();
135 
136  if (!error) {
137  m_transport.m_isConnected = true;
138 
139  if (!m_transmissionQueue.empty()) {
140  resume();
141  asyncWrite();
142  }
143  }
144  else {
145  m_transport.m_isConnected = false;
146  m_transport.close();
147  BOOST_THROW_EXCEPTION(Transport::Error(error, "error while connecting to the forwarder"));
148  }
149  }
150 
151  void
152  connectTimeoutHandler(const boost::system::error_code& error)
153  {
154  if (error) // e.g., cancelled timer
155  return;
156 
157  m_transport.close();
158  BOOST_THROW_EXCEPTION(Transport::Error(error, "error while connecting to the forwarder"));
159  }
160 
161  void
162  send(BlockSequence&& sequence)
163  {
164  m_transmissionQueue.emplace_back(sequence);
165 
166  if (m_transport.m_isConnected && m_transmissionQueue.size() == 1) {
167  asyncWrite();
168  }
169 
170  // if not connected or there is transmission in progress (m_transmissionQueue.size() > 1),
171  // next write will be scheduled either in connectHandler or in asyncWriteHandler
172  }
173 
174  void
176  {
177  BOOST_ASSERT(!m_transmissionQueue.empty());
178  boost::asio::async_write(m_socket, m_transmissionQueue.front(),
179  bind(&Impl::handleAsyncWrite, this->shared_from_this(), _1, m_transmissionQueue.begin()));
180  }
181 
182  void
183  handleAsyncWrite(const boost::system::error_code& error, TransmissionQueue::iterator queueItem)
184  {
185  if (error) {
186  if (error == boost::system::errc::operation_canceled) {
187  // async receive has been explicitly cancelled (e.g., socket close)
188  return;
189  }
190 
191  m_transport.close();
192  BOOST_THROW_EXCEPTION(Transport::Error(error, "error while sending data to socket"));
193  }
194 
195  if (!m_transport.m_isConnected) {
196  return; // queue has been already cleared
197  }
198 
199  m_transmissionQueue.erase(queueItem);
200 
201  if (!m_transmissionQueue.empty()) {
202  asyncWrite();
203  }
204  }
205 
206  void
208  {
209  m_socket.async_receive(boost::asio::buffer(m_inputBuffer + m_inputBufferSize,
211  bind(&Impl::handleAsyncReceive, this->shared_from_this(), _1, _2));
212  }
213 
214  void
215  handleAsyncReceive(const boost::system::error_code& error, std::size_t nBytesRecvd)
216  {
217  if (error) {
218  if (error == boost::system::errc::operation_canceled) {
219  // async receive has been explicitly cancelled (e.g., socket close)
220  return;
221  }
222 
223  m_transport.close();
224  BOOST_THROW_EXCEPTION(Transport::Error(error, "error while receiving data from socket"));
225  }
226 
227  m_inputBufferSize += nBytesRecvd;
228  // do magic
229 
230  std::size_t offset = 0;
231  bool hasProcessedSome = processAllReceived(m_inputBuffer, offset, m_inputBufferSize);
232  if (!hasProcessedSome && m_inputBufferSize == MAX_NDN_PACKET_SIZE && offset == 0) {
233  m_transport.close();
234  BOOST_THROW_EXCEPTION(Transport::Error(boost::system::error_code(),
235  "input buffer full, but a valid TLV cannot be "
236  "decoded"));
237  }
238 
239  if (offset > 0) {
240  if (offset != m_inputBufferSize) {
242  m_inputBufferSize -= offset;
243  }
244  else {
245  m_inputBufferSize = 0;
246  }
247  }
248 
249  asyncReceive();
250  }
251 
252  bool
253  processAllReceived(uint8_t* buffer, size_t& offset, size_t nBytesAvailable)
254  {
255  while (offset < nBytesAvailable) {
256  bool isOk = false;
257  Block element;
258  std::tie(isOk, element) = Block::fromBuffer(buffer + offset, nBytesAvailable - offset);
259  if (!isOk)
260  return false;
261 
262  m_transport.receive(element);
263  offset += element.size();
264  }
265  return true;
266  }
267 
268 protected:
269  BaseTransport& m_transport;
270 
271  typename Protocol::socket m_socket;
274 
275  TransmissionQueue m_transmissionQueue;
277 
278  boost::asio::deadline_timer m_connectTimer;
279 };
280 
281 } // namespace ndn
282 
283 #endif // NDN_TRANSPORT_STREAM_TRANSPORT_IMPL_HPP
bool processAllReceived(uint8_t *buffer, size_t &offset, size_t nBytesAvailable)
void connectTimeoutHandler(const boost::system::error_code &error)
Copyright (c) 2013-2017 Regents of the University of California.
Definition: common.hpp:66
static std::tuple< bool, Block > fromBuffer(ConstBufferPtr buffer, size_t offset)
Try to parse Block from a wire buffer.
Definition: block.cpp:197
void send(BlockSequence &&sequence)
StreamTransportImpl(BaseTransport &transport, boost::asio::io_service &ioService)
Represents a TLV element of NDN packet format.
Definition: block.hpp:42
implementation detail of a Boost.Asio-based stream-oriented transport
void handleAsyncReceive(const boost::system::error_code &error, std::size_t nBytesRecvd)
void send(const Block &wire)
size_t size() const
Get size of encoded wire, including Type-Length-Value.
Definition: block.cpp:301
StreamTransportImpl< BaseTransport, Protocol > Impl
void connect(const typename Protocol::endpoint &endpoint)
void handleAsyncWrite(const boost::system::error_code &error, TransmissionQueue::iterator queueItem)
uint8_t m_inputBuffer[MAX_NDN_PACKET_SIZE]
void send(const Block &header, const Block &payload)
boost::asio::deadline_timer m_connectTimer
std::list< BlockSequence > TransmissionQueue
void connectHandler(const boost::system::error_code &error)
const size_t MAX_NDN_PACKET_SIZE
practical limit of network layer packet size