diff options
Diffstat (limited to 'src/lib/util/client_ws.hpp')
-rw-r--r-- | src/lib/util/client_ws.hpp | 59 |
1 files changed, 46 insertions, 13 deletions
diff --git a/src/lib/util/client_ws.hpp b/src/lib/util/client_ws.hpp index f3c4ded6bb2..04af5026014 100644 --- a/src/lib/util/client_ws.hpp +++ b/src/lib/util/client_ws.hpp @@ -1,5 +1,21 @@ // license:MIT // copyright-holders:Ole Christian Eidheim, Miodrag Milanovic + +/* + Changes to the websocket client + + July 2022: (Michael Zapf, patch by Pete Eberlein [peberlein]) + Symptom: During the initial handshake, the ws client reads the HTTP header + from the server and then waits for more data to come. In cases where the + websocket server sends data immediately after the handshake to the client, + these data may already be available in the streambuf without triggering + a new reception event, so the client may wait forever for a new reception. + + Correction: When going into the async_read, the client only waits for so + many bytes as to satisfy the next read from the streambuf. For this end, + the Message::needed_for method has been added. Also, leftover bytes are + copied to the next message instance. +*/ #ifndef CLIENT_WS_HPP #define CLIENT_WS_HPP @@ -11,6 +27,7 @@ #include <atomic> #include <list> #include "crypto.hpp" +#include "base64.hpp" #ifndef CASE_INSENSITIVE_EQUALS_AND_HASH #define CASE_INSENSITIVE_EQUALS_AND_HASH @@ -65,7 +82,7 @@ namespace webpp { unsigned short remote_endpoint_port; private: - explicit Connection(socket_type* socket): remote_endpoint_port(0), socket(socket), strand(socket->get_io_context()), closed(false) { } + explicit Connection(asio::io_context &context, socket_type* socket): remote_endpoint_port(0), socket(socket), strand(context), closed(false) { } class SendData { public: @@ -122,16 +139,28 @@ namespace webpp { size_t size() const { return length; } - std::string string() const { - std::stringstream ss; - ss << rdbuf(); - return ss.str(); + std::string string() { + if (s.size() != length) { + asio::streambuf::const_buffers_type bufs = streambuf.data(); + s.assign(asio::buffers_begin(bufs), asio::buffers_begin(bufs)+length); + streambuf.consume(length); + } + return s; + } + + // Return the number of bytes that are missing in the streambuf + // before n bytes can be read from it + size_t needed_for(size_t n) const { + if (streambuf.size() >= n) + return 0; + return n - streambuf.size(); } private: Message(): std::istream(&streambuf), fin_rsv_opcode(0), length(0) { } size_t length; asio::streambuf streambuf; + std::string s; }; std::function<void()> on_open; @@ -345,10 +374,10 @@ namespace webpp { } void read_message(const std::shared_ptr<Message> &message) { - asio::async_read(*connection->socket, message->streambuf, asio::transfer_exactly(2), + asio::async_read(*connection->socket, message->streambuf, asio::transfer_exactly(message->needed_for(2)), [this, message](const std::error_code& ec, size_t bytes_transferred) { if(!ec) { - if(bytes_transferred==0) { //TODO: This might happen on server at least, might also happen here + if(message->streambuf.size()+bytes_transferred==0) { //TODO: This might happen on server at least, might also happen here read_message(message); return; } @@ -362,7 +391,7 @@ namespace webpp { if(first_bytes[1]>=128) { const std::string reason("message from server masked"); auto kept_connection=connection; - send_close(1002, reason, [this, kept_connection](const std::error_code& /*ec*/) {}); + send_close(1002, reason, [](const std::error_code& /*ec*/) {}); if(on_close) on_close(1002, reason); return; @@ -372,7 +401,7 @@ namespace webpp { if(length==126) { //2 next bytes is the size of content - asio::async_read(*connection->socket, message->streambuf, asio::transfer_exactly(2), + asio::async_read(*connection->socket, message->streambuf, asio::transfer_exactly(message->needed_for(2)), [this, message] (const std::error_code& ec, size_t /*bytes_transferred*/) { if(!ec) { @@ -394,7 +423,7 @@ namespace webpp { } else if(length==127) { //8 next bytes is the size of content - asio::async_read(*connection->socket, message->streambuf, asio::transfer_exactly(8), + asio::async_read(*connection->socket, message->streambuf, asio::transfer_exactly(message->needed_for(8)), [this, message] (const std::error_code& ec, size_t /*bytes_transferred*/) { if(!ec) { @@ -425,7 +454,7 @@ namespace webpp { } void read_message_content(const std::shared_ptr<Message> &message) { - asio::async_read(*connection->socket, message->streambuf, asio::transfer_exactly(message->length), + asio::async_read(*connection->socket, message->streambuf, asio::transfer_exactly(message->needed_for(message->length)), [this, message] (const std::error_code& ec, size_t /*bytes_transferred*/) { if(!ec) { @@ -440,7 +469,7 @@ namespace webpp { auto reason=message->string(); auto kept_connection=connection; - send_close(status, reason, [this, kept_connection](const std::error_code& /*ec*/) {}); + send_close(status, reason, [](const std::error_code& /*ec*/) {}); if(on_close) on_close(status, reason); return; @@ -457,6 +486,10 @@ namespace webpp { //Next message std::shared_ptr<Message> next_message(new Message()); + //Copy any leftover data in the streambuf into the next message + next_message->streambuf.commit(buffer_copy( + next_message->streambuf.prepare(message->streambuf.size()), + message->streambuf.data())); read_message(next_message); } else if(on_error) @@ -488,7 +521,7 @@ namespace webpp { resolver->async_resolve(query, [this] (const std::error_code &ec, asio::ip::tcp::resolver::iterator it){ if(!ec) { - connection=std::shared_ptr<Connection>(new Connection(new WS(*io_context))); + connection=std::shared_ptr<Connection>(new Connection(*io_context, new WS(*io_context))); asio::async_connect(*connection->socket, it, [this] (const std::error_code &ec, asio::ip::tcp::resolver::iterator /*it*/){ |