diff --git a/CMakeLists.txt b/CMakeLists.txt
index 10dc97edc..4382ad713 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -34,6 +34,7 @@ option(HAVE_ASSERT_PANIC "assert_panic disabled by default" OFF)
 option(HAVE_LOGGING "logging enabled by default" ON)
 option(HAVE_STATS "stats enabled by default" ON)
 option(TARGET_PINGSERVER "build pingserver binary" ON)
+option(TARGET_PUBSUB "build pubsub binary" ON)
 option(TARGET_SLIMCACHE "build slimcache binary" ON)
 option(TARGET_TWEMCACHE "build twemcache binary" ON)
 option(COVERAGE "code coverage" OFF)
diff --git a/config/pubsub.conf b/config/pubsub.conf
new file mode 100644
index 000000000..b0c646850
--- /dev/null
+++ b/config/pubsub.conf
@@ -0,0 +1,3 @@
+debug_log_level: 6
+# debug_log_file: pubsub.log
+# debug_log_nbuf: 16384
diff --git a/src/core/context.h b/src/core/context.h
index f3d215eb4..4d705bec4 100644
--- a/src/core/context.h
+++ b/src/core/context.h
@@ -12,5 +12,6 @@ struct context {
 };
 
 bool admin_init;
+bool pubsub_init;
 bool server_init;
 bool worker_init;
diff --git a/src/core/core.c b/src/core/core.c
index a4ea7433b..90b01ea72 100644
--- a/src/core/core.c
+++ b/src/core/core.c
@@ -10,7 +10,7 @@
 #include <sysexits.h>
 
 void
-core_run(void *arg_worker)
+worker_run(void *arg_worker)
 {
     pthread_t worker, server;
     int ret;
@@ -37,3 +37,32 @@ core_run(void *arg_worker)
 error:
     exit(EX_OSERR);
 }
+
+void
+pubsub_run(void *arg_pubsub)
+{
+    pthread_t pubsub, server;
+    int ret;
+
+    if (!admin_init || !server_init || !pubsub_init) {
+        log_crit("cannot run: admin/server/pubsub have to be initialized");
+        return;
+    }
+
+    ret = pthread_create(&pubsub, NULL, core_pubsub_evloop, arg_pubsub);
+    if (ret != 0) {
+        log_crit("pthread create failed for pubsub thread: %s", strerror(ret));
+        goto error;
+    }
+
+    ret = pthread_create(&server, NULL, core_server_evloop, NULL);
+    if (ret != 0) {
+        log_crit("pthread create failed for server thread: %s", strerror(ret));
+        goto error;
+    }
+
+    core_admin_evloop();
+
+error:
+    exit(EX_OSERR);
+}
diff --git a/src/core/core.h b/src/core/core.h
index c05542c13..ce8ec3f7b 100644
--- a/src/core/core.h
+++ b/src/core/core.h
@@ -5,8 +5,10 @@
  */
 
 #include "admin/admin.h"
+#include "data/pubsub.h"
 #include "data/shared.h"
 #include "data/server.h"
 #include "data/worker.h"
 
-void core_run(void *arg_worker);
+void worker_run(void *arg_worker);
+void pubsub_run(void *arg_pubsub);
diff --git a/src/core/data/CMakeLists.txt b/src/core/data/CMakeLists.txt
index 8b3b8afbe..061e06c68 100644
--- a/src/core/data/CMakeLists.txt
+++ b/src/core/data/CMakeLists.txt
@@ -1,5 +1,6 @@
 set(SOURCE
     ${SOURCE}
+    ${CMAKE_CURRENT_SOURCE_DIR}/pubsub.c
     ${CMAKE_CURRENT_SOURCE_DIR}/server.c
     ${CMAKE_CURRENT_SOURCE_DIR}/worker.c
     PARENT_SCOPE)
diff --git a/src/core/data/pubsub.c b/src/core/data/pubsub.c
new file mode 100644
index 000000000..2f77d0e88
--- /dev/null
+++ b/src/core/data/pubsub.c
@@ -0,0 +1,302 @@
+#include "pubsub.h"
+
+#include "shared.h"
+
+#include <time/time.h>
+
+#include <buffer/cc_buf.h>
+#include <buffer/cc_dbuf.h>
+#include <cc_debug.h>
+#include <cc_event.h>
+#include <cc_ring_array.h>
+#include <channel/cc_channel.h>
+#include <channel/cc_pipe.h>
+#include <channel/cc_tcp.h>
+
+#include <stream/cc_sockio.h>
+
+#include <sysexits.h>
+
+#define PUBSUB_MODULE_NAME "core::pubsub"
+
+pubsub_metrics_st *pubsub_metrics = NULL;
+
+static struct context context;
+struct context *ctx = &context;
+
+static channel_handler_st handlers;
+static channel_handler_st *hdl = &handlers;
+
+struct pubsub_processor *processor;
+
+static inline rstatus_i
+_pubsub_write(struct buf_sock *s)
+{
+    rstatus_i status;
+
+    log_verb("writing on buf_sock %p", s);
+
+    ASSERT(s != NULL);
+    ASSERT(s->wbuf != NULL && s->rbuf != NULL);
+
+    status = buf_tcp_write(s);
+
+    return status;
+}
+
+/* the caller only needs to check the return status of this function if
+ * it previously received a write event and wants to re-register the
+ * read event upon full, successful write.
+ */
+static inline rstatus_i
+_pubsub_event_write(struct buf_sock *s)
+{
+    rstatus_i status;
+    struct tcp_conn *c = s->ch;
+
+    status = _pubsub_write(s);
+    if (status == CC_ERETRY || status == CC_EAGAIN) { /* retry write */
+        /* by removing current masks and only listen to write event(s), we are
+         * effectively stopping processing incoming data until we can write
+         * something to the (kernel) buffer for the channel. This is sensible
+         * because either the local network or the client is backed up when
+         * kernel write buffer is full, and this allows us to propagate back
+         * pressure to the sending side.
+         */
+
+        event_del(ctx->evb, hdl->wid(c));
+        event_add_write(ctx->evb, hdl->wid(c), s);
+    } else if (status == CC_ERROR) {
+        c->state = CHANNEL_TERM;
+    }
+
+    if (processor->write(s) < 0) {
+        log_debug("handler signals channel termination");
+        c->state = CHANNEL_TERM;
+        return CC_ERROR;
+    }
+
+    return status;
+}
+
+static inline void
+_pubsub_read(struct buf_sock *s)
+{
+    log_verb("reading on buf_sock %p", s);
+
+    ASSERT(s != NULL);
+    ASSERT(s->wbuf != NULL && s->rbuf != NULL);
+
+    /* TODO(kyang): consider refactoring dbuf_tcp_read and buf_tcp_read to have no return status
+       at all, since the return status is already given by the connection state */
+    buf_tcp_read(s);
+}
+
+static inline void
+pubsub_close(struct buf_sock *s)
+{
+    log_info("pubsub core close on buf_sock %p", s);
+
+    processor->error(s);
+    event_del(ctx->evb, hdl->rid(s->ch));
+    hdl->term(s->ch);
+    buf_sock_return(&s);
+}
+
+/* read event over an existing connection */
+static inline void
+_pubsub_event_read(struct buf_sock *s)
+{
+    ASSERT(s != NULL);
+
+    _pubsub_read(s);
+    if (processor->read(s) < 0) {
+        log_debug("handler signals channel termination");
+        s->ch->state = CHANNEL_TERM;
+        return;
+    }
+    if (buf_rsize(s->wbuf) > 0) {
+        log_verb("attempt to write");
+        _pubsub_event_write(s);
+    }
+}
+
+static void
+pubsub_add_conn(void)
+{
+    struct buf_sock *s;
+    char buf[RING_ARRAY_DEFAULT_CAP]; /* buffer for discarding pipe data */
+    int i;
+    rstatus_i status;
+
+    /* server pushes connection on to the ring array before writing to the pipe,
+     * therefore, we should read from the pipe first and take the connections
+     * off the ring array to match the number of bytes received.
+     *
+     * Once we move server to its own thread, it is possible that there are more
+     * connections added to the queue when we are processing, it is OK to wait
+     * for the next read event in that case.
+     */
+
+    i = pipe_recv(pipe_c, buf, RING_ARRAY_DEFAULT_CAP);
+    if (i < 0) { /* errors, do not read from ring array */
+        log_warn("not adding new connections due to pipe error");
+        return;
+    }
+
+    /* each byte in the pipe corresponds to a new connection, which we will
+     * now get from the ring array
+     */
+    for (; i > 0; --i) {
+        status = ring_array_pop(&s, conn_arr);
+        if (status != CC_OK) {
+            log_warn("event number does not match conn queue: missing %d conns",
+                    i);
+            return;
+        }
+        log_verb("Adding new buf_sock %p to pubsub thread", s);
+        s->owner = ctx;
+        s->hdl = hdl;
+        event_add_read(ctx->evb, hdl->rid(s->ch), s);
+    }
+}
+
+static void
+_pubsub_event(void *arg, uint32_t events)
+{
+    struct buf_sock *s = arg;
+    log_verb("pubsub event %06"PRIX32" on buf_sock %p", events, s);
+
+    if (s == NULL) {
+        /* event on pipe_c, new connection */
+        if (events & EVENT_READ) {
+            pubsub_add_conn();
+        } else if (events & EVENT_ERR) {
+            log_error("error event received on conn_fds pipe");
+        } else {
+            /* there should never be any write events on the pipe from pubsub */
+            NOT_REACHED();
+        }
+    } else {
+        /* event on one of the connections */
+
+        if (events & EVENT_READ) {
+            log_verb("processing pubsub read event on buf_sock %p", s);
+            INCR(pubsub_metrics, pubsub_event_read);
+            _pubsub_event_read(s);
+        } else if (events & EVENT_WRITE) {
+            log_verb("processing pubsub write event on buf_sock %p", s);
+            INCR(pubsub_metrics, pubsub_event_write);
+            if (_pubsub_event_write(s) == CC_OK) {
+                /* write backlog cleared up, re-add read event (only) */
+                event_del(ctx->evb, hdl->wid(s->ch));
+                event_add_read(ctx->evb, hdl->rid(s->ch), s);
+            }
+        } else if (events & EVENT_ERR) {
+            s->ch->state = CHANNEL_TERM;
+            INCR(pubsub_metrics, pubsub_event_error);
+        } else {
+            NOT_REACHED();
+        }
+
+        /* TODO(yao): come up with a robust policy about channel connection
+         * and pending data. Since an error can either be server (usually
+         * memory) issues or client issues (bad syntax etc), or requested (quit)
+         * it is hard to determine whether the channel should be immediately
+         * closed or not. A simplistic approach might be to always close asap,
+         * and clients should not initiate closing unless they have received all
+         * their responses. This is not as nice as the TCP half-close behavior,
+         * but simpler to implement and probably fine initially.
+         */
+        if (s->ch->state == CHANNEL_TERM || s->ch->state == CHANNEL_ERROR) {
+            pubsub_close(s);
+        }
+    }
+}
+
+void
+core_pubsub_setup(pubsub_options_st *options, pubsub_metrics_st *metrics)
+{
+    int timeout = PUBSUB_TIMEOUT;
+    int nevent = PUBSUB_NEVENT;
+
+    log_info("set up the %s module", PUBSUB_MODULE_NAME);
+
+    if (pubsub_init) {
+        log_warn("pubsub has already been setup, re-creating");
+        core_pubsub_teardown();
+    }
+
+    pubsub_metrics = metrics;
+
+    if (options != NULL) {
+        timeout = option_uint(&options->pubsub_timeout);
+        nevent = option_uint(&options->pubsub_nevent);
+    }
+
+    ctx->timeout = timeout;
+    ctx->evb = event_base_create(nevent, _pubsub_event);
+    if (ctx->evb == NULL) {
+        log_crit("failed to setup pubsub thread core; could not create event_base");
+        exit(EX_CONFIG);
+    }
+
+    hdl->accept = (channel_accept_fn)tcp_accept;
+    hdl->reject = (channel_reject_fn)tcp_reject;
+    hdl->open = (channel_open_fn)tcp_listen;
+    hdl->term = (channel_term_fn)tcp_close;
+    hdl->recv = (channel_recv_fn)tcp_recv;
+    hdl->send = (channel_send_fn)tcp_send;
+    hdl->rid = (channel_id_fn)tcp_read_id;
+    hdl->wid = (channel_id_fn)tcp_write_id;
+
+    event_add_read(ctx->evb, pipe_read_id(pipe_c), NULL);
+
+    pubsub_init = true;
+}
+
+void
+core_pubsub_teardown(void)
+{
+    log_info("tear down the %s module", PUBSUB_MODULE_NAME);
+
+    if (!pubsub_init) {
+        log_warn("%s has never been setup", PUBSUB_MODULE_NAME);
+    } else {
+        event_base_destroy(&(ctx->evb));
+    }
+    pubsub_metrics = NULL;
+    pubsub_init = false;
+}
+
+static rstatus_i
+_pubsub_evwait(void)
+{
+    int n;
+
+    n = event_wait(ctx->evb, ctx->timeout);
+    if (n < 0) {
+        return n;
+    }
+
+    INCR(pubsub_metrics, pubsub_event_loop);
+    INCR_N(pubsub_metrics, pubsub_event_total, n);
+    time_update();
+
+    return CC_OK;
+}
+
+void *
+core_pubsub_evloop(void *arg)
+{
+    processor = arg;
+
+    for(;;) {
+        if (_pubsub_evwait() != CC_OK) {
+            log_crit("pubsub core event loop exited due to failure");
+            break;
+        }
+    }
+
+    exit(1);
+}
diff --git a/src/core/data/pubsub.h b/src/core/data/pubsub.h
new file mode 100644
index 000000000..d3ebf9f1d
--- /dev/null
+++ b/src/core/data/pubsub.h
@@ -0,0 +1,53 @@
+#pragma once
+
+#include "core/context.h"
+
+#include <cc_define.h>
+#include <cc_metric.h>
+#include <cc_option.h>
+
+#define PUBSUB_TIMEOUT   100     /* in ms */
+#define PUBSUB_NEVENT    1024
+
+/*          name            type                default         description */
+#define PUBSUB_OPTION(ACTION)                                                                   \
+    ACTION( pubsub_timeout, OPTION_TYPE_UINT,   PUBSUB_TIMEOUT, "evwait timeout"               )\
+    ACTION( pubsub_nevent,  OPTION_TYPE_UINT,   PUBSUB_NEVENT,  "evwait max nevent returned"   )
+
+typedef struct {
+    PUBSUB_OPTION(OPTION_DECLARE)
+} pubsub_options_st;
+
+/*          name                    type            description */
+#define CORE_PUBSUB_METRIC(ACTION)                                                   \
+    ACTION( pubsub_event_total,     METRIC_COUNTER, "# pubsub events returned"      )\
+    ACTION( pubsub_event_loop,      METRIC_COUNTER, "# pubsub event loops returned" )\
+    ACTION( pubsub_event_read,      METRIC_COUNTER, "# pubsub core_read events"     )\
+    ACTION( pubsub_event_write,     METRIC_COUNTER, "# pubsub core_write events"    )\
+    ACTION( pubsub_event_error,     METRIC_COUNTER, "# pubsub core_error events"    )
+
+typedef struct {
+    CORE_PUBSUB_METRIC(METRIC_DECLARE)
+} pubsub_metrics_st;
+
+/*
+ * To allow the use application-specific logic in the handling of read/write
+ * events, each application is expected to implement their own versions of
+ * (post) processing functions called after the channel-level read/write is done.
+ *
+ * Applications should set and pass their instance of processor as argument
+ * to core_pubsub_evloop().
+ */
+struct buf_sock;
+typedef int (*pubsub_fn)(struct buf_sock *);
+struct pubsub_processor {
+    pubsub_fn read;
+    pubsub_fn write;
+    pubsub_fn error;
+};
+
+extern struct context *ctx;
+
+void core_pubsub_setup(pubsub_options_st *options, pubsub_metrics_st *metrics);
+void core_pubsub_teardown(void);
+void *core_pubsub_evloop(void *arg);
diff --git a/src/protocol/data/redis/cmd_pubsub.h b/src/protocol/data/redis/cmd_pubsub.h
new file mode 100644
index 000000000..b7c5600c2
--- /dev/null
+++ b/src/protocol/data/redis/cmd_pubsub.h
@@ -0,0 +1,7 @@
+#pragma once
+
+/*          type                string          # of args */
+#define REQ_PUBSUB(ACTION)                          \
+    ACTION( REQ_PUBLISH,        "publish",      3  )\
+    ACTION( REQ_SUBSCRIBE,      "subscribe",    -2 )\
+    ACTION( REQ_UNSUBSCRIBE,    "unsubscribe",  -2 )
diff --git a/src/protocol/data/redis/request.c b/src/protocol/data/redis/request.c
index 2a38e5c46..2eec939df 100644
--- a/src/protocol/data/redis/request.c
+++ b/src/protocol/data/redis/request.c
@@ -19,6 +19,7 @@ struct command command_table[REQ_SENTINEL] = {
     REQ_HASH(CMD_INIT)
     REQ_ZSET(CMD_INIT)
     REQ_MISC(CMD_INIT)
+    REQ_PUBSUB(CMD_INIT)
 };
 #undef CMD_INIT
 
diff --git a/src/protocol/data/redis/request.h b/src/protocol/data/redis/request.h
index d5fb5f938..4289bd344 100644
--- a/src/protocol/data/redis/request.h
+++ b/src/protocol/data/redis/request.h
@@ -2,6 +2,7 @@
 
 #include "cmd_hash.h"
 #include "cmd_misc.h"
+#include "cmd_pubsub.h"
 #include "cmd_zset.h"
 
 #include <cc_array.h>
@@ -44,6 +45,7 @@ typedef enum cmd_type {
     REQ_HASH(GET_TYPE)
     REQ_ZSET(GET_TYPE)
     REQ_MISC(GET_TYPE)
+    REQ_PUBSUB(GET_TYPE)
     REQ_SENTINEL
 } cmd_type_e;
 #undef GET_TYPE
diff --git a/src/server/CMakeLists.txt b/src/server/CMakeLists.txt
index 9dff0246e..f6043c56e 100644
--- a/src/server/CMakeLists.txt
+++ b/src/server/CMakeLists.txt
@@ -2,6 +2,10 @@ if(TARGET_PINGSERVER)
     add_subdirectory(pingserver)
 endif()
 
+if(TARGET_PUBSUB)
+    add_subdirectory(pubsub)
+endif()
+
 if(TARGET_SLIMCACHE)
     add_subdirectory(slimcache)
 endif()
diff --git a/src/server/pingserver/main.c b/src/server/pingserver/main.c
index 27c578b87..7c0bf9712 100644
--- a/src/server/pingserver/main.c
+++ b/src/server/pingserver/main.c
@@ -93,10 +93,10 @@ setup(void)
     }
 
     /* setup top-level application options */
-    if (option_bool(&setting.pingserver.daemonize)) {
+    if (option_bool(&setting.main.daemonize)) {
         daemonize();
     }
-    fname = option_str(&setting.pingserver.pid_filename);
+    fname = option_str(&setting.main.pid_filename);
     if (fname != NULL) {
         /* to get the correct pid, call create_pidfile after daemonize */
         create_pidfile(fname);
@@ -121,7 +121,7 @@ setup(void)
     core_worker_setup(&setting.worker, &stats.worker);
 
     /* adding recurring events to maintenance/admin thread */
-    intvl = option_uint(&setting.pingserver.dlog_intvl);
+    intvl = option_uint(&setting.main.dlog_intvl);
     if (core_admin_register(intvl, debug_log_flush, NULL) == NULL) {
         log_stderr("Could not register timed event to flush debug log");
         goto error;
@@ -195,7 +195,7 @@ main(int argc, char **argv)
     setup();
     option_print_all((struct option *)&setting, nopt);
 
-    core_run(&worker_processor);
+    worker_run(&worker_processor);
 
     exit(EX_OK);
 }
diff --git a/src/server/pingserver/setting.c b/src/server/pingserver/setting.c
index b699a2f4f..1976105be 100644
--- a/src/server/pingserver/setting.c
+++ b/src/server/pingserver/setting.c
@@ -1,7 +1,7 @@
 #include "setting.h"
 
 struct setting setting = {
-    { PINGSERVER_OPTION(OPTION_INIT)},
+    { MAIN_OPTION(OPTION_INIT)      },
     { ADMIN_OPTION(OPTION_INIT)     },
     { SERVER_OPTION(OPTION_INIT)    },
     { WORKER_OPTION(OPTION_INIT)    },
diff --git a/src/server/pingserver/setting.h b/src/server/pingserver/setting.h
index 035bb7400..743be265c 100644
--- a/src/server/pingserver/setting.h
+++ b/src/server/pingserver/setting.h
@@ -19,18 +19,18 @@
 
 /* option related */
 /*          name            type                default description */
-#define PINGSERVER_OPTION(ACTION)                                                        \
+#define MAIN_OPTION(ACTION)                                                             \
     ACTION( daemonize,      OPTION_TYPE_BOOL,   false,  "daemonize the process"        )\
     ACTION( pid_filename,   OPTION_TYPE_STR,    NULL,   "file storing the pid"         )\
     ACTION( dlog_intvl,     OPTION_TYPE_UINT,   500,    "debug log flush interval(ms)" )
 
 typedef struct {
-    PINGSERVER_OPTION(OPTION_DECLARE)
-} pingserver_options_st;
+    MAIN_OPTION(OPTION_DECLARE)
+} main_options_st;
 
 struct setting {
     /* top-level */
-    pingserver_options_st   pingserver;
+    main_options_st         main;
     /* application modules */
     admin_options_st        admin;
     server_options_st       server;
diff --git a/src/server/pubsub/CMakeLists.txt b/src/server/pubsub/CMakeLists.txt
new file mode 100644
index 000000000..f30ed9f30
--- /dev/null
+++ b/src/server/pubsub/CMakeLists.txt
@@ -0,0 +1,29 @@
+add_subdirectory(admin)
+add_subdirectory(data)
+
+set(SOURCE
+    ${SOURCE}
+    main.c
+    setting.c
+    stats.c)
+
+set(MODULES
+    core
+    protocol_admin
+    protocol_redis
+    pubsub
+    time
+    util)
+
+set(LIBS
+    ccommon-static
+    ${CMAKE_THREAD_LIBS_INIT})
+
+set(CMAKE_RUNTIME_OUTPUT_DIRECTORY ${PROJECT_BINARY_DIR}/_bin)
+set(TARGET_NAME ${PROJECT_NAME}_pubsub)
+
+add_executable(${TARGET_NAME} ${SOURCE})
+target_link_libraries(${TARGET_NAME} ${MODULES} ${LIBS})
+
+install(TARGETS ${TARGET_NAME} RUNTIME DESTINATION bin)
+add_dependencies(service ${TARGET_NAME})
diff --git a/src/server/pubsub/admin/CMakeLists.txt b/src/server/pubsub/admin/CMakeLists.txt
new file mode 100644
index 000000000..31a7e65f3
--- /dev/null
+++ b/src/server/pubsub/admin/CMakeLists.txt
@@ -0,0 +1,4 @@
+set(SOURCE
+    ${SOURCE}
+    ${CMAKE_CURRENT_SOURCE_DIR}/process.c
+    PARENT_SCOPE)
diff --git a/src/server/pubsub/admin/process.c b/src/server/pubsub/admin/process.c
new file mode 100644
index 000000000..43fb64f89
--- /dev/null
+++ b/src/server/pubsub/admin/process.c
@@ -0,0 +1,70 @@
+#include "process.h"
+
+#include "protocol/admin/admin_include.h"
+#include "util/procinfo.h"
+
+#include <cc_mm.h>
+
+#define PUBSUB_ADMIN_MODULE_NAME "pubsub::admin"
+
+extern struct stats stats;
+extern unsigned int nmetric;
+
+static bool admin_init = false;
+static char *buf = NULL;
+static size_t cap;
+
+void
+admin_process_setup(void)
+{
+    log_info("set up the %s module", PUBSUB_ADMIN_MODULE_NAME);
+    if (admin_init) {
+        log_warn("%s has already been setup, overwrite",
+                 PUBSUB_ADMIN_MODULE_NAME);
+    }
+
+    cap = METRIC_PRINT_LEN * nmetric + METRIC_END_LEN;
+    buf = cc_alloc(cap);
+    /* TODO: check return status of cc_alloc */
+
+    admin_init = true;
+}
+
+void
+admin_process_teardown(void)
+{
+    log_info("tear down the %s module", PUBSUB_ADMIN_MODULE_NAME);
+    if (!admin_init) {
+        log_warn("%s has never been setup", PUBSUB_ADMIN_MODULE_NAME);
+    }
+
+    cc_free(buf);
+
+    admin_init = false;
+}
+
+static void
+_admin_stats(struct response *rsp, struct request *req)
+{
+    procinfo_update();
+    rsp->data.data = buf;
+    rsp->data.len = print_stats(buf, cap, (struct metric *)&stats, nmetric);
+}
+
+void
+admin_process_request(struct response *rsp, struct request *req)
+{
+    rsp->type = RSP_GENERIC;
+
+    switch (req->type) {
+    case REQ_STATS:
+        _admin_stats(rsp, req);
+        break;
+    case REQ_VERSION:
+        rsp->data = str2bstr(VERSION_PRINTED);
+        break;
+    default:
+        rsp->type = RSP_INVALID;
+        break;
+    }
+}
diff --git a/src/server/pubsub/admin/process.h b/src/server/pubsub/admin/process.h
new file mode 100644
index 000000000..361230004
--- /dev/null
+++ b/src/server/pubsub/admin/process.h
@@ -0,0 +1,4 @@
+#pragma once
+
+void admin_process_setup(void);
+void admin_process_teardown(void);
diff --git a/src/server/pubsub/data/CMakeLists.txt b/src/server/pubsub/data/CMakeLists.txt
new file mode 100644
index 000000000..31a7e65f3
--- /dev/null
+++ b/src/server/pubsub/data/CMakeLists.txt
@@ -0,0 +1,4 @@
+set(SOURCE
+    ${SOURCE}
+    ${CMAKE_CURRENT_SOURCE_DIR}/process.c
+    PARENT_SCOPE)
diff --git a/src/server/pubsub/data/process.c b/src/server/pubsub/data/process.c
new file mode 100644
index 000000000..5b8b6d1c6
--- /dev/null
+++ b/src/server/pubsub/data/process.c
@@ -0,0 +1,338 @@
+#include "process.h"
+
+#include "protocol/data/redis_include.h"
+#include "storage/pubsub/listener.h"
+#include "storage/pubsub/topic.h"
+
+#include "core/data/pubsub.h"
+
+#include <buffer/cc_dbuf.h>
+#include <cc_array.h>
+#include <cc_event.h>
+#include <stream/cc_sockio.h>
+
+#define PUBSUB_PROCESS_MODULE_NAME "pubsub::process"
+
+#define MESSAGE "message"
+#define SUBSCRIBE "subscribe"
+#define UNSUBSCRIBE "unsubscribe"
+
+static struct element el_message =
+    {.type = ELEM_BULK, .bstr = {sizeof(MESSAGE) - 1, MESSAGE}};
+static struct element el_subscribe =
+    {.type = ELEM_BULK, .bstr = {sizeof(SUBSCRIBE) - 1, SUBSCRIBE}};
+static struct element el_unsubscribe =
+    {.type = ELEM_BULK, .bstr = {sizeof(UNSUBSCRIBE) - 1, UNSUBSCRIBE}};
+
+
+typedef void (* command_fn)(struct response *, struct request *, struct buf_sock *);
+command_fn command_registry[REQ_SENTINEL];
+
+static bool process_init = false;
+static process_metrics_st *process_metrics = NULL;
+
+
+/* "publish topic msg"
+ * reply: # listeners subscribed
+ */
+static void
+command_publish(struct response *rsp, struct request *req, struct buf_sock *s)
+{
+    struct element *el_t, *el_m; /* topic & message */
+    struct listener *l;
+    struct topic *t;
+    uint32_t nsub = 0;
+
+    el_t = array_get(req->token, 1);
+    if (el_t->type != ELEM_BULK) {
+        /* handle error */
+    };
+    el_m = array_get(req->token, 2);
+    if (el_m->type != ELEM_BULK) {
+        /* handle error */
+    };
+
+    log_verb("publish from buf_sock %p", s);
+
+    l = listener_get(s);
+    if (l != NULL) {
+        log_error("found listener at %p: subscriber cannot publish", l);
+        /* handle error */
+    }
+
+    t = topic_get(&el_t->bstr);
+    if (t == NULL) {
+        log_verb("no listener on topic %.*s, ignore", el_t->bstr.len,
+                el_t->bstr.data);
+    } else { /* copy message to listener's buffer, this should be optimized later */
+        struct index_node *node;
+
+        nsub = t->nsub;
+        TAILQ_FOREACH(node, t->idx, i_tqe) {
+            l = (struct listener *)node->obj;
+            struct buf_sock *ss = l->s;
+            compose_array_header(&ss->wbuf, 3);
+            compose_element(&ss->wbuf, &el_message);
+            compose_element(&ss->wbuf, el_t);
+            compose_element(&ss->wbuf, el_m);
+            /* register listener's fd for write event */
+            event_add_write(ctx->evb, ss->hdl->wid(ss->ch), ss);
+        }
+    }
+
+    compose_element(&s->wbuf, &(struct element){.type = ELEM_INT, .num = nsub});
+}
+
+/* "subscribe topic [topic ...]"
+ * no reply
+ */
+static void
+command_subscribe(struct response *rsp, struct request *req, struct buf_sock *s)
+{
+    struct element *el;
+    struct element el_s = {.type = ELEM_INT}; /* # topics subscribed, an integer */
+    struct listener *l;
+    struct topic *t;
+    uint32_t ntopic = req->token->nelem - 1;
+
+    log_verb("subscribe buf_sock %p from topics", s);
+
+    l = listener_get(s);
+    if (l == NULL) {
+        l = listener_add(s);
+    }
+
+    for (int i = 1; i <= ntopic; i++) {
+        el = array_get(req->token, i);
+        if (el->type != ELEM_BULK) {
+            /* handle error */
+        };
+
+        t = topic_get(&el->bstr);
+        if (t == NULL) {
+            t = topic_add(&el->bstr);
+            /* handle error */
+        }
+
+        log_verb("subscribing to topic %.*s", el->bstr.len, el->bstr.data);
+        if (!topic_add_listener(t, l)) {
+            log_debug("listener not added");
+        }
+        if (!listener_add_topic(l, t)) {
+            log_debug("topic not added");
+        }
+
+        el_s.num = l->ntopic;
+        compose_array_header(&s->wbuf, 3);
+        compose_element(&s->wbuf, &el_subscribe);
+        compose_element(&s->wbuf, el);
+        compose_element(&s->wbuf, &el_s);
+    }
+}
+
+/* "unsubscribe topic [topic ...]"
+ * no reply
+ * (not supporting unsubscribe from all right now)
+ */
+static void
+command_unsubscribe(struct response *rsp, struct request *req, struct buf_sock *s)
+{
+    struct element *el;
+    struct element el_s = {.type = ELEM_INT}; /* # topics subscribed, an integer */
+    struct listener *l;
+    struct topic *t;
+    uint32_t ntopic = req->token->nelem - 1;
+
+    log_verb("unsubscribe buf_sock %p from topics", s);
+
+    l = listener_get(s);
+    if (l == NULL) {
+        log_info("listener not found for %p", s);
+        return;
+    }
+
+    for (int i = 1; i <= ntopic; i++) {
+        el = array_get(req->token, i);
+        if (el->type != ELEM_BULK) {
+            /* handle error */
+        };
+
+        t = topic_get(&el->bstr);
+        if (t == NULL) {
+            log_debug("topic %.*s does not exist", el->bstr.len, el->bstr.data);
+            continue;
+        }
+
+        log_verb("unsubscribing from topic %.*s", el->bstr.len, el->bstr.data);
+        listener_del_topic(l, t);
+        topic_del_listener(t, l);
+
+        if (t->nsub == 0) { /* remove topic that nobody listens to */
+            topic_delete(&t->name);
+        }
+
+        el_s.num = l->ntopic;
+        compose_array_header(&s->wbuf, 3);
+        compose_element(&s->wbuf, &el_unsubscribe);
+        compose_element(&s->wbuf, el);
+        compose_element(&s->wbuf, &el_s);
+    }
+}
+
+void
+process_setup(process_metrics_st *metrics)
+{
+    log_info("set up the %s module", PUBSUB_PROCESS_MODULE_NAME);
+
+    if (process_init) {
+        log_warn("%s has already been setup, overwrite",
+                 PUBSUB_PROCESS_MODULE_NAME);
+    }
+
+    listener_setup(16);
+    topic_setup(16);
+
+    command_registry[REQ_PUBLISH] = command_publish;
+    command_registry[REQ_SUBSCRIBE] = command_subscribe;
+    command_registry[REQ_UNSUBSCRIBE] = command_unsubscribe;
+
+    process_metrics = metrics;
+    process_init = true;
+}
+
+void
+process_teardown(void)
+{
+    log_info("tear down the %s module", PUBSUB_PROCESS_MODULE_NAME);
+    if (!process_init) {
+        log_warn("%s has never been setup", PUBSUB_PROCESS_MODULE_NAME);
+    }
+
+    listener_teardown();
+    topic_teardown();
+
+    command_registry[REQ_PUBLISH] = NULL;
+    command_registry[REQ_SUBSCRIBE] = NULL;
+    command_registry[REQ_UNSUBSCRIBE] = NULL;
+
+    process_metrics = NULL;
+    process_init = false;
+}
+
+static void
+process_request_sock(struct response *rsp, struct request *req, struct buf_sock *s)
+{
+    log_verb("processing req %p, write rsp to %p", req, rsp);
+    INCR(process_metrics, process_req);
+
+    if (command_registry[req->type] == NULL) {
+        /* return error */
+    }
+
+    log_verb("processing request type %d", req->type);
+    command_registry[req->type](rsp, req, s);
+}
+
+int
+pubsub_process_read(struct buf_sock *s)
+{
+    int status;
+    struct request *req;
+    struct response *rsp;
+
+    log_verb("post-read processing");
+
+    req = request_borrow();
+    rsp = response_borrow();
+
+    /* keep parse-process-compose until running out of data in rbuf */
+    while (buf_rsize(s->rbuf) > 0) {
+        /* stage 1: parsing */
+        log_verb("%"PRIu32" bytes left", buf_rsize(s->rbuf));
+
+        status = parse_req(req, s->rbuf);
+        if (status == PARSE_EUNFIN) {
+            buf_lshift(s->rbuf);
+            goto done;
+        }
+        if (status != PARSE_OK) {
+            /* parsing errors are all client errors, since we don't know
+             * how to recover from client errors in this condition (we do not
+             * have a valid request so we don't know where the invalid request
+             * ends), we should close the connection
+             */
+            log_warn("illegal request received, status: %d", status);
+            INCR(process_metrics, process_ex);
+            INCR(process_metrics, process_client_ex);
+            goto error;
+        }
+
+        /* stage 2: processing- check for quit, allocate response(s), process */
+
+        /* quit is special, no processing/resposne expected */
+        if (req->type == REQ_QUIT) {
+            log_info("peer called quit");
+            goto error;
+        }
+
+        /* actual processing */
+        process_request_sock(rsp, req, s);
+
+        /* stage 3: write response(s) if necessary */
+        status = compose_rsp(&s->wbuf, rsp);
+        if (status != PARSE_OK) {
+            log_error("composing rsp erred");
+            INCR(process_metrics, process_ex);
+            INCR(process_metrics, process_server_ex);
+            goto error;
+        }
+
+        /* noreply means no need to write to buffers */
+
+        /* logging, clean-up */
+    }
+
+done:
+    request_return(&req);
+    response_return(&rsp);
+
+    return 0;
+
+error:
+    request_return(&req);
+    response_return(&rsp);
+
+    return -1;
+}
+
+
+int
+pubsub_process_write(struct buf_sock *s)
+{
+    log_verb("post-write processing");
+
+    buf_lshift(s->rbuf);
+    buf_lshift(s->wbuf);
+    dbuf_shrink(&s->rbuf);
+    dbuf_shrink(&s->wbuf);
+
+    return 0;
+}
+
+
+int
+pubsub_process_error(struct buf_sock *s)
+{
+    log_verb("post-error processing");
+
+    listener_delete(s);
+    /* TODO: unsubscribe automatically from all */
+
+    /* normalize buffer size */
+    buf_reset(s->rbuf);
+    dbuf_shrink(&s->rbuf);
+    buf_reset(s->wbuf);
+    dbuf_shrink(&s->wbuf);
+
+    return 0;
+}
diff --git a/src/server/pubsub/data/process.h b/src/server/pubsub/data/process.h
new file mode 100644
index 000000000..63f06cc52
--- /dev/null
+++ b/src/server/pubsub/data/process.h
@@ -0,0 +1,31 @@
+#pragma once
+
+#include "protocol/data/redis_include.h"
+
+#include <buffer/cc_buf.h>
+#include <cc_metric.h>
+#include <cc_option.h>
+#include <stream/cc_sockio.h>
+
+/*          name                        type            description */
+#define PROCESS_METRIC(ACTION)                                          \
+    ACTION( process_req,       METRIC_COUNTER, "# requests processed"  )\
+    ACTION( process_ex,        METRIC_COUNTER, "# processing error"    )\
+    ACTION( process_client_ex, METRIC_COUNTER, "# client error"        )\
+    ACTION( process_server_ex, METRIC_COUNTER, "# server error"        )\
+    ACTION( publish,           METRIC_COUNTER, "# publish requests"    )\
+    ACTION( subscribe,         METRIC_COUNTER, "# subscribe requests"  )\
+    ACTION( unsubscribe,       METRIC_COUNTER, "# unsubscribe requests")
+
+typedef struct {
+    PROCESS_METRIC(METRIC_DECLARE)
+} process_metrics_st;
+
+void process_setup(process_metrics_st *metrics);
+void process_teardown(void);
+
+int pubsub_process_read(struct buf_sock *s);
+int pubsub_process_write(struct buf_sock *s);
+int pubsub_process_error(struct buf_sock *s);
+
+
diff --git a/src/server/pubsub/main.c b/src/server/pubsub/main.c
new file mode 100644
index 000000000..5d8e27d5d
--- /dev/null
+++ b/src/server/pubsub/main.c
@@ -0,0 +1,200 @@
+#include "setting.h"
+#include "stats.h"
+
+#include "time/time.h"
+#include "util/util.h"
+
+#include <cc_debug.h>
+
+#include <errno.h>
+#include <fcntl.h>
+#include <stdio.h>
+#include <string.h>
+#include <sys/socket.h>
+#include <sysexits.h>
+
+struct pubsub_processor pubsub_processor = {
+    pubsub_process_read,
+    pubsub_process_write,
+    pubsub_process_error,
+};
+
+static void
+show_usage(void)
+{
+    log_stdout(
+            "Usage:" CRLF
+            "  pelikan_pubsub [option|config]" CRLF
+            );
+    log_stdout(
+            "Description:" CRLF
+            "  pelikan_pubsub is an in-memory pub/sub server." CRLF
+            CRLF
+            "  It supports basic Redis pub/sub commands:" CRLF
+            "    subscribe, unsubscribe, publish,... " CRLF
+            );
+    log_stdout(
+            "Command-line options:" CRLF
+            "  -h, --help        show this message" CRLF
+            "  -v, --version     show version number" CRLF
+            "  -c, --config      list & describe all options in config" CRLF
+            "  -s, --stats       list & describe all metrics in stats" CRLF
+            );
+    log_stdout(
+            "Example:" CRLF
+            "  pelikan_pubsub pubsub.conf" CRLF CRLF
+            "Sample config files can be found under the config dir." CRLF
+            );
+}
+
+static void
+teardown(void)
+{
+    core_pubsub_teardown();
+    core_server_teardown();
+    core_admin_teardown();
+    admin_process_teardown();
+    compose_teardown();
+    parse_teardown();
+    procinfo_teardown();
+    time_teardown();
+
+    timing_wheel_teardown();
+    tcp_teardown();
+    sockio_teardown();
+    event_teardown();
+    dbuf_teardown();
+    buf_teardown();
+
+    debug_teardown();
+    log_teardown();
+}
+
+static void
+setup(void)
+{
+    char *fname = NULL;
+    uint64_t intvl;
+
+    if (atexit(teardown) != 0) {
+        log_stderr("cannot register teardown procedure with atexit()");
+        exit(EX_OSERR); /* only failure comes from NOMEM */
+    }
+
+    /* Setup logging first */
+    log_setup(&stats.log);
+    if (debug_setup(&setting.debug) < 0) {
+        log_stderr("debug log setup failed");
+        goto error;
+    }
+
+    /* setup top-level application options */
+    if (option_bool(&setting.main.daemonize)) {
+        daemonize();
+    }
+    fname = option_str(&setting.main.pid_filename);
+    if (fname != NULL) {
+        /* to get the correct pid, call create_pidfile after daemonize */
+        create_pidfile(fname);
+    }
+
+    /* setup library modules */
+    buf_setup(&setting.buf, &stats.buf);
+    dbuf_setup(&setting.dbuf, &stats.dbuf);
+    event_setup(&stats.event);
+    sockio_setup(&setting.sockio, &stats.sockio);
+    tcp_setup(&setting.tcp, &stats.tcp);
+    timing_wheel_setup(&stats.timing_wheel);
+
+    /* setup pelikan modules */
+    time_setup();
+    procinfo_setup(&stats.procinfo);
+    request_setup(&setting.request, &stats.request);
+    response_setup(&setting.response, &stats.response);
+    parse_setup(&stats.parse_req, NULL);
+    compose_setup(NULL, &stats.compose_rsp);
+    process_setup(&stats.process);
+    admin_process_setup();
+    core_admin_setup(&setting.admin);
+    core_server_setup(&setting.server, &stats.server);
+    core_pubsub_setup(&setting.pubsub, &stats.pubsub);
+
+    /* adding recurring events to maintenance/admin thread */
+    intvl = option_uint(&setting.main.dlog_intvl);
+    if (core_admin_register(intvl, debug_log_flush, NULL) == NULL) {
+        log_stderr("Could not register timed event to flush debug log");
+        goto error;
+    }
+
+    return;
+
+error:
+    if (fname != NULL) {
+        remove_pidfile(fname);
+    }
+
+    /* since we registered teardown with atexit, it'll be called upon exit */
+    exit(EX_CONFIG);
+}
+
+int
+main(int argc, char **argv)
+{
+    rstatus_i status = CC_OK;
+    FILE *fp = NULL;
+
+    if (argc > 2) {
+        show_usage();
+        exit(EX_USAGE);
+    }
+
+    if (argc == 1) {
+        log_stderr("launching server with default values.");
+    }
+
+    if (argc == 2) {
+        if (strcmp(argv[1], "-h") == 0 || strcmp(argv[1], "--help") == 0) {
+            show_usage();
+            exit(EX_OK);
+        }
+        if (strcmp(argv[1], "-v") == 0 || strcmp(argv[1], "--version") == 0) {
+            show_version();
+            exit(EX_OK);
+        }
+        if (strcmp(argv[1], "-c") == 0 || strcmp(argv[1], "--config") == 0) {
+            option_describe_all((struct option *)&setting, nopt);
+            exit(EX_OK);
+        }
+        if (strcmp(argv[1], "-s") == 0 || strcmp(argv[1], "--stats") == 0) {
+            metric_describe_all((struct metric *)&stats, nmetric);
+            exit(EX_OK);
+        }
+        fp = fopen(argv[1], "r");
+        if (fp == NULL) {
+            log_stderr("cannot open config: incorrect path or doesn't exist");
+            exit(EX_DATAERR);
+        }
+    }
+
+    if (option_load_default((struct option *)&setting, nopt) != CC_OK) {
+        log_stderr("failed to load default option values");
+        exit(EX_CONFIG);
+    }
+
+    if (fp != NULL) {
+        log_stderr("load config from %s", argv[1]);
+        status = option_load_file(fp, (struct option *)&setting, nopt);
+        fclose(fp);
+    }
+    if (status != CC_OK) {
+        log_stderr("failed to load config");
+        exit(EX_DATAERR);
+    }
+
+    setup();
+    option_print_all((struct option *)&setting, nopt);
+
+    pubsub_run(&pubsub_processor);
+
+    exit(EX_OK);
+}
diff --git a/src/server/pubsub/setting.c b/src/server/pubsub/setting.c
new file mode 100644
index 000000000..29489a43a
--- /dev/null
+++ b/src/server/pubsub/setting.c
@@ -0,0 +1,18 @@
+#include "setting.h"
+
+struct setting setting = {
+    { MAIN_OPTION(OPTION_INIT)      },
+    { ADMIN_OPTION(OPTION_INIT)     },
+    { SERVER_OPTION(OPTION_INIT)    },
+    { PUBSUB_OPTION(OPTION_INIT)    },
+    { REQUEST_OPTION(OPTION_INIT)   },
+    { RESPONSE_OPTION(OPTION_INIT)  },
+    { ARRAY_OPTION(OPTION_INIT)     },
+    { BUF_OPTION(OPTION_INIT)       },
+    { DBUF_OPTION(OPTION_INIT)      },
+    { DEBUG_OPTION(OPTION_INIT)     },
+    { SOCKIO_OPTION(OPTION_INIT)    },
+    { TCP_OPTION(OPTION_INIT)       },
+};
+
+unsigned int nopt = OPTION_CARDINALITY(setting);
diff --git a/src/server/pubsub/setting.h b/src/server/pubsub/setting.h
new file mode 100644
index 000000000..23d23d976
--- /dev/null
+++ b/src/server/pubsub/setting.h
@@ -0,0 +1,50 @@
+#pragma once
+
+#include "admin/process.h"
+#include "data/process.h"
+
+#include "core/core.h"
+#include "protocol/data/redis_include.h"
+
+#include <buffer/cc_buf.h>
+#include <buffer/cc_dbuf.h>
+#include <cc_array.h>
+#include <cc_debug.h>
+#include <cc_metric.h>
+#include <cc_option.h>
+#include <cc_ring_array.h>
+#include <cc_signal.h>
+#include <channel/cc_tcp.h>
+#include <stream/cc_sockio.h>
+
+/* option related */
+/*          name            type                default description */
+#define MAIN_OPTION(ACTION)                                                        \
+    ACTION( daemonize,      OPTION_TYPE_BOOL,   false,  "daemonize the process"        )\
+    ACTION( pid_filename,   OPTION_TYPE_STR,    NULL,   "file storing the pid"         )\
+    ACTION( dlog_intvl,     OPTION_TYPE_UINT,   500,    "debug log flush interval(ms)" )
+
+typedef struct {
+    MAIN_OPTION(OPTION_DECLARE)
+} main_options_st;
+
+struct setting {
+    /* top-level */
+    main_options_st         main;
+    /* application modules */
+    admin_options_st        admin;
+    server_options_st       server;
+    pubsub_options_st       pubsub;
+    request_options_st      request;
+    response_options_st     response;
+    /* ccommon libraries */
+    array_options_st        array;
+    buf_options_st          buf;
+    dbuf_options_st         dbuf;
+    debug_options_st        debug;
+    sockio_options_st       sockio;
+    tcp_options_st          tcp;
+};
+
+extern struct setting setting;
+extern unsigned int nopt;
diff --git a/src/server/pubsub/stats.c b/src/server/pubsub/stats.c
new file mode 100644
index 000000000..b58d972c8
--- /dev/null
+++ b/src/server/pubsub/stats.c
@@ -0,0 +1,21 @@
+#include "stats.h"
+
+struct stats stats = {
+    { PROCINFO_METRIC(METRIC_INIT)      },
+    { PROCESS_METRIC(METRIC_INIT)       },
+    { PARSE_REQ_METRIC(METRIC_INIT)     },
+    { COMPOSE_RSP_METRIC(METRIC_INIT)   },
+    { REQUEST_METRIC(METRIC_INIT)       },
+    { RESPONSE_METRIC(METRIC_INIT)      },
+    { CORE_SERVER_METRIC(METRIC_INIT)   },
+    { CORE_PUBSUB_METRIC(METRIC_INIT)   },
+    { BUF_METRIC(METRIC_INIT)           },
+    { DBUF_METRIC(METRIC_INIT)          },
+    { EVENT_METRIC(METRIC_INIT)         },
+    { LOG_METRIC(METRIC_INIT)           },
+    { SOCKIO_METRIC(METRIC_INIT)        },
+    { TCP_METRIC(METRIC_INIT)           },
+    { TIMING_WHEEL_METRIC(METRIC_INIT)  },
+};
+
+unsigned int nmetric = METRIC_CARDINALITY(stats);
diff --git a/src/server/pubsub/stats.h b/src/server/pubsub/stats.h
new file mode 100644
index 000000000..384b17bf5
--- /dev/null
+++ b/src/server/pubsub/stats.h
@@ -0,0 +1,38 @@
+#pragma once
+
+#include "data/process.h"
+
+#include "core/core.h"
+#include "protocol/data/redis_include.h"
+#include "util/procinfo.h"
+
+#include <buffer/cc_dbuf.h>
+#include <cc_event.h>
+#include <cc_log.h>
+#include <channel/cc_tcp.h>
+#include <stream/cc_sockio.h>
+#include <time/cc_wheel.h>
+
+struct stats {
+    /* perf info */
+    procinfo_metrics_st         procinfo;
+    /* application modules */
+    process_metrics_st          process;
+    parse_req_metrics_st        parse_req;
+    compose_rsp_metrics_st      compose_rsp;
+    request_metrics_st          request;
+    response_metrics_st         response;
+    server_metrics_st           server;
+    pubsub_metrics_st           pubsub;
+    /* ccommon libraries */
+    buf_metrics_st              buf;
+    dbuf_metrics_st             dbuf;
+    event_metrics_st            event;
+    log_metrics_st              log;
+    sockio_metrics_st           sockio;
+    tcp_metrics_st              tcp;
+    timing_wheel_metrics_st     timing_wheel;
+};
+
+extern struct stats stats;
+extern unsigned int nmetric;
diff --git a/src/server/slimcache/main.c b/src/server/slimcache/main.c
index 1d48de5b2..3a00232e4 100644
--- a/src/server/slimcache/main.c
+++ b/src/server/slimcache/main.c
@@ -98,10 +98,10 @@ setup(void)
     }
 
     /* setup top-level application options */
-    if (option_bool(&setting.slimcache.daemonize)) {
+    if (option_bool(&setting.main.daemonize)) {
         daemonize();
     }
-    fname = option_str(&setting.slimcache.pid_filename);
+    fname = option_str(&setting.main.pid_filename);
     if (fname != NULL) {
         /* to get the correct pid, call create_pidfile after daemonize */
         create_pidfile(fname);
@@ -131,13 +131,13 @@ setup(void)
     core_worker_setup(&setting.worker, &stats.worker);
 
     /* adding recurring events to maintenance/admin thread */
-    intvl = option_uint(&setting.slimcache.dlog_intvl);
+    intvl = option_uint(&setting.main.dlog_intvl);
     if (core_admin_register(intvl, debug_log_flush, NULL) == NULL) {
         log_stderr("Could not register timed event to flush debug log");
         goto error;
     }
 
-    intvl = option_uint(&setting.slimcache.klog_intvl);
+    intvl = option_uint(&setting.main.klog_intvl);
     if (core_admin_register(intvl, klog_flush, NULL) == NULL) {
         log_error("Could not register timed event to flush command log");
         goto error;
@@ -211,7 +211,7 @@ main(int argc, char **argv)
     setup();
     option_print_all((struct option *)&setting, nopt);
 
-    core_run(&worker_processor);
+    worker_run(&worker_processor);
 
     exit(EX_OK);
 }
diff --git a/src/server/slimcache/setting.c b/src/server/slimcache/setting.c
index e978d4667..09c74fa73 100644
--- a/src/server/slimcache/setting.c
+++ b/src/server/slimcache/setting.c
@@ -1,7 +1,7 @@
 #include "setting.h"
 
 struct setting setting = {
-    { SLIMCACHE_OPTION(OPTION_INIT) },
+    { MAIN_OPTION(OPTION_INIT)      },
     { ADMIN_OPTION(OPTION_INIT)     },
     { SERVER_OPTION(OPTION_INIT)    },
     { WORKER_OPTION(OPTION_INIT)    },
diff --git a/src/server/slimcache/setting.h b/src/server/slimcache/setting.h
index 5caf835fe..5f2495972 100644
--- a/src/server/slimcache/setting.h
+++ b/src/server/slimcache/setting.h
@@ -19,19 +19,19 @@
 
 /* option related */
 /*          name            type                default description */
-#define SLIMCACHE_OPTION(ACTION)                                                        \
+#define MAIN_OPTION(ACTION)                                                             \
     ACTION( daemonize,      OPTION_TYPE_BOOL,   false,  "daemonize the process"        )\
     ACTION( pid_filename,   OPTION_TYPE_STR,    NULL,   "file storing the pid"         )\
     ACTION( dlog_intvl,     OPTION_TYPE_UINT,   500,    "debug log flush interval(ms)" )\
     ACTION( klog_intvl,     OPTION_TYPE_UINT,   100,    "cmd log flush interval(ms)"   )
 
 typedef struct {
-    SLIMCACHE_OPTION(OPTION_DECLARE)
-} slimcache_options_st;
+    MAIN_OPTION(OPTION_DECLARE)
+} main_options_st;
 
 struct setting {
     /* top-level */
-    slimcache_options_st    slimcache;
+    main_options_st         main;
     /* application modules */
     admin_options_st        admin;
     server_options_st       server;
diff --git a/src/server/twemcache/main.c b/src/server/twemcache/main.c
index b3b8a88f5..4ceb9056d 100644
--- a/src/server/twemcache/main.c
+++ b/src/server/twemcache/main.c
@@ -94,10 +94,10 @@ setup(void)
     }
 
     /* setup top-level application options */
-    if (option_bool(&setting.twemcache.daemonize)) {
+    if (option_bool(&setting.main.daemonize)) {
         daemonize();
     }
-    fname = option_str(&setting.twemcache.pid_filename);
+    fname = option_str(&setting.main.pid_filename);
     if (fname != NULL) {
         /* to get the correct pid, call create_pidfile after daemonize */
         create_pidfile(fname);
@@ -127,13 +127,13 @@ setup(void)
     core_worker_setup(&setting.worker, &stats.worker);
 
     /* adding recurring events to maintenance/admin thread */
-    intvl = option_uint(&setting.twemcache.dlog_intvl);
+    intvl = option_uint(&setting.main.dlog_intvl);
     if (core_admin_register(intvl, debug_log_flush, NULL) == NULL) {
         log_stderr("Could not register timed event to flush debug log");
         goto error;
     }
 
-    intvl = option_uint(&setting.twemcache.klog_intvl);
+    intvl = option_uint(&setting.main.klog_intvl);
     if (core_admin_register(intvl, klog_flush, NULL) == NULL) {
         log_error("Could not register timed event to flush command log");
         goto error;
@@ -206,7 +206,7 @@ main(int argc, char **argv)
     setup();
     option_print_all((struct option *)&setting, nopt);
 
-    core_run(&worker_processor);
+    worker_run(&worker_processor);
 
     exit(EX_OK);
 }
diff --git a/src/server/twemcache/setting.c b/src/server/twemcache/setting.c
index ceb5b0847..e05ca0838 100644
--- a/src/server/twemcache/setting.c
+++ b/src/server/twemcache/setting.c
@@ -1,7 +1,7 @@
 #include "setting.h"
 
 struct setting setting = {
-    { TWEMCACHE_OPTION(OPTION_INIT) },
+    { MAIN_OPTION(OPTION_INIT)      },
     { ADMIN_OPTION(OPTION_INIT)     },
     { SERVER_OPTION(OPTION_INIT)    },
     { WORKER_OPTION(OPTION_INIT)    },
diff --git a/src/server/twemcache/setting.h b/src/server/twemcache/setting.h
index aea199ddf..c405d0a80 100644
--- a/src/server/twemcache/setting.h
+++ b/src/server/twemcache/setting.h
@@ -17,19 +17,19 @@
 
 /* option related */
 /*          name            type                default description */
-#define TWEMCACHE_OPTION(ACTION)                                                        \
+#define MAIN_OPTION(ACTION)                                                             \
     ACTION( daemonize,      OPTION_TYPE_BOOL,   false,  "daemonize the process"        )\
     ACTION( pid_filename,   OPTION_TYPE_STR,    NULL,   "file storing the pid"         )\
     ACTION( dlog_intvl,     OPTION_TYPE_UINT,   500,    "debug log flush interval(ms)" )\
     ACTION( klog_intvl,     OPTION_TYPE_UINT,   100,    "cmd log flush interval(ms)"   )
 
 typedef struct {
-    TWEMCACHE_OPTION(OPTION_DECLARE)
-} twemcache_options_st;
+    MAIN_OPTION(OPTION_DECLARE)
+} main_options_st;
 
 struct setting {
     /* top-level */
-    twemcache_options_st    twemcache;
+    main_options_st         main;
     /* application modules */
     admin_options_st        admin;
     server_options_st       server;
diff --git a/src/storage/CMakeLists.txt b/src/storage/CMakeLists.txt
index c06eede35..0756902af 100644
--- a/src/storage/CMakeLists.txt
+++ b/src/storage/CMakeLists.txt
@@ -1,2 +1,3 @@
 add_subdirectory(cuckoo)
+add_subdirectory(pubsub)
 add_subdirectory(slab)
diff --git a/src/storage/pubsub/CMakeLists.txt b/src/storage/pubsub/CMakeLists.txt
new file mode 100644
index 000000000..4578d1360
--- /dev/null
+++ b/src/storage/pubsub/CMakeLists.txt
@@ -0,0 +1,5 @@
+set(SOURCE
+    listener.c
+    topic.c)
+
+add_library(pubsub ${SOURCE})
diff --git a/src/storage/pubsub/README b/src/storage/pubsub/README
new file mode 100644
index 000000000..7def09c6a
--- /dev/null
+++ b/src/storage/pubsub/README
@@ -0,0 +1,5 @@
+Mapping #1: client -> channel, to clean-up when client quits
+Mapping #2: channel -> client, to handle subscription, fanout
+
+a channel with no subscriber should be deleted.
+a client with no channel subscription is kept.
diff --git a/src/storage/pubsub/hashtable.h b/src/storage/pubsub/hashtable.h
new file mode 100644
index 000000000..8de665265
--- /dev/null
+++ b/src/storage/pubsub/hashtable.h
@@ -0,0 +1,4 @@
+#pragma once
+
+#define HASHSIZE(_n) (1ULL << (_n))
+#define HASHMASK(_n) (HASHSIZE(_n) - 1)
diff --git a/src/storage/pubsub/index.h b/src/storage/pubsub/index.h
new file mode 100644
index 000000000..f9c8bc71a
--- /dev/null
+++ b/src/storage/pubsub/index.h
@@ -0,0 +1,10 @@
+#pragma once
+
+#include <cc_queue.h>
+
+struct index_node {
+    TAILQ_ENTRY(index_node) i_tqe;
+    void                    *obj;
+};
+
+TAILQ_HEAD(index_tqh, index_node);
diff --git a/src/storage/pubsub/listener.c b/src/storage/pubsub/listener.c
new file mode 100644
index 000000000..06fdadfc4
--- /dev/null
+++ b/src/storage/pubsub/listener.c
@@ -0,0 +1,231 @@
+#include "hashtable.h"
+#include "listener.h"
+
+#include <cc_mm.h>
+#include <cc_queue.h>
+#include <hash/cc_lookup3.h>
+
+#include <sysexits.h>
+
+SLIST_HEAD(listener_slh, listener);
+
+struct listener_ht {
+    struct listener_slh *table;
+    uint32_t nlistener;
+    uint32_t hash_power;
+};
+
+static struct listener_ht hashtable;
+static struct listener_ht *ht = &hashtable;
+
+
+static struct listener_slh *
+_get_bucket(const struct buf_sock *s)
+{
+    /* use the _address_ of the channel to hash */
+    uint32_t hval = hash_lookup3((char *)s, sizeof(struct buf_sock *), 0);
+    return &(ht->table[hval & HASHMASK(ht->hash_power)]);
+}
+
+static void
+_listener_reset(struct listener *l)
+{
+    l->s = NULL;
+    l->ntopic = 0;
+    TAILQ_INIT(l->idx);
+}
+
+static struct listener *
+_listener_create(const struct buf_sock *s)
+{
+    struct listener *l;
+
+    log_verb("create new listener for buf_sock %p", s);
+
+    l = cc_alloc(sizeof(struct listener));
+    if (l == NULL) {
+        return NULL;
+    }
+
+    l->idx = cc_alloc(sizeof(struct index_tqh));
+    if (l->idx == NULL) {
+        cc_free(l);
+        return NULL;
+    }
+
+    _listener_reset(l);
+    l->s = (struct buf_sock *)s;
+
+    return l;
+}
+
+static void
+_listener_destroy(struct listener **l)
+{
+    ASSERT(l != NULL && *l != NULL);
+
+    log_verb("destroy listener %p", *l);
+
+    struct index_node *curr, *next;
+    struct index_tqh *idx = (*l)->idx;
+
+    /* delete all elements of the index */
+    TAILQ_FOREACH_SAFE(curr, idx, i_tqe, next) {
+        TAILQ_REMOVE(idx, curr, i_tqe);
+        cc_free(curr);
+    }
+    cc_free(idx);
+
+    cc_free(*l);
+    *l = NULL;
+}
+
+
+struct listener *
+listener_get(const struct buf_sock *s)
+{
+    struct listener_slh *bucket;
+    struct listener *l;
+
+    bucket = _get_bucket(s);
+    for (l = SLIST_FIRST(bucket); l != NULL; l = SLIST_NEXT(l, l_sle)) {
+        if (l->s == s) {
+            return l;
+        }
+    }
+
+    log_verb("listener not found for buf_sock", s);
+    return NULL;
+}
+
+struct listener *
+listener_add(const struct buf_sock *s)
+{
+    ASSERT(listener_get(s) == NULL);
+
+    struct listener_slh *bucket;
+    struct listener *l = _listener_create(s);
+
+    log_verb("add listener %p for buf_sock %p", l, s);
+
+    bucket = _get_bucket(s);
+    SLIST_INSERT_HEAD(bucket, l, l_sle);
+
+    ht->nlistener++;
+    log_verb("total listeners: %"PRIu32, ht->nlistener);
+
+    return l;
+}
+
+void
+listener_delete(const struct buf_sock *s)
+{
+    struct listener_slh *bucket;
+    struct listener *l, *prev;
+
+    bucket = _get_bucket(s);
+    for (prev = NULL, l = SLIST_FIRST(bucket); l != NULL;
+        prev = l, l = SLIST_NEXT(l, l_sle)) {
+        if (l == NULL) {
+            log_debug("listener not found for buf_sock %p", s);
+            return;
+        }
+        if (l->s == s) {
+            break;
+        }
+    }
+
+    if (prev == NULL) {
+        SLIST_REMOVE_HEAD(bucket, l_sle);
+    } else {
+        SLIST_REMOVE_AFTER(prev, l_sle);
+    }
+
+    _listener_destroy(&l);
+    --(ht->nlistener);
+    log_verb("total listeners: %"PRIu32, ht->nlistener);
+}
+
+bool
+listener_add_topic(struct listener *l, const struct topic *t)
+{
+    struct index_node *node;
+
+    ASSERT(l != NULL && t != NULL);
+
+    /* do nothing if already subscribed */
+    TAILQ_FOREACH(node, l->idx, i_tqe) {
+        if (node->obj == t) {
+            log_debug("listener %p already subscribed to topic %p", l, t);
+            return false;
+        }
+    }
+
+    node = cc_alloc(sizeof(struct index_node));
+    if (node == NULL) {
+        log_error("cannot add topic: out of memory");
+        return false;
+    }
+    node->obj = (struct topic *)t;
+
+    TAILQ_INSERT_TAIL(l->idx, node, i_tqe);
+    l->ntopic++;
+    log_verb("listener %p subscribed to topic %p, total subscription: %"PRIu32,
+            l, t, l->ntopic);
+
+    return true;
+}
+
+void
+listener_del_topic(struct listener *l, const struct topic *t)
+{
+    struct index_node *node;
+
+    /* do nothing if not found */
+    TAILQ_FOREACH(node, l->idx, i_tqe) {
+        if (node->obj == t) {
+            break;
+        }
+    }
+    if (node == NULL) {
+        log_debug("listener %p not subscribed to topic %p", l, t);
+        return;
+    }
+
+    TAILQ_REMOVE(l->idx, node, i_tqe);
+    l->ntopic--;
+    cc_free(node);
+    log_verb("listener %p unsubscribed from topic %p, total subscription: %"
+            PRIu32, l, t, l->ntopic);
+}
+
+void
+listener_setup(uint32_t hash_power)
+{
+    uint32_t i, nentry;
+    struct listener_slh *table;
+
+    ASSERT(hash_power > 0);
+
+    ht->hash_power = hash_power;
+    ht->nlistener = 0;
+    nentry = HASHSIZE(ht->hash_power);
+
+    table = cc_alloc(sizeof(*table) * nentry);
+    if (table == NULL) {
+        log_crit("listener setup failed: OOM");
+        exit(EX_CONFIG);
+    }
+    ht->table = table;
+
+    for (i = 0; i < nentry; ++i) {
+        SLIST_INIT(&table[i]);
+    }
+}
+
+void listener_teardown(void)
+{
+    if (ht->table != NULL) {
+        cc_free(ht->table);
+    }
+}
diff --git a/src/storage/pubsub/listener.h b/src/storage/pubsub/listener.h
new file mode 100644
index 000000000..5ea91b5ea
--- /dev/null
+++ b/src/storage/pubsub/listener.h
@@ -0,0 +1,28 @@
+#pragma once
+
+#include "index.h"
+
+#include <stream/cc_sockio.h>
+
+/*
+ * a listener is a client that has subscribed to at least one channel
+ */
+struct listener {
+    SLIST_ENTRY(listener)   l_sle;
+
+    struct buf_sock         *s;
+    uint32_t                ntopic;
+    struct index_tqh        *idx; /* index of all topics */
+};
+
+struct topic;
+
+void listener_setup(uint32_t hash_power);
+void listener_teardown(void);
+
+struct listener *listener_get(const struct buf_sock *s);
+struct listener *listener_add(const struct buf_sock *s);
+void listener_delete(const struct buf_sock *s);
+
+bool listener_add_topic(struct listener *l, const struct topic *t);
+void listener_del_topic(struct listener *l, const struct topic *t);
diff --git a/src/storage/pubsub/topic.c b/src/storage/pubsub/topic.c
new file mode 100644
index 000000000..61609dd39
--- /dev/null
+++ b/src/storage/pubsub/topic.c
@@ -0,0 +1,236 @@
+#include "hashtable.h"
+#include "topic.h"
+
+#include <cc_mm.h>
+#include <cc_queue.h>
+#include <hash/cc_lookup3.h>
+
+#include <sysexits.h>
+
+SLIST_HEAD(topic_slh, topic);
+
+struct topic_ht {
+    struct topic_slh *table;
+    uint32_t ntopic;
+    uint32_t hash_power;
+};
+
+static struct topic_ht hashtable;
+static struct topic_ht *ht = &hashtable;
+
+
+static struct topic_slh *
+_get_bucket(const struct bstring *name)
+{
+    /* use the _address_ of the channel to hash */
+    uint32_t hval = hash_lookup3(name->data, name->len, 0);
+    return &(ht->table[hval & HASHMASK(ht->hash_power)]);
+}
+
+static void
+_topic_reset(struct topic *t)
+{
+    t->name = null_bstring;
+    t->nsub = 0;
+    TAILQ_INIT(t->idx);
+}
+
+static struct topic *
+_topic_create(const struct bstring *name)
+{
+    struct topic *t;
+
+    t = cc_alloc(sizeof(struct topic));
+    if (t == NULL) {
+        return NULL;
+    }
+
+    t->idx = cc_alloc(sizeof(struct index_tqh));
+    if (t->idx == NULL) {
+        cc_free(t);
+        return NULL;
+    }
+
+    _topic_reset(t);
+    t->name.len = name->len;
+    t->name.data = cc_alloc(t->name.len);
+    if (t->name.data == NULL) {
+        cc_free(t->idx);
+        cc_free(t);
+        return NULL;
+    }
+    cc_memcpy(t->name.data, name->data, name->len);
+
+    return t;
+}
+
+static void
+_topic_destroy(struct topic **t)
+{
+    ASSERT(t != NULL && *t != NULL);
+
+    struct index_node *curr, *next;
+    struct index_tqh *idx = (*t)->idx;
+
+    /* delete all elements of the index */
+    TAILQ_FOREACH_SAFE(curr, idx, i_tqe, next) {
+        TAILQ_REMOVE(idx, curr, i_tqe);
+        cc_free(curr);
+    }
+    cc_free(idx);
+    cc_free((*t)->name.data);
+
+    cc_free(*t);
+    *t = NULL;
+}
+
+
+struct topic *
+topic_get(const struct bstring *name)
+{
+    struct topic_slh *bucket;
+    struct topic *t;
+
+    bucket = _get_bucket(name);
+    for (t = SLIST_FIRST(bucket); t != NULL; t = SLIST_NEXT(t, t_sle)) {
+        if (bstring_compare(&t->name, name) == 0) {
+            return t;
+        }
+    }
+
+    log_verb("topic not found name %.*s", t->name.len, t->name.data);
+    return NULL;
+}
+
+struct topic *
+topic_add(const struct bstring *name)
+{
+    ASSERT(topic_get(name) == NULL);
+
+    struct topic_slh *bucket;
+    struct topic *t = _topic_create(name);
+
+    log_verb("add topic %p for name %.*s", t, name->len, name->data);
+
+    bucket = _get_bucket(&t->name);
+    SLIST_INSERT_HEAD(bucket, t, t_sle);
+
+    ht->ntopic++;
+    log_verb("total topics: %"PRIu32, ht->ntopic);
+
+    return t;
+}
+
+void
+topic_delete(const struct bstring *name)
+{
+    struct topic_slh *bucket;
+    struct topic *t, *prev;
+
+    bucket = _get_bucket(name);
+    for (prev = NULL, t = SLIST_FIRST(bucket); t != NULL;
+        prev = t, t = SLIST_NEXT(t, t_sle)) {
+        if (t == NULL) {
+            log_debug("topic not found for %.*s", name->len, name->data);
+            return;
+        }
+        if (bstring_compare(&t->name, name) == 0) {
+            break;
+        }
+    }
+
+    if (prev == NULL) {
+        SLIST_REMOVE_HEAD(bucket, t_sle);
+    } else {
+        SLIST_REMOVE_AFTER(prev, t_sle);
+    }
+
+    _topic_destroy(&t);
+    --(ht->ntopic);
+    log_verb("total topics: %"PRIu32, ht->ntopic);
+}
+
+bool
+topic_add_listener(struct topic *t, const struct listener *l)
+{
+    struct index_node *node;
+
+    ASSERT(t != NULL && l != NULL);
+
+    /* do nothing if already subscribed */
+    TAILQ_FOREACH(node, t->idx, i_tqe) {
+        if (node->obj == l) {
+            log_debug("topic %p already subscribed by listener %p", t, l);
+            return false;
+        }
+    }
+
+    node = cc_alloc(sizeof(struct index_node));
+    if (node == NULL) {
+        log_error("cannot add listener: out of memory");
+        return false;
+    }
+    node->obj = (struct listener *)l;
+
+    TAILQ_INSERT_TAIL(t->idx, node, i_tqe);
+    t->nsub++;
+    log_verb("topic %p subscribed by listener %p, total listeners: %"PRIu32,
+            t, l, t->nsub);
+
+    return true;
+}
+
+void
+topic_del_listener(struct topic *t, const struct listener *l)
+{
+    struct index_node *node;
+
+    /* do nothing if not found */
+    TAILQ_FOREACH(node, t->idx, i_tqe) {
+        if (node->obj == l) {
+            break;
+        }
+    }
+    if (node == NULL) {
+        log_debug("topic %p not subscribed by listener %p", t, l);
+        return;
+    }
+
+    TAILQ_REMOVE(t->idx, node, i_tqe);
+    t->nsub--;
+    cc_free(node);
+    log_verb("topic %p subscribed by listener %p, total listeners: %"PRIu32,
+            t, l, t->nsub);
+
+}
+
+void
+topic_setup(uint32_t hash_power)
+{
+    uint32_t i, nentry;
+    struct topic_slh *table;
+
+    ASSERT(hash_power > 0);
+
+    ht->hash_power = hash_power;
+    ht->ntopic = 0;
+    nentry = HASHSIZE(ht->hash_power);
+
+    table = cc_alloc(sizeof(*table) * nentry);
+    if (table == NULL) {
+        log_crit("topic setup failed: OOM");
+        exit(EX_CONFIG);
+    }
+    ht->table = table;
+
+    for (i = 0; i < nentry; ++i) {
+        SLIST_INIT(&table[i]);
+    }
+}
+
+void topic_teardown(void)
+{
+    if (ht->table != NULL) {
+        cc_free(ht->table);
+    }
+}
diff --git a/src/storage/pubsub/topic.h b/src/storage/pubsub/topic.h
new file mode 100644
index 000000000..b26a744a2
--- /dev/null
+++ b/src/storage/pubsub/topic.h
@@ -0,0 +1,29 @@
+#pragma once
+
+#include "index.h"
+
+#include <cc_bstring.h>
+
+/*
+ * a topic is an endpoint that clients can subscribe to, equivalent to
+ * "channel" in the original redis protocol.
+ */
+struct topic {
+    SLIST_ENTRY(topic)  t_sle;
+
+    struct bstring      name;
+    uint32_t            nsub;
+    struct index_tqh    *idx; /* index of all listeners */
+};
+
+struct listener;
+
+void topic_setup(uint32_t hash_power);
+void topic_teardown(void);
+
+struct topic *topic_get(const struct bstring *name);
+struct topic *topic_add(const struct bstring *name);
+void topic_delete(const struct bstring *name);
+
+bool topic_add_listener(struct topic *t, const struct listener *l);
+void topic_del_listener(struct topic *t, const struct listener *l);