Merge PR #853

12 files changed, 288 insertions, 152 deletions

diff --git a/src/nvim/os/channel.c b/src/nvim/os/channel.c
index 9a692cf9fe..653f09756a 100644
--- a/src/nvim/os/channel.c
+++ b/src/nvim/os/channel.c
@@ -24,7 +24,7 @@ typedef struct {
   msgpack_unpacker *unpacker;
   msgpack_sbuffer *sbuffer;
   union {
-    int job_id;
+    Job *job;
     struct {
       RStream *read;
       WStream *write;
@@ -68,11 +68,26 @@ void channel_teardown()
 /// stdin/stdout. stderr is forwarded to the editor error stream.
 ///
 /// @param argv The argument vector for the process
-void channel_from_job(char **argv)
+bool channel_from_job(char **argv)
 {
   Channel *channel = register_channel();
   channel->is_job = true;
-  channel->data.job_id = job_start(argv, channel, job_out, job_err, NULL);
+
+  int status;
+  channel->data.job = job_start(argv,
+                                channel,
+                                job_out,
+                                job_err,
+                                job_exit,
+                                true,
+                                &status);
+
+  if (status <= 0) {
+    close_channel(channel);
+    return false;
+  }
+
+  return true;
 }
 
 /// Creates an API channel from a libuv stream representing a tcp or
@@ -101,12 +116,13 @@ void channel_from_stream(uv_stream_t *stream)
 /// @param type The event type, an arbitrary string
 /// @param obj The event data
 /// @return True if the data was sent successfully, false otherwise.
-bool channel_send_event(uint64_t id, char *type, typval_T *data)
+bool channel_send_event(uint64_t id, char *type, Object data)
 {
   Channel *channel = NULL;
 
   if (id > 0) {
     if (!(channel = pmap_get(uint64_t)(channels, id))) {
+      msgpack_rpc_free_object(data);
       return false;
     }
     send_event(channel, type, data);
@@ -126,7 +142,7 @@ void channel_subscribe(uint64_t id, char *event)
   Channel *channel;
 
   if (!(channel = pmap_get(uint64_t)(channels, id))) {
-    return;
+    abort();
   }
 
   char *event_string = pmap_get(cstr_t)(event_strings, event);
@@ -148,7 +164,7 @@ void channel_unsubscribe(uint64_t id, char *event)
   Channel *channel;
 
   if (!(channel = pmap_get(uint64_t)(channels, id))) {
-    return;
+    abort();
   }
 
   unsubscribe(channel, event);
@@ -165,6 +181,11 @@ static void job_err(RStream *rstream, void *data, bool eof)
   // TODO(tarruda): plugin error messages should be sent to the error buffer
 }
 
+static void job_exit(Job *job, void *data)
+{
+  // TODO(tarruda): what should be done here?
+}
+
 static void parse_msgpack(RStream *rstream, void *data, bool eof)
 {
   Channel *channel = data;
@@ -183,30 +204,57 @@ static void parse_msgpack(RStream *rstream, void *data, bool eof)
 
   msgpack_unpacked unpacked;
   msgpack_unpacked_init(&unpacked);
+  UnpackResult result;
+  msgpack_packer response;
 
   // Deserialize everything we can.
-  while (msgpack_unpacker_next(channel->unpacker, &unpacked)) {
+  while ((result = msgpack_rpc_unpack(channel->unpacker, &unpacked))
+      == kUnpackResultOk) {
     // Each object is a new msgpack-rpc request and requires an empty response
-    msgpack_packer response;
     msgpack_packer_init(&response, channel->sbuffer, msgpack_sbuffer_write);
     // Perform the call
     msgpack_rpc_call(channel->id, &unpacked.data, &response);
     wstream_write(channel->data.streams.write,
-                  wstream_new_buffer(channel->sbuffer->data,
+                  wstream_new_buffer(xmemdup(channel->sbuffer->data,
+                                             channel->sbuffer->size),
                                      channel->sbuffer->size,
-                                     true));
+                                     free));
 
     // Clear the buffer for future calls
     msgpack_sbuffer_clear(channel->sbuffer);
   }
+
+  if (result == kUnpackResultFail) {
+    // See src/msgpack/unpack_template.h in msgpack source tree for
+    // causes for this error(search for 'goto _failed')
+    //
+    // A not so uncommon cause for this might be deserializing objects with
+    // a high nesting level: msgpack will break when it's internal parse stack
+    // size exceeds MSGPACK_EMBED_STACK_SIZE(defined as 32 by default)
+    msgpack_packer_init(&response, channel->sbuffer, msgpack_sbuffer_write);
+    msgpack_pack_array(&response, 4);
+    msgpack_pack_int(&response, 1);
+    msgpack_pack_int(&response, 0);
+    msgpack_rpc_error("Invalid msgpack payload. "
+                      "This error can also happen when deserializing "
+                      "an object with high level of nesting",
+                      &response);
+    wstream_write(channel->data.streams.write,
+                  wstream_new_buffer(xmemdup(channel->sbuffer->data,
+                                             channel->sbuffer->size),
+                                     channel->sbuffer->size,
+                                     free));
+    // Clear the buffer for future calls
+    msgpack_sbuffer_clear(channel->sbuffer);
+  }
 }
 
-static void send_event(Channel *channel, char *type, typval_T *data)
+static void send_event(Channel *channel, char *type, Object data)
 {
   wstream_write(channel->data.streams.write, serialize_event(type, data));
 }
 
-static void broadcast_event(char *type, typval_T *data)
+static void broadcast_event(char *type, Object data)
 {
   kvec_t(Channel *) subscribed;
   kv_init(subscribed);
@@ -219,6 +267,7 @@ static void broadcast_event(char *type, typval_T *data)
   });
 
   if (!kv_size(subscribed)) {
+    msgpack_rpc_free_object(data);
     goto end;
   }
 
@@ -255,7 +304,9 @@ static void close_channel(Channel *channel)
   msgpack_unpacker_free(channel->unpacker);
 
   if (channel->is_job) {
-    job_stop(channel->data.job_id);
+    if (channel->data.job) {
+      job_stop(channel->data.job);
+    }
   } else {
     rstream_free(channel->data.streams.read);
     wstream_free(channel->data.streams.write);
@@ -278,17 +329,17 @@ static void close_cb(uv_handle_t *handle)
   free(handle);
 }
 
-static WBuffer *serialize_event(char *type, typval_T *data)
+static WBuffer *serialize_event(char *type, Object data)
 {
   String event_type = {.size = strnlen(type, EVENT_MAXLEN), .data = type};
-  Object event_data = vim_to_object(data);
   msgpack_packer packer;
   msgpack_packer_init(&packer, &msgpack_event_buffer, msgpack_sbuffer_write);
-  msgpack_rpc_notification(event_type, event_data, &packer);
-  WBuffer *rv = wstream_new_buffer(msgpack_event_buffer.data,
+  msgpack_rpc_notification(event_type, data, &packer);
+  WBuffer *rv = wstream_new_buffer(xmemdup(msgpack_event_buffer.data,
+                                           msgpack_event_buffer.size),
                                    msgpack_event_buffer.size,
-                                   true);
-  msgpack_rpc_free_object(event_data);
+                                   free);
+  msgpack_rpc_free_object(data);
   msgpack_sbuffer_clear(&msgpack_event_buffer);
 
   return rv;
diff --git a/src/nvim/os/channel.h b/src/nvim/os/channel.h
index 240461d22e..f12d54cede 100644
--- a/src/nvim/os/channel.h
+++ b/src/nvim/os/channel.h
@@ -2,8 +2,8 @@
 #define NVIM_OS_CHANNEL_H
 
 #include <uv.h>
-#include <msgpack.h>
 
+#include "nvim/api/private/defs.h"
 #include "nvim/vim.h"
 
 #define EVENT_MAXLEN 512
diff --git a/src/nvim/os/event.c b/src/nvim/os/event.c
index 2ebf28f436..6723b97e0c 100644
--- a/src/nvim/os/event.c
+++ b/src/nvim/os/event.c
@@ -21,17 +21,22 @@
 #define _destroy_event(x)  // do nothing
 KLIST_INIT(Event, Event, _destroy_event)
 
+typedef struct {
+  bool timed_out;
+  int32_t ms;
+  uv_timer_t *timer;
+} TimerData;
+
 #ifdef INCLUDE_GENERATED_DECLARATIONS
 # include "os/event.c.generated.h"
 #endif
-static klist_t(Event) *event_queue;
-static uv_timer_t timer;
-static uv_prepare_t timer_prepare;
+static klist_t(Event) *deferred_events, *immediate_events;
 
 void event_init()
 {
-  // Initialize the event queue
-  event_queue = kl_init(Event);
+  // Initialize the event queues
+  deferred_events = kl_init(Event);
+  immediate_events = kl_init(Event);
   // Initialize input events
   input_init();
   // Timer to wake the event loop if a timeout argument is passed to
@@ -44,9 +49,6 @@ void event_init()
   channel_init();
   // Servers
   server_init();
-  uv_timer_init(uv_default_loop(), &timer);
-  // This prepare handle that actually starts the timer
-  uv_prepare_init(uv_default_loop(), &timer_prepare);
 }
 
 void event_teardown()
@@ -59,7 +61,6 @@ void event_teardown()
 // Wait for some event
 bool event_poll(int32_t ms)
 {
-  bool timed_out;
   uv_run_mode run_mode = UV_RUN_ONCE;
 
   if (input_ready()) {
@@ -67,15 +68,26 @@ bool event_poll(int32_t ms)
     return true;
   }
 
-  input_start();
-  timed_out = false;
+  static int recursive = 0;
+
+  if (!(recursive++)) {
+    // Only needs to start the libuv handle the first time we enter here
+    input_start();
+  }
+
+  uv_timer_t timer;
+  uv_prepare_t timer_prepare;
+  TimerData timer_data = {.ms = ms, .timed_out = false, .timer = &timer};
 
   if (ms > 0) {
+    uv_timer_init(uv_default_loop(), &timer);
+    // This prepare handle that actually starts the timer
+    uv_prepare_init(uv_default_loop(), &timer_prepare);
     // Timeout passed as argument to the timer
-    timer.data = &timed_out;
+    timer.data = &timer_data;
     // We only start the timer after the loop is running, for that we
     // use a prepare handle(pass the interval as data to it)
-    timer_prepare.data = &ms;
+    timer_prepare.data = &timer_data;
     uv_prepare_start(&timer_prepare, timer_prepare_cb);
   } else if (ms == 0) {
     // For ms == 0, we need to do a non-blocking event poll by
@@ -87,40 +99,51 @@ bool event_poll(int32_t ms)
     // Run one event loop iteration, blocking for events if run_mode is
     // UV_RUN_ONCE
     uv_run(uv_default_loop(), run_mode);
+    // Process immediate events outside uv_run since libuv event loop not
+    // support recursion(processing events may cause a recursive event_poll
+    // call)
+    event_process(false);
   } while (
       // Continue running if ...
       !input_ready() &&   // we have no input
-      kl_empty(event_queue) &&   // no events are waiting to be processed
+      !event_has_deferred() &&   // no events are waiting to be processed
       run_mode != UV_RUN_NOWAIT &&   // ms != 0
-      !timed_out);  // we didn't get a timeout
+      !timer_data.timed_out);  // we didn't get a timeout
 
-  input_stop();
+  if (!(--recursive)) {
+    // Again, only stop when we leave the top-level invocation
+    input_stop();
+  }
 
   if (ms > 0) {
-    // Stop the timer
-    uv_timer_stop(&timer);
+    // Ensure the timer-related handles are closed and run the event loop
+    // once more to let libuv perform it's cleanup
+    uv_close((uv_handle_t *)&timer, NULL);
+    uv_close((uv_handle_t *)&timer_prepare, NULL);
+    uv_run(uv_default_loop(), UV_RUN_NOWAIT);
+    event_process(false);
   }
 
-  return input_ready() || event_is_pending();
+  return input_ready() || event_has_deferred();
 }
 
-bool event_is_pending()
+bool event_has_deferred()
 {
-  return !kl_empty(event_queue);
+  return !kl_empty(get_queue(true));
 }
 
 // Push an event to the queue
-void event_push(Event event)
+void event_push(Event event, bool deferred)
 {
-  *kl_pushp(Event, event_queue) = event;
+  *kl_pushp(Event, get_queue(deferred)) = event;
 }
 
 // Runs the appropriate action for each queued event
-void event_process()
+void event_process(bool deferred)
 {
   Event event;
 
-  while (kl_shift(Event, event_queue, &event) == 0) {
+  while (kl_shift(Event, get_queue(deferred), &event) == 0) {
     switch (event.type) {
       case kEventSignal:
         signal_handle(event);
@@ -140,11 +163,19 @@ void event_process()
 // Set a flag in the `event_poll` loop for signaling of a timeout
 static void timer_cb(uv_timer_t *handle)
 {
-  *((bool *)handle->data) = true;
+  TimerData *data = handle->data;
+  data->timed_out = true;
 }
 
 static void timer_prepare_cb(uv_prepare_t *handle)
 {
-  uv_timer_start(&timer, timer_cb, *(uint32_t *)timer_prepare.data, 0);
-  uv_prepare_stop(&timer_prepare);
+  TimerData *data = handle->data;
+  assert(data->ms > 0);
+  uv_timer_start(data->timer, timer_cb, (uint32_t)data->ms, 0);
+  uv_prepare_stop(handle);
+}
+
+static klist_t(Event) *get_queue(bool deferred)
+{
+  return deferred ? deferred_events : immediate_events;
 }
diff --git a/src/nvim/os/input.c b/src/nvim/os/input.c
index 3e9751a4db..6e42cba4ad 100644
--- a/src/nvim/os/input.c
+++ b/src/nvim/os/input.c
@@ -67,7 +67,7 @@ int os_inchar(uint8_t *buf, int maxlen, int32_t ms, int tb_change_cnt)
 {
   InbufPollResult result;
 
-  if (event_is_pending()) {
+  if (event_has_deferred()) {
     // Return pending event bytes
     return push_event_key(buf, maxlen);
   }
@@ -91,8 +91,8 @@ int os_inchar(uint8_t *buf, int maxlen, int32_t ms, int tb_change_cnt)
     }
   }
 
-  // If there are pending events, return the keys directly
-  if (event_is_pending()) {
+  // If there are deferred events, return the keys directly
+  if (event_has_deferred()) {
     return push_event_key(buf, maxlen);
   }
 
diff --git a/src/nvim/os/job.c b/src/nvim/os/job.c
index f9f94158ae..b369004e47 100644
--- a/src/nvim/os/job.c
+++ b/src/nvim/os/job.c
@@ -37,6 +37,7 @@ struct job {
   int pending_closes;
   // If the job was already stopped
   bool stopped;
+  bool defer;
   // Data associated with the job
   void *data;
   // Callbacks
@@ -128,14 +129,18 @@ void job_teardown()
 /// @param stderr_cb Callback that will be invoked when data is available
 ///        on stderr
 /// @param exit_cb Callback that will be invoked when the job exits
-/// @return The job id if the job started successfully. If the the first item /
-///         of `argv`(the program) could not be executed, -1 will be returned.
-//          0 will be returned if the job table is full.
-int job_start(char **argv,
-              void *data,
-              rstream_cb stdout_cb,
-              rstream_cb stderr_cb,
-              job_exit_cb job_exit_cb)
+/// @param defer If the job callbacks invocation should be deferred to vim
+///         main loop
+/// @param[out] The job id if the job started successfully, 0 if the job table
+///             is full, -1 if the program could not be executed.
+/// @return The job pointer if the job started successfully, NULL otherwise
+Job *job_start(char **argv,
+               void *data,
+               rstream_cb stdout_cb,
+               rstream_cb stderr_cb,
+               job_exit_cb job_exit_cb,
+               bool defer,
+               int *status)
 {
   int i;
   Job *job;
@@ -149,12 +154,14 @@ int job_start(char **argv,
 
   if (i == MAX_RUNNING_JOBS) {
     // No free slots
-    return 0;
+    *status = 0;
+    return NULL;
   }
 
   job = xmalloc(sizeof(Job));
   // Initialize
   job->id = i + 1;
+  *status = job->id;
   job->pending_refs = 3;
   job->pending_closes = 4;
   job->data = data;
@@ -175,6 +182,7 @@ int job_start(char **argv,
   job->proc_stdin.data = NULL;
   job->proc_stdout.data = NULL;
   job->proc_stderr.data = NULL;
+  job->defer = defer;
 
   // Initialize the job std{in,out,err}
   uv_pipe_init(uv_default_loop(), &job->proc_stdin, 0);
@@ -192,7 +200,8 @@ int job_start(char **argv,
   // Spawn the job
   if (uv_spawn(uv_default_loop(), &job->proc, &job->proc_opts) != 0) {
     free_job(job);
-    return -1;
+    *status = -1;
+    return NULL;
   }
 
   // Give all handles a reference to the job
@@ -204,8 +213,8 @@ int job_start(char **argv,
   job->in = wstream_new(JOB_WRITE_MAXMEM);
   wstream_set_stream(job->in, (uv_stream_t *)&job->proc_stdin);
   // Start the readable streams
-  job->out = rstream_new(read_cb, JOB_BUFFER_SIZE, job, true);
-  job->err = rstream_new(read_cb, JOB_BUFFER_SIZE, job, true);
+  job->out = rstream_new(read_cb, JOB_BUFFER_SIZE, job, defer);
+  job->err = rstream_new(read_cb, JOB_BUFFER_SIZE, job, defer);
   rstream_set_stream(job->out, (uv_stream_t *)&job->proc_stdout);
   rstream_set_stream(job->err, (uv_stream_t *)&job->proc_stderr);
   rstream_start(job->out);
@@ -219,77 +228,64 @@ int job_start(char **argv,
   }
   job_count++;
 
-  return job->id;
+  return job;
 }
 
-/// Terminates a job. This is a non-blocking operation, but if the job exists
-/// it's guaranteed to succeed(SIGKILL will eventually be sent)
+/// Finds a job instance by id
 ///
 /// @param id The job id
-/// @return true if the stop request was successfully sent, false if the job
-///              id is invalid(probably because it has already stopped)
-bool job_stop(int id)
+/// @return the Job instance
+Job *job_find(int id)
 {
-  Job *job = find_job(id);
+  Job *job;
 
-  if (job == NULL || job->stopped) {
-    return false;
+  if (id <= 0 || id > MAX_RUNNING_JOBS || !(job = table[id - 1])
+      || job->stopped) {
+    return NULL;
   }
 
-  job->stopped = true;
+  return job;
+}
 
-  return true;
+/// Terminates a job. This is a non-blocking operation, but if the job exists
+/// it's guaranteed to succeed(SIGKILL will eventually be sent)
+///
+/// @param job The Job instance
+void job_stop(Job *job)
+{
+  job->stopped = true;
 }
 
 /// Writes data to the job's stdin. This is a non-blocking operation, it
 /// returns when the write request was sent.
 ///
-/// @param id The job id
-/// @param data Buffer containing the data to be written
-/// @param len Size of the data
-/// @return true if the write request was successfully sent, false if the job
-///              id is invalid(probably because it has already stopped)
-bool job_write(int id, char *data, uint32_t len)
+/// @param job The Job instance
+/// @param buffer The buffer which contains the data to be written
+/// @return true if the write request was successfully sent, false if writing
+///         to the job stream failed (possibly because the OS buffer is full)
+bool job_write(Job *job, WBuffer *buffer)
 {
-  Job *job = find_job(id);
-
-  if (job == NULL || job->stopped) {
-    free(data);
-    return false;
-  }
-
-  if (!wstream_write(job->in, wstream_new_buffer(data, len, false))) {
-    job_stop(job->id);
-    return false;
-  }
+  return wstream_write(job->in, buffer);
+}
 
-  return true;
+/// Sets the `defer` flag for a Job instance
+///
+/// @param rstream The Job id
+/// @param defer The new value for the flag
+void job_set_defer(Job *job, bool defer)
+{
+  job->defer = defer;
+  rstream_set_defer(job->out, defer);
+  rstream_set_defer(job->err, defer);
 }
 
+
 /// Runs the read callback associated with the job exit event
 ///
 /// @param event Object containing data necessary to invoke the callback
 void job_exit_event(Event event)
 {
-  Job *job = event.data.job;
-
-  // Free the slot now, 'exit_cb' may want to start another job to replace
-  // this one
-  table[job->id - 1] = NULL;
-
-  if (job->exit_cb) {
-    // Invoke the exit callback
-    job->exit_cb(job, job->data);
-  }
-
-  // Free the job resources
-  free_job(job);
-
-  // Stop polling job status if this was the last
-  job_count--;
-  if (job_count == 0) {
-    uv_prepare_stop(&job_prepare);
-  }
+  job_exit_callback(event.data.job);
 }
 
 /// Get the job id
@@ -310,18 +306,30 @@ void *job_data(Job *job)
   return job->data;
 }
 
-static bool is_alive(Job *job)
+static void job_exit_callback(Job *job)
 {
-  return uv_process_kill(&job->proc, 0) == 0;
-}
+  // Free the slot now, 'exit_cb' may want to start another job to replace
+  // this one
+  table[job->id - 1] = NULL;
 
-static Job * find_job(int id)
-{
-  if (id <= 0 || id > MAX_RUNNING_JOBS) {
-    return NULL;
+  if (job->exit_cb) {
+    // Invoke the exit callback
+    job->exit_cb(job, job->data);
+  }
+
+  // Free the job resources
+  free_job(job);
+
+  // Stop polling job status if this was the last
+  job_count--;
+  if (job_count == 0) {
+    uv_prepare_stop(&job_prepare);
   }
+}
 
-  return table[id - 1];
+static bool is_alive(Job *job)
+{
+  return uv_process_kill(&job->proc, 0) == 0;
 }
 
 static void free_job(Job *job)
@@ -385,7 +393,7 @@ static void emit_exit_event(Job *job)
   Event event;
   event.type = kEventJobExit;
   event.data.job = job;
-  event_push(event);
+  event_push(event, true);
 }
 
 static void close_cb(uv_handle_t *handle)
diff --git a/src/nvim/os/job.h b/src/nvim/os/job.h
index f48218ffe7..e0ca615626 100644
--- a/src/nvim/os/job.h
+++ b/src/nvim/os/job.h
@@ -12,6 +12,8 @@
 
 #include "nvim/os/rstream_defs.h"
 #include "nvim/os/event_defs.h"
+#include "nvim/os/wstream.h"
+#include "nvim/os/wstream_defs.h"
 
 #ifdef INCLUDE_GENERATED_DECLARATIONS
 # include "os/job.h.generated.h"
diff --git a/src/nvim/os/msgpack_rpc.c b/src/nvim/os/msgpack_rpc.c
index 932a7717fd..63e1245028 100644
--- a/src/nvim/os/msgpack_rpc.c
+++ b/src/nvim/os/msgpack_rpc.c
@@ -79,11 +79,13 @@ void msgpack_rpc_call(uint64_t id, msgpack_object *req, msgpack_packer *res)
              "Request array size is %u, it should be 4",
              req->via.array.size);
     msgpack_rpc_error(error_msg, res);
+    return;
   }
 
   if (req->via.array.ptr[1].type != MSGPACK_OBJECT_POSITIVE_INTEGER) {
     msgpack_pack_int(res, 0);  // no message id yet
     msgpack_rpc_error("Id must be a positive integer", res);
+    return;
   }
 
   // Set the response id, which is the same as the request
@@ -398,6 +400,31 @@ void msgpack_rpc_free_dictionary(Dictionary value)
   free(value.items);
 }
 
+UnpackResult msgpack_rpc_unpack(msgpack_unpacker* unpacker,
+                                msgpack_unpacked* result)
+{
+  if (result->zone != NULL) {
+    msgpack_zone_free(result->zone);
+  }
+
+  int res = msgpack_unpacker_execute(unpacker);
+
+  if (res > 0) {
+    result->zone = msgpack_unpacker_release_zone(unpacker);
+    result->data = msgpack_unpacker_data(unpacker);
+    msgpack_unpacker_reset(unpacker);
+    return kUnpackResultOk;
+  }
+
+  if (res < 0) {
+    // Since we couldn't parse it, destroy the data consumed so far
+    msgpack_unpacker_reset(unpacker);
+    return kUnpackResultFail;
+  }
+
+  return kUnpackResultNeedMore;
+}
+
 REMOTE_FUNCS_IMPL(Buffer, buffer)
 REMOTE_FUNCS_IMPL(Window, window)
 REMOTE_FUNCS_IMPL(Tabpage, tabpage)
diff --git a/src/nvim/os/msgpack_rpc.h b/src/nvim/os/msgpack_rpc.h
index c8f243e2cf..baabff20aa 100644
--- a/src/nvim/os/msgpack_rpc.h
+++ b/src/nvim/os/msgpack_rpc.h
@@ -9,6 +9,12 @@
 #include "nvim/func_attr.h"
 #include "nvim/api/private/defs.h"
 
+typedef enum {
+  kUnpackResultOk,        /// Successfully parsed a document
+  kUnpackResultFail,      /// Got unexpected input
+  kUnpackResultNeedMore   /// Need more data
+} UnpackResult;
+
 /// Validates the basic structure of the msgpack-rpc call and fills `res`
 /// with the basic response structure.
 ///
@@ -40,6 +46,19 @@ void msgpack_rpc_dispatch(uint64_t id,
                           msgpack_packer *res)
   FUNC_ATTR_NONNULL_ARG(2) FUNC_ATTR_NONNULL_ARG(3);
 
+/// Try to unpack a msgpack document from the data in the unpacker buffer. This
+/// function is a replacement to msgpack.h `msgpack_unpack_next` that lets
+/// the called know if the unpacking failed due to bad input or due to missing
+/// data.
+///
+/// @param unpacker The unpacker containing the parse buffer
+/// @param result The result which will contain the parsed object
+/// @return kUnpackResultOk      : An object was parsed
+///         kUnpackResultFail    : Got bad input
+///         kUnpackResultNeedMore: Need more data
+UnpackResult msgpack_rpc_unpack(msgpack_unpacker* unpacker,
+                                msgpack_unpacked* result);
+
 /// Finishes the msgpack-rpc call with an error message.
 ///
 /// @param msg The error message
diff --git a/src/nvim/os/rstream.c b/src/nvim/os/rstream.c
index 1025201f7a..81714f7bae 100644
--- a/src/nvim/os/rstream.c
+++ b/src/nvim/os/rstream.c
@@ -24,7 +24,7 @@ struct rstream {
   uv_file fd;
   rstream_cb cb;
   size_t buffer_size, rpos, wpos, fpos;
-  bool reading, free_handle, async;
+  bool reading, free_handle, defer;
 };
 
 #ifdef INCLUDE_GENERATED_DECLARATIONS
@@ -38,21 +38,19 @@ struct rstream {
 ///        for reading with `rstream_read`
 /// @param buffer_size Size in bytes of the internal buffer.
 /// @param data Some state to associate with the `RStream` instance
-/// @param async Flag that specifies if the callback should only be called
-///        outside libuv event loop(When processing async events with
-///        KE_EVENT). Only the RStream instance reading user input should set
-///        this to false
+/// @param defer Flag that specifies if callback invocation should be deferred
+///        to vim main loop(as a KE_EVENT special key)
 /// @return The newly-allocated `RStream` instance
 RStream * rstream_new(rstream_cb cb,
                       size_t buffer_size,
                       void *data,
-                      bool async)
+                      bool defer)
 {
   RStream *rv = xmalloc(sizeof(RStream));
   rv->buffer = xmalloc(buffer_size);
   rv->buffer_size = buffer_size;
   rv->data = data;
-  rv->async = async;
+  rv->defer = defer;
   rv->cb = cb;
   rv->rpos = rv->wpos = rv->fpos = 0;
   rv->stream = NULL;
@@ -213,6 +211,15 @@ size_t rstream_available(RStream *rstream)
   return rstream->wpos - rstream->rpos;
 }
 
+/// Sets the `defer` flag for a a RStream instance
+///
+/// @param rstream The RStream instance
+/// @param defer The new value for the flag
+void rstream_set_defer(RStream *rstream, bool defer)
+{
+  rstream->defer = defer;
+}
+
 /// Runs the read callback associated with the rstream
 ///
 /// @param event Object containing data necessary to invoke the callback
@@ -333,16 +340,9 @@ static void close_cb(uv_handle_t *handle)
 
 static void emit_read_event(RStream *rstream, bool eof)
 {
-  if (rstream->async) {
-    Event event;
-
-    event.type = kEventRStreamData;
-    event.data.rstream.ptr = rstream;
-    event.data.rstream.eof = eof;
-    event_push(event);
-  } else {
-    // Invoke the callback passing in the number of bytes available and data
-    // associated with the stream
-    rstream->cb(rstream, rstream->data, eof);
-  }
+  Event event;
+  event.type = kEventRStreamData;
+  event.data.rstream.ptr = rstream;
+  event.data.rstream.eof = eof;
+  event_push(event, rstream->defer);
 }
diff --git a/src/nvim/os/signal.c b/src/nvim/os/signal.c
index 85aa8ae5cb..cfdc8821a4 100644
--- a/src/nvim/os/signal.c
+++ b/src/nvim/os/signal.c
@@ -159,5 +159,5 @@ static void signal_cb(uv_signal_t *handle, int signum)
       .signum = signum
     }
   };
-  event_push(event);
+  event_push(event, true);
 }
diff --git a/src/nvim/os/wstream.c b/src/nvim/os/wstream.c
index c2ed05b78f..9a908a4348 100644
--- a/src/nvim/os/wstream.c
+++ b/src/nvim/os/wstream.c
@@ -21,8 +21,9 @@ struct wstream {
 };
 
 struct wbuffer {
-  size_t refcount, size;
+  size_t size, refcount;
   char *data;
+  wbuffer_data_finalizer cb;
 };
 
 typedef struct {
@@ -90,7 +91,7 @@ bool wstream_write(WStream *wstream, WBuffer *buffer)
   // This should not be called after a wstream was freed
   assert(!wstream->freed);
 
-  if (wstream->curmem + buffer->size > wstream->maxmem) {
+  if (wstream->curmem > wstream->maxmem) {
     return false;
   }
 
@@ -116,19 +117,16 @@ bool wstream_write(WStream *wstream, WBuffer *buffer)
 ///
 /// @param data Data stored by the WBuffer
 /// @param size The size of the data array
-/// @param copy If true, the data will be copied into the WBuffer
+/// @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, bool copy)
+WBuffer *wstream_new_buffer(char *data, size_t size, wbuffer_data_finalizer cb)
 {
   WBuffer *rv = xmalloc(sizeof(WBuffer));
   rv->size = size;
   rv->refcount = 0;
-
-  if (copy) {
-    rv->data = xmemdup(data, size);
-  } else {
-    rv->data = data;
-  }
+  rv->cb = cb;
+  rv->data = data;
 
   return rv;
 }
@@ -141,8 +139,7 @@ static void write_cb(uv_write_t *req, int status)
   data->wstream->curmem -= data->buffer->size;
 
   if (!--data->buffer->refcount) {
-    // Free the data written to the stream
-    free(data->buffer->data);
+    data->buffer->cb(data->buffer->data);
     free(data->buffer);
   }
 
diff --git a/src/nvim/os/wstream_defs.h b/src/nvim/os/wstream_defs.h
index a7565c9bc7..1bf61ffce1 100644
--- a/src/nvim/os/wstream_defs.h
+++ b/src/nvim/os/wstream_defs.h
@@ -3,6 +3,7 @@
 
 typedef struct wbuffer WBuffer;
 typedef struct wstream WStream;
+typedef void (*wbuffer_data_finalizer)(void *data);
 
 #endif  // NVIM_OS_WSTREAM_DEFS_H