blob: b4e3d09578d730e9274388e01a8dbc68223b84ff [file] [log] [blame]
Alexander Afanasyevc169a812014-05-20 20:37:29 -04001/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
Alexander Afanasyev5964fb72014-02-18 12:42:45 -08002/**
Alexander Afanasyevc169a812014-05-20 20:37:29 -04003 * Copyright (c) 2013-2014 Regents of the University of California.
Alexander Afanasyevdfa52c42014-04-24 21:10:11 -07004 *
5 * This file is part of ndn-cxx library (NDN C++ library with eXperimental eXtensions).
Alexander Afanasyevdfa52c42014-04-24 21:10:11 -07006 *
Alexander Afanasyevc169a812014-05-20 20:37:29 -04007 * 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.
Alexander Afanasyev5964fb72014-02-18 12:42:45 -080020 */
21
22#ifndef NDN_TRANSPORT_STREAM_TRANSPORT_HPP
23#define NDN_TRANSPORT_STREAM_TRANSPORT_HPP
24
Junxiao Shi71355d52014-12-11 09:20:44 -070025#include "transport.hpp"
Alexander Afanasyev5964fb72014-02-18 12:42:45 -080026
Alexander Afanasyev258ec2b2014-05-14 16:15:37 -070027#include <list>
28
Alexander Afanasyev5964fb72014-02-18 12:42:45 -080029namespace ndn {
30
Alexander Afanasyev5964fb72014-02-18 12:42:45 -080031template<class BaseTransport, class Protocol>
32class StreamTransportImpl
33{
34public:
Alexander Afanasyev49bb1fb2014-07-21 12:54:01 -070035 typedef StreamTransportImpl<BaseTransport,Protocol> Impl;
Alexander Afanasyev5964fb72014-02-18 12:42:45 -080036
Alexander Afanasyev6a05b4b2014-07-18 17:23:00 -070037 typedef std::list<Block> BlockSequence;
38 typedef std::list<BlockSequence> TransmissionQueue;
39
Alexander Afanasyev49bb1fb2014-07-21 12:54:01 -070040 StreamTransportImpl(BaseTransport& transport, boost::asio::io_service& ioService)
Alexander Afanasyev5964fb72014-02-18 12:42:45 -080041 : m_transport(transport)
42 , m_socket(ioService)
Alexander Afanasyev937aa782014-03-21 13:17:57 -070043 , m_inputBufferSize(0)
Alexander Afanasyev5964fb72014-02-18 12:42:45 -080044 , m_connectionInProgress(false)
45 , m_connectTimer(ioService)
46 {
47 }
Alexander Afanasyev937aa782014-03-21 13:17:57 -070048
Alexander Afanasyev5964fb72014-02-18 12:42:45 -080049 void
50 connectHandler(const boost::system::error_code& error)
51 {
52 m_connectionInProgress = false;
53 m_connectTimer.cancel();
54
55 if (!error)
56 {
Alexander Afanasyev52afb3f2014-03-07 09:05:35 +000057 resume();
Alexander Afanasyev5964fb72014-02-18 12:42:45 -080058 m_transport.m_isConnected = true;
59
Alexander Afanasyev6a05b4b2014-07-18 17:23:00 -070060 if (!m_transmissionQueue.empty()) {
61 boost::asio::async_write(m_socket, *m_transmissionQueue.begin(),
Alexander Afanasyev49bb1fb2014-07-21 12:54:01 -070062 bind(&Impl::handleAsyncWrite, this, _1,
Alexander Afanasyev6a05b4b2014-07-18 17:23:00 -070063 m_transmissionQueue.begin()));
64 }
Alexander Afanasyev5964fb72014-02-18 12:42:45 -080065 }
66 else
67 {
68 // may need to throw exception
69 m_transport.m_isConnected = false;
70 m_transport.close();
71 throw Transport::Error(error, "error while connecting to the forwarder");
72 }
73 }
74
75 void
76 connectTimeoutHandler(const boost::system::error_code& error)
77 {
78 if (error) // e.g., cancelled timer
79 return;
80
Alexander Afanasyevbc5830a2014-07-11 15:02:38 -070081 m_transport.close();
Alexander Afanasyev5964fb72014-02-18 12:42:45 -080082 throw Transport::Error(error, "error while connecting to the forwarder");
83 }
84
85 void
Alexander Afanasyev49bb1fb2014-07-21 12:54:01 -070086 connect(const typename Protocol::endpoint& endpoint)
Alexander Afanasyev5964fb72014-02-18 12:42:45 -080087 {
88 if (!m_connectionInProgress) {
89 m_connectionInProgress = true;
90
Alexander Afanasyev49bb1fb2014-07-21 12:54:01 -070091 // Wait at most 4 seconds to connect
Alexander Afanasyev5964fb72014-02-18 12:42:45 -080092 /// @todo Decide whether this number should be configurable
93 m_connectTimer.expires_from_now(boost::posix_time::seconds(4));
Alexander Afanasyev49bb1fb2014-07-21 12:54:01 -070094 m_connectTimer.async_wait(bind(&Impl::connectTimeoutHandler, this, _1));
Alexander Afanasyev5964fb72014-02-18 12:42:45 -080095
96 m_socket.open();
97 m_socket.async_connect(endpoint,
Alexander Afanasyev49bb1fb2014-07-21 12:54:01 -070098 bind(&Impl::connectHandler, this, _1));
Alexander Afanasyev5964fb72014-02-18 12:42:45 -080099 }
100 }
Alexander Afanasyev937aa782014-03-21 13:17:57 -0700101
102 void
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800103 close()
104 {
Alexander Afanasyevbc5830a2014-07-11 15:02:38 -0700105 m_connectionInProgress = false;
106
Alexander Afanasyev6507fb12014-04-28 23:18:56 -0700107 boost::system::error_code error; // to silently ignore all errors
108 m_connectTimer.cancel(error);
109 m_socket.cancel(error);
110 m_socket.close(error);
111
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800112 m_transport.m_isConnected = false;
Alexander Afanasyev6e0c5a52014-03-18 16:18:58 -0700113 m_transport.m_isExpectingData = false;
Alexander Afanasyev6a05b4b2014-07-18 17:23:00 -0700114 m_transmissionQueue.clear();
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800115 }
116
117 void
Alexander Afanasyev52afb3f2014-03-07 09:05:35 +0000118 pause()
119 {
Alexander Afanasyev6507fb12014-04-28 23:18:56 -0700120 if (m_connectionInProgress)
121 return;
122
Alexander Afanasyev52afb3f2014-03-07 09:05:35 +0000123 if (m_transport.m_isExpectingData)
124 {
125 m_transport.m_isExpectingData = false;
126 m_socket.cancel();
127 }
128 }
129
130 void
131 resume()
132 {
Alexander Afanasyev6507fb12014-04-28 23:18:56 -0700133 if (m_connectionInProgress)
134 return;
135
Alexander Afanasyev52afb3f2014-03-07 09:05:35 +0000136 if (!m_transport.m_isExpectingData)
137 {
138 m_transport.m_isExpectingData = true;
Alexander Afanasyev937aa782014-03-21 13:17:57 -0700139 m_inputBufferSize = 0;
Alexander Afanasyev49bb1fb2014-07-21 12:54:01 -0700140 m_socket.async_receive(boost::asio::buffer(m_inputBuffer, MAX_NDN_PACKET_SIZE), 0,
141 bind(&Impl::handleAsyncReceive, this, _1, _2));
Alexander Afanasyev52afb3f2014-03-07 09:05:35 +0000142 }
143 }
144
145 void
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800146 send(const Block& wire)
147 {
Alexander Afanasyev6a05b4b2014-07-18 17:23:00 -0700148 BlockSequence sequence;
149 sequence.push_back(wire);
150 m_transmissionQueue.push_back(sequence);
151
152 if (m_transport.m_isConnected && m_transmissionQueue.size() == 1) {
153 boost::asio::async_write(m_socket, *m_transmissionQueue.begin(),
Alexander Afanasyev49bb1fb2014-07-21 12:54:01 -0700154 bind(&Impl::handleAsyncWrite, this, _1,
Alexander Afanasyev6a05b4b2014-07-18 17:23:00 -0700155 m_transmissionQueue.begin()));
156 }
157
158 // if not connected or there is transmission in progress (m_transmissionQueue.size() > 1),
159 // next write will be scheduled either in connectHandler or in asyncWriteHandler
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800160 }
161
162 void
163 send(const Block& header, const Block& payload)
164 {
Alexander Afanasyev6a05b4b2014-07-18 17:23:00 -0700165 BlockSequence sequence;
166 sequence.push_back(header);
167 sequence.push_back(payload);
168 m_transmissionQueue.push_back(sequence);
Alexander Afanasyev937aa782014-03-21 13:17:57 -0700169
Alexander Afanasyev6a05b4b2014-07-18 17:23:00 -0700170 if (m_transport.m_isConnected && m_transmissionQueue.size() == 1) {
171 boost::asio::async_write(m_socket, *m_transmissionQueue.begin(),
Alexander Afanasyev49bb1fb2014-07-21 12:54:01 -0700172 bind(&Impl::handleAsyncWrite, this, _1,
Alexander Afanasyev6a05b4b2014-07-18 17:23:00 -0700173 m_transmissionQueue.begin()));
174 }
175
176 // if not connected or there is transmission in progress (m_transmissionQueue.size() > 1),
177 // next write will be scheduled either in connectHandler or in asyncWriteHandler
178 }
179
180 void
181 handleAsyncWrite(const boost::system::error_code& error,
182 TransmissionQueue::iterator queueItem)
183 {
184 if (error)
185 {
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 throw Transport::Error(error, "error while sending data to socket");
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800193 }
Alexander Afanasyev6a05b4b2014-07-18 17:23:00 -0700194
195 m_transmissionQueue.erase(queueItem);
196
197 if (!m_transmissionQueue.empty()) {
198 boost::asio::async_write(m_socket, *m_transmissionQueue.begin(),
Alexander Afanasyev49bb1fb2014-07-21 12:54:01 -0700199 bind(&Impl::handleAsyncWrite, this, _1,
Alexander Afanasyev6a05b4b2014-07-18 17:23:00 -0700200 m_transmissionQueue.begin()));
201 }
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800202 }
Alexander Afanasyev937aa782014-03-21 13:17:57 -0700203
Alexander Afanasyev49bb1fb2014-07-21 12:54:01 -0700204 bool
205 processAll(uint8_t* buffer, size_t& offset, size_t nBytesAvailable)
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800206 {
Alexander Afanasyev49bb1fb2014-07-21 12:54:01 -0700207 while (offset < nBytesAvailable)
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800208 {
Alexander Afanasyev49bb1fb2014-07-21 12:54:01 -0700209 Block element;
210 bool ok = Block::fromBuffer(buffer + offset, nBytesAvailable - offset, element);
Alexander Afanasyev937aa782014-03-21 13:17:57 -0700211 if (!ok)
212 return false;
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800213
Alexander Afanasyev937aa782014-03-21 13:17:57 -0700214 m_transport.receive(element);
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800215 offset += element.size();
216 }
Alexander Afanasyev937aa782014-03-21 13:17:57 -0700217 return true;
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800218 }
Alexander Afanasyev937aa782014-03-21 13:17:57 -0700219
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800220 void
Alexander Afanasyev49bb1fb2014-07-21 12:54:01 -0700221 handleAsyncReceive(const boost::system::error_code& error, std::size_t nBytesRecvd)
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800222 {
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800223 if (error)
224 {
225 if (error == boost::system::errc::operation_canceled) {
226 // async receive has been explicitly cancelled (e.g., socket close)
227 return;
228 }
Alexander Afanasyev937aa782014-03-21 13:17:57 -0700229
Alexander Afanasyevbc5830a2014-07-11 15:02:38 -0700230 m_transport.close();
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800231 throw Transport::Error(error, "error while receiving data from socket");
232 }
Alexander Afanasyev937aa782014-03-21 13:17:57 -0700233
Alexander Afanasyev49bb1fb2014-07-21 12:54:01 -0700234 m_inputBufferSize += nBytesRecvd;
Alexander Afanasyev937aa782014-03-21 13:17:57 -0700235 // do magic
236
237 std::size_t offset = 0;
Alexander Afanasyev49bb1fb2014-07-21 12:54:01 -0700238 bool hasProcessedSome = processAll(m_inputBuffer, offset, m_inputBufferSize);
239 if (!hasProcessedSome && m_inputBufferSize == MAX_NDN_PACKET_SIZE && offset == 0)
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800240 {
Alexander Afanasyevbc5830a2014-07-11 15:02:38 -0700241 m_transport.close();
Alexander Afanasyev937aa782014-03-21 13:17:57 -0700242 throw Transport::Error(boost::system::error_code(),
243 "input buffer full, but a valid TLV cannot be decoded");
244 }
245
246 if (offset > 0)
247 {
248 if (offset != m_inputBufferSize)
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800249 {
Alexander Afanasyev937aa782014-03-21 13:17:57 -0700250 std::copy(m_inputBuffer + offset, m_inputBuffer + m_inputBufferSize,
251 m_inputBuffer);
252 m_inputBufferSize -= offset;
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800253 }
254 else
255 {
Alexander Afanasyev937aa782014-03-21 13:17:57 -0700256 m_inputBufferSize = 0;
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800257 }
258 }
259
Alexander Afanasyev937aa782014-03-21 13:17:57 -0700260 m_socket.async_receive(boost::asio::buffer(m_inputBuffer + m_inputBufferSize,
Alexander Afanasyev49bb1fb2014-07-21 12:54:01 -0700261 MAX_NDN_PACKET_SIZE - m_inputBufferSize), 0,
262 bind(&Impl::handleAsyncReceive, this, _1, _2));
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800263 }
264
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800265protected:
Alexander Afanasyev49bb1fb2014-07-21 12:54:01 -0700266 BaseTransport& m_transport;
Alexander Afanasyev937aa782014-03-21 13:17:57 -0700267
Alexander Afanasyev49bb1fb2014-07-21 12:54:01 -0700268 typename Protocol::socket m_socket;
269 uint8_t m_inputBuffer[MAX_NDN_PACKET_SIZE];
Alexander Afanasyev937aa782014-03-21 13:17:57 -0700270 size_t m_inputBufferSize;
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800271
Alexander Afanasyev6a05b4b2014-07-18 17:23:00 -0700272 TransmissionQueue m_transmissionQueue;
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800273 bool m_connectionInProgress;
274
275 boost::asio::deadline_timer m_connectTimer;
276};
277
278
279template<class BaseTransport, class Protocol>
280class StreamTransportWithResolverImpl : public StreamTransportImpl<BaseTransport, Protocol>
281{
282public:
Alexander Afanasyev49bb1fb2014-07-21 12:54:01 -0700283 typedef StreamTransportWithResolverImpl<BaseTransport,Protocol> Impl;
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800284
Alexander Afanasyev49bb1fb2014-07-21 12:54:01 -0700285 StreamTransportWithResolverImpl(BaseTransport& transport, boost::asio::io_service& ioService)
286 : StreamTransportImpl<BaseTransport, Protocol>(transport, ioService)
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800287 {
288 }
Alexander Afanasyev937aa782014-03-21 13:17:57 -0700289
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800290 void
291 resolveHandler(const boost::system::error_code& error,
Alexander Afanasyev49bb1fb2014-07-21 12:54:01 -0700292 typename Protocol::resolver::iterator endpoint,
293 const shared_ptr<typename Protocol::resolver>&)
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800294 {
295 if (error)
296 {
297 if (error == boost::system::errc::operation_canceled)
298 return;
Alexander Afanasyev937aa782014-03-21 13:17:57 -0700299
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800300 throw Transport::Error(error, "Error during resolution of host or port");
301 }
Alexander Afanasyev937aa782014-03-21 13:17:57 -0700302
Alexander Afanasyev49bb1fb2014-07-21 12:54:01 -0700303 typename Protocol::resolver::iterator end;
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800304 if (endpoint == end)
305 {
Alexander Afanasyevbc5830a2014-07-11 15:02:38 -0700306 this->m_transport.close();
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800307 throw Transport::Error(error, "Unable to resolve because host or port");
308 }
309
310 this->m_socket.async_connect(*endpoint,
Alexander Afanasyev49bb1fb2014-07-21 12:54:01 -0700311 bind(&Impl::connectHandler, this, _1));
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800312 }
Alexander Afanasyev937aa782014-03-21 13:17:57 -0700313
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800314 void
Alexander Afanasyev49bb1fb2014-07-21 12:54:01 -0700315 connect(const typename Protocol::resolver::query& query)
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800316 {
317 if (!this->m_connectionInProgress) {
318 this->m_connectionInProgress = true;
319
Alexander Afanasyev49bb1fb2014-07-21 12:54:01 -0700320 // Wait at most 4 seconds to connect
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800321 /// @todo Decide whether this number should be configurable
322 this->m_connectTimer.expires_from_now(boost::posix_time::seconds(4));
Alexander Afanasyev49bb1fb2014-07-21 12:54:01 -0700323 this->m_connectTimer.async_wait(bind(&Impl::connectTimeoutHandler, this, _1));
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800324
Alexander Afanasyev49bb1fb2014-07-21 12:54:01 -0700325 // typename boost::asio::ip::basic_resolver< Protocol > resolver;
326 shared_ptr<typename Protocol::resolver> resolver =
327 make_shared<typename Protocol::resolver>(ref(this->m_socket.get_io_service()));
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800328
Alexander Afanasyev49bb1fb2014-07-21 12:54:01 -0700329 resolver->async_resolve(query, bind(&Impl::resolveHandler, this, _1, _2, resolver));
Alexander Afanasyev5964fb72014-02-18 12:42:45 -0800330 }
331 }
332};
333
334
335} // namespace ndn
336
337#endif // NDN_TRANSPORT_STREAM_TRANSPORT_HPP