recoonect fix

Signed-off-by: elestrias <[email protected]>

Author: Elestrias

core/api/rpc/wsc.cpp

@@ -42,20 +42,20 @@ namespace fc::api::rpc {
                                           const std::string &target,
                                           const std::string &token) {
     boost::system::error_code ec;
-    socket.next_layer().connect({boost::asio::ip::make_address(host),
+    socket->next_layer().connect({boost::asio::ip::make_address(host),
                                  boost::lexical_cast<uint16_t>(port)},
                                 ec);
     if (ec) {
       return ec;
     }
     if (not token.empty()) {
-      socket.set_option(
+      socket->set_option(
           boost::beast::websocket::stream_base::decorator([&](auto &req) {
             req.set(boost::beast::http::field::authorization,
                     "Bearer " + token);
           }));
     }
-    socket.handshake(host, target, ec);
+    socket->handshake(host, target, ec);
     client_data = ClientData{host, port, target, token};
     if (ec) {
       return ec;
@@ -92,10 +92,10 @@ namespace fc::api::rpc {
   }
 
   void Client::_flush() {
-    if (!writing && !write_queue.empty() && !reconnecting){
+    if (!writing && !write_queue.empty() && not reconnecting){
       auto &[id, buffer] = write_queue.front();
       writing = true;
-      socket.async_write(boost::asio::buffer(buffer.data(), buffer.size()),
+      socket->async_write(boost::asio::buffer(buffer.data(), buffer.size()),
                          [=](auto &&ec, auto) {
                            std::lock_guard lock{mutex};
                            if (ec) {
@@ -109,7 +109,7 @@ namespace fc::api::rpc {
   }
 
   void Client::_read() {
-    socket.async_read(buffer, [=](auto &&ec, auto) {
+    socket->async_read(buffer, [=](auto &&ec, auto) {
       if (ec) {
         std::lock_guard lock{mutex};
         return _error(ec);
@@ -189,10 +189,13 @@ namespace fc::api::rpc {
   }
 
   void Client::reconnect(int counter, std::chrono::milliseconds wait) {
-    if(reconnecting.exchange(true)) return;
+    if(reconnecting) return;
+    reconnecting = true;
     logger_->info("Starting reconnect to {}:{}", client_data.host, client_data.port);
     for(int i = 0; i < counter; i++){
-      std::this_thread::sleep_for(wait*(i+1));
+      std::this_thread::sleep_for(wait);
+      socket.reset();
+      socket.emplace(io);
       auto res = connect(client_data.host,
                          client_data.port,
                          client_data.target,
@@ -201,7 +204,7 @@ namespace fc::api::rpc {
         break;
       }
     }
-    reconnecting.store(false);
+    reconnecting = false;
     logger_->info("Reconnect to {}:{} was successful", client_data.host, client_data.port);
     _flush();
   }

core/api/rpc/wsc.hpp

@@ -76,15 +76,15 @@ namespace fc::api::rpc {
     io_context io;
     io_context &io2;
     boost::asio::executor_work_guard<io_context::executor_type> work_guard;
-    boost::beast::websocket::stream<boost::asio::ip::tcp::socket> socket;
+    boost::optional<boost::beast::websocket::stream<boost::asio::ip::tcp::socket>> socket;
     boost::beast::flat_buffer buffer;
     std::mutex mutex;
     uint64_t next_req{};
     std::map<uint64_t, ResultCb> result_queue;
     std::map<uint64_t, ChanCb> chans;
     std::queue<std::pair<uint64_t, Bytes>> write_queue;
     bool writing{false};
-    std::atomic<bool> reconnecting;
+    bool reconnecting{false};
 
     template <typename M>
     void _setup(Client &c, M &m);