src/websocketstream.cpp - master - Stormbery/CEX

src/websocketstream.cpp
122 строки4 KB
stormbery
обновлена функция парсинга
25 июн 2024, 19:22
25 июн 2024, 19:22521174c
#include "websocketstream.h"
#include "coin.h"

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

    {
            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;
     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()))
         {
             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));

         // 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;
        while ( m_RequestLoopEnable) {
            auto [ec, bytes] = co_await ws.async_read(buffer, as_tuple(use_awaitable));
                if (count_ping > 3) {
                    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();
}
             co_await ws.async_close(
                 websocket::close_code::normal, use_awaitable);
}
     catch(beast::system_error const& se)
     {
         std::cerr << "Handled: " << se.code().message() << "\n";
         throw; // handled at the spawn site instead
     }
          }, net::detached);
}