HTTPServer: read requests from connected clients

diff --git a/src/httpserver.cpp b/src/httpserver.cpp
index c155544d..1e785a5b 100644
--- a/src/httpserver.cpp
+++ b/src/httpserver.cpp
@@ -1072,6 +1072,65 @@ void HTTPServer::NewSockAccepted(std::unique_ptr<Sock>&& sock, const CService& a
              addr.ToStringAddrPort(), id);
 }
 
+void HTTPServer::SocketHandlerConnected(const IOReadiness& io_readiness) const
+{
+    for (const auto& [sock, events] : io_readiness.events_per_sock) {
+        if (m_interrupt_net) {
+            return;
+        }
+
+        auto it{io_readiness.httpclients_per_sock.find(sock)};
+        if (it == io_readiness.httpclients_per_sock.end()) {
+            continue;
+        }
+        const std::shared_ptr<HTTPRemoteClient>& client{it->second};
+
+        bool send_ready = events.occurred & Sock::SEND;
+        bool recv_ready = events.occurred & Sock::RECV;
+        bool err_ready = events.occurred & Sock::ERR;
+
+        if (send_ready) {
+            // TODO: send data
+        }
+
+        if (recv_ready || err_ready) {
+            std::byte buf[0x10000]; // typical socket buffer is 8K-64K
+
+            const ssize_t nrecv{WITH_LOCK(
+                client->m_sock_mutex,
+                return client->m_sock->Recv(buf, sizeof(buf), MSG_DONTWAIT);)};
+
+            if (nrecv < 0) {
+                const int err = WSAGetLastError();
+                if (IOErrorIsPermanent(err)) {
+                    LogDebug(
+                        BCLog::HTTP,
+                        "Permanent read error from %s (id=%llu): %s",
+                        client->m_origin,
+                        client->m_id,
+                        NetworkErrorString(err));
+                    // TODO: Disconnect
+                }
+            } else if (nrecv == 0) {
+                LogDebug(
+                    BCLog::HTTP,
+                    "Received EOF from %s (id=%llu)",
+                    client->m_origin,
+                    client->m_id);
+                // TODO: Disconnect
+            } else {
+                // Copy data from socket buffer to client receive buffer
+                client->m_recv_buffer.insert(
+                    client->m_recv_buffer.end(),
+                    buf,
+                    buf + nrecv);
+                // Process as much received data as we can
+                MaybeDispatchRequestsFromClient(client);
+            }
+        }
+    }
+}
+
 void HTTPServer::SocketHandlerListening(const Sock::EventsPerSock& events_per_sock)
 {
     for (const auto& sock : m_listen) {
@@ -1131,8 +1190,66 @@ void HTTPServer::ThreadSocketHandler()
             m_interrupt_net.sleep_for(SELECT_TIMEOUT);
         }
 
+        // Service (send/receive) each of the already connected sockets.
+        SocketHandlerConnected(io_readiness);
+
         // Accept new connections from listening sockets.
         SocketHandlerListening(io_readiness.events_per_sock);
     }
 }
+
+void HTTPServer::MaybeDispatchRequestsFromClient(const std::shared_ptr<HTTPRemoteClient>& client) const
+{
+    // Try reading (potentially multiple) HTTP requests from the buffer
+    while (!client->m_recv_buffer.empty()) {
+        // Create a new request object and try to fill it with data from the receive buffer
+        auto req = std::make_unique<HTTPRequest>(client);
+        try {
+            // Stop reading if we need more data from the client to parse a complete request
+            if (!client->ReadRequest(*req)) break;
+        } catch (const std::runtime_error& e) {
+            LogDebug(
+                BCLog::HTTP,
+                "Error reading HTTP request from client %s (id=%llu): %s",
+                client->m_origin,
+                client->m_id,
+                e.what());
+
+            // We failed to read a complete request from the buffer
+            // TODO: respond with HTTP_BAD_REQUEST and disconnect
+
+            return;
+        }
+
+        // We read a complete request from the buffer into the queue
+        LogDebug(
+            BCLog::HTTP,
+            "Received a %s request for %s from %s (id=%llu)",
+            req->m_method,
+            req->m_target,
+            client->m_origin,
+            client->m_id);
+
+        // handle request
+        m_request_dispatcher(std::move(req));
+    }
+}
+
+bool HTTPRemoteClient::ReadRequest(HTTPRequest& req)
+{
+    LineReader reader(m_recv_buffer, MAX_HEADERS_SIZE);
+
+    if (!req.LoadControlData(reader)) return false;
+    if (!req.LoadHeaders(reader)) return false;
+    if (!req.LoadBody(reader)) return false;
+
+    // Remove the bytes read out of the buffer.
+    // If one of the above calls throws an error, the caller must
+    // catch it and disconnect the client.
+    m_recv_buffer.erase(
+        m_recv_buffer.begin(),
+        m_recv_buffer.begin() + reader.Consumed());
+
+    return true;
+}
 } // namespace http_bitcoin
diff --git a/src/httpserver.h b/src/httpserver.h
index 73a06441..092c5935 100644
--- a/src/httpserver.h
+++ b/src/httpserver.h
@@ -281,6 +281,8 @@ public:
     std::string StringifyHeaders() const;
 };
 
+class HTTPRemoteClient;
+
 class HTTPRequest
 {
 public:
@@ -290,6 +292,13 @@ public:
     HTTPHeaders m_headers;
     std::string m_body;
 
+    //! Pointer to the client that made the request so we know who to respond to.
+    std::shared_ptr<HTTPRemoteClient> m_client;
+
+    explicit HTTPRequest(std::shared_ptr<HTTPRemoteClient> client) : m_client{std::move(client)} {}
+    //! Construct with a null client for unit tests
+    explicit HTTPRequest() : m_client{} {}
+
     /**
      * Methods that attempt to parse HTTP request fields line-by-line
      * from a receive buffer.
@@ -305,8 +314,6 @@ public:
     /// @}
 };
 
-class HTTPRemoteClient;
-
 class HTTPServer
 {
 public:
@@ -315,6 +322,8 @@ public:
      */
     using Id = uint64_t;
 
+    explicit HTTPServer(std::function<void(std::unique_ptr<HTTPRequest>&&)> func) : m_request_dispatcher{std::move(func)} {}
+
     virtual ~HTTPServer()
     {
         Assume(!m_thread_socket_handler.joinable()); // Missing call to JoinSocketsThreads()
@@ -344,13 +353,6 @@ public:
      */
     size_t GetConnectionsCount() const { return m_connected_size.load(std::memory_order_acquire); }
 
-    /**
-     * This is a temporary method used to get a pointer to an HTTPRemoteClient
-     * so its connection can be inspected in a unit test.
-     * It will be removed in a future commit.
-     */
-    const HTTPRemoteClient& GetFirstConnection() { return *m_connected.front(); }
-
     /**
      * Start the necessary threads for sockets IO.
      */
@@ -436,6 +438,11 @@ private:
      */
     std::thread m_thread_socket_handler;
 
+    /*
+     * What to do with HTTP requests once received, validated and parsed
+     */
+    std::function<void(std::unique_ptr<HTTPRequest>&&)> m_request_dispatcher;
+
     /**
      * Accept a connection.
      * @param[in] listen_sock Socket on which to accept the connection.
@@ -457,6 +464,12 @@ private:
      */
     void NewSockAccepted(std::unique_ptr<Sock>&& sock, const CService& addr);
 
+    /**
+     * Do the read/write for connected sockets that are ready for IO.
+     * @param[in] io_readiness Which sockets are ready and their corresponding HTTPRemoteClients.
+     */
+    void SocketHandlerConnected(const IOReadiness& io_readiness) const;
+
     /**
      * Accept incoming connections, one from each read-ready listening socket.
      * @param[in] events_per_sock Sockets that are ready for IO.
@@ -475,6 +488,15 @@ private:
      * This is the main I/O loop of the server.
      */
     void ThreadSocketHandler();
+
+    /**
+     * Try to read HTTPRequests from a client's receive buffer.
+     * Complete requests are dispatched, incomplete requests are
+     * left in the buffer to wait for more data. Some read errors
+     * will mark this client for disconnection.
+     * @param[in] client The HTTPRemoteClient to read requests from
+     */
+    void MaybeDispatchRequestsFromClient(const std::shared_ptr<HTTPRemoteClient>& client) const;
 };
 
 class HTTPRemoteClient
@@ -489,6 +511,13 @@ public:
     //! IP:port of connected client, cached for logging purposes
     const std::string m_origin;
 
+    /**
+     * In lieu of an intermediate transport class like p2p uses,
+     * we copy data from the socket buffer to the client object
+     * and attempt to read HTTP requests from here.
+     */
+    std::vector<std::byte> m_recv_buffer{};
+
     /**
      * Mutex that serializes the Send() and Recv() calls on `m_sock`. Reading
      * from the client occurs in the I/O thread but writing back to a client
@@ -511,6 +540,13 @@ public:
     // Disable copies (should only be used as shared pointers)
     HTTPRemoteClient(const HTTPRemoteClient&) = delete;
     HTTPRemoteClient& operator=(const HTTPRemoteClient&) = delete;
+
+    /**
+     * Try to read an HTTP request from the receive buffer.
+     * @param[in]   req     A HTTPRequest to read into
+     * @returns true upon reading a complete request, otherwise false (may throw).
+     */
+    bool ReadRequest(HTTPRequest& req);
 };
 } // namespace http_bitcoin
 
diff --git a/src/test/httpserver_tests.cpp b/src/test/httpserver_tests.cpp
index b30a76f0..bb9da546 100644
--- a/src/test/httpserver_tests.cpp
+++ b/src/test/httpserver_tests.cpp
@@ -19,6 +19,16 @@ using http_bitcoin::MAX_BODY_SIZE;
 using http_bitcoin::MAX_HEADERS_SIZE;
 using util::LineReader;
 
+// HTTP request captured from bitcoin-cli
+constexpr std::string_view full_request = "POST / HTTP/1.1\r\n"
+                                          "Host: 127.0.0.1\r\n"
+                                          "Connection: close\r\n"
+                                          "Content-Type: application/json\r\n"
+                                          "Authorization: Basic X19jb29raWVfXzo5OGQ5ODQ3MWNmNjg0NzAzYTkzN2EzNzk0ZDFlODQ1NjZmYTRkZjJiMzFkYjhhODI4ZGY4MjVjOTg5ZGI4OTVl\r\n"
+                                          "Content-Length: 46\r\n"
+                                          "\r\n"
+                                          R"({"method":"getblockcount","params":[],"id":1})""\n";
+
 BOOST_FIXTURE_TEST_SUITE(httpserver_tests, SocketTestingSetup)
 
 BOOST_AUTO_TEST_CASE(test_query_parameters)
@@ -174,15 +184,6 @@ BOOST_AUTO_TEST_CASE(http_response_tests)
 BOOST_AUTO_TEST_CASE(http_request_tests)
 {
     {
-        // Reading request captured from bitcoin-cli
-        constexpr std::string_view full_request = "POST / HTTP/1.1\r\n"
-                                                  "Host: 127.0.0.1\r\n"
-                                                  "Connection: close\r\n"
-                                                  "Content-Type: application/json\r\n"
-                                                  "Authorization: Basic X19jb29raWVfXzo5OGQ5ODQ3MWNmNjg0NzAzYTkzN2EzNzk0ZDFlODQ1NjZmYTRkZjJiMzFkYjhhODI4ZGY4MjVjOTg5ZGI4OTVl\r\n"
-                                                  "Content-Length: 46\r\n"
-                                                  "\r\n"
-                                                  R"({"method":"getblockcount","params":[],"id":1})""\n";
         HTTPRequest req;
         LineReader reader(full_request, MAX_HEADERS_SIZE);
         BOOST_CHECK(req.LoadControlData(reader));
@@ -377,7 +378,16 @@ BOOST_AUTO_TEST_CASE(http_request_tests)
 
 BOOST_AUTO_TEST_CASE(http_server_socket_tests)
 {
-    HTTPServer server;
+    // Prepare a request handler that just stores received requests so we can examine them.
+    // Mutex is required to prevent a race between this test's main thread and the server's I/O loop.
+    Mutex requests_mutex;
+    std::deque<std::unique_ptr<HTTPRequest>> requests;
+    auto StoreRequest = [&](std::unique_ptr<HTTPRequest>&& req) {
+        LOCK(requests_mutex);
+        requests.push_back(std::move(req));
+    };
+
+    HTTPServer server{StoreRequest};
 
     {
         // We can only bind to NET_IPV4 and NET_IPV6
@@ -403,8 +413,8 @@ BOOST_AUTO_TEST_CASE(http_server_socket_tests)
     // No connections yet
     BOOST_CHECK_EQUAL(server.GetConnectionsCount(), 0);
 
-    // Create a mock client and add it to the local CreateSock queue
-    ConnectClient();
+    // Create a mock client with pre-loaded request data and add it to the local CreateSock queue
+    ConnectClient(std::as_bytes(std::span(full_request)));
 
     // Wait up to a minute to find and connect the client in the I/O loop
     int attempts{6000};
@@ -413,9 +423,31 @@ BOOST_AUTO_TEST_CASE(http_server_socket_tests)
         BOOST_REQUIRE(--attempts > 0);
     }
 
-    // Inspect the connection
-    const HTTPRemoteClient& client = server.GetFirstConnection();
-    BOOST_CHECK_EQUAL(client.m_origin, "5.5.5.5:6789");
+    // Prepare a pointer to the client, we'll assign it from the request itself.
+    std::shared_ptr<HTTPRemoteClient> client;
+
+    // Wait up to a minute to read the request from the client.
+    // Given that the mock client is itself a mock socket
+    // with hard-coded data it should only take a fraction of that.
+    attempts = 6000;
+    while (true) {
+        {
+            LOCK(requests_mutex);
+            // Connected client should have one request already from the static content.
+            if (requests.size() == 1) {
+                // Check the received request
+                BOOST_CHECK_EQUAL(requests.front()->m_body, R"({"method":"getblockcount","params":[],"id":1})""\n");
+
+                // Inspect the connection pointed to from the request
+                client = requests.front()->m_client;
+                BOOST_CHECK_EQUAL(client->m_origin, "5.5.5.5:6789");
+
+                break;
+            }
+        }
+        std::this_thread::sleep_for(10ms);
+        BOOST_REQUIRE(--attempts > 0);
+    }
 
     // Stop the I/O thread before modifying m_connected. CloseConnection() is a
     // temporary test-only API that is not thread-safe against GenerateWaitSockets(),
@@ -423,7 +455,7 @@ BOOST_AUTO_TEST_CASE(http_server_socket_tests)
     server.InterruptNet();
     server.JoinSocketsThreads();
     // Close connection
-    BOOST_REQUIRE(server.CloseConnection(client));
+    BOOST_REQUIRE(server.CloseConnection(*client));
     // Close all listening sockets
     server.StopListening();
 }
diff --git a/src/test/util/setup_common.cpp b/src/test/util/setup_common.cpp
index 846f4703..6a4f8ff1 100644
--- a/src/test/util/setup_common.cpp
+++ b/src/test/util/setup_common.cpp
@@ -658,12 +658,13 @@ SocketTestingSetup::~SocketTestingSetup()
     CreateSock = m_create_sock_orig;
 }
 
-void SocketTestingSetup::ConnectClient()
+void SocketTestingSetup::ConnectClient(std::span<const std::byte> data)
 {
     // I/O pipes for a mock Connected Socket we can read and write to.
     auto connected_socket_pipes(std::make_shared<DynSock::Pipes>());
 
-    // TODO: Insert a payload
+    // Insert the payload
+    connected_socket_pipes->recv.PushBytes(data.data(), data.size());
 
     // Create the Mock Connected Socket that represents a client.
     // It needs I/O pipes but its queue can remain empty
diff --git a/src/test/util/setup_common.h b/src/test/util/setup_common.h
index bf9ec42e..6773a689 100644
--- a/src/test/util/setup_common.h
+++ b/src/test/util/setup_common.h
@@ -256,8 +256,8 @@ public:
     explicit SocketTestingSetup();
     ~SocketTestingSetup();
 
-    //! Connect to the socket with a mock client (a DynSock)
-    void ConnectClient();
+    //! Connect to the socket with a mock client (a DynSock) and send pre-loaded data.
+    void ConnectClient(std::span<const std::byte> data);
 
 private:
     //! Save the original value of CreateSock here and restore it when the test ends.