websocket: add WebsocketClient

examples/meson.build

@@ -33,6 +33,7 @@ endif
 
 websocket_examples_sources = [
   'websocket_server.cpp',
+  'websocket_client.cpp',
 ]
 if openssl_dep.found()
   foreach source: websocket_examples_sources

examples/websocket_client.cpp

@@ -0,0 +1,24 @@
+#include "uvio/net/http.hpp"
+#include "uvio/net/websocket.hpp"
+
+using namespace uvio::net::http;
+using namespace uvio::net;
+
+auto process_message(websocket::WebsocketFramed &channel) -> Task<> {
+    // Ignore errors
+    for (int i = 0; i < 64; i++) {
+        auto msg = std::format("test message {} from client", i);
+        co_await channel.send(msg);
+
+        auto msg2 = (co_await channel.recv()).value();
+        LOG_INFO("Received: `{}`", std::string_view{msg2.data(), msg2.size()});
+    }
+    co_await channel.close();
+    co_return;
+}
+
+auto main() -> int {
+    websocket::WebsocketClient client{"127.0.0.1", 8000};
+    client.handle_message(process_message);
+    client.run();
+}

examples/websocket_server.cpp

@@ -9,6 +9,7 @@ auto process_message(websocket::WebsocketFramed &channel) -> Task<> {
     for (int i = 0; i < 64; i++) {
         auto msg = (co_await channel.recv()).value();
         LOG_INFO("Received: `{}`", std::string_view{msg.data(), msg.size()});
+
         co_await channel.send(msg);
     }
     co_await channel.close();

uvio/codec/http.hpp

@@ -44,7 +44,7 @@ class HttpCodec : public Codec<HttpCodec> {
         // Host: xxx
         // Content-Type: application/html
         auto headers = parse_headers(request_headers);
-        req.headers = headers;
+        req.headers = std::move(headers);
 
         if (auto has_length = headers.find("Content-Length")) {
             std::string body;
@@ -75,7 +75,54 @@ class HttpCodec : public Codec<HttpCodec> {
     template <typename Reader>
     auto decode(http::HttpResponse &resp, Reader &reader)
         -> Task<Result<void>> {
-        // TODO(x)
+        std::string status_line;
+        if (auto ret = co_await reader.read_until(status_line, "\r\n"); !ret) {
+            co_return unexpected{ret.error()};
+        }
+        LOG_DEBUG("{}", status_line);
+        // HTTP/1.1 200 OK
+        auto ret = parse_status_line(status_line);
+        if (!ret) {
+            co_return unexpected{ret.error()};
+        }
+        auto [version, status_code, status_text] = ret.value();
+        LOG_DEBUG("version: {}", version);
+        LOG_DEBUG("status_code: {}", status_code);
+        LOG_DEBUG("status_text: {}", status_text);
+        resp.status_code = std::stoi(std::string{status_code});
+
+        std::string request_headers(2, 0);
+        if (auto ret = co_await reader.read_exact(request_headers); !ret) {
+            co_return unexpected{ret.error()};
+        } else if (request_headers == "\r\n") {
+            // no response body
+            co_return Result<void>{};
+        }
+
+        if (auto ret = co_await reader.read_until(request_headers, "\r\n\r\n");
+            !ret) {
+            co_return unexpected{ret.error()};
+        }
+        LOG_DEBUG("{}", request_headers);
+        // Server: Apache
+        // Content-Length: 81
+        // Connection: Keep-Alive
+        // Content-Type: text/html
+        auto headers = parse_headers(request_headers);
+        resp.headers = std::move(headers);
+
+        if (auto has_length = headers.find("Content-Length")) {
+            std::string body;
+            auto        length = std::stoi(has_length.value());
+            LOG_DEBUG("Parsed body length: {}", length);
+            body.resize(length);
+            if (auto ret = co_await reader.read_exact(body); !ret) {
+                co_return unexpected{ret.error()};
+            }
+            LOG_DEBUG("body: `{}`", body);
+            resp.body = std::move(body);
+        }
+
         co_return Result<void>{};
     }
 
@@ -170,6 +217,45 @@ class HttpCodec : public Codec<HttpCodec> {
         return std::tuple{parts[0], parts[1], parts[2]};
     }
 
+    auto parse_status_line(std::string_view line) -> Result<
+        std::tuple<std::string_view, std::string_view, std::string_view>> {
+        std::vector<std::string_view> parts;
+
+        std::size_t start = 0;
+        std::size_t end = 0;
+
+        // HTTP version
+        while (end < line.size() && line[end] != ' ') {
+            ++end;
+        }
+        if (line[end] != ' ') {
+            return unexpected{make_uvio_error(Error::Unclassified)};
+        }
+        parts.emplace_back(line.data() + start, end - start);
+        start = ++end;
+
+        // status code
+        while (end < line.size() && line[end] != ' ') {
+            ++end;
+        }
+        if (line[end] != ' ') {
+            return unexpected{make_uvio_error(Error::Unclassified)};
+        }
+        parts.emplace_back(line.data() + start, end - start);
+        start = ++end;
+
+        // status text
+        while (end < line.size() && line[end] != '\r' && line[end] != '\n') {
+            ++end;
+        }
+        if (line[end] != '\r' && line[end] != '\n') {
+            return unexpected{make_uvio_error(Error::Unclassified)};
+        }
+        parts.emplace_back(line.data() + start, end - start);
+
+        return std::tuple{parts[0], parts[1], parts[2]};
+    }
+
     auto trim(std::string_view str) -> std::string_view {
         std::size_t start = 0;
         std::size_t end = str.size();

uvio/net/websocket/websocket_client.hpp

@@ -0,0 +1,80 @@
+#pragma once
+
+#include "uvio/common/base64.hpp"
+#include "uvio/core.hpp"
+#include "uvio/crypto/secure_hash.hpp"
+#include "uvio/debug.hpp"
+#include "uvio/net/tcp_listener.hpp"
+#include "uvio/net/websocket/websocket_frame.hpp"
+
+namespace uvio::net::websocket {
+
+class WebsocketClient {
+    using WebsocketHandlerFunc
+        = std::function<Task<>(WebsocketFramed &websocket_framed)>;
+
+public:
+    WebsocketClient(std::string_view host, int port)
+        : host_{host}
+        , port_{port} {}
+
+public:
+    auto handle_message(WebsocketHandlerFunc &&func) {
+        websocket_handler_ = std::move(func);
+    }
+
+    auto run() {
+        block_on([this]() -> Task<> {
+            console.info("Connecting to {}:{} ...", this->host_, this->port_);
+            auto has_stream
+                = co_await TcpStream::connect(this->host_, this->port_);
+            if (!has_stream) {
+                LOG_ERROR("{}", has_stream.error().message());
+            }
+            auto stream = std::move(has_stream.value());
+            spawn(this->handle_websocket(std::move(stream)));
+        }());
+    }
+
+private:
+    auto handle_websocket(TcpStream stream) -> Task<void> {
+        WebsocketFramed websocket_framed{std::move(stream)};
+        websocket_framed.client_side();
+
+        // GET␣/␣HTTP/1.1\r\nHost:␣localhost:8000\r\nUpgrade:␣websocket\r\nConnection:␣Upgrade\r\nSec-WebSocket-Key:␣8AVhXO1zSlflZw1LfElVnw==\r\nSec-WebSocket-Version:␣13\r\nSec-WebSocket-Extensions:␣permessage-deflate;␣client_max_window_bits\r\nUser-Agent:␣Python/3.12␣websockets/12.0\r\n\r\n
+        http::HttpRequest req{
+            .method = "GET",
+            .uri = "/",
+        };
+        req.headers.add("Upgrade", "websocket");
+        req.headers.add("Connection", "Upgrade");
+        // req.headers.add("Sec-WebSocket-Key", security_key);
+        req.headers.add("Sec-WebSocket-Key", "8AVhXO1zSlflZw1LfElVnw==");
+        req.headers.add("Sec-WebSocket-Version", "13");
+        // req.headers.add("Sec-WebSocket-Extensions", "permessage-deflate;
+        // client_max_window_bits");
+        req.headers.add("User-Agent", "UVIO/0.0.1");
+
+        if (auto ok = co_await websocket_framed.send_request(req); !ok) {
+            console.info("handshake failed");
+            co_return;
+        }
+        if (auto ok = co_await websocket_framed.recv_response(); !ok) {
+            console.info("handshake failed");
+            co_return;
+        }
+
+        if (websocket_handler_) {
+            co_await websocket_handler_(websocket_framed);
+        }
+        co_return;
+    }
+
+private:
+    std::string          host_;
+    int                  port_;
+    TcpStream            stream_{nullptr};
+    WebsocketHandlerFunc websocket_handler_;
+};
+
+} // namespace uvio::net::websocket