test: add a mocked Sock that allows inspecting what has been Send() to it

And also allows gradually providing the data to be returned by `Recv()`
and sending and receiving net messages (`CNetMessage`).

src/test/util/net.cpp

@@ -14,7 +14,10 @@
 #include <random.h>
 #include <serialize.h>
 #include <span.h>
+#include <sync.h>
 
+#include <chrono>
+#include <optional>
 #include <vector>
 
 void ConnmanTestMsg::Handshake(CNode& node,
@@ -240,3 +243,168 @@ StaticContentsSock& StaticContentsSock::operator=(Sock&& other)
     assert(false && "Move of Sock into StaticContentsSock not allowed.");
     return *this;
 }
+
+ssize_t DynSock::Pipe::GetBytes(void* buf, size_t len, int flags)
+{
+    WAIT_LOCK(m_mutex, lock);
+
+    if (m_data.empty()) {
+        if (m_eof) {
+            return 0;
+        }
+        errno = EAGAIN; // Same as recv(2) on a non-blocking socket.
+        return -1;
+    }
+
+    const size_t read_bytes{std::min(len, m_data.size())};
+
+    std::memcpy(buf, m_data.data(), read_bytes);
+    if ((flags & MSG_PEEK) == 0) {
+        m_data.erase(m_data.begin(), m_data.begin() + read_bytes);
+    }
+
+    return read_bytes;
+}
+
+std::optional<CNetMessage> DynSock::Pipe::GetNetMsg()
+{
+    V1Transport transport{NodeId{0}};
+
+    {
+        WAIT_LOCK(m_mutex, lock);
+
+        WaitForDataOrEof(lock);
+        if (m_eof && m_data.empty()) {
+            return std::nullopt;
+        }
+
+        for (;;) {
+            Span<const uint8_t> s{m_data};
+            if (!transport.ReceivedBytes(s)) {  // Consumed bytes are removed from the front of s.
+                return std::nullopt;
+            }
+            m_data.erase(m_data.begin(), m_data.begin() + m_data.size() - s.size());
+            if (transport.ReceivedMessageComplete()) {
+                break;
+            }
+            if (m_data.empty()) {
+                WaitForDataOrEof(lock);
+                if (m_eof && m_data.empty()) {
+                    return std::nullopt;
+                }
+            }
+        }
+    }
+
+    bool reject{false};
+    CNetMessage msg{transport.GetReceivedMessage(/*time=*/{}, reject)};
+    if (reject) {
+        return std::nullopt;
+    }
+    return std::make_optional<CNetMessage>(std::move(msg));
+}
+
+void DynSock::Pipe::PushBytes(const void* buf, size_t len)
+{
+    LOCK(m_mutex);
+    const uint8_t* b = static_cast<const uint8_t*>(buf);
+    m_data.insert(m_data.end(), b, b + len);
+    m_cond.notify_all();
+}
+
+void DynSock::Pipe::Eof()
+{
+    LOCK(m_mutex);
+    m_eof = true;
+    m_cond.notify_all();
+}
+
+void DynSock::Pipe::WaitForDataOrEof(UniqueLock<Mutex>& lock)
+{
+    Assert(lock.mutex() == &m_mutex);
+
+    m_cond.wait(lock, [&]() EXCLUSIVE_LOCKS_REQUIRED(m_mutex) {
+        AssertLockHeld(m_mutex);
+        return !m_data.empty() || m_eof;
+    });
+}
+
+DynSock::DynSock(std::shared_ptr<Pipes> pipes, std::shared_ptr<Queue> accept_sockets)
+    : m_pipes{pipes}, m_accept_sockets{accept_sockets}
+{
+}
+
+DynSock::~DynSock()
+{
+    m_pipes->send.Eof();
+}
+
+ssize_t DynSock::Recv(void* buf, size_t len, int flags) const
+{
+    return m_pipes->recv.GetBytes(buf, len, flags);
+}
+
+ssize_t DynSock::Send(const void* buf, size_t len, int) const
+{
+    m_pipes->send.PushBytes(buf, len);
+    return len;
+}
+
+std::unique_ptr<Sock> DynSock::Accept(sockaddr* addr, socklen_t* addr_len) const
+{
+    ZeroSock::Accept(addr, addr_len);
+    return m_accept_sockets->Pop().value_or(nullptr);
+}
+
+bool DynSock::Wait(std::chrono::milliseconds timeout,
+                   Event requested,
+                   Event* occurred) const
+{
+    EventsPerSock ev;
+    ev.emplace(this, Events{requested});
+    const bool ret{WaitMany(timeout, ev)};
+    if (occurred != nullptr) {
+        *occurred = ev.begin()->second.occurred;
+    }
+    return ret;
+}
+
+bool DynSock::WaitMany(std::chrono::milliseconds timeout, EventsPerSock& events_per_sock) const
+{
+    const auto deadline = std::chrono::steady_clock::now() + timeout;
+    bool at_least_one_event_occurred{false};
+
+    for (;;) {
+        // Check all sockets for readiness without waiting.
+        for (auto& [sock, events] : events_per_sock) {
+            if ((events.requested & Sock::SEND) != 0) {
+                // Always ready for Send().
+                events.occurred |= Sock::SEND;
+                at_least_one_event_occurred = true;
+            }
+
+            if ((events.requested & Sock::RECV) != 0) {
+                auto dyn_sock = reinterpret_cast<const DynSock*>(sock.get());
+                uint8_t b;
+                if (dyn_sock->m_pipes->recv.GetBytes(&b, 1, MSG_PEEK) == 1 || !dyn_sock->m_accept_sockets->Empty()) {
+                    events.occurred |= Sock::RECV;
+                    at_least_one_event_occurred = true;
+                }
+            }
+        }
+
+        if (at_least_one_event_occurred || std::chrono::steady_clock::now() > deadline) {
+            break;
+        }
+
+        std::this_thread::sleep_for(10ms);
+    }
+
+    return true;
+}
+
+DynSock& DynSock::operator=(Sock&&)
+{
+    assert(false && "Move of Sock into DynSock not allowed.");
+    return *this;
+}