src/websocketstream.cpp - develop - Stormbery/CEX

src/websocketstream.cpp
157 строк5 KB
stormbery
deleted plots objects
20 сен 2024, 21:17
20 сен 2024, 21:174e0e245
#include "websocketstream.h"
#include "coin.h"

WebSocketStream::WebSocketStream():
        m_iocStrand(net::make_strand(m_ioc)),
        m_RequestLoopEnable {true},
        m_workguard(m_iocStrand)
    //    m_WebSocketOpen { false }

    {
             m_WebSocketOpen = 0;
            for (uint16_t i = 1; i <= 5;i ++) {
            m_threadIOC_[i] = boost::make_shared<boost::thread>([this] () {m_ioc.run();});
            m_threadIOC_[i]->detach();
        }

    }

WebSocketStream::~WebSocketStream() {

  std::cout << "DeStructor WebSocketStream starting..." << std::endl;

}

void WebSocketStream::SetPtr(boost::shared_ptr<Coin> m_ptr){
    m_ptr_coin = m_ptr ;
}

void WebSocketStream::SetPair(std::string  pair){
    m_pair = pair;
}

 void WebSocketStream::StopWebSocket() {
    m_RequestLoopEnable = false;
    while ( m_WebSocketOpen >=1) {}
    m_ptr_coin -> WebSocketLoggingTrace ("WebSocketStream DeStructor starting..." );
    m_ptr_coin = nullptr;
    m_workguard.reset();
    m_ioc.stop();
    while (!m_ioc.stopped()) {}
}

void WebSocketStream::AddPublicStream(std::string  host, std::string    stream,std::string    subscribe){

     net::co_spawn(
          m_iocStrand,// для выполнения последовательной цепочки
         //   m_ioc,
           [this, host, stream, subscribe]() mutable
    -> net::awaitable<void>
          {

     tcp::resolver resolver(m_ioc);
     websocket::stream<
     beast::ssl_stream<beast::tcp_stream>> ws(m_ioc, m_ctx);
  try
     {
         // Look up the domain name
         auto const results = co_await resolver.async_resolve(
             host, "443", use_awaitable);

         // Set a timeout on the operation
         beast::get_lowest_layer(ws).expires_after(
             std::chrono::seconds(30));

         // Make the connection on the IP address we get from a lookup
         auto ep = co_await beast::get_lowest_layer(ws).async_connect(
             results, use_awaitable);

         // Set SNI Hostname (many hosts need this to handshake
         // successfully)
         if(! SSL_set_tlsext_host_name(
             ws.next_layer().native_handle(), host.c_str()))
         {
             m_ptr_coin -> WebSocketLoggingError ("WebSocket SSL Error ");
             throw beast::system_error(
                 static_cast<int>(::ERR_get_error()),
                                       net::error::get_ssl_category());

         }

         // Update the host string. This will provide the value of the
         // Host HTTP header during the WebSocket handshake.
         // See https://tools.ietf.org/html/rfc7230#section-5.4
         host += ':' + std::to_string(ep.port());

         // Set a timeout on the operation
         beast::get_lowest_layer(ws).expires_after(
             std::chrono::seconds(30));

         // Perform the SSL handshake
         co_await ws.next_layer().async_handshake(
             ssl::stream_base::client, use_awaitable);

         // Turn off the timeout on the tcp_stream, because
         // the websocket stream has its own timeout system.
         beast::get_lowest_layer(ws).expires_never();

         // Set suggested timeout settings for the websocket
         ws.set_option(websocket::stream_base::timeout::suggested(
             beast::role_type::client));
/*
         websocket::stream_base::timeout opt {
            std::chrono::seconds (30),
            std::chrono::seconds(5),
            true
        };
        */
     //   ws.set_option(opt);
         // Perform the websocket handshake
     //    co_await ws.async_handshake(host, "/v5/public/spot", use_awaitable);
        co_await ws.async_handshake(host, stream, use_awaitable);
 //       if (ws.is_open()) {}
         co_await ws.async_write(
             net::buffer(std::string(subscribe)), use_awaitable);
         beast::flat_buffer buffer;
        uint16_t count_ping;
        m_ptr_coin -> WebSocketLoggingTrace ("Run Loop.... ");
        //m_WebSocketOpen = true;
         m_WebSocketOpen ++;
        while ( m_RequestLoopEnable) {
            auto [ec, bytes] = co_await ws.async_read(buffer, as_tuple(use_awaitable));
            if (ec == beast::error::timeout)
                {
                    std::cout << "WebSocketOpen timeout.... " << std::endl;
                 //   StopWebSocket();
                  beast::get_lowest_layer(ws).cancel();
                //m_ioc.reset();

                  m_WebSocketOpen --;
                   m_ptr_coin -> AddPublicStream();
                 //  break;
                    co_return;
                }
                if (count_ping > 10000) {
                    co_await ws.async_write(net::buffer(R"({"req_id": "100001", "op": "ping"})"), use_awaitable);
                    count_ping = 0;
                }
            m_ptr_coin -> ParsePublicStream(std::move(buffer));
        //     buffer.clear();
}
           m_ptr_coin -> WebSocketLoggingTrace ("Stop Loop.... ");
             co_await ws.async_close(
                websocket::close_code::normal, use_awaitable);
                m_ptr_coin -> WebSocketLoggingTrace ("WebSocket Close " );
               // m_WebSocketOpen = false;
        m_ioc.reset();
        m_WebSocketOpen --;

}
     catch(beast::system_error const& se)
     {
         std::cerr << "Handled: " << se.code().message() << "\n";
         m_ptr_coin -> WebSocketLoggingError ("WebSocketError code: " + se.code().message());
         throw; // handled at the spawn site instead
     }
          }, net::detached);
}