C ++-Boost streambuf放气的行为不一致? [英] C++ - Inconsistent behavior with Boost streambuf deflating?
问题描述
我正在尝试使用他们的 websocket与OKEX交易所建立Websocket连接API .我正在使用 Boost :: Beast websockets.
I'm trying to build a websocket connection to the OKEX exchange using their websocket API. I am using Boost::Beast websockets.
问题是OKEX的服务器未遵循正确的 permessage_deflate 压缩协议，并且发送的邮件
The problem is that OKEX's servers don't follow the correct permessage_deflate compression protocol, and send messages that are incorrectly deflated. So I'm trying to inflate the messages myself. The problem is that it's not working... and what's driving me insane is that the behavior I'm getting is somewhat inconsistent.
我的代码主要是从先前链接到线程复制并粘贴的.为简单起见，我删除了所有预处理器宏，并对套接字值进行了硬编码.
My code is mostly copy and pasted from the previously linked to thread. For simplicity, I removed all the preprocessor macros and I hardcoded the socket values.
膨胀代码来自 Raj Advani的回答.
这是 main.cpp 文件:
#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/ssl.hpp>
#include <boost/asio.hpp>
#include <boost/beast.hpp>
#include <boost/beast/websocket.hpp>
#include <boost/beast/websocket/ssl.hpp>
#include <istream>
#include "zlib.h"
#include <iostream>
namespace net = boost::asio;
namespace ssl = net::ssl;
namespace beast = boost::beast;
namespace http = beast::http;
namespace websocket = beast::websocket;
using tcp = net::ip::tcp;
using stream_t = websocket::stream<ssl::stream<tcp::socket>>;
int inflate(const void *src, int srcLen, void *dst, int dstLen) {
z_stream strm = {0};
strm.total_in = strm.avail_in = srcLen;
strm.total_out = strm.avail_out = dstLen;
strm.next_in = (Bytef *) src;
strm.next_out = (Bytef *) dst;
strm.zalloc = Z_NULL;
strm.zfree = Z_NULL;
strm.opaque = Z_NULL;
int err = -1;
int ret = -1;
err = inflateInit2(&strm, (15 + 32)); //15 window bits, and the +32 tells zlib to to detect if using gzip or zlib
if (err == Z_OK) {
err = inflate(&strm, Z_FINISH);
if (err == Z_STREAM_END) {
ret = strm.total_out;
}
else {
inflateEnd(&strm);
return err;
}
}
else {
inflateEnd(&strm);
return err;
}
inflateEnd(&strm);
return ret;
}
int main(int argc, char** argv) {
std::string host = "real.okex.com";
auto const port = "8443";
auto const path = "/ws/v3";
net::io_context ioc;
ssl::context ctx{ ssl::context::sslv23 };
tcp::resolver resolver{ ioc };
stream_t s{ ioc, ctx };
ctx.set_verify_mode(ssl::verify_none);
tcp::resolver::results_type results = resolver.resolve(host, port);
net::connect(
beast::get_lowest_layer(s),
//s.next_layer().next_layer(),
results.begin());
// SSL handshake
s.next_layer().handshake(ssl::stream_base::client);
s.handshake(host + ":" + port, path);
std::cout << "connected." << std::endl;
// send request to the websocket
s.write(net::buffer("{'op':'subscribe', 'args':['spot/ticker:ETH-USDT']}"));
{
net::streambuf buffer;
s.read(buffer);
// auto data_it = buffer.data().begin();
// std::cout<<"Iterating over data of size:" << buffer.data().size()<<endl; // LINE 85
// int i = 0;
// while (data_it != buffer.data().end()) {
// std::cout << "buffer data["<<i++<<"] size:" << (data_it->size())<<endl;
// data_it++;
// }
net::streambuf out_buffer;
const int error_code_out = inflate(&buffer, buffer.size(), &out_buffer, 10000000);
std::cout << "received. size:"<<buffer.size()<<" data: "<< &buffer << std::endl;
std::cout << "deflated. error?"<< error_code_out << " data: " << &out_buffer << std::endl;
}
}
代码输出+问题
通货膨胀说， buffer 的大小是117.我认为这是合理的，但是由于某些原因，我在解压缩时会收到 Z_DATA_ERROR ，这使我相信还有更多要解析的数据....
Code Output+Question
The inflation says that the buffer's size is 117. I thought that was reasonable, but for some reason, I get Z_DATA_ERROR when decompressing, leading me to believe there is more data to be parsed....
因此，我查阅了 net :: streambuf 的文档，发现显然可以读取多个缓冲区，所以也许我只使用了一个缓冲区?我运行了注释掉的代码(中间不包括 LINE 85 行)，但是它从未经历过循环……我认为这很奇怪.我放在那行中，然后突然间我有几百个缓冲区?(截断的)输出类似于:
So I looked up net::streambuf's documentation and I found that there are apparently multiple buffers that can be read from, so maybe I was using just one buffer? I ran the commented out code (excluding the LINE 85 line in the middle) and it never went through the loop... which I thought was odd. I put in that line, and then all of a sudden I have a couple hundred buffers? The (truncated) output is something like:
connected.
Iterating over data of size:117
buffer data[0] size:117
buffer data[1] size:72198326954657960
buffer data[2] size:140735485986592
buffer data[3] size:140618848326656
buffer data[4] size:140618848326656
.. many more lines of this...
buffer data[121] size:7089075335985461349
buffer data[122] size:3472329396561475632
buffer data[123] size:8747116609081390898
buffer data[124] size:3472329396561475632
buffer data[125] size:3472387902693336678
buffer data[126] size:
Process finished with exit code 139 (interrupted by signal 11: SIGSEGV)
您可以看到它崩溃了.我不知道发生了什么事.我现在不知道如何解码 streambuf ...和
as you can see it crashes. I have no clue what is going on. I have no clue how to decode the streambuf at this point... and the documentation seems to assume a lot of background knowledge i don't have. I tried using buffer.data(), converting the buffer to a char* array, all of which lead me to exactly the same behavior...
不知道该怎么办.欢迎任何帮助
Not sure what to do. Any help welcome
import websockets
import asyncio
import zlib
def inflate(data):
decompress = zlib.decompressobj(-zlib.MAX_WBITS)
inflated = decompress.decompress(data)
inflated += decompress.flush()
return inflated
async def main():
client = await websockets.connect("wss://real.okex.com:8443/ws/v3")
await client.send("{'op':'subscribe', 'args':['spot/ticker:ETH-USDT']}")
r = await client.recv()
print(len(r), r)
print(inflate(r))
if __name__ == '__main__':
asyncio.run(main())
推荐答案
我有一个较早的答案，说明了这种代码的问题.此后，我找到了解决剩余问题的解决方案:服务器响应.
I have an older answer up that kind of hashes out the problems with the code. I have since found the solution to the remaining issue: the server responses.
解决方案
OKEX不仅不遵守WS标准以启用按消息放气，而且还突然终止了数据.但是，事实证明，如果保留部分夸大的结果，那实际上很好.
Solution
Not only does OKEX not adhere to WS standards to enable per-message deflation, but also it abruptly ends the data. However, it turns out that it is actually fine if you keep the partially inflated results.
我实现它的方法是直接不使用ZLIB，而是使用 beast :: zlib :: inflate_stream .它具有更灵活的界面，可让我们获得所需的结果:
The way I made it work was by not using ZLIB directly but instead using beast::zlib::inflate_stream. This has a more flexible interface, which allows us to get the results we need:
namespace mylib {
auto inflate(std::vector<uint8_t> const& in, std::vector<uint8_t>& out) {
boost::system::error_code ec;
beast::zlib::z_params zp{};
zp.next_in = (Bytef*)in.data();
zp.avail_in = in.size();
zp.next_out = out.data();
zp.avail_out = out.size();
beast::zlib::inflate_stream zs;
zs.write(zp, beast::zlib::Flush::full, ec);
return ec;
}
}
现在我们像这样使用它
std::vector<uint8_t> in, out;
auto in_buffer = net::dynamic_buffer(in);
s.read(in_buffer);
out.resize(1024); // make sure it's enough
auto ec = mylib::inflate(in, out);
std::cout << "deflated. " << ec.message() << std::endl;
std::cout << std::string(out.begin(), out.end()) << std::endl;
它会打印
connected.
deflated. unexpected end of deflate stream
{"event":"error","message":"Unrecognized request: {'op':'subscribe', 'args':['spot/ticker:ETH-USDT']}\u0000","errorCode":30039}
因此，尽管deflate流意外结束，但数据是有效且完整的JSON.
So, despite the unexpected end of deflate stream, the data is valid and complete JSON.
Live On Compiler Exporer
#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/ssl.hpp>
#include <boost/asio.hpp>
#include <boost/beast.hpp>
#include <boost/beast/websocket.hpp>
#include <boost/beast/websocket/ssl.hpp>
#include <istream>
#include "zlib.h"
#include <iostream>
namespace net = boost::asio;
namespace ssl = net::ssl;
namespace beast = boost::beast;
namespace http = beast::http;
namespace websocket = beast::websocket;
using tcp = net::ip::tcp;
using stream_t = websocket::stream<ssl::stream<tcp::socket>>;
namespace mylib {
auto inflate(std::vector<uint8_t> const& in, std::vector<uint8_t>& out) {
boost::system::error_code ec;
beast::zlib::z_params zp{};
zp.next_in = (Bytef*)in.data();
zp.avail_in = in.size();
zp.next_out = out.data();
zp.avail_out = out.size();
beast::zlib::inflate_stream zs;
zs.write(zp, beast::zlib::Flush::full, ec);
return ec;
}
}
int main() {
std::string host = "real.okex.com";
auto const port = "8443";
auto const path = "/ws/v3";
net::io_context ioc;
tcp::resolver resolver{ ioc };
ssl::context ctx { ssl::context::sslv23 };
ctx.set_verify_mode(ssl::verify_none);
stream_t s{ ioc, ctx };
net::connect(beast::get_lowest_layer(s), resolver.resolve(host, port));
// SSL handshake
s.next_layer().handshake(ssl::stream_base::client);
{
websocket::permessage_deflate opt;
opt.client_enable = true; // for clients
opt.server_enable = true; // for servers
s.set_option(opt);
}
s.handshake(host + ":" + port, path);
std::cout << "connected." << std::endl;
// send request to the websocket
s.write(net::buffer("{'op':'subscribe', 'args':['spot/ticker:ETH-USDT']}"));
{
std::vector<uint8_t> in, out;
auto in_buffer = net::dynamic_buffer(in);
s.read(in_buffer);
out.resize(1024); // make sure it's enough
auto ec = mylib::inflate(in, out);
std::cout << "deflated. " << ec.message() << std::endl;
std::cout << std::string(out.begin(), out.end()) << std::endl;
}
}
这篇关于C ++-Boost streambuf放气的行为不一致?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
