LibCore+LibIPC+Everywhere: Return Stream::LocalSocket from LocalServer

This change unfortunately cannot be atomically made without a single
commit changing everything.

Most of the important changes are in LibIPC/Connection.cpp,
LibIPC/ServerConnection.cpp and LibCore/LocalServer.cpp.

The notable changes are:
- IPCCompiler now generates the decode and decode_message functions such
  that they take a Core::Stream::LocalSocket instead of the socket fd.
- IPC::Decoder now uses the receive_fd method of LocalSocket instead of
  doing system calls directly on the fd.
- IPC::ConnectionBase and related classes now use the Stream API
  functions.
- IPC::ServerConnection no longer constructs the socket itself; instead,
  a convenience macro, IPC_CLIENT_CONNECTION, is used in place of
  C_OBJECT and will generate a static try_create factory function for
  the ServerConnection subclass. The subclass is now responsible for
  passing the socket constructed in this function to its
  ServerConnection base; the socket is passed as the first argument to
  the constructor (as a NonnullOwnPtr<Core::Stream::LocalServer>) before
  any other arguments.
- The functionality regarding taking over sockets from SystemServer has
  been moved to LibIPC/SystemServerTakeover.cpp. The Core::LocalSocket
  implementation of this functionality hasn't been deleted due to my
  intention of removing this class in the near future and to reduce
  noise on this (already quite noisy) PR.

5 files changed, 33 insertions, 25 deletions

diff --git a/Userland/Services/InspectorServer/ClientConnection.cpp b/Userland/Services/InspectorServer/ClientConnection.cpp
index f289272882..ee5346b951 100644
--- a/Userland/Services/InspectorServer/ClientConnection.cpp
+++ b/Userland/Services/InspectorServer/ClientConnection.cpp
@@ -12,7 +12,7 @@ namespace InspectorServer {
 
 static HashMap<int, RefPtr<ClientConnection>> s_connections;
 
-ClientConnection::ClientConnection(NonnullRefPtr<Core::LocalSocket> socket, int client_id)
+ClientConnection::ClientConnection(NonnullOwnPtr<Core::Stream::LocalSocket> socket, int client_id)
     : IPC::ClientConnection<InspectorClientEndpoint, InspectorServerEndpoint>(*this, move(socket), client_id)
 {
     s_connections.set(client_id, *this);
diff --git a/Userland/Services/InspectorServer/ClientConnection.h b/Userland/Services/InspectorServer/ClientConnection.h
index 3b27c7cd83..e90086683e 100644
--- a/Userland/Services/InspectorServer/ClientConnection.h
+++ b/Userland/Services/InspectorServer/ClientConnection.h
@@ -23,7 +23,7 @@ public:
     virtual void die() override;
 
 private:
-    explicit ClientConnection(NonnullRefPtr<Core::LocalSocket>, int client_id);
+    explicit ClientConnection(NonnullOwnPtr<Core::Stream::LocalSocket>, int client_id);
 
     virtual Messages::InspectorServer::GetAllObjectsResponse get_all_objects(pid_t) override;
     virtual Messages::InspectorServer::SetInspectedObjectResponse set_inspected_object(pid_t, u64 object_id) override;
diff --git a/Userland/Services/InspectorServer/InspectableProcess.cpp b/Userland/Services/InspectorServer/InspectableProcess.cpp
index 9e2c98f5f1..dc5d9cb01e 100644
--- a/Userland/Services/InspectorServer/InspectableProcess.cpp
+++ b/Userland/Services/InspectorServer/InspectableProcess.cpp
@@ -16,14 +16,17 @@ InspectableProcess* InspectableProcess::from_pid(pid_t pid)
     return g_processes.get(pid).value_or(nullptr);
 }
 
-InspectableProcess::InspectableProcess(pid_t pid, NonnullRefPtr<Core::LocalSocket> socket)
+InspectableProcess::InspectableProcess(pid_t pid, NonnullOwnPtr<Core::Stream::LocalSocket> socket)
     : m_pid(pid)
     , m_socket(move(socket))
 {
-    m_socket->set_blocking(true);
+    // FIXME: Propagate errors
+    MUST(m_socket->set_blocking(true));
+
     m_socket->on_ready_to_read = [this] {
-        [[maybe_unused]] auto buffer = m_socket->read(1);
-        if (m_socket->eof()) {
+        char c;
+        [[maybe_unused]] auto buffer = m_socket->read({ &c, 1 });
+        if (m_socket->is_eof()) {
             g_processes.remove(m_pid);
             return;
         }
@@ -36,46 +39,51 @@ InspectableProcess::~InspectableProcess()
 
 String InspectableProcess::wait_for_response()
 {
-    if (m_socket->eof()) {
+    if (m_socket->is_eof()) {
         dbgln("InspectableProcess disconnected: PID {}", m_pid);
         m_socket->close();
         return {};
     }
 
     u32 length {};
-    auto nread = m_socket->read((u8*)&length, sizeof(length));
+    auto nread = m_socket->read({ (u8*)&length, sizeof(length) }).release_value_but_fixme_should_propagate_errors();
     if (nread != sizeof(length)) {
         dbgln("InspectableProcess got malformed data: PID {}", m_pid);
         m_socket->close();
         return {};
     }
 
-    ByteBuffer data;
-    size_t remaining_bytes = length;
+    auto data_buffer = ByteBuffer::create_uninitialized(length).release_value();
+    auto remaining_data_buffer = data_buffer.bytes();
 
-    while (remaining_bytes) {
-        auto packet = m_socket->read(remaining_bytes);
-        if (packet.size() == 0)
-            break;
-        if (auto result = data.try_append(packet.data(), packet.size()); result.is_error()) {
-            dbgln("Failed to append {} bytes to data buffer: {}", packet.size(), result.error());
+    while (!remaining_data_buffer.is_empty()) {
+        auto maybe_nread = m_socket->read(remaining_data_buffer);
+        if (maybe_nread.is_error()) {
+            dbgln("InspectableProcess::wait_for_response: Failed to read data: {}", maybe_nread.error());
             break;
         }
-        remaining_bytes -= packet.size();
+
+        auto nread = maybe_nread.release_value();
+        if (nread == 0)
+            break;
+
+        remaining_data_buffer = remaining_data_buffer.slice(nread);
     }
 
-    VERIFY(data.size() == length);
+    VERIFY(data_buffer.size() == length);
     dbgln("Got data size {} and read that many bytes", length);
 
-    return String::copy(data);
+    return String::copy(data_buffer);
 }
 
 void InspectableProcess::send_request(JsonObject const& request)
 {
     auto serialized = request.to_string();
     u32 length = serialized.length();
-    m_socket->write((u8 const*)&length, sizeof(length));
-    m_socket->write(serialized);
+
+    // FIXME: Propagate errors
+    MUST(m_socket->write({ (u8 const*)&length, sizeof(length) }));
+    MUST(m_socket->write(serialized.bytes()));
 }
 
 }
diff --git a/Userland/Services/InspectorServer/InspectableProcess.h b/Userland/Services/InspectorServer/InspectableProcess.h
index afefa0ba87..ed1547cce0 100644
--- a/Userland/Services/InspectorServer/InspectableProcess.h
+++ b/Userland/Services/InspectorServer/InspectableProcess.h
@@ -6,13 +6,13 @@
 
 #pragma once
 
-#include <LibCore/LocalSocket.h>
+#include <LibCore/Stream.h>
 
 namespace InspectorServer {
 
 class InspectableProcess {
 public:
-    InspectableProcess(pid_t, NonnullRefPtr<Core::LocalSocket>);
+    InspectableProcess(pid_t, NonnullOwnPtr<Core::Stream::LocalSocket>);
     ~InspectableProcess();
 
     void send_request(JsonObject const& request);
@@ -22,7 +22,7 @@ public:
 
 private:
     pid_t m_pid { 0 };
-    NonnullRefPtr<Core::LocalSocket> m_socket;
+    NonnullOwnPtr<Core::Stream::LocalSocket> m_socket;
 };
 
 extern HashMap<pid_t, NonnullOwnPtr<InspectorServer::InspectableProcess>> g_processes;
diff --git a/Userland/Services/InspectorServer/main.cpp b/Userland/Services/InspectorServer/main.cpp
index d9770f32fa..316238ef4c 100644
--- a/Userland/Services/InspectorServer/main.cpp
+++ b/Userland/Services/InspectorServer/main.cpp
@@ -25,7 +25,7 @@ ErrorOr<int> serenity_main(Main::Arguments)
     TRY(inspectables_server->take_over_from_system_server("/tmp/portal/inspectables"));
 
     inspectables_server->on_accept = [&](auto client_socket) {
-        auto pid = client_socket->peer_pid();
+        auto pid = client_socket->peer_pid().release_value_but_fixme_should_propagate_errors();
         InspectorServer::g_processes.set(pid, make<InspectorServer::InspectableProcess>(pid, move(client_socket)));
     };