Cleanup server.h/c.
authorFlorent Bruneau <florent.bruneau@polytechnique.org>
Thu, 16 Oct 2008 21:02:22 +0000 (23:02 +0200)
committerFlorent Bruneau <florent.bruneau@polytechnique.org>
Thu, 16 Oct 2008 21:02:22 +0000 (23:02 +0200)
Signed-off-by: Florent Bruneau <florent.bruneau@polytechnique.org>
common/rbl.c
common/server.c
common/server.h
pfix-srsd/main-srsd.c
postlicyd/main-postlicyd.c

index e7c79fb..c118f15 100644 (file)
@@ -48,7 +48,7 @@ typedef struct rbl_context_t {
 ARRAY(rbl_context_t);
 
 static struct ub_ctx *ctx = NULL;
-static server_t *async_event = NULL;
+static client_t *async_event = NULL;
 static PA(rbl_context_t) ctx_pool = ARRAY_INIT;
 
 static rbl_context_t *rbl_context_new(void)
@@ -90,8 +90,7 @@ static void rbl_exit(void)
         ctx = NULL;
     }
     if (async_event != NULL) {
-        async_event->fd = -1;
-        server_release(async_event);
+        client_release(async_event);
         async_event = NULL;
     }
     array_deep_wipe(ctx_pool, rbl_context_delete);
@@ -121,15 +120,15 @@ static void rbl_callback(void *arg, int err, struct ub_result *result)
     rbl_context_release(context);
 }
 
-static int rbl_handler(server_t *event, void *config)
+static int rbl_handler(client_t *event, void *config)
 {
     int retval = 0;
     debug("rbl_handler called: ub_fd triggered");
-    server_none(event);
+    client_io_none(event);
     if ((retval = ub_process(ctx)) != 0) {
         err("error in DNS resolution: %s", ub_strerror(retval));
     }
-    server_ro(event);
+    client_io_ro(event);
     return 0;
 }
 
@@ -139,7 +138,7 @@ static inline bool rbl_dns_check(const char *hostname, rbl_result_t *result,
     if (ctx == NULL) {
         ctx = ub_ctx_create();
         ub_ctx_async(ctx, true);
-        if ((async_event = server_register(ub_fd(ctx), rbl_handler, NULL)) == NULL) {
+        if ((async_event = client_register(ub_fd(ctx), rbl_handler, NULL)) == NULL) {
             crit("cannot register asynchronous DNS event handler");
             abort();
         }
index b47da2f..f142cb7 100644 (file)
  * Copyright © 2008 Florent Bruneau
  */
 
+#include <ev.h>
 #include "server.h"
 #include "common.h"
 
-static PA(server_t) listeners   = ARRAY_INIT;
-static PA(server_t) server_pool = ARRAY_INIT;
+typedef struct server_io_t {
+    struct ev_io io;
+    int fd;
+} server_io_t;
 
-struct ev_loop *global_loop    = NULL;
-static start_client_t  client_start   = NULL;
-static delete_client_t client_delete  = NULL;
-static run_client_t    client_run     = NULL;
-static refresh_t       config_refresh = NULL;
-static void           *config_ptr     = NULL;
+struct listener_t {
+    server_io_t io;
+};
 
-static server_t* server_new(void)
+struct client_t {
+    server_io_t io;
+
+    buffer_t ibuf;
+    buffer_t obuf;
+
+    run_client_t run;
+    delete_client_t clear_data;
+    void* data;
+};
+
+
+static PA(listener_t) listeners = ARRAY_INIT;
+static PA(client_t) client_pool = ARRAY_INIT;
+
+static struct ev_loop *gl_loop           = NULL;
+static start_client_t  gl_client_start   = NULL;
+static delete_client_t gl_client_delete  = NULL;
+static run_client_t    gl_client_run     = NULL;
+static refresh_t       gl_config_refresh = NULL;
+static void           *gl_config         = NULL;
+
+
+/* Server io structure methods.
+ */
+
+static inline void server_io_wipe(server_io_t *io)
+{
+    if (io->fd >= 0) {
+        ev_io_stop(gl_loop, &io->io);
+        close(io->fd);
+        io->fd = -1;
+    }
+}
+
+
+/* Client methods.
+ */
+
+/* 1 - managing clients */
+
+static client_t* client_new(void)
 {
-    server_t* server = p_new(server_t, 1);
-    server->fd  = -1;
+    client_t* server = p_new(client_t, 1);
+    server->io.fd  = -1;
     return server;
 }
 
-static void server_wipe(server_t *server)
+static void client_wipe(client_t *server)
 {
-    if (server->fd >= 0) {
-        ev_io_stop(global_loop, &server->io);
-        close(server->fd);
-        server->fd = -1;
-    }
+    server_io_wipe(&server->io);
     if (server->data && server->clear_data) {
         server->clear_data(&server->data);
     }
+    server->data = NULL;
+    server->clear_data = NULL;
+    server->run = NULL;
 }
 
-void server_delete(server_t **server)
+void client_delete(client_t **server)
 {
     if (*server) {
         buffer_wipe(&(*server)->ibuf);
         buffer_wipe(&(*server)->obuf);
-        server_wipe(*server);
+        client_wipe(*server);
         p_delete(server);
     }
 }
 
-static server_t* server_acquire(void)
+static client_t* client_acquire(void)
 {
-    if (server_pool.len != 0) {
-        return array_elt(server_pool, --server_pool.len);
+    if (client_pool.len != 0) {
+        return array_pop_last(client_pool);
     } else {
-        return server_new();
+        return client_new();
     }
 }
 
-void server_release(server_t *server)
+void client_release(client_t *server)
 {
-    server_wipe(server);
-    array_add(server_pool, server);
+    client_wipe(server);
+    array_add(client_pool, server);
 }
 
-static int server_init(void)
+/* 2 - Doing I/O */
+
+void client_io_none(client_t *server)
 {
-    global_loop = ev_default_loop(0);
-    return 0;
+    ev_io_stop(gl_loop, &server->io.io);
 }
 
-static void server_shutdown(void)
+void client_io_rw(client_t *server)
 {
-    array_deep_wipe(listeners, server_delete);
-    array_deep_wipe(server_pool, server_delete);
+    ev_io_stop(gl_loop, &server->io.io);
+    ev_io_set(&server->io.io, server->io.fd, EV_READ | EV_WRITE);
+    ev_io_start(gl_loop, &server->io.io);
+}
+
+void client_io_ro(client_t *server)
+{
+    ev_io_stop(gl_loop, &server->io.io);
+    ev_io_set(&server->io.io, server->io.fd, EV_READ);
+    ev_io_start(gl_loop, &server->io.io);
+}
+
+ssize_t client_read(client_t *client)
+{
+    return buffer_read(&client->ibuf, client->io.fd, -1);
+}
+
+buffer_t *client_input_buffer(client_t *client)
+{
+    return &client->ibuf;
+}
+
+buffer_t *client_output_buffer(client_t *client)
+{
+    return &client->obuf;
+}
+
+void *client_data(client_t *client)
+{
+    return client->data;
 }
-module_init(server_init);
-module_exit(server_shutdown);
+
 
 static void client_cb(EV_P_ struct ev_io *w, int events)
 {
-    server_t *server = (server_t*)w;
+    client_t *server = (client_t*)w;
 
     debug("Entering client_cb for %p, %d (%d | %d)", w, events, EV_WRITE, EV_READ);
 
     if (events & EV_WRITE && server->obuf.len) {
-        if (buffer_write(&server->obuf, server->fd) < 0) {
-            server_release(server);
+        if (buffer_write(&server->obuf, server->io.fd) < 0) {
+            client_release(server);
             return;
         }
         if (!server->obuf.len) {
-            ev_io_set(&server->io, server->fd, EV_READ);
+            client_io_ro(server);
         }
     }
 
     if (events & EV_READ) {
-        if (server->run(server, config_ptr) < 0) {
-            server_release(server);
+        if (server->run(server, gl_config) < 0) {
+            client_release(server);
             return;
         }
     }
 }
 
-static int start_client(server_t *server, start_client_t starter,
-                        run_client_t runner, delete_client_t deleter)
+client_t *client_register(int fd, run_client_t runner, void *data)
 {
-    server_t *tmp;
+    if (fd < 0) {
+        return NULL;
+    }
+
+    client_t *tmp   = client_acquire();
+    tmp->io.fd      = fd;
+    tmp->data       = data;
+    tmp->run        = runner;
+    tmp->clear_data = NULL;
+    ev_io_init(&tmp->io.io, client_cb, tmp->io.fd, EV_READ);
+    ev_io_start(gl_loop, &tmp->io.io);
+    return tmp;
+}
+
+
+/* Listeners management.
+ */
+
+/* 1 - Allocation */
+
+static listener_t *listener_new(void)
+{
+    listener_t *io = p_new(listener_t, 1);
+    io->io.fd = -1;
+    return io;
+}
+
+static inline void listener_wipe(listener_t *io)
+{
+    server_io_wipe(&io->io);
+}
+
+static inline void listener_delete(listener_t **io)
+{
+    if (*io) {
+        listener_wipe(*io);
+        p_delete(io);
+    }
+}
+
+
+/* 2 - Management */
+
+static void listener_cb(EV_P_ struct ev_io *w, int events)
+{
+    listener_t *server = (listener_t*)w;
+    client_t *tmp;
     void* data = NULL;
     int sock;
 
-    sock = accept_nonblock(server->fd);
+    sock = accept_nonblock(server->io.fd);
     if (sock < 0) {
         UNIXERR("accept");
-        return -1;
+        ev_unloop(EV_A_ EVUNLOOP_ALL);
+        return;
     }
 
-    if (starter) {
-        data = starter(server);
+    if (gl_client_start) {
+        data = gl_client_start(server);
         if (data == NULL) {
             close(sock);
-            return -1;
+            ev_unloop(EV_A_ EVUNLOOP_ALL);
+            return;
         }
     }
 
-    tmp             = server_acquire();
-    tmp->fd         = sock;
+    tmp             = client_acquire();
+    tmp->io.fd      = sock;
     tmp->data       = data;
-    tmp->run        = runner;
-    tmp->clear_data = deleter;
-    ev_io_init(&tmp->io, client_cb, tmp->fd, EV_READ);
-    ev_io_start(global_loop, &tmp->io);
-    return 0;
+    tmp->run        = gl_client_run;
+    tmp->clear_data = gl_client_delete;
+    ev_io_init(&tmp->io.io, client_cb, tmp->io.fd, EV_READ);
+    ev_io_start(gl_loop, &tmp->io.io);
 }
 
-static void server_cb(EV_P_ struct ev_io *w, int events)
-{
-    server_t *server = (server_t*)w;
-    if (start_client(server, client_start, client_run, client_delete) != 0) {
-        ev_unloop(EV_A_ EVUNLOOP_ALL);
-    }
-}
-
-int start_server(int port, start_listener_t starter, delete_client_t deleter)
+listener_t *start_listener(int port)
 {
     struct sockaddr_in addr = {
         .sin_family = AF_INET,
         .sin_addr   = { htonl(INADDR_LOOPBACK) },
     };
-    server_t *tmp;
-    void* data = NULL;
+    listener_t *tmp;
     int sock;
 
     addr.sin_port = htons(port);
     sock = tcp_listen_nonblock((const struct sockaddr *)&addr, sizeof(addr));
     if (sock < 0) {
-        return -1;
-    }
-
-    if (starter) {
-      data = starter();
-      if (data == NULL) {
-        close(sock);
-        return -1;
-      }
+        return NULL;
     }
 
-    tmp             = server_acquire();
-    tmp->fd         = sock;
-    tmp->data       = data;
-    tmp->run        = NULL;
-    tmp->clear_data = deleter;
-    ev_io_init(&tmp->io, server_cb, tmp->fd, EV_READ);
-    ev_io_start(global_loop, &tmp->io);
+    tmp             = listener_new();
+    tmp->io.fd      = sock;
+    ev_io_init(&tmp->io.io, listener_cb, tmp->io.fd, EV_READ);
+    ev_io_start(gl_loop, &tmp->io.io);
     array_add(listeners, tmp);
-    return 0;
+    return tmp;
 }
 
-server_t *server_register(int fd, run_client_t runner, void *data)
+
+
+
+/* Server runtime stuff.
+ */
+
+static int server_init(void)
 {
-    if (fd < 0) {
-        return NULL;
-    }
+    gl_loop = ev_default_loop(0);
+    return 0;
+}
 
-    server_t *tmp   = server_acquire();
-    tmp->fd         = fd;
-    tmp->data       = data;
-    tmp->run        = runner;
-    tmp->clear_data = NULL;
-    ev_io_init(&tmp->io, client_cb, tmp->fd, EV_READ);
-    ev_io_start(global_loop, &tmp->io);
-    return tmp;
+static void server_shutdown(void)
+{
+    array_deep_wipe(listeners, listener_delete);
+    array_deep_wipe(client_pool, client_delete);
 }
+module_init(server_init);
+module_exit(server_shutdown);
+
 
 static void refresh_cb(EV_P_ struct ev_signal *w, int event)
 {
-    if (!config_refresh(config_ptr)) {
+    if (!gl_config_refresh(gl_config)) {
         ev_unloop(EV_A_ EVUNLOOP_ALL);
     }
 }
@@ -237,23 +336,23 @@ int server_loop(start_client_t starter, delete_client_t deleter,
     struct ev_signal ev_sigint;
     struct ev_signal ev_sigterm;
 
-    client_start   = starter;
-    client_delete  = deleter;
-    client_run     = runner;
-    config_refresh = refresh;
-    config_ptr     = config;
+    gl_client_start   = starter;
+    gl_client_delete  = deleter;
+    gl_client_run     = runner;
+    gl_config_refresh = refresh;
+    gl_config         = config;
 
     if (refresh != NULL) {
         ev_signal_init(&ev_sighup, refresh_cb, SIGHUP);
-        ev_signal_start(global_loop, &ev_sighup);
+        ev_signal_start(gl_loop, &ev_sighup);
     }
     ev_signal_init(&ev_sigint, exit_cb, SIGINT);
-    ev_signal_start(global_loop, &ev_sigint);
+    ev_signal_start(gl_loop, &ev_sigint);
     ev_signal_init(&ev_sigterm, exit_cb, SIGTERM);
-    ev_signal_start(global_loop, &ev_sigterm);
+    ev_signal_start(gl_loop, &ev_sigterm);
 
     info("entering processing loop");
-    ev_loop(global_loop, 0);
+    ev_loop(gl_loop, 0);
     info("exit requested");
     return EXIT_SUCCESS;
 }
index 2757eae..7dec88e 100644 (file)
 #ifndef PFIXTOOLS_SERVER_H
 #define PFIXTOOLS_SERVER_H
 
-#include <ev.h>
 #include "buffer.h"
 
-typedef struct server_t server_t;
+typedef struct client_t client_t;
+typedef struct listener_t listener_t;
+PARRAY(client_t)
+PARRAY(listener_t)
 
 typedef void *(*start_listener_t)(void);
 typedef void  (*delete_client_t)(void*);
-typedef void *(*start_client_t)(server_t*);
-typedef int   (*run_client_t)(server_t*, void*);
+typedef void *(*start_client_t)(listener_t*);
+typedef int   (*run_client_t)(client_t*, void*);
 typedef bool   (*refresh_t)(void*);
 
-extern struct ev_loop *global_loop;
 
-struct server_t {
-    struct ev_io io;
-    int fd;
+listener_t *start_listener(int port);
 
-    buffer_t ibuf;
-    buffer_t obuf;
+client_t *client_register(int fd, run_client_t runner, void *data);
+void client_delete(client_t **client);
+void client_release(client_t *client);
 
-    run_client_t run;
-    delete_client_t clear_data;
-    void* data;
-};
-ARRAY(server_t);
+void client_io_none(client_t *client);
+void client_io_rw(client_t *client);
+void client_io_ro(client_t *client);
 
-int start_server(int port, start_listener_t starter, delete_client_t deleter);
-
-server_t *server_register(int fd, run_client_t runner, void *data);
-void server_delete(server_t **server);
-void server_release(server_t *server);
-
-static inline void server_none(server_t *server)
-{
-    ev_io_stop(global_loop, &server->io);
-}
-
-static inline void server_rw(server_t *server)
-{
-    ev_io_stop(global_loop, &server->io);
-    ev_io_set(&server->io, server->fd, EV_READ | EV_WRITE);
-    ev_io_start(global_loop, &server->io);
-}
-
-static inline void server_ro(server_t *server)
-{
-    ev_io_stop(global_loop, &server->io);
-    ev_io_set(&server->io, server->fd, EV_READ);
-    ev_io_start(global_loop, &server->io);
-}
+ssize_t client_read(client_t *client);
+buffer_t *client_input_buffer(client_t *client);
+buffer_t *client_output_buffer(client_t *client);
+void *client_data(client_t *client);
 
 int server_loop(start_client_t starter, delete_client_t deleter,
                 run_client_t runner, refresh_t refresh, void *config);
index 440e56f..49339c1 100644 (file)
@@ -60,27 +60,12 @@ typedef struct srs_config_t {
 /* Server {{{1
  */
 
-static const char* const decoder_ptr = "decoder";
-static const char* const encoder_ptr = "encoder";
+static listener_t *decoder_ptr = NULL;
+static listener_t *encoder_ptr = NULL;
 
-static void *srsd_new_decoder(void)
+static void *srsd_starter(listener_t *server)
 {
-    return (void*)decoder_ptr;
-}
-
-static void *srsd_new_encoder(void)
-{
-    return (void*)encoder_ptr;
-}
-
-static void *srsd_starter(server_t *server)
-{
-    return server->data;
-}
-
-int start_listener(int port, bool decoder)
-{
-    return start_server(port, decoder ? srsd_new_decoder : srsd_new_encoder, NULL);
+    return server;
 }
 
 
@@ -107,74 +92,77 @@ void urldecode(char *s, char *end)
     *s++ = '\0';
 }
 
-int process_srs(server_t *srsd, void* vconfig)
+int process_srs(client_t *srsd, void* vconfig)
 {
     srs_config_t* config = vconfig;
-    int res = buffer_read(&srsd->ibuf, srsd->fd, -1);
+    buffer_t *ibuf = client_input_buffer(srsd);
+    buffer_t *obuf = client_output_buffer(srsd);
+    bool decoder = (client_data(srsd) == decoder_ptr);
+    int res = client_read(srsd);
 
     if ((res < 0 && errno != EINTR && errno != EAGAIN) || res == 0)
         return -1;
 
-    while (srsd->ibuf.len > 4) {
+    while (ibuf->len > 4) {
         char buf[BUFSIZ], *p, *q, *nl;
         int err;
 
-        nl = strchr(srsd->ibuf.data + 4, '\n');
+        nl = strchr(ibuf->data + 4, '\n');
         if (!nl) {
-            if (srsd->ibuf.len > BUFSIZ) {
+            if (ibuf->len > BUFSIZ) {
                 err("unreasonnable amount of data without a \\n");
                 return -1;
             }
-            if (srsd->obuf.len) {
-                server_rw(srsd);
+            if (obuf->len) {
+                client_io_rw(srsd);
             }
             return 0;
         }
 
-        if (strncmp("get ", srsd->ibuf.data, 4)) {
+        if (strncmp("get ", ibuf->data, 4)) {
                                                err("bad request, not starting with \"get \"");
             return -1;
         }
 
-        for (p = srsd->ibuf.data + 4; p < nl && isspace(*p); p++);
+        for (p = ibuf->data + 4; p < nl && isspace(*p); p++);
         for (q = nl++; q >= p && isspace(*q); *q-- = '\0');
 
         if (p == q) {
-            buffer_addstr(&srsd->obuf, "400 empty request ???\n");
+            buffer_addstr(&*obuf, "400 empty request ???\n");
             warn("empty request");
             goto skip;
         }
 
         urldecode(p, q);
 
-        if (srsd->data == (void*)decoder_ptr) {
+        if (decoder) {
             err = srs_reverse(config->srs, buf, ssizeof(buf), p);
         } else {
             err = srs_forward(config->srs, buf, ssizeof(buf), p, config->domain);
         }
 
         if (err == 0) {
-            buffer_addstr(&srsd->obuf, "200 ");
-            buffer_addstr(&srsd->obuf, buf);
+            buffer_addstr(&*obuf, "200 ");
+            buffer_addstr(&*obuf, buf);
         } else {
             switch (SRS_ERROR_TYPE(err)) {
               case SRS_ERRTYPE_SRS:
               case SRS_ERRTYPE_SYNTAX:
-                buffer_addstr(&srsd->obuf, "500 ");
+                buffer_addstr(&*obuf, "500 ");
                 break;
               default:
-                buffer_addstr(&srsd->obuf, "400 ");
+                buffer_addstr(&*obuf, "400 ");
                 break;
             }
-            buffer_addstr(&srsd->obuf, srs_strerror(err));
+            buffer_addstr(obuf, srs_strerror(err));
         }
-        buffer_addch(&srsd->obuf, '\n');
+        buffer_addch(obuf, '\n');
 
       skip:
-        buffer_consume(&srsd->ibuf, nl - srsd->ibuf.data);
+        buffer_consume(ibuf, nl - ibuf->data);
     }
-    if (srsd->obuf.len) {
-        server_rw(srsd);
+    if (obuf->len) {
+        client_io_rw(srsd);
     }
     return 0;
 }
@@ -323,8 +311,8 @@ int main(int argc, char *argv[])
     if (!config.srs
         || common_setup(pidfile, unsafe, RUNAS_USER, RUNAS_GROUP,
                         daemonize) != EXIT_SUCCESS
-        || start_listener(port_enc, false) < 0
-        || start_listener(port_dec, true) < 0) {
+        || (encoder_ptr = start_listener(port_enc)) == NULL
+        || (decoder_ptr = start_listener(port_dec)) == NULL) {
         return EXIT_FAILURE;
     }
     return server_loop(srsd_starter, NULL, process_srs, NULL, &config);
index 9baf34a..50c19d5 100644 (file)
@@ -41,7 +41,7 @@
 #include "policy_tokens.h"
 #include "server.h"
 #include "config.h"
-#include "postlicyd.h"
+#include "query.h"
 
 #define DAEMON_NAME             "postlicyd"
 #define DAEMON_VERSION          "0.3"
 
 DECLARE_MAIN
 
+typedef struct query_context_t {
+    query_t query;
+    filter_context_t context;
+    client_t *client;
+} query_context_t;
+
 static config_t *config  = NULL;
 static bool refresh      = false;
-static PA(server_t) busy = ARRAY_INIT;
+static PA(client_t) busy = ARRAY_INIT;
 
-static void *query_starter(server_t* server)
+static void *query_starter(listener_t* server)
 {
     query_context_t *context = p_new(query_context_t, 1);
     filter_context_prepare(&context->context, context);
@@ -78,45 +84,50 @@ static bool config_refresh(void *mconfig)
         return true;
     }
     bool ret = config_reload(mconfig);
-    foreach (server_t **server, busy) {
-        server_ro(*server);
+    foreach (client_t **server, busy) {
+        client_io_ro(*server);
     }}
     array_len(busy) = 0;
     refresh = false;
     return ret;
 }
 
-static void policy_answer(server_t *pcy, const char *message)
+static void policy_answer(client_t *pcy, const char *message)
 {
-    query_context_t *context = pcy->data;
+    query_context_t *context = client_data(pcy);
     const query_t* query = &context->query;
+    buffer_t *buf = client_output_buffer(pcy);
 
-    buffer_addstr(&pcy->obuf, "action=");
-    buffer_ensure(&pcy->obuf, m_strlen(message) + 64);
+    /* Write reply "action=ACTION [text]" */
+    buffer_addstr(buf, "action=");
+    buffer_ensure(buf, m_strlen(message) + 64);
 
-    ssize_t size = array_size(pcy->obuf) - array_len(pcy->obuf);
-    ssize_t format_size = query_format(array_ptr(pcy->obuf, array_len(pcy->obuf)),
+    ssize_t size = array_size(*buf) - array_len(*buf);
+    ssize_t format_size = query_format(array_ptr(*buf, array_len(*buf)),
                                        size, message, query);
     if (format_size == -1) {
-        buffer_addstr(&pcy->obuf, message);
+        buffer_addstr(buf, message);
     } else if (format_size > size) {
-        buffer_ensure(&pcy->obuf, format_size + 1);
-        query_format(array_ptr(pcy->obuf, array_len(pcy->obuf)),
-                     array_size(pcy->obuf) - array_len(pcy->obuf),
+        buffer_ensure(buf, format_size + 1);
+        query_format(array_ptr(*buf, array_len(*buf)),
+                     array_size(*buf) - array_len(*buf),
                      message, query);
-        array_len(pcy->obuf) += format_size;
+        array_len(*buf) += format_size;
     } else {
-        array_len(pcy->obuf) += format_size;
+        array_len(*buf) += format_size;
     }
-    buffer_addstr(&pcy->obuf, "\n\n");
-    buffer_consume(&pcy->ibuf, query->eoq - pcy->ibuf.data);
-    server_rw(pcy);
+    buffer_addstr(buf, "\n\n");
+
+    /* Finalize query. */
+    buf = client_input_buffer(pcy);
+    buffer_consume(buf, query->eoq - buf->data);
+    client_io_rw(pcy);
 }
 
-static const filter_t *next_filter(server_t *pcy, const filter_t *filter,
+static const filter_t *next_filter(client_t *pcy, const filter_t *filter,
                                    const query_t *query, const filter_hook_t *hook, bool *ok) {
     if (hook != NULL) {
-        query_context_t *context = pcy->data;
+        query_context_t *context = client_data(pcy);
         if (hook->counter >= 0 && hook->counter < MAX_COUNTERS && hook->cost > 0) {
             context->context.counters[hook->counter] += hook->cost;
             debug("request client=%s, from=<%s>, to=<%s>: added %d to counter %d (now %u)",
@@ -164,9 +175,9 @@ static const filter_t *next_filter(server_t *pcy, const filter_t *filter,
     }
 }
 
-static bool policy_process(server_t *pcy, const config_t *mconfig)
+static bool policy_process(client_t *pcy, const config_t *mconfig)
 {
-    query_context_t *context = pcy->data;
+    query_context_t *context = client_data(pcy);
     const query_t* query = &context->query;
     const filter_t *filter;
     if (mconfig->entry_points[query->state] == -1) {
@@ -189,20 +200,22 @@ static bool policy_process(server_t *pcy, const config_t *mconfig)
     }
 }
 
-static int policy_run(server_t *pcy, void* vconfig)
+static int policy_run(client_t *pcy, void* vconfig)
 {
+    const config_t *mconfig = vconfig;
     if (refresh) {
         array_add(busy, pcy);
         return 0;
     }
 
-    int search_offs = MAX(0, (int)(pcy->ibuf.len - 1));
-    int nb = buffer_read(&pcy->ibuf, pcy->fd, -1);
+    query_context_t *context = client_data(pcy);
+    query_t         *query   = &context->query;
+    context->client = pcy;
+
+    buffer_t *buf   = client_input_buffer(pcy);
+    int search_offs = MAX(0, (int)(buf->len - 1));
+    int nb          = client_read(pcy);
     const char *eoq;
-    query_context_t *context = pcy->data;
-    query_t  *query  = &context->query;
-    context->server = pcy;
-    const config_t *mconfig = vconfig;
 
     if (nb < 0) {
         if (errno == EAGAIN || errno == EINTR)
@@ -211,22 +224,26 @@ static int policy_run(server_t *pcy, void* vconfig)
         return -1;
     }
     if (nb == 0) {
-        if (pcy->ibuf.len)
+        if (buf->len)
             err("unexpected end of data");
         return -1;
     }
 
-    if (!(eoq = strstr(pcy->ibuf.data + search_offs, "\n\n")))
+    if (!(eoq = strstr(buf->data + search_offs, "\n\n"))) {
         return 0;
+    }
 
-    if (!query_parse(pcy->data, pcy->ibuf.data))
+    if (!query_parse(query, buf->data)) {
         return -1;
+    }
     query->eoq = eoq + strlen("\n\n");
+
+    /* The instance changed => reset the static context */
     if (query->instance == NULL || strcmp(context->context.instance, query->instance) != 0) {
         filter_context_clean(&context->context);
         m_strcat(context->context.instance, 64, query->instance);
     }
-    server_none(pcy);
+    client_io_none(pcy);
     return policy_process(pcy, mconfig) ? 0 : -1;
 }
 
@@ -237,14 +254,14 @@ static void policy_async_handler(filter_context_t *context,
     const filter_t *filter = context->current_filter;
     query_context_t *qctx  = context->data;
     query_t         *query = &qctx->query;
-    server_t        *server = qctx->server;
+    client_t        *server = qctx->client;
 
     context->current_filter = next_filter(server, filter, query, hook, &ok);
     if (context->current_filter != NULL) {
         ok = policy_process(server, config);
     }
     if (!ok) {
-        server_release(server);
+        client_release(server);
     }
     if (refresh && filter_running == 0) {
         config_refresh(config);
@@ -259,16 +276,11 @@ static int postlicyd_init(void)
 
 static void postlicyd_shutdown(void)
 {
-    array_deep_wipe(busy, server_delete);
+    array_deep_wipe(busy, client_delete);
 }
 module_init(postlicyd_init);
 module_exit(postlicyd_shutdown);
 
-int start_listener(int port)
-{
-    return start_server(port, NULL, NULL);
-}
-
 /* administrivia {{{ */
 
 void usage(void)
@@ -354,7 +366,7 @@ int main(int argc, char *argv[])
 
     pidfile_refresh();
 
-    if (start_listener(config->port) < 0) {
+    if (start_listener(config->port) == NULL) {
         return EXIT_FAILURE;
     } else {
         return server_loop(query_starter, query_stopper,