rstream/wstream: Unify structures and simplify API

- Simplify RStream/WStream API and make it more consistent with libuv.
- Move into the event loop layer(event subdirectory)
- Remove uv_helpers module.
- Simplify job/process internal modules/API.
- Unify RStream and WStream into a single structure. This is necessary because
  libuv streams can be readable and writable at the same time(and because the
  uv_helpers.c hack to associate multiple streams with libuv handle was removed)
- Make struct definition public, allowing more flexible/simple memory
  management by users of the module.
- Adapt channel/job modules to cope with the changes.

33 files changed, 765 insertions, 991 deletions

diff --git a/src/nvim/eval.c b/src/nvim/eval.c
index 4173acefd9..ae6b99c336 100644
--- a/src/nvim/eval.c
+++ b/src/nvim/eval.c
@@ -83,8 +83,7 @@
 #include "nvim/window.h"
 #include "nvim/os/os.h"
 #include "nvim/os/job.h"
-#include "nvim/os/rstream.h"
-#include "nvim/os/rstream_defs.h"
+#include "nvim/event/rstream.h"
 #include "nvim/os/time.h"
 #include "nvim/msgpack_rpc/channel.h"
 #include "nvim/msgpack_rpc/server.h"
@@ -20353,19 +20352,19 @@ static inline void push_job_event(Job *job, ufunc_T *callback,
   }, !disable_job_defer);
 }
 
-static void on_job_stdout(RStream *rstream, RBuffer *buf, void *job, bool eof)
+static void on_job_stdout(Stream *stream, RBuffer *buf, void *job, bool eof)
 {
   TerminalJobData *data = job_data(job);
-  on_job_output(rstream, job, buf, eof, data->on_stdout, "stdout");
+  on_job_output(stream, job, buf, eof, data->on_stdout, "stdout");
 }
 
-static void on_job_stderr(RStream *rstream, RBuffer *buf, void *job, bool eof)
+static void on_job_stderr(Stream *stream, RBuffer *buf, void *job, bool eof)
 {
   TerminalJobData *data = job_data(job);
-  on_job_output(rstream, job, buf, eof, data->on_stderr, "stderr");
+  on_job_output(stream, job, buf, eof, data->on_stderr, "stderr");
 }
 
-static void on_job_output(RStream *rstream, Job *job, RBuffer *buf, bool eof,
+static void on_job_output(Stream *stream, Job *job, RBuffer *buf, bool eof,
     ufunc_T *callback, const char *type)
 {
   if (eof) {
diff --git a/src/nvim/event/rstream.c b/src/nvim/event/rstream.c
new file mode 100644
index 0000000000..78c044347f
--- /dev/null
+++ b/src/nvim/event/rstream.c
@@ -0,0 +1,167 @@
+#include <assert.h>
+#include <stdint.h>
+#include <stdbool.h>
+#include <stdlib.h>
+
+#include <uv.h>
+
+#include "nvim/event/rstream.h"
+#include "nvim/ascii.h"
+#include "nvim/vim.h"
+#include "nvim/memory.h"
+#include "nvim/log.h"
+#include "nvim/misc1.h"
+#include "nvim/event/loop.h"
+
+#ifdef INCLUDE_GENERATED_DECLARATIONS
+# include "event/rstream.c.generated.h"
+#endif
+
+void rstream_init_fd(Loop *loop, Stream *stream, int fd, size_t bufsize,
+    void *data)
+  FUNC_ATTR_NONNULL_ARG(1)
+  FUNC_ATTR_NONNULL_ARG(2)
+{
+  stream_init(loop, stream, fd, NULL, data);
+  rstream_init(stream, bufsize);
+}
+
+void rstream_init_stream(Stream *stream, uv_stream_t *uvstream, size_t bufsize,
+    void *data)
+  FUNC_ATTR_NONNULL_ARG(1)
+  FUNC_ATTR_NONNULL_ARG(2)
+{
+  stream_init(NULL, stream, -1, uvstream, data);
+  rstream_init(stream, bufsize);
+}
+
+void rstream_init(Stream *stream, size_t bufsize)
+  FUNC_ATTR_NONNULL_ARG(1)
+{
+  stream->buffer = rbuffer_new(bufsize);
+  stream->buffer->data = stream;
+  stream->buffer->full_cb = on_rbuffer_full;
+  stream->buffer->nonfull_cb = on_rbuffer_nonfull;
+}
+
+
+/// Starts watching for events from a `Stream` instance.
+///
+/// @param stream The `Stream` instance
+void rstream_start(Stream *stream, stream_read_cb cb)
+{
+  stream->read_cb = cb;
+  if (stream->uvstream) {
+    uv_read_start(stream->uvstream, alloc_cb, read_cb);
+  } else {
+    uv_idle_start(&stream->uv.idle, fread_idle_cb);
+  }
+}
+
+/// Stops watching for events from a `Stream` instance.
+///
+/// @param stream The `Stream` instance
+void rstream_stop(Stream *stream)
+{
+  if (stream->uvstream) {
+    uv_read_stop(stream->uvstream);
+  } else {
+    uv_idle_stop(&stream->uv.idle);
+  }
+}
+
+static void on_rbuffer_full(RBuffer *buf, void *data)
+{
+  rstream_stop(data);
+}
+
+static void on_rbuffer_nonfull(RBuffer *buf, void *data)
+{
+  Stream *stream = data;
+  assert(stream->read_cb);
+  rstream_start(stream, stream->read_cb);
+}
+
+// Callbacks used by libuv
+
+// Called by libuv to allocate memory for reading.
+static void alloc_cb(uv_handle_t *handle, size_t suggested, uv_buf_t *buf)
+{
+  Stream *stream = handle->data;
+  buf->base = rbuffer_write_ptr(stream->buffer, &buf->len);
+}
+
+// Callback invoked by libuv after it copies the data into the buffer provided
+// by `alloc_cb`. This is also called on EOF or when `alloc_cb` returns a
+// 0-length buffer.
+static void read_cb(uv_stream_t *uvstream, ssize_t cnt, const uv_buf_t *buf)
+{
+  Stream *stream = uvstream->data;
+
+  if (cnt <= 0) {
+    if (cnt != UV_ENOBUFS
+        // cnt == 0 means libuv asked for a buffer and decided it wasn't needed:
+        // http://docs.libuv.org/en/latest/stream.html#c.uv_read_start.
+        //
+        // We don't need to do anything with the RBuffer because the next call
+        // to `alloc_cb` will return the same unused pointer(`rbuffer_produced`
+        // won't be called)
+        && cnt != 0) {
+      DLOG("Closing Stream(%p) because of %s(%zd)", stream,
+           uv_strerror((int)cnt), cnt);
+      // Read error or EOF, either way stop the stream and invoke the callback
+      // with eof == true
+      uv_read_stop(uvstream);
+      stream->read_cb(stream, stream->buffer, stream->data, true);
+    }
+    return;
+  }
+
+  // at this point we're sure that cnt is positive, no error occurred
+  size_t nread = (size_t)cnt;
+  // Data was already written, so all we need is to update 'wpos' to reflect
+  // the space actually used in the buffer.
+  rbuffer_produced(stream->buffer, nread);
+  stream->read_cb(stream, stream->buffer, stream->data, false);
+}
+
+// Called by the by the 'idle' handle to emulate a reading event
+static void fread_idle_cb(uv_idle_t *handle)
+{
+  uv_fs_t req;
+  Stream *stream = handle->data;
+
+  stream->uvbuf.base = rbuffer_write_ptr(stream->buffer, &stream->uvbuf.len);
+
+  // the offset argument to uv_fs_read is int64_t, could someone really try
+  // to read more than 9 quintillion (9e18) bytes?
+  // upcast is meant to avoid tautological condition warning on 32 bits
+  uintmax_t fpos_intmax = stream->fpos;
+  if (fpos_intmax > INT64_MAX) {
+    ELOG("stream offset overflow");
+    preserve_exit();
+  }
+
+  // Synchronous read
+  uv_fs_read(
+      handle->loop,
+      &req,
+      stream->fd,
+      &stream->uvbuf,
+      1,
+      (int64_t) stream->fpos,
+      NULL);
+
+  uv_fs_req_cleanup(&req);
+
+  if (req.result <= 0) {
+    uv_idle_stop(&stream->uv.idle);
+    stream->read_cb(stream, stream->buffer, stream->data, true);
+    return;
+  }
+
+  // no errors (req.result (ssize_t) is positive), it's safe to cast.
+  size_t nread = (size_t) req.result;
+  rbuffer_produced(stream->buffer, nread);
+  stream->fpos += nread;
+}
diff --git a/src/nvim/event/rstream.h b/src/nvim/event/rstream.h
new file mode 100644
index 0000000000..f30ad79ee5
--- /dev/null
+++ b/src/nvim/event/rstream.h
@@ -0,0 +1,16 @@
+#ifndef NVIM_EVENT_RSTREAM_H
+#define NVIM_EVENT_RSTREAM_H
+
+#include <stdbool.h>
+#include <stddef.h>
+
+#include <uv.h>
+
+#include "nvim/event/loop.h"
+#include "nvim/event/stream.h"
+
+
+#ifdef INCLUDE_GENERATED_DECLARATIONS
+# include "event/rstream.h.generated.h"
+#endif
+#endif  // NVIM_EVENT_RSTREAM_H
diff --git a/src/nvim/event/stream.c b/src/nvim/event/stream.c
new file mode 100644
index 0000000000..959b532146
--- /dev/null
+++ b/src/nvim/event/stream.c
@@ -0,0 +1,108 @@
+#include <assert.h>
+#include <stdio.h>
+#include <stdbool.h>
+
+#include <uv.h>
+
+#include "nvim/rbuffer.h"
+#include "nvim/event/stream.h"
+
+#ifdef INCLUDE_GENERATED_DECLARATIONS
+# include "event/stream.c.generated.h"
+#endif
+
+/// Sets the stream associated with `fd` to "blocking" mode.
+///
+/// @return `0` on success, or `-errno` on failure.
+int stream_set_blocking(int fd, bool blocking)
+{
+  // Private loop to avoid conflict with existing watcher(s):
+  //    uv__io_stop: Assertion `loop->watchers[w->fd] == w' failed.
+  uv_loop_t loop;
+  uv_pipe_t stream;
+  uv_loop_init(&loop);
+  uv_pipe_init(&loop, &stream, 0);
+  uv_pipe_open(&stream, fd);
+  int retval = uv_stream_set_blocking((uv_stream_t *)&stream, blocking);
+  uv_close((uv_handle_t *)&stream, NULL);
+  uv_run(&loop, UV_RUN_NOWAIT);  // not necessary, but couldn't hurt.
+  uv_loop_close(&loop);
+  return retval;
+}
+
+void stream_init(Loop *loop, Stream *stream, int fd, uv_stream_t *uvstream,
+    void *data)
+{
+  stream->uvstream = uvstream;
+
+  if (fd >= 0) {
+    uv_handle_type type = uv_guess_handle(fd);
+    stream->fd = fd;
+
+    if (type == UV_FILE) {
+      // Non-blocking file reads are simulated with an idle handle that reads in
+      // chunks of the ring buffer size, giving time for other events to be
+      // processed between reads.
+      uv_idle_init(&loop->uv, &stream->uv.idle);
+      stream->uv.idle.data = stream;
+    } else {
+      assert(type == UV_NAMED_PIPE || type == UV_TTY);
+      uv_pipe_init(&loop->uv, &stream->uv.pipe, 0);
+      uv_pipe_open(&stream->uv.pipe, fd);
+      stream->uvstream = (uv_stream_t *)&stream->uv.pipe;
+    }
+  }
+
+  if (stream->uvstream) {
+    stream->uvstream->data = stream;
+  }
+
+  stream->data = data;
+  stream->internal_data = NULL;
+  stream->fpos = 0;
+  stream->curmem = 0;
+  stream->maxmem = 0;
+  stream->pending_reqs = 0;
+  stream->read_cb = NULL;
+  stream->write_cb = NULL;
+  stream->close_cb = NULL;
+  stream->internal_close_cb = NULL;
+  stream->closed = false;
+  stream->buffer = NULL;
+}
+
+void stream_close(Stream *stream, stream_close_cb on_stream_close)
+{
+  assert(!stream->closed);
+
+  if (stream->buffer) {
+    rbuffer_free(stream->buffer);
+  }
+
+  stream->closed = true;
+  stream->close_cb = on_stream_close;
+
+  if (!stream->pending_reqs) {
+    stream_close_handle(stream);
+  }
+}
+
+void stream_close_handle(Stream *stream)
+{
+  if (stream->uvstream) {
+    uv_close((uv_handle_t *)stream->uvstream, close_cb);
+  } else {
+    uv_close((uv_handle_t *)&stream->uv.idle, close_cb);
+  }
+}
+
+static void close_cb(uv_handle_t *handle)
+{
+  Stream *stream = handle->data;
+  if (stream->close_cb) {
+    stream->close_cb(stream, stream->data);
+  }
+  if (stream->internal_close_cb) {
+    stream->internal_close_cb(stream, stream->internal_data);
+  }
+}
diff --git a/src/nvim/event/stream.h b/src/nvim/event/stream.h
new file mode 100644
index 0000000000..37410b2036
--- /dev/null
+++ b/src/nvim/event/stream.h
@@ -0,0 +1,55 @@
+#ifndef NVIM_EVENT_STREAM_H
+#define NVIM_EVENT_STREAM_H
+
+#include <stdbool.h>
+#include <stddef.h>
+
+#include <uv.h>
+
+#include "nvim/event/loop.h"
+#include "nvim/rbuffer.h"
+
+typedef struct stream Stream;
+/// Type of function called when the Stream buffer is filled with data
+///
+/// @param stream The Stream instance
+/// @param rbuffer The associated RBuffer instance
+/// @param data User-defined data
+/// @param eof If the stream reached EOF.
+typedef void (*stream_read_cb)(Stream *stream, RBuffer *buf, void *data,
+    bool eof);
+
+/// Type of function called when the Stream has information about a write
+/// request.
+///
+/// @param wstream The Stream instance
+/// @param data User-defined data
+/// @param status 0 on success, anything else indicates failure
+typedef void (*stream_write_cb)(Stream *stream, void *data, int status);
+typedef void (*stream_close_cb)(Stream *stream, void *data);
+
+struct stream {
+  union {
+    uv_pipe_t pipe;
+    uv_tcp_t tcp;
+    uv_idle_t idle;
+  } uv;
+  uv_stream_t *uvstream;
+  uv_buf_t uvbuf;
+  RBuffer *buffer;
+  uv_file fd;
+  stream_read_cb read_cb;
+  stream_write_cb write_cb;
+  stream_close_cb close_cb, internal_close_cb;
+  size_t fpos;
+  size_t curmem;
+  size_t maxmem;
+  size_t pending_reqs;
+  void *data, *internal_data;
+  bool closed;
+};
+
+#ifdef INCLUDE_GENERATED_DECLARATIONS
+# include "event/stream.h.generated.h"
+#endif
+#endif  // NVIM_EVENT_STREAM_H
diff --git a/src/nvim/event/wstream.c b/src/nvim/event/wstream.c
new file mode 100644
index 0000000000..5fcb724fe3
--- /dev/null
+++ b/src/nvim/event/wstream.c
@@ -0,0 +1,162 @@
+#include <assert.h>
+#include <stdint.h>
+#include <stdbool.h>
+#include <stdlib.h>
+
+#include <uv.h>
+
+#include "nvim/event/loop.h"
+#include "nvim/event/wstream.h"
+#include "nvim/vim.h"
+#include "nvim/memory.h"
+
+#define DEFAULT_MAXMEM 1024 * 1024 * 10
+
+typedef struct {
+  Stream *stream;
+  WBuffer *buffer;
+  uv_write_t uv_req;
+} WRequest;
+
+#ifdef INCLUDE_GENERATED_DECLARATIONS
+# include "event/wstream.c.generated.h"
+#endif
+
+void wstream_init_fd(Loop *loop, Stream *stream, int fd, size_t maxmem,
+    void *data)
+  FUNC_ATTR_NONNULL_ARG(1) FUNC_ATTR_NONNULL_ARG(2)
+{
+  stream_init(loop, stream, fd, NULL, data);
+  wstream_init(stream, maxmem);
+}
+
+void wstream_init_stream(Stream *stream, uv_stream_t *uvstream, size_t maxmem,
+    void *data)
+  FUNC_ATTR_NONNULL_ARG(1) FUNC_ATTR_NONNULL_ARG(2)
+{
+  stream_init(NULL, stream, -1, uvstream, data);
+  wstream_init(stream, maxmem);
+}
+
+void wstream_init(Stream *stream, size_t maxmem)
+{
+  stream->maxmem = maxmem ? maxmem : DEFAULT_MAXMEM;
+}
+
+/// Sets a callback that will be called on completion of a write request,
+/// indicating failure/success.
+///
+/// This affects all requests currently in-flight as well. Overwrites any
+/// possible earlier callback.
+///
+/// @note This callback will not fire if the write request couldn't even be
+///       queued properly (i.e.: when `wstream_write() returns an error`).
+///
+/// @param stream The `Stream` instance
+/// @param cb The callback
+void wstream_set_write_cb(Stream *stream, stream_write_cb cb)
+  FUNC_ATTR_NONNULL_ALL
+{
+  stream->write_cb = cb;
+}
+
+/// Queues data for writing to the backing file descriptor of a `Stream`
+/// instance. This will fail if the write would cause the Stream use more
+/// memory than specified by `maxmem`.
+///
+/// @param stream The `Stream` instance
+/// @param buffer The buffer which contains data to be written
+/// @return false if the write failed
+bool wstream_write(Stream *stream, WBuffer *buffer)
+  FUNC_ATTR_NONNULL_ALL
+{
+  assert(stream->maxmem);
+  // This should not be called after a stream was freed
+  assert(!stream->closed);
+
+  if (stream->curmem > stream->maxmem) {
+    goto err;
+  }
+
+  stream->curmem += buffer->size;
+
+  WRequest *data = xmalloc(sizeof(WRequest));
+  data->stream = stream;
+  data->buffer = buffer;
+  data->uv_req.data = data;
+
+  uv_buf_t uvbuf;
+  uvbuf.base = buffer->data;
+  uvbuf.len = buffer->size;
+
+  if (uv_write(&data->uv_req, stream->uvstream, &uvbuf, 1, write_cb)) {
+    xfree(data);
+    goto err;
+  }
+
+  stream->pending_reqs++;
+  return true;
+
+err:
+  wstream_release_wbuffer(buffer);
+  return false;
+}
+
+/// Creates a WBuffer object for holding output data. Instances of this
+/// object can be reused across Stream instances, and the memory is freed
+/// automatically when no longer needed(it tracks the number of references
+/// internally)
+///
+/// @param data Data stored by the WBuffer
+/// @param size The size of the data array
+/// @param refcount The number of references for the WBuffer. This will be used
+///        by Stream instances to decide when a WBuffer should be freed.
+/// @param cb Pointer to function that will be responsible for freeing
+///        the buffer data(passing 'free' will work as expected).
+/// @return The allocated WBuffer instance
+WBuffer *wstream_new_buffer(char *data,
+                            size_t size,
+                            size_t refcount,
+                            wbuffer_data_finalizer cb)
+{
+  WBuffer *rv = xmalloc(sizeof(WBuffer));
+  rv->size = size;
+  rv->refcount = refcount;
+  rv->cb = cb;
+  rv->data = data;
+
+  return rv;
+}
+
+static void write_cb(uv_write_t *req, int status)
+{
+  WRequest *data = req->data;
+
+  data->stream->curmem -= data->buffer->size;
+
+  wstream_release_wbuffer(data->buffer);
+
+  if (data->stream->write_cb) {
+    data->stream->write_cb(data->stream, data->stream->data, status);
+  }
+
+  data->stream->pending_reqs--;
+
+  if (data->stream->closed && data->stream->pending_reqs == 0) {
+    // Last pending write, free the stream;
+    stream_close_handle(data->stream);
+  }
+
+  xfree(data);
+}
+
+void wstream_release_wbuffer(WBuffer *buffer)
+{
+  if (!--buffer->refcount) {
+    if (buffer->cb) {
+      buffer->cb(buffer->data);
+    }
+
+    xfree(buffer);
+  }
+}
diff --git a/src/nvim/event/wstream.h b/src/nvim/event/wstream.h
new file mode 100644
index 0000000000..9008de0d97
--- /dev/null
+++ b/src/nvim/event/wstream.h
@@ -0,0 +1,24 @@
+#ifndef NVIM_EVENT_WSTREAM_H
+#define NVIM_EVENT_WSTREAM_H
+
+#include <stdint.h>
+#include <stdbool.h>
+
+#include <uv.h>
+
+#include "nvim/event/loop.h"
+#include "nvim/event/stream.h"
+
+typedef struct wbuffer WBuffer;
+typedef void (*wbuffer_data_finalizer)(void *data);
+
+struct wbuffer {
+  size_t size, refcount;
+  char *data;
+  wbuffer_data_finalizer cb;
+};
+
+#ifdef INCLUDE_GENERATED_DECLARATIONS
+# include "event/wstream.h.generated.h"
+#endif
+#endif  // NVIM_EVENT_WSTREAM_H
diff --git a/src/nvim/ex_docmd.c b/src/nvim/ex_docmd.c
index e53b2e47c5..3b6e05fd8a 100644
--- a/src/nvim/ex_docmd.c
+++ b/src/nvim/ex_docmd.c
@@ -73,8 +73,8 @@
 #include "nvim/os/time.h"
 #include "nvim/ex_cmds_defs.h"
 #include "nvim/mouse.h"
-#include "nvim/os/rstream.h"
-#include "nvim/os/wstream.h"
+#include "nvim/event/rstream.h"
+#include "nvim/event/wstream.h"
 
 static int quitmore = 0;
 static int ex_pressedreturn = FALSE;
diff --git a/src/nvim/if_cscope.c b/src/nvim/if_cscope.c
index 99926ecf16..d6c5cf4fd5 100644
--- a/src/nvim/if_cscope.c
+++ b/src/nvim/if_cscope.c
@@ -31,6 +31,7 @@
 #include "nvim/window.h"
 #include "nvim/os/os.h"
 #include "nvim/os/input.h"
+#include "nvim/event/stream.h"
 
 #include <sys/types.h>
 #include <sys/stat.h>
diff --git a/src/nvim/misc1.c b/src/nvim/misc1.c
index 0737caec5d..4f23b3da63 100644
--- a/src/nvim/misc1.c
+++ b/src/nvim/misc1.c
@@ -61,6 +61,7 @@
 #include "nvim/os/shell.h"
 #include "nvim/os/input.h"
 #include "nvim/os/time.h"
+#include "nvim/event/stream.h"
 
 #ifdef INCLUDE_GENERATED_DECLARATIONS
 # include "misc1.c.generated.h"
diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index 3932fa1f36..05badc72d4 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -10,10 +10,8 @@
 #include "nvim/msgpack_rpc/channel.h"
 #include "nvim/msgpack_rpc/remote_ui.h"
 #include "nvim/event/loop.h"
-#include "nvim/os/rstream.h"
-#include "nvim/os/rstream_defs.h"
-#include "nvim/os/wstream.h"
-#include "nvim/os/wstream_defs.h"
+#include "nvim/event/rstream.h"
+#include "nvim/event/wstream.h"
 #include "nvim/os/job.h"
 #include "nvim/os/job_defs.h"
 #include "nvim/msgpack_rpc/helpers.h"
@@ -34,6 +32,12 @@
 #define log_server_msg(...)
 #endif
 
+typedef enum {
+  kChannelTypeSocket,
+  kChannelTypeJob,
+  kChannelTypeStdio
+} ChannelType;
+
 typedef struct {
   uint64_t request_id;
   bool returned, errored;
@@ -45,15 +49,16 @@ typedef struct {
   size_t refcount;
   size_t pending_requests;
   PMap(cstr_t) *subscribed_events;
-  bool is_job, closed;
+  bool closed;
+  ChannelType type;
   msgpack_unpacker *unpacker;
   union {
     Job *job;
+    Stream stream;
     struct {
-      RStream *read;
-      WStream *write;
-      uv_stream_t *uv;
-    } streams;
+      Stream in;
+      Stream out;
+    } std;
   } data;
   uint64_t next_request_id;
   kvec_t(ChannelCallFrame *) call_stack;
@@ -112,8 +117,7 @@ void channel_teardown(void)
 ///         0, on error.
 uint64_t channel_from_job(char **argv)
 {
-  Channel *channel = register_channel();
-  channel->is_job = true;
+  Channel *channel = register_channel(kChannelTypeJob);
   incref(channel);  // job channels are only closed by the exit_cb
 
   int status;
@@ -140,21 +144,15 @@ uint64_t channel_from_job(char **argv)
 /// pipe/socket client connection
 ///
 /// @param stream The established connection
-void channel_from_stream(uv_stream_t *stream)
+void channel_from_stream(uv_stream_t *uvstream)
 {
-  Channel *channel = register_channel();
-  stream->data = NULL;
-  channel->is_job = false;
-  // read stream
-  channel->data.streams.read = rstream_new(parse_msgpack,
-                                           rbuffer_new(CHANNEL_BUFFER_SIZE),
-                                           channel);
-  rstream_set_stream(channel->data.streams.read, stream);
-  rstream_start(channel->data.streams.read);
+  Channel *channel = register_channel(kChannelTypeSocket);
+  stream_init(NULL, &channel->data.stream, -1, uvstream, channel);
   // write stream
-  channel->data.streams.write = wstream_new(0);
-  wstream_set_stream(channel->data.streams.write, stream);
-  channel->data.streams.uv = stream;
+  wstream_init(&channel->data.stream, 0);
+  // read stream
+  rstream_init(&channel->data.stream, CHANNEL_BUFFER_SIZE);
+  rstream_start(&channel->data.stream, parse_msgpack);
 }
 
 /// Sends event/arguments to channel
@@ -313,28 +311,23 @@ bool channel_close(uint64_t id)
 /// Neovim
 static void channel_from_stdio(void)
 {
-  Channel *channel = register_channel();
+  Channel *channel = register_channel(kChannelTypeStdio);
   incref(channel);  // stdio channels are only closed on exit
-  channel->is_job = false;
   // read stream
-  channel->data.streams.read = rstream_new(parse_msgpack,
-                                           rbuffer_new(CHANNEL_BUFFER_SIZE),
-                                           channel);
-  rstream_set_file(channel->data.streams.read, 0);
-  rstream_start(channel->data.streams.read);
+  rstream_init_fd(&loop, &channel->data.std.in, 0, CHANNEL_BUFFER_SIZE,
+      channel);
+  rstream_start(&channel->data.std.in, parse_msgpack);
   // write stream
-  channel->data.streams.write = wstream_new(0);
-  wstream_set_file(channel->data.streams.write, 1);
-  channel->data.streams.uv = NULL;
+  wstream_init_fd(&loop, &channel->data.std.out, 1, 0, NULL);
 }
 
-static void job_out(RStream *rstream, RBuffer *buf, void *data, bool eof)
+static void job_out(Stream *stream, RBuffer *buf, void *data, bool eof)
 {
   Job *job = data;
-  parse_msgpack(rstream, buf, job_data(job), eof);
+  parse_msgpack(stream, buf, job_data(job), eof);
 }
 
-static void job_err(RStream *rstream, RBuffer *rbuf, void *data, bool eof)
+static void job_err(Stream *stream, RBuffer *rbuf, void *data, bool eof)
 {
   while (rbuffer_size(rbuf)) {
     char buf[256];
@@ -350,7 +343,7 @@ static void job_exit(Job *job, int status, void *data)
   decref(data);
 }
 
-static void parse_msgpack(RStream *rstream, RBuffer *rbuf, void *data, bool eof)
+static void parse_msgpack(Stream *stream, RBuffer *rbuf, void *data, bool eof)
 {
   Channel *channel = data;
   incref(channel);
@@ -362,9 +355,9 @@ static void parse_msgpack(RStream *rstream, RBuffer *rbuf, void *data, bool eof)
   }
 
   size_t count = rbuffer_size(rbuf);
-  DLOG("Feeding the msgpack parser with %u bytes of data from RStream(%p)",
+  DLOG("Feeding the msgpack parser with %u bytes of data from Stream(%p)",
        count,
-       rstream);
+       stream);
 
   // Feed the unpacker with data
   msgpack_unpacker_reserve_buffer(channel->unpacker, count);
@@ -516,10 +509,18 @@ static bool channel_write(Channel *channel, WBuffer *buffer)
     return false;
   }
 
-  if (channel->is_job) {
-    success = job_write(channel->data.job, buffer);
-  } else {
-    success = wstream_write(channel->data.streams.write, buffer);
+  switch (channel->type) {
+    case kChannelTypeSocket:
+      success = wstream_write(&channel->data.stream, buffer);
+      break;
+    case kChannelTypeJob:
+      success = job_write(channel->data.job, buffer);
+      break;
+    case kChannelTypeStdio:
+      success = wstream_write(&channel->data.std.out, buffer);
+      break;
+    default:
+      abort();
   }
 
   if (!success) {
@@ -637,20 +638,23 @@ static void close_channel(Channel *channel)
 
   channel->closed = true;
 
-  if (channel->is_job) {
-    if (channel->data.job) {
-      job_stop(channel->data.job);
-    }
-  } else {
-    rstream_free(channel->data.streams.read);
-    wstream_free(channel->data.streams.write);
-    uv_handle_t *handle = (uv_handle_t *)channel->data.streams.uv;
-    if (handle) {
-      uv_close(handle, close_cb);
-    } else {
+  switch (channel->type) {
+    case kChannelTypeSocket:
+      stream_close(&channel->data.stream, close_cb);
+      break;
+    case kChannelTypeJob:</