blob: 1bde91b6fcca78b6b44529e458359e33e2b2dad1 [file] [log] [blame]
Alexander Afanasyev5964fb72014-02-18 12:42:45 -08001/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil -*- */
2/**
3 * Copyright (C) 2013-2014 Regents of the University of California.
4 * See COPYING for copyright and distribution information.
5 */
6
7#ifndef NDN_TRANSPORT_STREAM_TRANSPORT_HPP
8#define NDN_TRANSPORT_STREAM_TRANSPORT_HPP
9
10#include "../common.hpp"
11
12namespace ndn {
13
14const size_t MAX_LENGTH = 9000;
15
16template<class BaseTransport, class Protocol>
17class StreamTransportImpl
18{
19public:
20 typedef BaseTransport base_transport;
21 typedef Protocol protocol;
22 typedef StreamTransportImpl<BaseTransport,Protocol> impl;
23
24 StreamTransportImpl(base_transport& transport, boost::asio::io_service& ioService)
25 : m_transport(transport)
26 , m_socket(ioService)
27 , m_partialDataSize(0)
28 , m_connectionInProgress(false)
29 , m_connectTimer(ioService)
30 {
31 }
32
33 void
34 connectHandler(const boost::system::error_code& error)
35 {
36 m_connectionInProgress = false;
37 m_connectTimer.cancel();
38
39 if (!error)
40 {
Alexander Afanasyev52afb3f2014-03-07 09:05:35 +000041 resume();
Alexander Afanasyev5964fb72014-02-18 12:42:45 -080042 m_transport.m_isConnected = true;
43
44 for (std::list<Block>::iterator i = m_sendQueue.begin(); i != m_sendQueue.end(); ++i)
45 m_socket.async_send(boost::asio::buffer(i->wire(), i->size()),
46 bind(&impl::handle_async_send, this, _1, *i));
47
48 for (std::list< std::pair<Block,Block> >::iterator i = m_sendPairQueue.begin();
49 i != m_sendPairQueue.end(); ++i)
50 {
51 std::vector<boost::asio::const_buffer> buffer;
52 buffer.reserve(2);
53 buffer.push_back(boost::asio::buffer(i->first.wire(), i->first.size()));
54 buffer.push_back(boost::asio::buffer(i->second.wire(), i->second.size()));
55 m_socket.async_send(buffer,
56 bind(&impl::handle_async_send, this, _1, i->first, i->second));
57 }
58
59 m_sendQueue.clear();
60 m_sendPairQueue.clear();
61 }
62 else
63 {
64 // may need to throw exception
65 m_transport.m_isConnected = false;
66 m_transport.close();
67 throw Transport::Error(error, "error while connecting to the forwarder");
68 }
69 }
70
71 void
72 connectTimeoutHandler(const boost::system::error_code& error)
73 {
74 if (error) // e.g., cancelled timer
75 return;
76
77 m_connectionInProgress = false;
78 m_transport.m_isConnected = false;
79 m_socket.close();
80 throw Transport::Error(error, "error while connecting to the forwarder");
81 }
82
83 void
84 connect(const typename protocol::endpoint& endpoint)
85 {
86 if (!m_connectionInProgress) {
87 m_connectionInProgress = true;
88
89 // Wait at most 4 seconds to connect
90 /// @todo Decide whether this number should be configurable
91 m_connectTimer.expires_from_now(boost::posix_time::seconds(4));
92 m_connectTimer.async_wait(bind(&impl::connectTimeoutHandler, this, _1));
93
94 m_socket.open();
95 m_socket.async_connect(endpoint,
96 bind(&impl::connectHandler, this, _1));
97 }
98 }
99
100 void
101 close()
102 {
103 m_connectTimer.cancel();
104 m_socket.close();
105 m_transport.m_isConnected = false;
Alexander Afanasyev6e0c5a52014-03-18 16:18:58 -0700106 m_transport.m_isExpectingData = false;
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800107 m_sendQueue.clear();
108 m_sendPairQueue.clear();
109 }
110
111 void
Alexander Afanasyev52afb3f2014-03-07 09:05:35 +0000112 pause()
113 {
114 if (m_transport.m_isExpectingData)
115 {
116 m_transport.m_isExpectingData = false;
117 m_socket.cancel();
118 }
119 }
120
121 void
122 resume()
123 {
124 if (!m_transport.m_isExpectingData)
125 {
126 m_transport.m_isExpectingData = true;
127 m_partialDataSize = 0;
128 m_socket.async_receive(boost::asio::buffer(m_inputBuffer, MAX_LENGTH), 0,
129 bind(&impl::handle_async_receive, this, _1, _2));
130 }
131 }
132
133 void
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800134 send(const Block& wire)
135 {
136 if (!m_transport.m_isConnected)
137 m_sendQueue.push_back(wire);
138 else
139 m_socket.async_send(boost::asio::buffer(wire.wire(), wire.size()),
140 bind(&impl::handle_async_send, this, _1, wire));
141 }
142
143 void
144 send(const Block& header, const Block& payload)
145 {
146 if (!m_transport.m_isConnected)
147 {
148 m_sendPairQueue.push_back(std::make_pair(header, payload));
149 }
150 else
151 {
152 std::vector<boost::asio::const_buffer> buffers;
153 buffers.reserve(2);
154 buffers.push_back(boost::asio::buffer(header.wire(), header.size()));
155 buffers.push_back(boost::asio::buffer(payload.wire(), payload.size()));
156
157 m_socket.async_send(buffers,
158 bind(&impl::handle_async_send, this, _1, header, payload));
159 }
160 }
161
162 inline void
163 processAll(uint8_t* buffer, size_t& offset, size_t availableSize)
164 {
165 while(offset < availableSize)
166 {
167 Block element(buffer + offset, availableSize - offset);
168 m_transport.receive(element);
169
170 offset += element.size();
171 }
172 }
173
174 void
175 handle_async_receive(const boost::system::error_code& error, std::size_t bytes_recvd)
176 {
177 /// @todo The socket is not datagram, so need to have internal buffer to handle partial data reception
178
179 if (error)
180 {
181 if (error == boost::system::errc::operation_canceled) {
182 // async receive has been explicitly cancelled (e.g., socket close)
183 return;
184 }
185
186 m_socket.close(); // closing at this point may not be that necessary
187 m_transport.m_isConnected = true;
188 throw Transport::Error(error, "error while receiving data from socket");
189 }
190
191 if (!error && bytes_recvd > 0)
192 {
193 // m_inputBuffer has bytes_recvd received bytes of data
194 if (m_partialDataSize > 0)
195 {
196 size_t newDataSize = std::min(bytes_recvd, MAX_LENGTH-m_partialDataSize);
197 std::copy(m_inputBuffer, m_inputBuffer + newDataSize, m_partialData + m_partialDataSize);
198
199 m_partialDataSize += newDataSize;
200
201 size_t offset = 0;
202 try
203 {
204 processAll(m_partialData, offset, m_partialDataSize);
205
206 // no exceptions => processed the whole thing
207 if (bytes_recvd - newDataSize > 0)
208 {
209 // there is a little bit more data available
210
211 offset = 0;
212 m_partialDataSize = bytes_recvd - newDataSize;
213 std::copy(m_inputBuffer + newDataSize, m_inputBuffer + newDataSize + m_partialDataSize, m_partialData);
214
215 processAll(m_partialData, offset, m_partialDataSize);
216
217 // no exceptions => processed the whole thing
218 m_partialDataSize = 0;
219 }
220 else
221 {
222 // done processing
223 m_partialDataSize = 0;
224 }
225 }
226 catch(Tlv::Error &)
227 {
228 if (offset > 0)
229 {
230 m_partialDataSize -= offset;
231 std::copy(m_partialData + offset, m_partialData + offset + m_partialDataSize, m_partialData);
232 }
233 else if (offset == 0 && m_partialDataSize == MAX_LENGTH)
234 {
235 // very bad... should close connection
236 m_socket.close();
237 m_transport.m_isConnected = true;
238 throw Transport::Error(boost::system::error_code(),
239 "input buffer full, but a valid TLV cannot be decoded");
240 }
241 }
242 }
243 else
244 {
245 size_t offset = 0;
246 try
247 {
248 processAll(m_inputBuffer, offset, bytes_recvd);
249 }
250 catch(Tlv::Error &error)
251 {
252 if (offset > 0)
253 {
254 m_partialDataSize = bytes_recvd - offset;
255 std::copy(m_inputBuffer + offset, m_inputBuffer + offset + m_partialDataSize, m_partialData);
256 }
257 }
258 }
259 }
260
261 m_socket.async_receive(boost::asio::buffer(m_inputBuffer, MAX_LENGTH), 0,
262 bind(&impl::handle_async_receive, this, _1, _2));
263 }
264
265 void
266 handle_async_send(const boost::system::error_code& error, const Block& wire)
267 {
268 // pass (needed to keep data block alive during the send)
269 }
270
271 void
272 handle_async_send(const boost::system::error_code& error,
273 const Block& header, const Block& payload)
274 {
275 // pass (needed to keep data blocks alive during the send)
276 }
277
278protected:
279 base_transport& m_transport;
280
281 typename protocol::socket m_socket;
282 uint8_t m_inputBuffer[MAX_LENGTH];
283
284 uint8_t m_partialData[MAX_LENGTH];
285 size_t m_partialDataSize;
286
287 std::list< Block > m_sendQueue;
288 std::list< std::pair<Block, Block> > m_sendPairQueue;
289 bool m_connectionInProgress;
290
291 boost::asio::deadline_timer m_connectTimer;
292};
293
294
295template<class BaseTransport, class Protocol>
296class StreamTransportWithResolverImpl : public StreamTransportImpl<BaseTransport, Protocol>
297{
298public:
299 typedef BaseTransport base_transport;
300 typedef Protocol protocol;
301 typedef StreamTransportWithResolverImpl<BaseTransport,Protocol> impl;
302
303 StreamTransportWithResolverImpl(base_transport& transport, boost::asio::io_service& ioService)
304 : StreamTransportImpl<base_transport, protocol>(transport, ioService)
305 {
306 }
307
308 void
309 resolveHandler(const boost::system::error_code& error,
310 typename protocol::resolver::iterator endpoint,
311 const shared_ptr<typename protocol::resolver>&)
312 {
313 if (error)
314 {
315 if (error == boost::system::errc::operation_canceled)
316 return;
317
318 throw Transport::Error(error, "Error during resolution of host or port");
319 }
320
321 typename protocol::resolver::iterator end;
322 if (endpoint == end)
323 {
324 this->m_connectionInProgress = false;
325 this->m_transport.m_isConnected = false;
326 this->m_socket.close();
327 throw Transport::Error(error, "Unable to resolve because host or port");
328 }
329
330 this->m_socket.async_connect(*endpoint,
331 bind(&impl::connectHandler, this, _1));
332 }
333
334 void
335 connect(const typename protocol::resolver::query& query)
336 {
337 if (!this->m_connectionInProgress) {
338 this->m_connectionInProgress = true;
339
340 // Wait at most 4 seconds to connect
341 /// @todo Decide whether this number should be configurable
342 this->m_connectTimer.expires_from_now(boost::posix_time::seconds(4));
343 this->m_connectTimer.async_wait(bind(&impl::connectTimeoutHandler, this, _1));
344
345 // typename boost::asio::ip::basic_resolver< protocol > resolver;
346 shared_ptr<typename protocol::resolver> resolver =
347 make_shared<typename protocol::resolver>(boost::ref(this->m_socket.get_io_service()));
348
349 resolver->async_resolve(query, bind(&impl::resolveHandler, this, _1, _2, resolver));
350 }
351 }
352};
353
354
355} // namespace ndn
356
357#endif // NDN_TRANSPORT_STREAM_TRANSPORT_HPP