Introduce CIPCServerSideClient

As a new base class of IPC connections server-side.
Port ASClientConnection to CIPCServerSideClient.

Libraries/LibCore/CIPCServerSideClient.h

@@ -0,0 +1,206 @@
+#pragma once
+
+#include <LibCore/CObject.h>
+#include <LibCore/CEvent.h>
+#include <LibCore/CEventLoop.h>
+#include <LibCore/CIODevice.h>
+#include <LibCore/CNotifier.h>
+
+#include <errno.h>
+#include <unistd.h>
+#include <sys/uio.h>
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <stdio.h>
+
+class CIPCServerEvent : public CEvent {
+public:
+    enum Type {
+        Invalid = 2000,
+        ClientDisconnected,
+    };
+    CIPCServerEvent() {}
+    explicit CIPCServerEvent(Type type)
+        : CEvent(type)
+    {
+    }
+};
+
+class ASClientDisconnectedNotification : public CIPCServerEvent {
+public:
+    explicit ASClientDisconnectedNotification(int client_id)
+        : CIPCServerEvent(ClientDisconnected)
+        , m_client_id(client_id)
+    {
+    }
+
+    int client_id() const { return m_client_id; }
+
+private:
+    int m_client_id { 0 };
+};
+
+template <typename T, class... Args>
+T* CIPCServerSideClientCreator(Args&& ... args)
+{
+    auto conn = new T(AK::forward<Args>(args)...) /* arghs */;
+    conn->send_greeting();
+    return conn;
+};
+
+template <typename ServerMessage, typename ClientMessage>
+class CIPCServerSideClient : public CObject
+{
+public:
+    CIPCServerSideClient(int fd, int client_id)
+        : m_socket(fd)
+        , m_notifier(CNotifier(fd, CNotifier::Read))
+        , m_client_id(client_id)
+    {
+        m_notifier.on_ready_to_read = [this] { drain_client(); };
+        dbg() << "********** S: Created new CIPCServerSideClient " << fd << client_id << " and said hello";
+    }
+
+    ~CIPCServerSideClient()
+    {
+        dbg() << "********** S: Destroyed CIPCServerSideClient " << m_socket.fd() << client_id();
+    }
+
+    void post_message(const ServerMessage& message, const ByteBuffer& extra_data = {})
+    {
+        if (!extra_data.is_empty())
+            const_cast<ServerMessage&>(message).extra_size = extra_data.size();
+
+        struct iovec iov[2];
+        int iov_count = 1;
+
+        iov[0].iov_base = const_cast<ServerMessage*>(&message);
+        iov[0].iov_len = sizeof(message);
+
+        if (!extra_data.is_empty()) {
+            iov[1].iov_base = const_cast<u8*>(extra_data.data());
+            iov[1].iov_len = extra_data.size();
+            ++iov_count;
+        }
+
+        int nwritten = writev(m_socket.fd(), iov, iov_count);
+        if (nwritten < 0) {
+            switch (errno) {
+            case EPIPE:
+                dbgprintf("WSClientConnection::post_message: Disconnected from peer.\n");
+                delete_later();
+                return;
+                break;
+            case EAGAIN:
+                dbgprintf("WSClientConnection::post_message: Client buffer overflowed.\n");
+                did_misbehave();
+                return;
+                break;
+            default:
+                perror("WSClientConnection::post_message writev");
+                ASSERT_NOT_REACHED();
+            }
+        }
+
+        ASSERT(nwritten == (int)(sizeof(message) + extra_data.size()));
+    }
+
+    void drain_client()
+    {
+        unsigned messages_received = 0;
+        for (;;) {
+            ClientMessage message;
+            // FIXME: Don't go one message at a time, that's so much context switching, oof.
+            ssize_t nread = recv(m_socket.fd(), &message, sizeof(ClientMessage), MSG_DONTWAIT);
+            if (nread == 0 || (nread == -1 && errno == EAGAIN)) {
+                if (!messages_received) {
+                    // TODO: is delete_later() sufficient?
+                    CEventLoop::current().post_event(*this, make<ASClientDisconnectedNotification>(client_id()));
+                }
+                break;
+            }
+            if (nread < 0) {
+                perror("recv");
+                ASSERT_NOT_REACHED();
+            }
+            ByteBuffer extra_data;
+            if (message.extra_size) {
+                if (message.extra_size >= 32768) {
+                    dbgprintf("message.extra_size is way too large\n");
+                    return did_misbehave();
+                }
+                extra_data = ByteBuffer::create_uninitialized(message.extra_size);
+                // FIXME: We should allow this to time out. Maybe use a socket timeout?
+                int extra_nread = read(m_socket.fd(), extra_data.data(), extra_data.size());
+                if (extra_nread != (int)message.extra_size) {
+                    dbgprintf("extra_nread(%d) != extra_size(%d)\n", extra_nread, extra_data.size());
+                    if (extra_nread < 0)
+                        perror("read");
+                    return did_misbehave();
+                }
+            }
+            if (!handle_message(message, move(extra_data)))
+                return;
+            ++messages_received;
+        }
+    }
+
+    void did_misbehave()
+    {
+        dbgprintf("CIPCServerSideClient{%p} (id=%d, pid=%d) misbehaved, disconnecting.\n", this, client_id(), m_pid);
+        delete_later();
+        m_notifier.set_enabled(false);
+    }
+
+    const char* class_name() const override { return "CIPCServerSideClient"; }
+
+    int client_id() const { return m_client_id; }
+    pid_t client_pid() const { return m_pid; }
+    void set_client_pid(pid_t pid) { m_pid = pid; }
+
+    // ### having this public is sad
+    virtual void send_greeting() = 0;
+
+protected:
+    void event(CEvent& event)
+    {
+        if (event.type() == CIPCServerEvent::ClientDisconnected) {
+            int client_id = static_cast<const ASClientDisconnectedNotification&>(event).client_id();
+            dbgprintf("CIPCServerSideClient: Client disconnected: %d\n", client_id);
+            delete this;
+            return;
+        }
+
+        CObject::event(event);
+    }
+
+    virtual bool handle_message(const ClientMessage&, const ByteBuffer& = {}) = 0;
+
+private:
+    // TODO: A way to create some kind of CIODevice with an open FD would be nice.
+    class COpenedSocket : public CIODevice
+    {
+    public:
+        const char* class_name() const override { return "COpenedSocket"; }
+        COpenedSocket(int fd)
+        {
+            set_fd(fd);
+            set_mode(CIODevice::OpenMode::ReadWrite);
+        }
+
+        bool open(CIODevice::OpenMode) override
+        {
+            ASSERT_NOT_REACHED();
+            return true;
+        };
+
+        int fd() const { return CIODevice::fd(); }
+    };
+
+    COpenedSocket m_socket;
+    CNotifier m_notifier;
+    int m_client_id;
+    int m_pid;
+};
+
+

Servers/AudioServer/ASClientConnection.cpp

@@ -14,121 +14,29 @@
 #include <stdio.h>
 
 ASClientConnection::ASClientConnection(int fd, int client_id, ASMixer& mixer)
-    : m_socket(fd)
+    : CIPCServerSideClient(fd, client_id)
-    , m_notifier(CNotifier(fd, CNotifier::Read))
-    , m_client_id(client_id)
     , m_mixer(mixer)
 {
-    m_notifier.on_ready_to_read = [this] { drain_client(); };
-    ASAPI_ServerMessage message;
-    message.type = ASAPI_ServerMessage::Type::Greeting;
-    message.greeting.server_pid = getpid();
-    message.greeting.your_client_id = m_client_id;
-    post_message(message);
-    dbg() << "********** S: Created new ASClientConnection " << fd << client_id << " and said hello";
 }
 
 ASClientConnection::~ASClientConnection()
 {
-    dbg() << "********** S: Destroyed ASClientConnection " << m_socket.fd() << m_client_id;
 }
 
-void ASClientConnection::post_message(const ASAPI_ServerMessage& message, const ByteBuffer& extra_data)
+void ASClientConnection::send_greeting()
 {
-    if (!extra_data.is_empty())
+    ASAPI_ServerMessage message;
-        const_cast<ASAPI_ServerMessage&>(message).extra_size = extra_data.size();
+    message.type = ASAPI_ServerMessage::Type::Greeting;
-
+    message.greeting.server_pid = getpid();
-    struct iovec iov[2];
+    message.greeting.your_client_id = client_id();
-    int iov_count = 1;
+    post_message(message);
-
-    iov[0].iov_base = const_cast<ASAPI_ServerMessage*>(&message);
-    iov[0].iov_len = sizeof(message);
-
-    if (!extra_data.is_empty()) {
-        iov[1].iov_base = const_cast<u8*>(extra_data.data());
-        iov[1].iov_len = extra_data.size();
-        ++iov_count;
-    }
-
-    int nwritten = writev(m_socket.fd(), iov, iov_count);
-    if (nwritten < 0) {
-        switch (errno) {
-        case EPIPE:
-            dbgprintf("WSClientConnection::post_message: Disconnected from peer.\n");
-            delete_later();
-            return;
-            break;
-        case EAGAIN:
-            dbgprintf("WSClientConnection::post_message: Client buffer overflowed.\n");
-            did_misbehave();
-            return;
-            break;
-        default:
-            perror("WSClientConnection::post_message writev");
-            ASSERT_NOT_REACHED();
-        }
-    }
-
-    ASSERT(nwritten == (int)(sizeof(message) + extra_data.size()));
 }
 
-void ASClientConnection::event(CEvent& event)
+bool ASClientConnection::handle_message(const ASAPI_ClientMessage& message, const ByteBuffer&&)
-{
-    if (event.type() == ASEvent::WM_ClientDisconnected) {
-        int client_id = static_cast<const ASClientDisconnectedNotification&>(event).client_id();
-        dbgprintf("ASClientConnection: Client disconnected: %d\n", client_id);
-        delete this;
-        return;
-    }
-
-    CObject::event(event);
-}
- 
-void ASClientConnection::drain_client()
-{
-    unsigned messages_received = 0;
-    for (;;) {
-        ASAPI_ClientMessage message;
-        // FIXME: Don't go one message at a time, that's so much context switching, oof.
-        ssize_t nread = recv(m_socket.fd(), &message, sizeof(ASAPI_ClientMessage), MSG_DONTWAIT);
-        if (nread == 0 || (nread == -1 && errno == EAGAIN)) {
-            if (!messages_received) {
-                // TODO: is delete_later() sufficient?
-                CEventLoop::current().post_event(*this, make<ASClientDisconnectedNotification>(m_client_id));
-            }
-            break;
-        }
-        if (nread < 0) {
-            perror("recv");
-            ASSERT_NOT_REACHED();
-        }
-        ByteBuffer extra_data;
-        if (message.extra_size) {
-            if (message.extra_size >= 32768) {
-                dbgprintf("message.extra_size is way too large\n");
-                return did_misbehave();
-            }
-            extra_data = ByteBuffer::create_uninitialized(message.extra_size);
-            // FIXME: We should allow this to time out. Maybe use a socket timeout?
-            int extra_nread = read(m_socket.fd(), extra_data.data(), extra_data.size());
-            if (extra_nread != (int)message.extra_size) {
-                dbgprintf("extra_nread(%d) != extra_size(%d)\n", extra_nread, extra_data.size());
-                if (extra_nread < 0)
-                    perror("read");
-                return did_misbehave();
-            }
-        }
-        if (!handle_message(message, move(extra_data)))
-            return;
-        ++messages_received;
-    }
-}
-
-bool ASClientConnection::handle_message(const ASAPI_ClientMessage& message, const ByteBuffer&)
 {
     switch (message.type) {
     case ASAPI_ClientMessage::Type::Greeting:
-        m_pid = message.greeting.client_pid;
+        set_client_pid(message.greeting.client_pid);
         break;
     case ASAPI_ClientMessage::Type::PlayBuffer: {
         // ### ensure that the size is that of a Vector<ASample>
@@ -168,9 +76,3 @@ bool ASClientConnection::handle_message(const ASAPI_ClientMessage& message, cons
     return true;
 }
 
-void ASClientConnection::did_misbehave()
-{
-    dbgprintf("ASClientConnection{%p} (id=%d, pid=%d) misbehaved, disconnecting.\n", this, m_client_id, m_pid);
-    delete_later();
-    m_notifier.set_enabled(false);
-}

Servers/AudioServer/ASClientConnection.h

@@ -1,84 +1,20 @@
 #pragma once
 
-#include <LibCore/CObject.h>
+#include <LibCore/CIPCServerSideClient.h>
-#include <LibCore/CEvent.h>
+#include <LibAudio/ASAPI.h>
-#include <LibCore/CIODevice.h>
-#include <LibCore/CNotifier.h>
-
-struct ASAPI_ServerMessage;
-struct ASAPI_ClientMessage;
-
-class ASEvent : public CEvent {
-public:
-    enum Type {
-        Invalid = 2000,
-        WM_ClientDisconnected,
-    };
-    ASEvent() {}
-    explicit ASEvent(Type type)
-        : CEvent(type)
-    {
-    }
-};
-
-class ASClientDisconnectedNotification : public ASEvent {
-public:
-    explicit ASClientDisconnectedNotification(int client_id)
-        : ASEvent(WM_ClientDisconnected)
-        , m_client_id(client_id)
-    {
-    }
-
-    int client_id() const { return m_client_id; }
-
-private:
-    int m_client_id { 0 };
-};
 
 class ASMixer;
 
-class ASClientConnection : public CObject
+class ASClientConnection final : public CIPCServerSideClient<ASAPI_ServerMessage, ASAPI_ClientMessage>
 {
 public:
-    ASClientConnection(int fd, int client_id, ASMixer& mixer);
+    explicit ASClientConnection(int fd, int client_id, ASMixer& mixer);
-    ~ASClientConnection();
+    ~ASClientConnection() override;
-
+    void send_greeting() override;
-    void post_message(const ASAPI_ServerMessage&, const ByteBuffer& = {});
+    bool handle_message(const ASAPI_ClientMessage&, const ByteBuffer& = {}) override;
-    bool handle_message(const ASAPI_ClientMessage&, const ByteBuffer& = {});
-
-    void drain_client();
-
-    void did_misbehave();
-
     const char* class_name() const override { return "ASClientConnection"; }
 
-protected:
-    void event(CEvent& event) override;
 private:
-    // TODO: A way to create some kind of CIODevice with an open FD would be nice.
-    class ASOpenedSocket : public CIODevice
-    {
-    public:
-        const char* class_name() const override { return "ASOpenedSocket"; }
-        ASOpenedSocket(int fd)
-        {
-            set_fd(fd);
-            set_mode(CIODevice::OpenMode::ReadWrite);
-        }
-
-        bool open(CIODevice::OpenMode) override
-        {
-            ASSERT_NOT_REACHED();
-            return true;
-        };
-
-        int fd() const { return CIODevice::fd(); }
-    };
-
-    ASOpenedSocket m_socket;
-    CNotifier m_notifier;
-    int m_client_id;
-    int m_pid;
     ASMixer& m_mixer;
 };
 

Servers/AudioServer/ASEventLoop.cpp

@@ -31,7 +31,8 @@ void ASEventLoop::drain_server()
     } else {
         dbgprintf("AudioServer: accept()ed client %d\n", client_fd);
         static int s_next_client_id = 0;
-        new ASClientConnection(client_fd, s_next_client_id++, m_mixer);
+        CIPCServerSideClientCreator<ASClientConnection>(client_fd, s_next_client_id++, m_mixer);
+        //new ASClientConnection(client_fd, s_next_client_id++, m_mixer);
     }
 }