source: trunk/FACT++/src/Connection.cc@ 12105

Last change on this file since 12105 was 11528, checked in by tbretz, 13 years ago
Replace a v6 loopback address by a v4 loopback address; we do not support v6.
File size: 13.6 KB
Line 
1// **************************************************************************
2/** @class Connection
3
4@brief Maintains an ansynchronous TCP/IP client connection
5
6@todo
7 Unify with ConnectionUSB
8
9*/
10// **************************************************************************
11#include "Connection.h"
12
13using namespace std;
14
15namespace ba = boost::asio;
16namespace bs = boost::system;
17namespace dummy = ba::placeholders;
18
19using ba::ip::tcp;
20
21 // -------- Abbreviations for starting async tasks ---------
22
23int Connection::Write(const Time &t, const string &txt, int qos)
24{
25 if (fLog)
26 return fLog->Write(t, txt, qos);
27
28 return MessageImp::Write(t, txt, qos);
29}
30
31void Connection::AsyncRead(const ba::mutable_buffers_1 buffers, int type)
32{
33 ba::async_read(*this, buffers,
34 boost::bind(&Connection::HandleReceivedData, this,
35 dummy::error, dummy::bytes_transferred, type));
36}
37
38void Connection::AsyncWrite(const ba::const_buffers_1 &buffers)
39{
40 ba::async_write(*this, buffers,
41 boost::bind(&Connection::HandleSentData, this,
42 dummy::error, dummy::bytes_transferred));
43}
44
45/*
46void Connection::AsyncWait(ba::deadline_timer &timer, int millisec,
47 void (Connection::*handler)(const bs::error_code&))
48{
49 // - The boost::asio::basic_deadline_timer::expires_from_now()
50 // function cancels any pending asynchronous waits, and returns
51 // the number of asynchronous waits that were cancelled. If it
52 // returns 0 then you were too late and the wait handler has
53 // already been executed, or will soon be executed. If it
54 // returns 1 then the wait handler was successfully cancelled.
55 // - If a wait handler is cancelled, the bs::error_code passed to
56 // it contains the value bs::error::operation_aborted.
57 timer.expires_from_now(boost::posix_time::milliseconds(millisec));
58
59 timer.async_wait(boost::bind(handler, this, dummy::error));
60}
61*/
62
63void Connection::AsyncConnect(tcp::resolver::iterator iterator)
64{
65 tcp::endpoint endpoint = *iterator;
66
67 // AsyncConnect + Deadline
68 async_connect(endpoint,
69 boost::bind(&Connection::ConnectIter,
70 this, iterator, ba::placeholders::error));
71
72 // We will get a "Connection timeout anyway"
73 //AsyncWait(fConnectTimeout, 5, &Connection::HandleConnectTimeout);
74}
75
76void Connection::AsyncConnect()
77{
78 // AsyncConnect + Deadline
79 async_connect(fEndpoint,
80 boost::bind(&Connection::ConnectAddr,
81 this, fEndpoint, ba::placeholders::error));
82
83 // We will get a "Connection timeout anyway"
84 //AsyncWait(fConnectTimeout, 5, &Connection::HandleConnectTimeout);
85}
86
87// ------------------------ close --------------------------
88// close from another thread
89void Connection::CloseImp(bool restart)
90{
91 if (IsConnected())
92 {
93 ostringstream str;
94 str << "Connection closed to " << URL() << ".";
95 Info(str);
96 }
97
98 // Stop any pending connection attempt
99 fConnectionTimer.cancel();
100
101 // Close possible open connections
102 close();
103
104 // Reset the connection status
105 fConnectionStatus = kDisconnected;
106
107 // Stop deadline counters
108 fInTimeout.cancel();
109 fOutTimeout.cancel();
110
111 // Empty output queue
112 fOutQueue.clear();
113
114 if (!restart || IsConnecting())
115 return;
116
117 // We need some timeout before reconnecting!
118 // And we have to check if we are alreayd trying to connect
119 // We shoudl wait until all operations in progress were canceled
120
121 // Start trying to reconnect
122 fMsgConnect = "";
123 fErrConnect = "";
124 StartConnect();
125}
126
127void Connection::PostClose(bool restart)
128{
129 get_io_service().post(boost::bind(&Connection::CloseImp, this, restart));
130}
131
132// ------------------------ write --------------------------
133void Connection::HandleWriteTimeout(const bs::error_code &error)
134{
135 if (error==ba::error::basic_errors::operation_aborted)
136 return;
137
138 // 125: Operation canceled (bs::error_code(125, bs::system_category))
139 if (error)
140 {
141 ostringstream str;
142 str << "Write timeout of " << URL() << ": " << error.message() << " (" << error << ")";// << endl;
143 Error(str);
144
145 CloseImp();
146 return;
147 }
148
149 if (!is_open())
150 {
151 // For example: Here we could schedule a new accept if we
152 // would not want to allow two connections at the same time.
153 return;
154 }
155
156 // Check whether the deadline has passed. We compare the deadline
157 // against the current time since a new asynchronous operation
158 // may have moved the deadline before this actor had a chance
159 // to run.
160 if (fOutTimeout.expires_at() > ba::deadline_timer::traits_type::now())
161 return;
162
163 Error("fOutTimeout has expired, writing data to "+URL());
164
165 CloseImp();
166}
167
168void Connection::HandleSentData(const bs::error_code& error, size_t n)
169{
170 if (error && error != ba::error::not_connected)
171 {
172 ostringstream str;
173 str << "Writing to " << URL() << ": " << error.message() << " (" << error << ")";// << endl;
174 Error(str);
175
176 CloseImp();
177 return;
178 }
179
180 if (error == ba::error::not_connected)
181 {
182 ostringstream msg;
183 msg << n << " bytes could not be sent to " << URL() << " due to missing connection.";
184 Warn(msg);
185 }
186 else
187 {
188 if (fDebugTx)
189 {
190 ostringstream msg;
191 msg << n << " bytes successfully sent to " << URL();
192 Message(msg);
193 }
194 }
195
196 // This is "thread" safe because SendMessage and HandleSentMessage
197 // are serialized in the EventQueue. Note: Do not call these
198 // functions directly from any other place then Handlers, use
199 // PostMessage instead
200 fOutQueue.pop_front();
201
202 if (fOutQueue.empty())
203 {
204 // Queue went empty, remove deadline
205 fOutTimeout.cancel();
206 return;
207 }
208
209 // AsyncWrite + Deadline
210 AsyncWrite(ba::const_buffers_1(fOutQueue.front().data(), fOutQueue.front().size())/*, &Connection::HandleSentData*/);
211 AsyncWait(fOutTimeout, 5000, &Connection::HandleWriteTimeout);
212}
213
214// It is important that when SendMessageImp is called, or to be more
215// precise boost::bind is called, teh data is copied!
216void Connection::SendMessageImp(const vector<char> msg)
217{
218 /*
219 if (!fConnectionEstablished)
220 {
221 UpdateWarn("SendMessageImp, but no connection to "+fAddress+":"+fPort+".");
222 return;
223 }*/
224
225 const bool first_message_in_queue = fOutQueue.empty();
226
227 // This is "thread" safe because SendMessage and HandleSentMessage
228 // are serialized in the EventQueue. Note: Do not call these
229 // functions directly from any other place then Handlers, use
230 // PostMessage instead
231 fOutQueue.push_back(msg);
232
233 if (!first_message_in_queue)
234 return;
235
236 // AsyncWrite + Deadline
237 AsyncWrite(ba::const_buffers_1(fOutQueue.front().data(), fOutQueue.front().size())/*, &Connection::HandleSentData*/);
238 AsyncWait(fOutTimeout, 5000, &Connection::HandleWriteTimeout);
239}
240
241void Connection::PostMessage(const void *ptr, size_t max)
242{
243 const vector<char> msg(reinterpret_cast<const char*>(ptr),
244 reinterpret_cast<const char*>(ptr)+max);
245
246 get_io_service().post(boost::bind(&Connection::SendMessageImp, this, msg));
247}
248
249void Connection::PostMessage(const string &cmd, size_t max)
250{
251 if (max==size_t(-1))
252 max = cmd.length()+1;
253
254 vector <char>msg(max);
255
256 copy(cmd.begin(), cmd.begin()+min(cmd.length()+1, max), msg.begin());
257
258 PostMessage(msg);
259}
260
261void Connection::HandleConnectionTimer(const bs::error_code &error)
262{
263 if (error==ba::error::basic_errors::operation_aborted)
264 return;
265
266 if (error)
267 {
268 ostringstream str;
269 str << "Connetion timer of " << URL() << ": " << error.message() << " (" << error << ")";// << endl;
270 Error(str);
271 }
272
273 if (is_open())
274 {
275 // For example: Here we could schedule a new accept if we
276 // would not want to allow two connections at the same time.
277 return;
278 }
279
280 // Check whether the deadline has passed. We compare the deadline
281 // against the current time since a new asynchronous operation
282 // may have moved the deadline before this actor had a chance
283 // to run.
284 if (fConnectionTimer.expires_at() < ba::deadline_timer::traits_type::now())
285 StartConnect();
286}
287
288bool Connection::ConnectImp(const tcp::endpoint &endpoint, const bs::error_code& error)
289{
290 const string host = endpoint.port()==0 ? "" :
291 endpoint.address().to_string()+':'+to_string((long long unsigned int)endpoint.port());
292
293 // Connection established
294 if (!error)
295 {
296 set_option(socket_base::keep_alive(true));
297
298 const int optval = 10;
299 // First keep alive after 10s
300 setsockopt(native(), SOL_TCP, TCP_KEEPIDLE, &optval, sizeof(optval));
301 // New keep alive after 10s
302 setsockopt(native(), SOL_TCP, TCP_KEEPINTVL, &optval, sizeof(optval));
303
304 Info("Connection established to "+host+"...");
305
306 fConnectionStatus = kConnected;
307
308 ConnectionEstablished();
309 return true;
310 }
311
312 // If returning from run will lead to deletion of this
313 // instance, close() is not needed (maybe implicitly called).
314 // If run is called again, close() is needed here. Otherwise:
315 // Software caused connection abort when we try to resolve
316 // the endpoint again.
317 CloseImp(false);
318
319 ostringstream msg;
320 if (!host.empty())
321 msg << "Connecting to " << host << ": " << error.message() << " (" << error << ")";
322
323 if (fErrConnect!=msg.str())
324 {
325 if (error!=ba::error::basic_errors::connection_refused)
326 fMsgConnect = "";
327 fErrConnect = msg.str();
328 Warn(fErrConnect);
329 }
330
331 if (error==ba::error::basic_errors::operation_aborted)
332 return true;
333
334 fConnectionStatus = kConnecting;
335
336 return false;
337/*
338 // Go on with the next
339 if (++iterator != tcp::resolver::iterator())
340 {
341 AsyncConnect(iterator);
342 return;
343 }
344*/
345 // No more entries to try, if we would not put anything else
346 // into the queue anymore it would now return (run() would return)
347
348 // Since we don't want to block the main loop, we wait using an
349 // asnychronous timer
350
351 // FIXME: Should we move this before AsyncConnect() ?
352// AsyncWait(fConnectionTimer, 250, &Connection::HandleConnectionTimer);
353}
354
355void Connection::ConnectIter(tcp::resolver::iterator iterator, const bs::error_code& error)
356{
357 if (ConnectImp(*iterator, error))
358 return;
359
360 // Go on with the next
361 if (++iterator != tcp::resolver::iterator())
362 {
363 AsyncConnect(iterator);
364 return;
365 }
366
367 // No more entries to try, if we would not put anything else
368 // into the queue anymore it would now return (run() would return)
369 AsyncWait(fConnectionTimer, 250, &Connection::HandleConnectionTimer);
370}
371
372void Connection::ConnectAddr(const tcp::endpoint &endpoint, const bs::error_code& error)
373{
374 if (ConnectImp(endpoint, error))
375 return;
376
377 AsyncWait(fConnectionTimer, 250, &Connection::HandleConnectionTimer);
378}
379
380// FIXME: Async connect should get address and port as an argument
381void Connection::StartConnect()
382{
383 fConnectionStatus = kConnecting;
384
385 if (fEndpoint!=tcp::endpoint())
386 {
387 ostringstream msg;
388 msg << "Trying to connect to " << fEndpoint << "...";
389 if (fMsgConnect!=msg.str())
390 {
391 fMsgConnect = msg.str();
392 Info(msg);
393 }
394
395 AsyncConnect();
396 return;
397 }
398
399 tcp::resolver resolver(get_io_service());
400
401 boost::system::error_code ec;
402
403 tcp::resolver::query query(fAddress, fPort);
404 tcp::resolver::iterator iterator = resolver.resolve(query, ec);
405
406 ostringstream msg;
407 if (!fAddress.empty() || !fPort.empty() || ec)
408 msg << "Trying to connect to " << URL() << "...";
409
410 if (ec)
411 msg << " " << ec.message() << " (" << ec << ")";
412
413 // Only output message if it has changed
414 if (fMsgConnect!=msg.str())
415 {
416 fMsgConnect = msg.str();
417 ec ? Error(msg) : Info(msg);
418 }
419
420 if (ec)
421 AsyncWait(fConnectionTimer, 250, &Connection::HandleConnectionTimer);
422 else
423 // Start connection attempts (will also reset deadline counter)
424 AsyncConnect(iterator);
425}
426
427void Connection::SetEndpoint(const string &addr, int port)
428{
429 if (fConnectionStatus>=1)
430 Warn("Connection or connection attempt in progress. New endpoint only valid for next connection.");
431
432 fAddress = addr;
433 fPort = to_string((long long)port);
434}
435
436void Connection::SetEndpoint(const string &addr, const string &port)
437{
438 if (fConnectionStatus>=1 && URL()!=":")
439 Warn("Connection or connection attempt in progress. New endpoint only valid for next connection.");
440
441 fAddress = addr;
442 fPort = port;
443}
444
445void Connection::SetEndpoint(const string &addr)
446{
447 const size_t p0 = addr.find_first_of(':');
448 const size_t p1 = addr.find_last_of(':');
449
450 if (p0==string::npos || p0!=p1)
451 {
452 Error("Connection::SetEndpoint - Wrong format of argument '"+addr+"' ('host:port' expected)");
453 return;
454 }
455
456 SetEndpoint(addr.substr(0, p0), addr.substr(p0+1));
457}
458
459void Connection::SetEndpoint(const tcp::endpoint &ep)
460{
461 const ba::ip::address addr = ep.address();
462
463 const ba::ip::address use =
464 addr.is_v6() && addr.to_v6().is_loopback() ?
465 ba::ip::address(ba::ip::address_v4::loopback()) :
466 addr;
467
468 SetEndpoint(use.to_string(), ep.port());
469
470 fEndpoint = tcp::endpoint(use, ep.port());
471}
472
473
474Connection::Connection(ba::io_service& ioservice, ostream &out) :
475MessageImp(out), tcp::socket(ioservice), fLog(0),
476fInTimeout(ioservice), fOutTimeout(ioservice), fConnectionTimer(ioservice),
477fConnectionStatus(kDisconnected)
478{
479}
Note: See TracBrowser for help on using the repository browser.