Line data Source code
1 : // Copyright (c) 2012-2024 Made to Order Software Corp. All Rights Reserved
2 : //
3 : // https://snapwebsites.org/project/eventdispatcher
4 : // contact@m2osw.com
5 : //
6 : // This program is free software; you can redistribute it and/or modify
7 : // it under the terms of the GNU General Public License as published by
8 : // the Free Software Foundation; either version 2 of the License, or
9 : // (at your option) any later version.
10 : //
11 : // This program is distributed in the hope that it will be useful,
12 : // but WITHOUT ANY WARRANTY; without even the implied warranty of
13 : // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14 : // GNU General Public License for more details.
15 : //
16 : // You should have received a copy of the GNU General Public License along
17 : // with this program; if not, write to the Free Software Foundation, Inc.,
18 : // 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
19 :
20 : /** \file
21 : * \brief Implementation client permanent message connection.
22 : *
23 : * This class implements a permanent connection to a Unix socket. This
24 : * means if the server is restarted, this class is capable to automatically
25 : * reconnect under the hood. This can be done using a thread so if the
26 : * connect() command is slow (needs to time out), it won't block the
27 : * rest of your event loop.
28 : */
29 :
30 :
31 : // self
32 : //
33 : #include "eventdispatcher/local_stream_client_permanent_message_connection.h"
34 :
35 : #include "eventdispatcher/communicator.h"
36 : #include "eventdispatcher/exception.h"
37 : #include "eventdispatcher/local_stream_client_message_connection.h"
38 : #include "eventdispatcher/thread_done_signal.h"
39 :
40 :
41 : // snaplogger
42 : //
43 : #include <snaplogger/message.h>
44 :
45 :
46 : // cppthread
47 : //
48 : #include <cppthread/exception.h>
49 : #include <cppthread/guard.h>
50 : #include <cppthread/runner.h>
51 : #include <cppthread/thread.h>
52 :
53 :
54 : // last include
55 : //
56 : #include <snapdev/poison.h>
57 :
58 :
59 :
60 : namespace ed
61 : {
62 :
63 :
64 :
65 : namespace detail
66 : {
67 :
68 :
69 : /** \brief Internal implementation of the local_stream_client_permanent_message_connection class.
70 : *
71 : * This class is used to handle a thread that will process a connection for
72 : * us. This allows us to connect in any amount of time required by the
73 : * Unix system to obtain the connection with the remote server.
74 : *
75 : * \todo
76 : * Having threads at the time we do a fork() is not safe. We may
77 : * want to reconsider offering this functionality here. Because at
78 : * this time we would have no control of when the thread is created
79 : * and thus a way to make sure that no such thread is running when
80 : * we call fork().
81 : */
82 : class local_stream_client_permanent_message_connection_impl
83 : {
84 : public:
85 : class messenger
86 : : public local_stream_client_message_connection
87 : {
88 : public:
89 : typedef std::shared_ptr<messenger> pointer_t;
90 :
91 1 : messenger(
92 : local_stream_client_permanent_message_connection * parent
93 : , addr::addr_unix const & address
94 : , bool const blocking
95 : , bool const close_on_exec)
96 1 : : local_stream_client_message_connection(
97 : address
98 : , blocking
99 : , close_on_exec)
100 1 : , f_parent(parent)
101 : {
102 3 : set_name("local_stream_client_permanent_message_connection_impl::messenger");
103 1 : }
104 :
105 : messenger(messenger const & rhs) = delete;
106 : messenger & operator = (messenger const & rhs) = delete;
107 :
108 : // connection implementation
109 2 : virtual void process_empty_buffer()
110 : {
111 2 : local_stream_client_message_connection::process_empty_buffer();
112 2 : f_parent->process_empty_buffer();
113 2 : }
114 :
115 : // connection implementation
116 0 : virtual void process_error()
117 : {
118 0 : local_stream_client_message_connection::process_error();
119 0 : f_parent->process_error();
120 0 : }
121 :
122 : // connection implementation
123 0 : virtual void process_hup()
124 : {
125 0 : local_stream_client_message_connection::process_hup();
126 0 : f_parent->process_hup();
127 0 : }
128 :
129 : // connection implementation
130 0 : virtual void process_invalid()
131 : {
132 0 : local_stream_client_message_connection::process_invalid();
133 0 : f_parent->process_invalid();
134 0 : }
135 :
136 : // local_stream_server_client_message_connection implementation
137 1 : virtual void process_message(message & msg)
138 : {
139 : // We call the dispatcher from our parent since the child
140 : // (this messenger) is not given a dispatcher
141 : //
142 1 : f_parent->dispatch_message(msg);
143 1 : }
144 :
145 : private:
146 : local_stream_client_permanent_message_connection * f_parent = nullptr;
147 : };
148 :
149 : class thread_signal_handler
150 : : public thread_done_signal
151 : {
152 : public:
153 : typedef std::shared_ptr<thread_signal_handler> pointer_t;
154 :
155 1 : thread_signal_handler(local_stream_client_permanent_message_connection_impl * parent_impl)
156 1 : : f_parent_impl(parent_impl)
157 : {
158 3 : set_name("local_stream_client_permanent_message_connection_impl::thread_signal_handler");
159 1 : }
160 :
161 : thread_signal_handler(thread_signal_handler const & rhs) = delete;
162 : thread_signal_handler & operator = (thread_signal_handler const & rhs) = delete;
163 :
164 : /** \brief This signal was emitted.
165 : *
166 : * This function gets called whenever the thread is just about to
167 : * quit. Calling f_thread.is_running() may still return true when
168 : * you get in the 'thread_done()' callback. However, an
169 : * f_thread.stop() will return very quickly.
170 : */
171 1 : virtual void process_read()
172 : {
173 1 : thread_done_signal::process_read();
174 :
175 1 : f_parent_impl->thread_done();
176 1 : }
177 :
178 : private:
179 : local_stream_client_permanent_message_connection_impl * f_parent_impl = nullptr;
180 : };
181 :
182 : class runner
183 : : public cppthread::runner
184 : {
185 : public:
186 1 : runner(
187 : local_stream_client_permanent_message_connection_impl * parent_impl
188 : , addr::addr_unix const & address
189 : , bool const blocking = false
190 : , bool const close_on_exec = true)
191 1 : : cppthread::runner("background local_stream_client_permanent_message_connection for asynchronous connections")
192 1 : , f_parent_impl(parent_impl)
193 1 : , f_address(address)
194 1 : , f_blocking(blocking)
195 3 : , f_close_on_exec(close_on_exec)
196 : {
197 1 : }
198 :
199 : runner(runner const & rhs) = delete;
200 : runner & operator = (runner const & rhs) = delete;
201 :
202 :
203 : /** \brief This is the actual function run by the thread.
204 : *
205 : * This function calls the connect() function and then
206 : * tells the main thread we are done.
207 : */
208 1 : virtual void run()
209 : {
210 1 : connect();
211 :
212 : // tell the main thread that we are done
213 : //
214 1 : f_parent_impl->trigger_thread_done();
215 1 : }
216 :
217 :
218 : /** \brief This function attempts to connect.
219 : *
220 : * This function attempts a connection to the specified address
221 : * and port with the specified mode (i.e. plain or encrypted.)
222 : *
223 : * The function may take a long time to succeed connecting with
224 : * the server. The main thread will be awaken whenever this
225 : * thread dies.
226 : *
227 : * If an error occurs, then the f_socket variable member will
228 : * be set to -1. Otherwise it represents the socket that we
229 : * just connected with.
230 : */
231 1 : void connect()
232 : {
233 1 : char const * error_name(nullptr);
234 : try
235 : {
236 3 : f_messenger = std::make_shared<messenger>(
237 2 : f_parent_impl->parent()
238 1 : , f_address
239 1 : , f_blocking
240 2 : , f_close_on_exec);
241 1 : return;
242 : }
243 0 : catch(initialization_error const & e)
244 : {
245 0 : error_name = "initialization_error";
246 0 : f_last_error = e.what();
247 0 : }
248 0 : catch(runtime_error const & e)
249 : {
250 0 : error_name = "runtime_error";
251 0 : f_last_error = e.what();
252 0 : }
253 0 : catch(std::exception const & e)
254 : {
255 0 : error_name = "std::exception";
256 0 : f_last_error = e.what();
257 0 : }
258 0 : catch(...)
259 : {
260 0 : error_name = "... (any other exception)";
261 0 : f_last_error = "Unknown exception";
262 0 : }
263 0 : f_messenger.reset();
264 :
265 : // connection failed... we will have to try again later
266 : //
267 0 : SNAP_LOG_ERROR
268 : << "connection to "
269 0 : << f_address.to_string()
270 : << " failed with: "
271 0 : << f_last_error
272 : << " ("
273 : << error_name
274 : << ")"
275 : << SNAP_LOG_SEND;
276 : }
277 :
278 :
279 : /** \brief Retrieve the address to connect to.
280 : *
281 : * This function returns the address passed in on creation.
282 : *
283 : * \note
284 : * Since the variable is constant, it is likely to never change.
285 : * However, the c_str() function may change the buffer pointer.
286 : * Hence, to be 100% safe, you cannot call this function until
287 : * you make sure that the thread is fully stopped.
288 : *
289 : * \return The destination address.
290 : */
291 0 : addr::addr_unix get_address() const
292 : {
293 0 : return f_address;
294 : }
295 :
296 :
297 : /** \brief Retrieve the client allocated and connected by the thread.
298 : *
299 : * This function returns the TCP connection object resulting from
300 : * connection attempts of the background thread.
301 : *
302 : * If the pointer is null, then you may get the corresponding
303 : * error message using the get_last_error() function.
304 : *
305 : * You can get the client TCP connection pointer once. After that
306 : * you always get a null pointer.
307 : *
308 : * \note
309 : * This function is guarded so the pointer and the object it
310 : * points to will be valid in another thread that retrieves it.
311 : *
312 : * \return The connection pointer.
313 : */
314 1 : messenger::pointer_t release_client()
315 : {
316 1 : cppthread::guard g(f_mutex);
317 1 : messenger::pointer_t release;
318 1 : release.swap(f_messenger);
319 2 : return release;
320 1 : }
321 :
322 :
323 : /** \brief Retrieve the last error message that happened.
324 : *
325 : * This function returns the last error message that was captured
326 : * when trying to connect to the socket. The message is the
327 : * e.what() message from the exception we captured.
328 : *
329 : * The message does not get cleared so the function can be called
330 : * any number of times. To know whether an error was generated
331 : * on the last attempt, make sure to first get the get_socket()
332 : * and if it returns -1, then this message is significant,
333 : * otherwise it is from a previous error.
334 : *
335 : * \warning
336 : * Remember that if the background thread was used the error will
337 : * NOT be available in the main thread until a full memory barrier
338 : * was executed. For that reason we make sure that the thread
339 : * was stopped when we detect an error.
340 : *
341 : * \return The last error message.
342 : */
343 0 : std::string const & get_last_error() const
344 : {
345 0 : return f_last_error;
346 : }
347 :
348 :
349 : /** \brief Close the connection.
350 : *
351 : * This function closes the connection. Since the f_local_stream_connection
352 : * holds the socket to the remote server, we have get this function
353 : * called in order to completely disconnect.
354 : *
355 : * \note
356 : * This function does not clear the f_last_error parameter so it
357 : * can be read later.
358 : */
359 1 : void close()
360 : {
361 1 : f_messenger.reset();
362 1 : }
363 :
364 :
365 : private:
366 : local_stream_client_permanent_message_connection_impl *
367 : f_parent_impl = nullptr;
368 : addr::addr_unix const f_address;
369 : bool const f_blocking;
370 : bool const f_close_on_exec;
371 : messenger::pointer_t f_messenger = messenger::pointer_t();
372 : std::string f_last_error = std::string();
373 : };
374 :
375 :
376 : /** \brief Initialize a permanent message connection implementation object.
377 : *
378 : * This object manages the thread used to asynchronically connect to
379 : * the specified address and port.
380 : *
381 : * This class and its sub-classes may end up executing callbacks
382 : * of the local_stream_client_permanent_message_connection object.
383 : * However, in all cases these are never run from the thread.
384 : *
385 : * \param[in] parent A pointer to the owner of this
386 : * local_stream_client_permanent_message_connection_impl object.
387 : * \param[in] address The address we are to connect to.
388 : * \param[in] blocking Whether to open in blocking mode or not.
389 : * \param[in] close_on_exec Whether to mark the socket as requiring to
390 : * be closed on an exec() call.
391 : */
392 1 : local_stream_client_permanent_message_connection_impl(
393 : local_stream_client_permanent_message_connection * parent
394 : , addr::addr_unix const & address
395 : , bool const blocking
396 : , bool const close_on_exec)
397 1 : : f_parent(parent)
398 1 : , f_thread_runner(this, address, blocking, close_on_exec)
399 4 : , f_thread("background connection handler thread", &f_thread_runner)
400 : {
401 1 : }
402 :
403 :
404 : local_stream_client_permanent_message_connection_impl(local_stream_client_permanent_message_connection_impl const & rhs) = delete;
405 : local_stream_client_permanent_message_connection_impl & operator = (local_stream_client_permanent_message_connection_impl const & rhs) = delete;
406 :
407 : /** \brief Destroy the permanent message connection.
408 : *
409 : * This function makes sure that the messenger was lost.
410 : */
411 1 : ~local_stream_client_permanent_message_connection_impl()
412 : {
413 : // to make sure we can lose the messenger, first we want to be sure
414 : // that we do not have a thread running
415 : //
416 : try
417 : {
418 1 : f_thread.stop();
419 : }
420 0 : catch(cppthread::mutex_failed_error const &)
421 : {
422 0 : }
423 0 : catch(cppthread::invalid_error const &)
424 : {
425 0 : }
426 :
427 : // in this case we may still have an instance of the f_thread_done
428 : // which linger around, we want it out
429 : //
430 : // Note: the call is safe even if the f_thread_done is null
431 : //
432 1 : communicator::instance()->remove_connection(f_thread_done);
433 :
434 : // although the f_messenger variable gets reset automatically in
435 : // the destructor, it would not get removed from the
436 : // communicator instance if we were not doing it explicitly
437 : //
438 1 : disconnect();
439 1 : }
440 :
441 :
442 : /** \brief Direct connect to the messenger.
443 : *
444 : * In this case we try to connect without the thread. This allows
445 : * us to avoid the thread problems, but we are blocked until the
446 : * OS decides to time out or the connection worked.
447 : */
448 0 : void connect()
449 : {
450 0 : if(f_done)
451 : {
452 0 : SNAP_LOG_ERROR
453 : << "Permanent connection marked done. Cannot attempt to reconnect."
454 : << SNAP_LOG_SEND;
455 0 : return;
456 : }
457 :
458 : // call the thread connect() function from the main thread
459 : //
460 0 : f_thread_runner.connect();
461 :
462 : // simulate receiving the thread_done() signal
463 : //
464 0 : thread_done();
465 : }
466 :
467 :
468 : /** \brief Check whether the permanent connection is currently connected.
469 : *
470 : * This function returns true if the messenger exists, which means that
471 : * the connection is up.
472 : *
473 : * \return true if the connection is up.
474 : */
475 0 : bool is_connected()
476 : {
477 0 : return f_messenger != nullptr;
478 : }
479 :
480 :
481 : /** \brief Try to start the thread runner.
482 : *
483 : * This function tries to start the thread runner in order to initiate
484 : * a connection in the background. If the thread could not be started,
485 : * then the function returns false.
486 : *
487 : * If the thread started, then the function returns true. This does
488 : * not mean that the connection was obtained. This is known once
489 : * the process_connected() function is called.
490 : *
491 : * \return true if the thread was successfully started.
492 : */
493 1 : bool background_connect()
494 : {
495 1 : if(f_done)
496 : {
497 0 : SNAP_LOG_ERROR
498 : << "Permanent connection marked done. Cannot attempt to reconnect."
499 : << SNAP_LOG_SEND;
500 0 : return false;
501 : }
502 :
503 1 : if(f_thread.is_running())
504 : {
505 0 : SNAP_LOG_ERROR
506 : << "A background connection attempt is already in progress. Further requests are ignored."
507 : << SNAP_LOG_SEND;
508 0 : return false;
509 : }
510 :
511 : // create the f_thread_done only when required
512 : //
513 1 : if(f_thread_done == nullptr)
514 : {
515 1 : f_thread_done = std::make_shared<thread_signal_handler>(this);
516 : }
517 :
518 1 : communicator::instance()->add_connection(f_thread_done);
519 :
520 1 : if(!f_thread.start())
521 : {
522 0 : SNAP_LOG_ERROR
523 : << "The thread used to run the background connection process did not start."
524 : << SNAP_LOG_SEND;
525 0 : return false;
526 : }
527 :
528 1 : return true;
529 : }
530 :
531 :
532 : /** \brief Tell the main thread that the background thread is done.
533 : *
534 : * This function is called by the thread so the thread_done()
535 : * function of the thread done object gets called. Only the
536 : * thread should call this function.
537 : *
538 : * As a result the thread_done() function of this class will be
539 : * called by the main thread.
540 : */
541 1 : void trigger_thread_done()
542 : {
543 1 : f_thread_done->thread_done();
544 1 : }
545 :
546 :
547 : /** \brief Signal that the background thread is done.
548 : *
549 : * This callback is called whenever the background thread sends
550 : * a signal to us. This is used to avoid calling end user functions
551 : * that would certainly cause a lot of problem if called from the
552 : * thread.
553 : *
554 : * The function calls the process_connection_failed() if the
555 : * connection did not happen.
556 : *
557 : * The function calls the process_connected() if the connection
558 : * did happen.
559 : *
560 : * \note
561 : * This is used only if the user requested that the connection
562 : * happen in the background (i.e. use_thread was set to true
563 : * in the local_stream_client_permanent_message_connection object
564 : * constructor.)
565 : */
566 1 : void thread_done()
567 : {
568 : // if we used the thread we have to remove the signal used
569 : // to know that the thread was done
570 : //
571 1 : communicator::instance()->remove_connection(f_thread_done);
572 :
573 : // we will access the f_last_error member of the thread runner
574 : // which may not be available to the main thread yet, calling
575 : // stop forces a memory barrier so we are all good.
576 : //
577 : // calling stop() has no effect if we did not use the thread,
578 : // however, not calling stop() when we did use the thread
579 : // causes all sorts of other problems (especially, the thread
580 : // never gets joined)
581 : //
582 1 : f_thread.stop();
583 :
584 1 : messenger::pointer_t client(f_thread_runner.release_client());
585 1 : if(f_done)
586 : {
587 : // already marked done, ignore the result and lose the
588 : // connection immediately
589 : //
590 : //f_thread_running.close(); -- not necessary, 'client' is the connection
591 0 : return;
592 : }
593 :
594 1 : if(client == nullptr)
595 : {
596 : // TODO: fix address in error message using a addr::addr so
597 : // as to handle IPv6 seamlessly.
598 : //
599 0 : SNAP_LOG_ERROR
600 : << "connection to "
601 0 : << f_thread_runner.get_address().to_uri()
602 : << " failed with: "
603 0 : << f_thread_runner.get_last_error()
604 : << SNAP_LOG_SEND;
605 :
606 : // signal that an error occurred
607 : //
608 0 : f_parent->process_connection_failed(f_thread_runner.get_last_error());
609 : }
610 : else
611 : {
612 1 : f_messenger = client;
613 :
614 : // add the messenger to the communicator
615 : //
616 1 : communicator::instance()->add_connection(f_messenger);
617 :
618 : // if some messages were cached, process them immediately
619 : //
620 2 : while(!f_message_cache.empty())
621 : {
622 1 : f_messenger->send_message(f_message_cache[0]);
623 1 : f_message_cache.erase(f_message_cache.begin());
624 : }
625 :
626 : // let the client know we are now connected
627 : //
628 1 : f_parent->process_connected();
629 : }
630 1 : }
631 :
632 : /** \brief Send a message to the connection.
633 : *
634 : * This implementation function actually sends the message to the
635 : * connection, assuming that the connection exists. Otherwise, it
636 : * may cache the message (if cache is true.)
637 : *
638 : * Note that the message does not get cached if mark_done() was
639 : * called earlier since we are trying to close the whole connection.
640 : *
641 : * \param[in] msg The message to send.
642 : * \param[in] cache Whether to cache the message if the connection is
643 : * currently down.
644 : *
645 : * \return true if the message was forwarded, false if the message
646 : * was ignored or cached.
647 : */
648 2 : bool send_message(message & msg, bool cache)
649 : {
650 2 : if(f_messenger != nullptr)
651 : {
652 1 : return f_messenger->send_message(msg);
653 : }
654 :
655 1 : if(cache && !f_done)
656 : {
657 1 : f_message_cache.push_back(msg);
658 : }
659 :
660 1 : return false;
661 : }
662 :
663 :
664 : /** \brief Forget about the messenger connection.
665 : *
666 : * This function is used to fully disconnect from the messenger.
667 : *
668 : * If there is a messenger, this means:
669 : *
670 : * \li Removing the messenger from the communicator instance.
671 : * \li Closing the connection in the thread object.
672 : *
673 : * In most cases, it is called when an error occur, also it happens
674 : * that we call it explicitly through the disconnect() function
675 : * of the permanent connection class.
676 : *
677 : * \note
678 : * This is safe, even though it is called from the messenger itself
679 : * because it will not get deleted yet. This is because the run()
680 : * loop has a copy in its own temporary copy of the vector of
681 : * connections.
682 : */
683 2 : void disconnect()
684 : {
685 2 : if(f_messenger != nullptr)
686 : {
687 1 : communicator::instance()->remove_connection(f_messenger);
688 1 : f_messenger.reset();
689 :
690 : // just the messenger does not close the TCP connection because
691 : // we may have another in the thread runner
692 : //
693 1 : f_thread_runner.close();
694 : }
695 2 : }
696 :
697 :
698 : /** \brief Return the address and size of the remote computer.
699 : *
700 : * This function retrieve the socket address.
701 : *
702 : * \warning
703 : * If the socket is not currently connected, the function returns
704 : * a default Unix address. This means it returns a valid unnamed
705 : * address.
706 : *
707 : * \return The Unix address used to connect.
708 : */
709 0 : addr::addr_unix get_address() const
710 : {
711 0 : if(f_messenger != nullptr)
712 : {
713 0 : return f_messenger->get_address();
714 : }
715 0 : return addr::addr_unix();
716 : }
717 :
718 :
719 : /** \brief Mark the messenger as done.
720 : *
721 : * This function is used to mark the messenger as done. This means it
722 : * will get removed from the communicator instance as soon as it
723 : * is done with its current write buffer if there is one.
724 : *
725 : * You may also want to call the disconnection() function to actually
726 : * reset the pointer along the way.
727 : */
728 1 : void mark_done()
729 : {
730 1 : f_done = true;
731 :
732 : // once done we don't attempt to reconnect so we can as well
733 : // get rid of our existing cache immediately to save some
734 : // memory
735 : //
736 1 : f_message_cache.clear();
737 :
738 1 : if(f_messenger != nullptr)
739 : {
740 1 : f_messenger->mark_done();
741 : }
742 1 : }
743 :
744 :
745 : /** \brief Retrieve the parent of the impl.
746 : *
747 : * This function returns the parent of the impl, which is the main
748 : * local_stream_client_permanent_message_connection point. This is
749 : * saved in the messenger so we can relay events from our internal
750 : * messenger implementation to the main class owned by the user.
751 : *
752 : * \return The pointer to the parent of the impl.
753 : */
754 1 : local_stream_client_permanent_message_connection * parent() const
755 : {
756 1 : return f_parent;
757 : }
758 :
759 :
760 : private:
761 : local_stream_client_permanent_message_connection *
762 : f_parent = nullptr;
763 : thread_signal_handler::pointer_t f_thread_done = thread_signal_handler::pointer_t();
764 : runner f_thread_runner;
765 : cppthread::thread f_thread;
766 : messenger::pointer_t f_messenger = messenger::pointer_t();
767 : message::vector_t f_message_cache = message::vector_t();
768 : bool f_done = false;
769 : };
770 :
771 :
772 :
773 : }
774 : // namespace detail
775 :
776 :
777 :
778 : /** \brief Initializes this TCP client message connection.
779 : *
780 : * This implementation creates what we call a permanent connection.
781 : * Such a connection may fail once in a while. In such circumstances,
782 : * the class automatically requests for a reconnection (see various
783 : * parameters in the regard below.) However, this causes one issue:
784 : * by default, the connection just never ends. When you are about
785 : * ready to close the connection, you must call the mark_done()
786 : * function first. This will tell the various error functions to
787 : * drop this connection instead of restarting it after a small pause.
788 : *
789 : * This constructor makes sure to initialize the timer and saves
790 : * the address, port, mode, pause, and use_thread parameters.
791 : *
792 : * The timer is first set to trigger immediately. This means the TCP
793 : * connection will be attempted as soon as possible (the next time
794 : * the run() loop is entered, it will time out immediately.) You
795 : * are free to call set_timeout_date() with a date in the future if
796 : * you prefer that the connect be attempted a little later.
797 : *
798 : * The \p pause parameter is used if the connection is lost and this
799 : * timer is used again to attempt a new connection. It will be reused
800 : * as long as the connection fails (as a delay). It has to be at least
801 : * 10 microseconds, although really you should not use less than 1
802 : * second (1000000). You may set the pause parameter to 0 in which case
803 : * you are responsible to set the delay (by default there will be no
804 : * delay and thus the timer will never time out.)
805 : *
806 : * To start with a delay, instead of trying to connect immediately,
807 : * you may pass a negative pause parameter. So for example to get the
808 : * first attempt 5 seconds after you created this object, you use
809 : * -5000000LL as the pause parameter.
810 : *
811 : * The \p use_thread parameter determines whether the connection should
812 : * be attempted in a thread (asynchronously) or immediately (which means
813 : * the timeout callback may block for a while.) If the connection is to
814 : * a local server with an IP address specified as numbers (i.e. 127.0.0.1),
815 : * the thread is probably not required. For connections to a remote
816 : * computer, though, it certainly is important.
817 : *
818 : * \param[in] address The address to listen on. It may be set to "0.0.0.0".
819 : * \param[in] pause The amount of time to wait before attempting a new
820 : * connection after a failure, in microseconds, or 0.
821 : * \param[in] use_thread Whether a thread is used to connect to the
822 : * server.
823 : * \param[in] blocking Whether the socket is going to be blocking or not.
824 : * \param[in] close_on_exec Automatically close the connection if the process
825 : * execute an exec() call.
826 : * \param[in] service_name The name of your daemon service. Only use once
827 : * on your permanent connection to snapcommunicator.
828 : */
829 1 : local_stream_client_permanent_message_connection::local_stream_client_permanent_message_connection(
830 : addr::addr_unix const & address
831 : , std::int64_t const pause
832 : , bool const use_thread
833 : , bool const blocking
834 : , bool const close_on_exec
835 1 : , std::string const & service_name)
836 : : timer(pause < 0 ? -pause : 0)
837 : , connection_with_send_message(service_name)
838 2 : , f_impl(std::make_shared<detail::local_stream_client_permanent_message_connection_impl>(
839 1 : this
840 : , address
841 : , blocking
842 : , close_on_exec))
843 1 : , f_pause(llabs(pause))
844 1 : , f_use_thread(use_thread)
845 : {
846 1 : }
847 :
848 :
849 : /** \brief Destroy instance.
850 : *
851 : * This function cleans up everything in the permanent message object.
852 : */
853 1 : local_stream_client_permanent_message_connection::~local_stream_client_permanent_message_connection()
854 : {
855 : // Does nothing
856 1 : }
857 :
858 :
859 : /** \brief Attempt to send a message to this connection.
860 : *
861 : * If the connection is currently enabled, the message is sent immediately.
862 : * Otherwise, it may be cached if the \p cache parameter is set to true.
863 : * A cached message is forwarded as soon as a new successful connection
864 : * happens, which can be a problem if messages need to happen in a very
865 : * specific order (For example, after a reconnection to snapcommunicator
866 : * you first need to REGISTER or CONNECT...)
867 : *
868 : * \param[in] msg The message to send to the connected server.
869 : * \param[in] cache Whether the message should be cached.
870 : *
871 : * \return true if the message was sent, false if it was not sent, although
872 : * if cache was true, it was cached
873 : */
874 2 : bool local_stream_client_permanent_message_connection::send_message(message & msg, bool cache)
875 : {
876 2 : return f_impl->send_message(msg, cache);
877 : }
878 :
879 :
880 : /** \brief Check whether the connection is up.
881 : *
882 : * This function returns true if the connection is considered to be up.
883 : * This means sending messages will work quickly instead of being
884 : * cached up until an actual TCP/IP connection gets established.
885 : *
886 : * Note that the connection may have hanged up since, and the system
887 : * may not have yet detected the fact (i.e. the connection is going
888 : * to receive the process_hup() call after the event in which you are
889 : * working.)
890 : *
891 : * \return true if connected
892 : */
893 0 : bool local_stream_client_permanent_message_connection::is_connected() const
894 : {
895 0 : return f_impl->is_connected();
896 : }
897 :
898 :
899 : /** \brief Disconnect the messenger now.
900 : *
901 : * This function kills the current connection.
902 : *
903 : * There are a few cases where two daemons communicate between each others
904 : * and at some point one of them wants to exit and needs to disconnect. This
905 : * function can be used in that one situation assuming that you have an
906 : * acknowledgement from the other daemon.
907 : *
908 : * Say you have daemon A and B. B wants to quit and before doing so sends
909 : * a form of "I'm quitting" message to A. In that situation, B is not closing
910 : * the messenger connection, A is responsible for that (i.e. A acknowledges
911 : * receipt of the "I'm quitting" message from B by closing the connection.)
912 : *
913 : * B also wants to call the mark_done() function to make sure that it
914 : * does not reconnected a split second later and instead the permanent
915 : * connection gets removed from the communicator list of connections.
916 : */
917 0 : void local_stream_client_permanent_message_connection::disconnect()
918 : {
919 0 : f_impl->disconnect();
920 0 : }
921 :
922 :
923 : /** \brief Overload so we do not have to use namespace everywhere.
924 : *
925 : * This function overloads the connection::mark_done() function so
926 : * we can call it without the need to use timer::mark_done()
927 : * everywhere.
928 : */
929 0 : void local_stream_client_permanent_message_connection::mark_done()
930 : {
931 0 : timer::mark_done();
932 0 : }
933 :
934 :
935 : /** \brief Mark connection as done.
936 : *
937 : * This function allows you to mark the permanent connection and the
938 : * messenger as done.
939 : *
940 : * Note that calling this function with false is the same as calling the
941 : * base class mark_done() function.
942 : *
943 : * If the \p message parameter is set to true, we suggest you also call
944 : * the disconnect() function. That way the messenger will truly get
945 : * removed from everyone quickly.
946 : *
947 : * \param[in] messenger If true, also mark the messenger as done.
948 : */
949 1 : void local_stream_client_permanent_message_connection::mark_done(bool messenger)
950 : {
951 1 : timer::mark_done();
952 1 : if(messenger)
953 : {
954 1 : f_impl->mark_done();
955 : }
956 1 : }
957 :
958 :
959 : /** \brief Retrieve a copy of the client's address.
960 : *
961 : * This function makes a copy of the address of this client connection
962 : * to the \p address parameter and returns the length.
963 : *
964 : * \return Return a copy of the Unix address.
965 : */
966 0 : addr::addr_unix local_stream_client_permanent_message_connection::get_address() const
967 : {
968 0 : return f_impl->get_address();
969 : }
970 :
971 :
972 : /** \brief Internal timeout callback implementation.
973 : *
974 : * This callback implements the guts of this class: it attempts to connect
975 : * to the specified address and port, optionally after creating a thread
976 : * so the attempt can happen asynchronously.
977 : *
978 : * When the connection fails, the timer is used to try again pause
979 : * microseconds later (pause as specified in the constructor).
980 : *
981 : * When a connection succeeds, the timer is disabled until you detect
982 : * an error while using the connection and re-enable the timer.
983 : *
984 : * \warning
985 : * This function changes the timeout delay to the pause amount
986 : * as defined with the constructor. If you want to change that
987 : * amount, you can do so an any point after this function call
988 : * using the set_timeout_delay() function. If the pause parameter
989 : * was set to -1, then the timeout never gets changed.
990 : * However, you should not use a permanent message timer as your
991 : * own or you will interfere with the internal use of the timer.
992 : */
993 1 : void local_stream_client_permanent_message_connection::process_timeout()
994 : {
995 : // got a spurious call when already marked done
996 : //
997 1 : if(is_done())
998 : {
999 0 : return;
1000 : }
1001 :
1002 : // change the timeout delay although we will not use it immediately
1003 : // if we start the thread or attempt an immediate connection, but
1004 : // that way the user can change it by calling set_timeout_delay()
1005 : // at any time after the first process_timeout() call
1006 : //
1007 1 : if(f_pause > 0)
1008 : {
1009 1 : set_timeout_delay(f_pause);
1010 1 : f_pause = 0;
1011 : }
1012 :
1013 1 : if(f_use_thread)
1014 : {
1015 : // in this case we create a thread, run it and know whether the
1016 : // connection succeeded only when the thread tells us it did
1017 : //
1018 : // TODO: the background_connect() may return false in two situations:
1019 : // 1) when the thread is already running and then the behavior
1020 : // we have below is INCORRECT
1021 : // 2) when the thread cannot be started (i.e. could not
1022 : // allocate the stack?) in which case the if() below
1023 : // is the correct behavior
1024 : //
1025 1 : if(f_impl->background_connect())
1026 : {
1027 : // we started the thread successfully, so block the timer
1028 : //
1029 1 : set_enable(false);
1030 : }
1031 : }
1032 : else
1033 : {
1034 : // the success is noted when we receive a call to
1035 : // process_connected(); there we do set_enable(false)
1036 : // so the timer stops
1037 : //
1038 0 : f_impl->connect();
1039 : }
1040 : }
1041 :
1042 :
1043 : /** \brief Process an error.
1044 : *
1045 : * When an error occurs, we restart the timer so we can attempt to reconnect
1046 : * to that server.
1047 : *
1048 : * If you overload this function, make sure to either call this
1049 : * implementation or enable the timer yourselves.
1050 : *
1051 : * \warning
1052 : * This function does not call the timer::process_error() function
1053 : * which means that this connection is not automatically removed from
1054 : * the communicator object on failures.
1055 : */
1056 0 : void local_stream_client_permanent_message_connection::process_error()
1057 : {
1058 0 : if(is_done())
1059 : {
1060 0 : timer::process_error();
1061 : }
1062 : else
1063 : {
1064 0 : f_impl->disconnect();
1065 0 : set_enable(true);
1066 : }
1067 0 : }
1068 :
1069 :
1070 : /** \brief Process a hang up.
1071 : *
1072 : * When a hang up occurs, we restart the timer so we can attempt to reconnect
1073 : * to that server.
1074 : *
1075 : * If you overload this function, make sure to either call this
1076 : * implementation or enable the timer yourselves.
1077 : *
1078 : * \warning
1079 : * This function does not call the timer::process_hup() function
1080 : * which means that this connection is not automatically removed from
1081 : * the communicator object on failures.
1082 : */
1083 0 : void local_stream_client_permanent_message_connection::process_hup()
1084 : {
1085 0 : if(is_done())
1086 : {
1087 0 : timer::process_hup();
1088 : }
1089 : else
1090 : {
1091 0 : f_impl->disconnect();
1092 0 : set_enable(true);
1093 : }
1094 0 : }
1095 :
1096 :
1097 : /** \brief Process an invalid signal.
1098 : *
1099 : * When an invalid signal occurs, we restart the timer so we can attempt
1100 : * to reconnect to that server.
1101 : *
1102 : * If you overload this function, make sure to either call this
1103 : * implementation or enable the timer yourselves.
1104 : *
1105 : * \warning
1106 : * This function does not call the timer::process_invalid() function
1107 : * which means that this connection is not automatically removed from
1108 : * the communicator object on failures.
1109 : */
1110 0 : void local_stream_client_permanent_message_connection::process_invalid()
1111 : {
1112 0 : if(is_done())
1113 : {
1114 0 : timer::process_invalid();
1115 : }
1116 : else
1117 : {
1118 0 : f_impl->disconnect();
1119 0 : set_enable(true);
1120 : }
1121 0 : }
1122 :
1123 :
1124 : /** \brief Make sure that the messenger connection gets removed.
1125 : *
1126 : * This function makes sure that the messenger sub-connection also gets
1127 : * removed from the communicator. Otherwise it would lock the system
1128 : * since connections are saved in the communicator object as shared
1129 : * pointers.
1130 : */
1131 1 : void local_stream_client_permanent_message_connection::connection_removed()
1132 : {
1133 1 : f_impl->disconnect();
1134 1 : }
1135 :
1136 :
1137 : /** \brief Process a connection failed callback.
1138 : *
1139 : * When a connection attempt fails, we restart the timer so we can
1140 : * attempt to reconnect to that server.
1141 : *
1142 : * If you overload this function, make sure to either call this
1143 : * implementation or enable the timer yourselves.
1144 : *
1145 : * \param[in] error_message The error message that triggered this callback.
1146 : */
1147 0 : void local_stream_client_permanent_message_connection::process_connection_failed(std::string const & error_message)
1148 : {
1149 0 : snapdev::NOT_USED(error_message);
1150 0 : set_enable(true);
1151 0 : }
1152 :
1153 :
1154 : /** \brief The connection is ready.
1155 : *
1156 : * This callback gets called whenever the connection succeeded and is
1157 : * ready to be used.
1158 : *
1159 : * You should implement this virtual function if you have to initiate
1160 : * the communication. For example, the snapserver has to send a
1161 : * REGISTER to the snapcommunicator system and thus implements this
1162 : * function.
1163 : *
1164 : * The default implementation makes sure that the timer gets turned off
1165 : * so we do not try to reconnect every minute or so. Make sure you call
1166 : * the default function so you get the proper behavior.
1167 : */
1168 1 : void local_stream_client_permanent_message_connection::process_connected()
1169 : {
1170 1 : set_enable(false);
1171 1 : }
1172 :
1173 :
1174 :
1175 : } // namespace ed
1176 : // vim: ts=4 sw=4 et
|
