Kurlyk
Loading...
Searching...
No Matches
BaseWebSocketClient.hpp
Go to the documentation of this file.
1#pragma once
2#ifndef _KURLYK_BASE_WEB_SOCKET_CLIENT_HPP_INCLUDED
3#define _KURLYK_BASE_WEB_SOCKET_CLIENT_HPP_INCLUDED
4
7
8#include "BaseWebSocketClient/IWebSocketSender.hpp"
9#include "BaseWebSocketClient/WebSocketEventData.hpp"
10#include "BaseWebSocketClient/IWebSocketClient.hpp"
11#include "BaseWebSocketClient/WebSocketRateLimiter.hpp"
12
13namespace kurlyk {
14
17 class BaseWebSocketClient : public IWebSocketClient, public std::enable_shared_from_this<BaseWebSocketClient> {
18 public:
19
21 BaseWebSocketClient() = default;
22
24 virtual ~BaseWebSocketClient() = default;
25
30 std::function<void(std::unique_ptr<WebSocketEventData>)>& event_handler() override final {
31 return m_on_event;
32 }
33
39 std::function<void()>& notify_handler() override final {
40 return m_on_event_notify;
41 }
42
46 void set_config(std::unique_ptr<WebSocketConfig> config, std::function<void(bool)> callback) override final {
47 m_fsm_event_queue.push_event(FSMEventData(FsmEvent::UpdateConfig, std::move(config), std::move(callback)));
48 }
49
52 void connect(std::function<void(bool)> callback) override final {
53 m_fsm_event_queue.push_event(FSMEventData(FsmEvent::RequestConnect, std::move(callback)));
54 }
55
58 void disconnect(std::function<void(bool)> callback) override final {
59 m_fsm_event_queue.push_event(FSMEventData(FsmEvent::RequestDisconnect, std::move(callback)));
60 }
61
64 bool is_connected() const override final {
65 return m_is_connected;
66 }
67
70 bool is_running() const override final {
71 return m_is_running || m_fsm_event_queue.has_events();
72 }
73
78 std::list<std::unique_ptr<WebSocketEventData>> receive_events() const override final {
79 std::lock_guard<std::mutex> lock(m_event_queue_mutex);
80 auto event_queue = std::move(m_event_queue);
81 m_event_queue.clear();
82 return event_queue;
83 }
84
89 std::unique_ptr<WebSocketEventData> receive_event() const override final {
90 std::lock_guard<std::mutex> lock(m_event_queue_mutex);
91 if (m_event_queue.empty()) return nullptr;
92 std::unique_ptr<WebSocketEventData> event = std::move(*m_event_queue.begin());
93 m_event_queue.erase(m_event_queue.begin());
94 return event;
95 }
96
102 SubmitResult submit_message(
103 const std::string &message,
104 long rate_limit_id,
105 std::function<void(const std::error_code& ec)> callback = nullptr) override final {
106 if (message.empty()) {
107 return SubmitResult{false, utils::make_error_code(utils::ClientError::InvalidConfiguration)};
108 }
109 if (!is_connected()) {
110 return SubmitResult{false, utils::make_error_code(utils::ClientError::NotConnected)};
111 }
112 std::lock_guard<std::mutex> lock(m_message_queue_mutex);
113 const std::size_t queue_limit = m_max_send_queue_size.load();
114 if (queue_limit && m_message_queue.size() >= queue_limit) {
115 return SubmitResult{false, utils::make_error_code(utils::ClientError::QueueLimitExceeded)};
116 }
117# if __cplusplus >= 201402L
118 m_message_queue.push_back(std::make_shared<WebSocketSendInfo>(message, rate_limit_id, false, 0, std::move(callback)));
119# else
120 m_message_queue.push_back(std::shared_ptr<WebSocketSendInfo>(new WebSocketSendInfo(message, rate_limit_id, false, 0, std::move(callback))));
121# endif
122 return SubmitResult{true, std::error_code()};
123 }
124
130 bool send_message(
131 const std::string &message,
132 long rate_limit_id,
133 std::function<void(const std::error_code& ec)> callback = nullptr) override final {
134 return submit_message(message, rate_limit_id, std::move(callback)).accepted;
135 }
136
142 SubmitResult submit_close(
143 const int status = 1000,
144 const std::string &reason = std::string(),
145 std::function<void(const std::error_code& ec)> callback = nullptr) override final {
146 if (!is_connected()) {
147 return SubmitResult{false, utils::make_error_code(utils::ClientError::NotConnected)};
148 }
149 std::lock_guard<std::mutex> lock(m_message_queue_mutex);
150 const std::size_t queue_limit = m_max_send_queue_size.load();
151 if (queue_limit && m_message_queue.size() >= queue_limit) {
152 return SubmitResult{false, utils::make_error_code(utils::ClientError::QueueLimitExceeded)};
153 }
154# if __cplusplus >= 201402L
155 m_message_queue.push_back(std::make_shared<WebSocketSendInfo>(reason, 0, true, status, std::move(callback)));
156# else
157 m_message_queue.push_back(std::shared_ptr<WebSocketSendInfo>(new WebSocketSendInfo(reason, 0, true, status, std::move(callback))));
158# endif
159 return SubmitResult{true, std::error_code()};
160 }
161
167 bool send_close(
168 const int status = 1000,
169 const std::string &reason = std::string(),
170 std::function<void(const std::error_code& ec)> callback = nullptr) override final {
171 return submit_close(status, reason, std::move(callback)).accepted;
172 }
173
174
179 void process() override final {
180 process_fsm_state();
181 process_message_queue();
182 process_send_callback_queue();
183 }
184
187 void shutdown() override final {
188 m_fsm_event_queue.push_event(FSMEventData(FsmEvent::RequestDisconnect));
189 while (is_running()) {
190 process_fsm_state();
191 process_send_callback_queue();
192 }
193 }
194
195 protected:
196
197 std::unique_ptr<WebSocketConfig> m_config;
198
207
218
219
221 virtual bool init_websocket() = 0;
222
224 virtual void deinit_websocket() = 0;
225
228 virtual void send_message(std::shared_ptr<WebSocketSendInfo>& send_info) = 0;
229
232 virtual void send_close(std::shared_ptr<WebSocketSendInfo>& send_info) = 0;
233
236 std::unique_ptr<WebSocketEventData> create_websocket_event() {
237# if __cplusplus >= 201402L
238 auto event = std::make_unique<WebSocketEventData>();
239# else
240 auto event = std::unique_ptr<WebSocketEventData>(new WebSocketEventData());
241# endif
242 event->sender = shared_from_this();
243 return event;
244 }
245
252 std::unique_ptr<WebSocketEventData> create_websocket_close_event(
253 const std::string& reason = "Normal Closure",
254 int status_code = 1000) {
255 auto websocket_event = create_websocket_event();
256 websocket_event->event_type = WebSocketEventType::WS_CLOSE;
257 websocket_event->message = reason;
258 websocket_event->status_code = status_code;
259 return websocket_event;
260 }
261
267 std::unique_ptr<WebSocketEventData> create_websocket_error_event(
268 const std::error_code& error_code) {
269 auto websocket_event = create_websocket_event();
270 websocket_event->event_type = WebSocketEventType::WS_ERROR;
271 websocket_event->error_code = error_code;
272 return websocket_event;
273 }
274
278 void add_send_callback(
279 const std::error_code& error_code,
280 const std::function<void(const std::error_code& ec)> &callback) {
281 std::lock_guard<std::mutex> lock(m_send_callback_queue_mutex);
282 m_send_callback_queue.push_back(std::make_pair(error_code, callback));
283 }
284
288 void add_fsm_event(FsmEvent event_type, std::unique_ptr<WebSocketEventData> event_data) {
289 m_fsm_event_queue.push_event(FSMEventData(event_type, std::move(event_data)));
290 m_on_event_notify();
291 }
292
293 private:
294
295 std::function<void(std::unique_ptr<WebSocketEventData>)> m_on_event;
296 std::function<void()> m_on_event_notify;
297
300 struct FSMEventData {
301 FsmEvent event_type;
302 std::unique_ptr<WebSocketEventData> event_data;
303 std::unique_ptr<WebSocketConfig> config_data;
304 std::function<void(bool)> callback;
305
308 FSMEventData(FSMEventData&& other) noexcept
309 : event_type(other.event_type),
310 event_data(std::move(other.event_data)),
311 config_data(std::move(other.config_data)),
312 callback(std::move(other.callback)) {
313 }
314
318 FSMEventData& operator=(FSMEventData&& other) noexcept {
319 if (this != &other) {
320 event_type = other.event_type;
321 event_data = std::move(other.event_data);
322 config_data = std::move(other.config_data);
323 callback = std::move(other.callback);
324 }
325 return *this;
326 }
327
329 FSMEventData(const FSMEventData&) = delete;
330
332 FSMEventData& operator=(const FSMEventData&) = delete;
333
337 FSMEventData(
338 FsmEvent event_type,
339 std::unique_ptr<WebSocketEventData> &&event_data) :
340 event_type(event_type),
341 event_data(std::move(event_data)) {
342 }
343
348 FSMEventData(
349 FsmEvent event_type,
350 std::unique_ptr<WebSocketConfig> &&config_data,
351 std::function<void(bool)> &&callback) :
352 event_type(event_type),
353 config_data(std::move(config_data)),
354 callback(std::move(callback)) {
355 }
356
360 FSMEventData(
361 FsmEvent event_type,
362 std::function<void(bool)> &&callback) :
363 event_type(event_type),
364 callback(std::move(callback)) {
365 }
366
372
373 }; // FSMEventData
374
375 utils::EventQueue<FSMEventData> m_fsm_event_queue;
376 long m_reconnect_attempt = 0;
377 std::atomic<bool> m_is_running = ATOMIC_VAR_INIT(false);
378 std::atomic<bool> m_is_connected = ATOMIC_VAR_INIT(false);
379 std::atomic<std::size_t> m_max_send_queue_size = ATOMIC_VAR_INIT(0);
380
381 WebSocketRateLimiter m_rate_limiter;
382 std::chrono::steady_clock::time_point m_close_time;
383
384 mutable std::mutex m_event_queue_mutex;
385 using event_data_ptr_t = std::unique_ptr<WebSocketEventData>;
386 mutable std::list<event_data_ptr_t> m_event_queue;
387
388 std::mutex m_message_queue_mutex;
389 using send_info_ptr_t = std::shared_ptr<WebSocketSendInfo>;
390 std::list<send_info_ptr_t> m_message_queue;
391
392 std::mutex m_send_callback_queue_mutex;
393 using send_callback_t = std::pair<std::error_code, std::function<void(const std::error_code& ec)>>;
394 std::list<send_callback_t> m_send_callback_queue;
395
397 void process_fsm_state() {
398 switch (m_fsm_state) {
399 case FsmState::INIT:
400 process_state_init();
401 break;
402 case FsmState::CONNECTING:
403 process_state_connecting();
404 break;
405 case FsmState::WORKING:
406 process_state_working();
407 break;
408 case FsmState::RECONNECTING:
409 process_state_reconnecting();
410 break;
411 case FsmState::STOPPED:
412 process_state_stopped();
413 break;
414 };
415 }
416
418 void process_state_init() {
419 if (!m_fsm_event_queue.has_events()) return;
420
421 auto event = m_fsm_event_queue.pop_event();
422 switch (event.event_type) {
423 case FsmEvent::RequestConnect:
424 if (!m_config) {
425 handle_error_event(utils::make_error_code(utils::ClientError::InvalidConfiguration));
426 if (event.callback) event.callback(false);
427 m_fsm_state = FsmState::STOPPED;
428 break;
429 }
430 if (!init_websocket()) {
431 handle_error_event(utils::make_error_code(utils::ClientError::InvalidConfiguration));
432 if (event.callback) event.callback(false);
433 m_fsm_state = FsmState::STOPPED;
434 break;
435 }
436 m_is_running = true;
437 if (event.callback) event.callback(true);
438 m_fsm_state = FsmState::CONNECTING;
439 break;
440 case FsmEvent::UpdateConfig:
441 m_config = std::move(event.config_data);
442 if (m_config) {
443 m_rate_limiter.set_limit(m_config->rate_limits);
444 m_max_send_queue_size.store(m_config->max_send_queue_size);
445 if (event.callback) event.callback(true);
446 } else {
447 if (event.callback) event.callback(false);
448 }
449 break;
450 default:
451 if (event.callback) event.callback(false);
452 break;
453 };
454 }
455
457 void process_state_connecting() {
458 if (!m_fsm_event_queue.has_events()) return;
459
460 auto event = m_fsm_event_queue.pop_event();
461 switch (event.event_type) {
462 case FsmEvent::ConnectionOpened:
463 handle_open_event(std::move(event.event_data));
464 m_reconnect_attempt = 0;
465 m_is_running = true;
466 m_fsm_state = FsmState::WORKING;
467 break;
468 case FsmEvent::ConnectionError:
469 handle_error_event(std::move(event.event_data));
470 case FsmEvent::ConnectionClosed:
471 deinit_websocket();
472 if (event.event_type == FsmEvent::ConnectionClosed) {
473 handle_close_event(std::move(event.event_data));
474 } else {
475 handle_close_event(create_websocket_close_event("Going Away", 1001));
476 }
477
478 if (!m_config->reconnect) {
479 m_is_running = false;
480 m_fsm_state = FsmState::INIT;
481 break;
482 }
483
484 m_reconnect_attempt++;
485 m_close_time = std::chrono::steady_clock::now();
486 m_is_running = true;
487 m_fsm_state = FsmState::RECONNECTING;
488 break;
489 case FsmEvent::RequestDisconnect: {
490 deinit_websocket();
491 handle_close_event();
492
493 m_reconnect_attempt = 0;
494 m_is_running = false;
495 if (event.callback) event.callback(true);
496 m_fsm_state = FsmState::INIT;
497 break;
498 }
499 case FsmEvent::UpdateConfig:
500 deinit_websocket();
501 handle_close_event();
502
503 m_config = std::move(event.config_data);
504 if (!m_config) {
505 handle_error_event(utils::make_error_code(utils::ClientError::InvalidConfiguration));
506 if (event.callback) event.callback(false);
507 m_fsm_state = FsmState::STOPPED;
508 break;
509 }
510 m_rate_limiter.set_limit(m_config->rate_limits);
511 m_max_send_queue_size.store(m_config->max_send_queue_size);
512
513 m_reconnect_attempt = 0;
514 if (!init_websocket()) {
515 handle_error_event(utils::make_error_code(utils::ClientError::InvalidConfiguration));
516 if (event.callback) event.callback(false);
517 m_fsm_state = FsmState::STOPPED;
518 break;
519 }
520
521 m_is_running = true;
522 if (event.callback) event.callback(true);
523 m_fsm_state = FsmState::CONNECTING;
524 break;
525 default:
526 if (event.callback) event.callback(false);
527 break;
528 };
529 }
530
532 void process_state_working() {
533 bool is_message;
534 while (m_fsm_event_queue.has_events()) {
535 is_message = false;
536 auto event = m_fsm_event_queue.pop_event();
537 switch (event.event_type) {
538 case FsmEvent::RequestDisconnect:
539 deinit_websocket();
540 handle_close_event();
541
542 m_reconnect_attempt = 0;
543 m_is_running = false;
544 if (event.callback) event.callback(true);
545 m_fsm_state = FsmState::INIT;
546 break;
547 case FsmEvent::ConnectionError:
548 handle_error_event(std::move(event.event_data));
549 case FsmEvent::ConnectionClosed:
550 deinit_websocket();
551 if (event.event_type == FsmEvent::ConnectionClosed) {
552 handle_close_event(std::move(event.event_data));
553 } else {
554 handle_close_event(create_websocket_close_event("Going Away", 1001));
555 }
556
557 if (!m_config->reconnect) {
558 m_is_running = false;
559 m_fsm_state = FsmState::INIT;
560 break;
561 }
562
563 m_reconnect_attempt++;
564 m_close_time = std::chrono::steady_clock::now();
565 m_is_running = true;
566 m_fsm_state = FsmState::RECONNECTING;
567 break;
568 case FsmEvent::UpdateConfig:
569 deinit_websocket();
570 handle_close_event();
571
572 m_config = std::move(event.config_data);
573 if (!m_config) {
574 handle_error_event(utils::make_error_code(utils::ClientError::InvalidConfiguration));
575 if (event.callback) event.callback(false);
576 m_fsm_state = FsmState::STOPPED;
577 break;
578 }
579 m_rate_limiter.set_limit(m_config->rate_limits);
580 m_max_send_queue_size.store(m_config->max_send_queue_size);
581
582 m_reconnect_attempt = 0;
583 if (!init_websocket()) {
584 handle_error_event(utils::make_error_code(utils::ClientError::InvalidConfiguration));
585 if (event.callback) event.callback(false);
586 m_fsm_state = FsmState::STOPPED;
587 break;
588 }
589
590 m_is_running = true;
591 if (event.callback) event.callback(true);
592 m_fsm_state = FsmState::CONNECTING;
593 break;
594 case FsmEvent::MessageReceived:
595 handle_message_event(std::move(event.event_data));
596 is_message = true;
597 break;
598 default:
599 if (event.callback) event.callback(false);
600 break;
601 };
602 if (!is_message) break;
603 }
604 }
605
607 void process_state_reconnecting() {
608 if (m_fsm_event_queue.has_events()){
609 auto event = m_fsm_event_queue.pop_event();
610 switch (event.event_type) {
611 case FsmEvent::RequestDisconnect:
612 m_is_running = false;
613 if (event.callback) event.callback(true);
614 m_fsm_state = FsmState::INIT;
615 break;
616 case FsmEvent::UpdateConfig:
617 m_config = std::move(event.config_data);
618 if (!m_config) {
619 handle_error_event(utils::make_error_code(utils::ClientError::InvalidConfiguration));
620 if (event.callback) event.callback(false);
621 m_fsm_state = FsmState::STOPPED;
622 break;
623 }
624 m_rate_limiter.set_limit(m_config->rate_limits);
625 m_max_send_queue_size.store(m_config->max_send_queue_size);
626
627 m_reconnect_attempt = 0;
628 if (!init_websocket()) {
629 handle_error_event(utils::make_error_code(utils::ClientError::InvalidConfiguration));
630 if (event.callback) event.callback(false);
631 m_fsm_state = FsmState::STOPPED;
632 break;
633 }
634
635 m_is_running = true;
636 if (event.callback) event.callback(true);
637 m_fsm_state = FsmState::CONNECTING;
638 break;
639 case FsmEvent::MessageReceived:
640 handle_message_event(std::move(event.event_data));
641 break;
642 default:
643 if (event.callback) event.callback(false);
644 break;
645 };
646 }
647
648 if (!m_config) {
649 handle_error_event(utils::make_error_code(utils::ClientError::InvalidConfiguration));
650 m_fsm_state = FsmState::STOPPED;
651 return;
652 }
653
654 if (m_config->reconnect) {
655 if (m_config->reconnect_attempts &&
656 m_reconnect_attempt >= m_config->reconnect_attempts) {
657 m_is_running = false;
658 m_fsm_state = FsmState::INIT;
659 return;
660 }
661
662 auto now = std::chrono::steady_clock::now();
663 auto duration = std::chrono::duration_cast<std::chrono::seconds>(now - m_close_time);
664 if (duration.count() >= m_config->reconnect_delay) {
665 if (!init_websocket()) {
666 handle_error_event(utils::make_error_code(utils::ClientError::InvalidConfiguration));
667 m_fsm_state = FsmState::STOPPED;
668 return;
669 }
670 m_is_running = true;
671 m_fsm_state = FsmState::CONNECTING;
672 }
673 return;
674 }
675
676 m_is_running = false;
677 m_fsm_state = FsmState::INIT;
678 }
679
681 void process_state_stopped() {
682 if (!m_fsm_event_queue.has_events()) return;
683
684 auto event = m_fsm_event_queue.pop_event();
685 switch (event.event_type) {
686 case FsmEvent::RequestConnect:
687 if (!m_config) {
688 handle_error_event(utils::make_error_code(utils::ClientError::InvalidConfiguration));
689 if (event.callback) event.callback(false);
690 m_fsm_state = FsmState::STOPPED;
691 break;
692 }
693 if (!init_websocket()) {
694 handle_error_event(utils::make_error_code(utils::ClientError::InvalidConfiguration));
695 if (event.callback) event.callback(false);
696 m_fsm_state = FsmState::STOPPED;
697 break;
698 }
699 m_is_running = true;
700 if (event.callback) event.callback(true);
701 m_fsm_state = FsmState::CONNECTING;
702 break;
703 case FsmEvent::UpdateConfig:
704 m_config = std::move(event.config_data);
705 if (!m_config) {
706 handle_error_event(utils::make_error_code(utils::ClientError::InvalidConfiguration));
707 if (event.callback) event.callback(false);
708 m_fsm_state = FsmState::STOPPED;
709 break;
710 }
711 m_rate_limiter.set_limit(m_config->rate_limits);
712 m_max_send_queue_size.store(m_config->max_send_queue_size);
713
714 m_reconnect_attempt = 0;
715 if (!init_websocket()) {
716 handle_error_event(utils::make_error_code(utils::ClientError::InvalidConfiguration));
717 if (event.callback) event.callback(false);
718 m_fsm_state = FsmState::STOPPED;
719 break;
720 }
721
722 m_is_running = true;
723 if (event.callback) event.callback(true);
724 m_fsm_state = FsmState::CONNECTING;
725 break;
726 default:
727 if (event.callback) event.callback(false);
728 break;
729 };
730 }
731
735 void process_message_queue() {
736 std::unique_lock<std::mutex> lock(m_message_queue_mutex);
737 if (m_message_queue.empty()) return;
738
739 std::list<send_info_ptr_t> message_queue;
740 auto it = m_message_queue.begin();
741 while (it != m_message_queue.end()) {
742 auto& send_info = *it;
743 // Check if the message is allowed by the rate limiter.
744 if (!m_rate_limiter.allow_request(send_info->rate_limit_id)) {
745 ++it;
746 continue;
747 }
748 message_queue.push_back(std::move(send_info));
749 it = m_message_queue.erase(it);
750 }
751 lock.unlock();
752 if (message_queue.empty()) return;
753
754 for (auto &send_info : message_queue) {
755 if (!send_info->is_send_close) {
756 send_message(send_info);
757 continue;
758 }
759 send_close(send_info);
760 }
761 }
762
766 void process_send_callback_queue() {
767 std::unique_lock<std::mutex> lock(m_send_callback_queue_mutex);
768 if (m_send_callback_queue.empty()) return;
769 auto send_callback_queue = std::move(m_send_callback_queue);
770 m_send_callback_queue.clear();
771
772 for (auto &item : send_callback_queue) {
773 item.second(item.first);
774 }
775 }
776
781 void handle_open_event(std::unique_ptr<WebSocketEventData> event) {
782 if (!m_is_connected) {
783 m_is_connected = true;
784 if (m_on_event) {
785 m_on_event(std::move(event));
786 } else {
787 std::lock_guard<std::mutex> lock(m_event_queue_mutex);
788 m_event_queue.push_back(std::move(event));
789 }
790 }
791 }
792
797 void handle_close_event(std::unique_ptr<WebSocketEventData> event = nullptr) {
798 if (!event) {
799 event = create_websocket_close_event();
800 }
801 if (m_is_connected) {
802 m_is_connected = false;
803 if (m_on_event) {
804 m_on_event(std::move(event));
805 } else {
806 std::lock_guard<std::mutex> lock(m_event_queue_mutex);
807 m_event_queue.push_back(std::move(event));
808 }
809 }
810 }
811
815 void handle_error_event(std::unique_ptr<WebSocketEventData> event) {
816 if (m_on_event) {
817 m_on_event(std::move(event));
818 return;
819 }
820 std::lock_guard<std::mutex> lock(m_event_queue_mutex);
821 m_event_queue.push_back(std::move(event));
822 }
823
826 void handle_error_event(const std::error_code& error_code) {
827 handle_error_event(create_websocket_error_event(error_code));
828 }
829
833 void handle_message_event(std::unique_ptr<WebSocketEventData> event) {
834 if (m_on_event) {
835 m_on_event(std::move(event));
836 return;
837 }
838 std::lock_guard<std::mutex> lock(m_event_queue_mutex);
839 m_event_queue.push_back(std::move(event));
840 }
841
842 }; // BaseWebSocketClient
843
844}; // namespace kurlyk
845
846
847#endif // _KURLYK_BASE_WEB_SOCKET_CLIENT_HPP_INCLUDED
IWebSocketClient.hpp
Defines a legacy internal interface for WebSocket client functionality.
IWebSocketSender.hpp
Defines the public sender abstraction exposed through WebSocket events.
WebSocketEventData.hpp
Defines the WebSocketEventData class, which encapsulates data related to WebSocket events.
WebSocketRateLimiter.hpp
Defines WebSocket rate limiting to control the frequency of WebSocket requests.
kurlyk::BaseWebSocketClient::~BaseWebSocketClient
virtual ~BaseWebSocketClient()=default
Virtual destructor.
kurlyk::BaseWebSocketClient::process_state_working
void process_state_working()
Handles the WORKING state. Processes incoming events and manages connection health.
Definition BaseWebSocketClient.hpp:532
kurlyk::BaseWebSocketClient::m_close_time
std::chrono::steady_clock::time_point m_close_time
Timestamp of the last WebSocket close event, used for reconnection timing.
Definition BaseWebSocketClient.hpp:382
kurlyk::BaseWebSocketClient::init_websocket
virtual bool init_websocket()=0
Initializes the WebSocket connection. Must be implemented in derived classes.
kurlyk::BaseWebSocketClient::handle_message_event
void handle_message_event(std::unique_ptr< WebSocketEventData > event)
Handles incoming WebSocket message events and queues them if no event handler is set.
Definition BaseWebSocketClient.hpp:833
kurlyk::BaseWebSocketClient::send_message
bool send_message(const std::string &message, long rate_limit_id, std::function< void(const std::error_code &ec)> callback=nullptr) override final
Sends a message through the WebSocket.
Definition BaseWebSocketClient.hpp:130
kurlyk::BaseWebSocketClient::event_data_ptr_t
std::unique_ptr< WebSocketEventData > event_data_ptr_t
Alias for unique pointers to WebSocketEventData.
Definition BaseWebSocketClient.hpp:385
kurlyk::BaseWebSocketClient::create_websocket_event
std::unique_ptr< WebSocketEventData > create_websocket_event()
Creates a generic WebSocket event.
Definition BaseWebSocketClient.hpp:236
kurlyk::BaseWebSocketClient::BaseWebSocketClient
BaseWebSocketClient()=default
Default constructor.
kurlyk::BaseWebSocketClient::is_connected
bool is_connected() const override final
Checks if the WebSocket client is actively running.
Definition BaseWebSocketClient.hpp:64
kurlyk::BaseWebSocketClient::connect
void connect(std::function< void(bool)> callback) override final
Initiates a connection to the WebSocket server.
Definition BaseWebSocketClient.hpp:52
kurlyk::BaseWebSocketClient::m_message_queue_mutex
std::mutex m_message_queue_mutex
Mutex for synchronizing access to the message queue.
Definition BaseWebSocketClient.hpp:388
kurlyk::BaseWebSocketClient::m_send_callback_queue_mutex
std::mutex m_send_callback_queue_mutex
Mutex for synchronizing access to the send callback queue.
Definition BaseWebSocketClient.hpp:392
kurlyk::BaseWebSocketClient::m_event_queue_mutex
std::mutex m_event_queue_mutex
Mutex for synchronizing access to the event queue.
Definition BaseWebSocketClient.hpp:384
kurlyk::BaseWebSocketClient::is_running
bool is_running() const override final
Checks if the WebSocket client is actively running.
Definition BaseWebSocketClient.hpp:70
kurlyk::BaseWebSocketClient::process_state_connecting
void process_state_connecting()
Handles the CONNECTING state. Manages connection attempt, errors, or disconnection.
Definition BaseWebSocketClient.hpp:457
kurlyk::BaseWebSocketClient::process_state_init
void process_state_init()
Handles the INIT state. Initializes connection or updates configuration.
Definition BaseWebSocketClient.hpp:418
kurlyk::BaseWebSocketClient::m_fsm_state
enum kurlyk::BaseWebSocketClient::FsmState m_fsm_state
kurlyk::BaseWebSocketClient::submit_message
SubmitResult submit_message(const std::string &message, long rate_limit_id, std::function< void(const std::error_code &ec)> callback=nullptr) override final
Attempts to submit a message through the WebSocket.
Definition BaseWebSocketClient.hpp:102
kurlyk::BaseWebSocketClient::send_close
bool send_close(const int status=1000, const std::string &reason=std::string(), std::function< void(const std::error_code &ec)> callback=nullptr) override final
Sends a close request through the WebSocket.
Definition BaseWebSocketClient.hpp:167
kurlyk::BaseWebSocketClient::handle_error_event
void handle_error_event(const std::error_code &error_code)
Overloaded method to handle WebSocket error events using an error code.
Definition BaseWebSocketClient.hpp:826
kurlyk::BaseWebSocketClient::m_rate_limiter
WebSocketRateLimiter m_rate_limiter
Rate limiter for controlling the frequency of message sending.
Definition BaseWebSocketClient.hpp:381
kurlyk::BaseWebSocketClient::m_fsm_event_queue
utils::EventQueue< FSMEventData > m_fsm_event_queue
Queue for FSM events, managing the event sequence for the FSM.
Definition BaseWebSocketClient.hpp:375
kurlyk::BaseWebSocketClient::handle_close_event
void handle_close_event(std::unique_ptr< WebSocketEventData > event=nullptr)
Handles the event when the WebSocket connection is closed.
Definition BaseWebSocketClient.hpp:797
kurlyk::BaseWebSocketClient::deinit_websocket
virtual void deinit_websocket()=0
Deinitializes the WebSocket connection. Must be implemented in derived classes.
kurlyk::BaseWebSocketClient::process_state_stopped
void process_state_stopped()
Processes the STOPPED state in the FSM.
Definition BaseWebSocketClient.hpp:681
kurlyk::BaseWebSocketClient::m_event_queue
std::list< event_data_ptr_t > m_event_queue
Queue holding pending WebSocket events.
Definition BaseWebSocketClient.hpp:386
kurlyk::BaseWebSocketClient::create_websocket_close_event
std::unique_ptr< WebSocketEventData > create_websocket_close_event(const std::string &reason="Normal Closure", int status_code=1000)
Creates a WebSocket close event with a specified reason and status code.
Definition BaseWebSocketClient.hpp:252
kurlyk::BaseWebSocketClient::send_callback_t
std::pair< std::error_code, std::function< void(const std::error_code &ec)> > send_callback_t
Alias for callback pairs with error codes.
Definition BaseWebSocketClient.hpp:393
kurlyk::BaseWebSocketClient::set_config
void set_config(std::unique_ptr< WebSocketConfig > config, std::function< void(bool)> callback) override final
Sets the configuration for the WebSocket client.
Definition BaseWebSocketClient.hpp:46
kurlyk::BaseWebSocketClient::handle_error_event
void handle_error_event(std::unique_ptr< WebSocketEventData > event)
Handles WebSocket error events and queues them if no event handler is set.
Definition BaseWebSocketClient.hpp:815
kurlyk::BaseWebSocketClient::process_message_queue
void process_message_queue()
Processes the queue of messages to be sent over the WebSocket.
Definition BaseWebSocketClient.hpp:735
kurlyk::BaseWebSocketClient::send_message
virtual void send_message(std::shared_ptr< WebSocketSendInfo > &send_info)=0
Sends a WebSocket message.
kurlyk::BaseWebSocketClient::receive_event
std::unique_ptr< WebSocketEventData > receive_event() const override final
Retrieves the next available WebSocket event, if any.
Definition BaseWebSocketClient.hpp:89
kurlyk::BaseWebSocketClient::FsmState
FsmState
Finite State Machine (FSM) states for the WebSocket connection.
Definition BaseWebSocketClient.hpp:200
kurlyk::BaseWebSocketClient::FsmState::WORKING
@ WORKING
Connection active and working.
Definition BaseWebSocketClient.hpp:203
kurlyk::BaseWebSocketClient::process_fsm_state
void process_fsm_state()
Processes the current FSM state and transitions to the appropriate next state.
Definition BaseWebSocketClient.hpp:397
kurlyk::BaseWebSocketClient::m_on_event
std::function< void(std::unique_ptr< WebSocketEventData >)> m_on_event
Function to handle WebSocket events. Called when a new event is received.
Definition BaseWebSocketClient.hpp:295
kurlyk::BaseWebSocketClient::receive_events
std::list< std::unique_ptr< WebSocketEventData > > receive_events() const override final
Retrieves all pending WebSocket events in a batch.
Definition BaseWebSocketClient.hpp:78
kurlyk::BaseWebSocketClient::process_send_callback_queue
void process_send_callback_queue()
Processes the queue of send callbacks.
Definition BaseWebSocketClient.hpp:766
kurlyk::BaseWebSocketClient::event_handler
std::function< void(std::unique_ptr< WebSocketEventData >)> & event_handler() override final
Accessor for the event handler function.
Definition BaseWebSocketClient.hpp:30
kurlyk::BaseWebSocketClient::process_state_reconnecting
void process_state_reconnecting()
Handles the RECONNECTING state. Attempts to reconnect based on configuration settings.
Definition BaseWebSocketClient.hpp:607
kurlyk::BaseWebSocketClient::add_send_callback
void add_send_callback(const std::error_code &error_code, const std::function< void(const std::error_code &ec)> &callback)
Adds a send callback to the queue.
Definition BaseWebSocketClient.hpp:278
kurlyk::BaseWebSocketClient::m_is_connected
std::atomic< bool > m_is_connected
Atomic flag indicating if the client is connected.
Definition BaseWebSocketClient.hpp:378
kurlyk::BaseWebSocketClient::process
void process() override final
Processes internal operations such as event handling and state updates.
Definition BaseWebSocketClient.hpp:179
kurlyk::BaseWebSocketClient::m_max_send_queue_size
std::atomic< std::size_t > m_max_send_queue_size
Maximum number of queued outbound send operations, or zero if unbounded.
Definition BaseWebSocketClient.hpp:379
kurlyk::BaseWebSocketClient::submit_close
SubmitResult submit_close(const int status=1000, const std::string &reason=std::string(), std::function< void(const std::error_code &ec)> callback=nullptr) override final
Attempts to submit a close request through the WebSocket.
Definition BaseWebSocketClient.hpp:142
kurlyk::BaseWebSocketClient::disconnect
void disconnect(std::function< void(bool)> callback) override final
Closes the connection to the WebSocket server.
Definition BaseWebSocketClient.hpp:58
kurlyk::BaseWebSocketClient::m_send_callback_queue
std::list< send_callback_t > m_send_callback_queue
Queue holding send callbacks with their respective error codes.
Definition BaseWebSocketClient.hpp:394
kurlyk::BaseWebSocketClient::add_fsm_event
void add_fsm_event(FsmEvent event_type, std::unique_ptr< WebSocketEventData > event_data)
Adds an FSM event to the event queue and triggers the notify handler.
Definition BaseWebSocketClient.hpp:288
kurlyk::BaseWebSocketClient::m_config
std::unique_ptr< WebSocketConfig > m_config
Current configuration for the WebSocket.
Definition BaseWebSocketClient.hpp:197
kurlyk::BaseWebSocketClient::send_info_ptr_t
std::shared_ptr< WebSocketSendInfo > send_info_ptr_t
Alias for shared pointers to WebSocketSendInfo.
Definition BaseWebSocketClient.hpp:389
kurlyk::BaseWebSocketClient::FsmEvent
FsmEvent
Represents events in the finite state machine.
Definition BaseWebSocketClient.hpp:209
kurlyk::BaseWebSocketClient::FsmEvent::MessageReceived
@ MessageReceived
Incoming WebSocket message.
Definition BaseWebSocketClient.hpp:215
kurlyk::BaseWebSocketClient::FsmEvent::ConnectionOpened
@ ConnectionOpened
Connection opened successfully.
Definition BaseWebSocketClient.hpp:212
kurlyk::BaseWebSocketClient::notify_handler
std::function< void()> & notify_handler() override final
Accesses the notification handler for WebSocket events.
Definition BaseWebSocketClient.hpp:39
kurlyk::BaseWebSocketClient::m_message_queue
std::list< send_info_ptr_t > m_message_queue
Queue holding messages to be sent over the WebSocket.
Definition BaseWebSocketClient.hpp:390
kurlyk::BaseWebSocketClient::send_close
virtual void send_close(std::shared_ptr< WebSocketSendInfo > &send_info)=0
Sends a close request.
kurlyk::BaseWebSocketClient::create_websocket_error_event
std::unique_ptr< WebSocketEventData > create_websocket_error_event(const std::error_code &error_code)
Creates a WebSocket error event with a specified error code.
Definition BaseWebSocketClient.hpp:267
kurlyk::BaseWebSocketClient::m_reconnect_attempt
long m_reconnect_attempt
Counter for the number of reconnection attempts.
Definition BaseWebSocketClient.hpp:376
kurlyk::BaseWebSocketClient::m_is_running
std::atomic< bool > m_is_running
Atomic flag indicating if the client is running.
Definition BaseWebSocketClient.hpp:377
kurlyk::BaseWebSocketClient::m_on_event_notify
std::function< void()> m_on_event_notify
Function to notify about new events in the FSM.
Definition BaseWebSocketClient.hpp:296
kurlyk::BaseWebSocketClient::handle_open_event
void handle_open_event(std::unique_ptr< WebSocketEventData > event)
Handles the event when the WebSocket connection is opened.
Definition BaseWebSocketClient.hpp:781
kurlyk::BaseWebSocketClient::shutdown
void shutdown() override final
Shuts down the WebSocket client, disconnecting and clearing all pending events.
Definition BaseWebSocketClient.hpp:187
kurlyk::IWebSocketClient::IWebSocketClient
IWebSocketClient()=default
Default constructor.
kurlyk::WebSocketEventData
Encapsulates data for a WebSocket event, providing information about event type, message,...
Definition WebSocketEventData.hpp:15
kurlyk::WebSocketRateLimiter
Manages rate limiting for WebSocket requests based on predefined limits.
Definition WebSocketRateLimiter.hpp:13
kurlyk::WebSocketSendInfo
Holds information for sending a WebSocket message, including rate limiting, close status,...
Definition WebSocketSendInfo.hpp:12
kurlyk::utils::EventQueue
A thread-safe event queue that supports blocking and non-blocking event retrieval.
Definition EventQueue.hpp:17
kurlyk::utils::ClientError::NotConnected
@ NotConnected
Operation requires an active connection but none exists.
Definition ClientErrorCategory.hpp:17
kurlyk::utils::ClientError::QueueLimitExceeded
@ QueueLimitExceeded
Operation was rejected because the bounded queue is already full.
Definition ClientErrorCategory.hpp:18
kurlyk::utils::ClientError::InvalidConfiguration
@ InvalidConfiguration
Provided configuration is incomplete or invalid.
Definition ClientErrorCategory.hpp:16
kurlyk::utils::make_error_code
std::error_code make_error_code(ClientError e)
Creates a std::error_code from a ClientError value.
Definition ClientErrorCategory.hpp:62
kurlyk
Primary namespace for the Kurlyk library, encompassing initialization, request management,...
kurlyk::WebSocketEventType::WS_ERROR
@ WS_ERROR
Error occurred.
Definition enums.hpp:35
kurlyk::WebSocketEventType::WS_CLOSE
@ WS_CLOSE
Connection closed.
Definition enums.hpp:34
std
Enables use of ClientError with std::error_code.
Definition ClientErrorCategory.hpp:69
kurlyk::BaseWebSocketClient::FSMEventData
Represents an event in the finite state machine (FSM) with optional associated data and callback.
Definition BaseWebSocketClient.hpp:300
kurlyk::BaseWebSocketClient::FSMEventData::callback
std::function< void(bool)> callback
Optional callback function to execute on event completion.
Definition BaseWebSocketClient.hpp:304
kurlyk::BaseWebSocketClient::FSMEventData::operator=
FSMEventData & operator=(const FSMEventData &)=delete
Deleted copy assignment operator to prevent copying.
kurlyk::BaseWebSocketClient::FSMEventData::FSMEventData
FSMEventData(FsmEvent event_type, std::unique_ptr< WebSocketEventData > &&event_data)
Constructs FSMEventData with an event type and WebSocket event data.
Definition BaseWebSocketClient.hpp:337
kurlyk::BaseWebSocketClient::FSMEventData::FSMEventData
FSMEventData(const FSMEventData &)=delete
Deleted copy constructor to prevent copying.
kurlyk::BaseWebSocketClient::FSMEventData::event_data
std::unique_ptr< WebSocketEventData > event_data
Optional WebSocket event data associated with the FSM event.
Definition BaseWebSocketClient.hpp:302
kurlyk::BaseWebSocketClient::FSMEventData::FSMEventData
FSMEventData(FsmEvent event_type)
Constructs FSMEventData with only an event type.
Definition BaseWebSocketClient.hpp:369
kurlyk::BaseWebSocketClient::FSMEventData::FSMEventData
FSMEventData(FSMEventData &&other) noexcept
Move constructor for FSMEventData.
Definition BaseWebSocketClient.hpp:308
kurlyk::BaseWebSocketClient::FSMEventData::FSMEventData
FSMEventData(FsmEvent event_type, std::function< void(bool)> &&callback)
Constructs FSMEventData with an event type and a callback.
Definition BaseWebSocketClient.hpp:360
kurlyk::BaseWebSocketClient::FSMEventData::operator=
FSMEventData & operator=(FSMEventData &&other) noexcept
Move assignment operator for FSMEventData.
Definition BaseWebSocketClient.hpp:318
kurlyk::BaseWebSocketClient::FSMEventData::FSMEventData
FSMEventData(FsmEvent event_type, std::unique_ptr< WebSocketConfig > &&config_data, std::function< void(bool)> &&callback)
Constructs FSMEventData with an event type, configuration data, and a callback.
Definition BaseWebSocketClient.hpp:348
kurlyk::BaseWebSocketClient::FSMEventData::event_type
FsmEvent event_type
The type of the FSM event.
Definition BaseWebSocketClient.hpp:301
kurlyk::BaseWebSocketClient::FSMEventData::config_data
std::unique_ptr< WebSocketConfig > config_data
Optional configuration data for FSM settings.
Definition BaseWebSocketClient.hpp:303
kurlyk::SubmitResult
Represents the synchronous result of trying to enqueue or submit work.
Definition SubmitResult.hpp:14
kurlyk::SubmitResult::accepted
bool accepted
Indicates whether the work item was accepted for processing.
Definition SubmitResult.hpp:15