msgpack: Allow notifications to execute commands.

Consider: `let vim = rpcstart('nvim', ['--embed'])`

Allows `rpcnotify(vim, ...)` to work like an asynchronous
`rpcrequest(nvim, ...)`.

Helped-by: Michael Reed <m.reed@mykolab.com>
Helped-by: Justin M. Keyes <>

src/nvim/api/private/defs.h

@@ -22,6 +22,15 @@ typedef enum {
   kErrorTypeValidation
 } ErrorType;
 
+typedef enum {
+  kMessageTypeRequest,
+  kMessageTypeResponse,
+  kMessageTypeNotification
+} MessageType;
+
+/// Used as the message ID of notifications.
+#define NO_RESPONSE UINT64_MAX
+
 typedef struct {
   ErrorType type;
   char msg[1024];

src/nvim/msgpack_rpc/channel.c

@@ -435,18 +435,18 @@ static void handle_request(Channel *channel, msgpack_object *request)
 
   // Retrieve the request handler
   MsgpackRpcRequestHandler handler;
-  msgpack_object method = request->via.array.ptr[2];
+  msgpack_object *method = msgpack_rpc_method(request);
 
-  if (method.type == MSGPACK_OBJECT_BIN || method.type == MSGPACK_OBJECT_STR) {
-    handler = msgpack_rpc_get_handler_for(method.via.bin.ptr,
-                                          method.via.bin.size);
+  if (method) {
+    handler = msgpack_rpc_get_handler_for(method->via.bin.ptr,
+                                          method->via.bin.size);
   } else {
     handler.fn = msgpack_rpc_handle_missing_method;
     handler.defer = false;
   }
 
   Array args = ARRAY_DICT_INIT;
-  msgpack_rpc_to_array(request->via.array.ptr + 3, &args);
+  msgpack_rpc_to_array(msgpack_rpc_args(request), &args);
   bool defer = (!kv_size(channel->call_stack) && handler.defer);
   RequestEvent *event_data = xmalloc(sizeof(RequestEvent));
   event_data->channel = channel;
@@ -469,6 +469,7 @@ static void on_request_event(Event event)
   uint64_t request_id = e->request_id;
   Error error = ERROR_INIT;
   Object result = handler.fn(channel->id, request_id, args, &error);
+  if (request_id != NO_RESPONSE) {
     // send the response
     msgpack_packer response;
     msgpack_packer_init(&response, &out_buffer, msgpack_sbuffer_write);
@@ -477,6 +478,9 @@ static void on_request_event(Event event)
                                               &error,
                                               result,
                                               &out_buffer));
+  } else {
+    api_free_object(result);
+  }
   // All arguments were freed already, but we still need to free the array
   xfree(args.items);
   decref(channel);

src/nvim/msgpack_rpc/helpers.c

@@ -351,49 +351,86 @@ void msgpack_rpc_serialize_response(uint64_t response_id,
   }
 }
 
+static bool msgpack_rpc_is_notification(msgpack_object *req)
+{
+  return req->via.array.ptr[0].via.u64 == 2;
+}
+
+msgpack_object *msgpack_rpc_method(msgpack_object *req)
+{
+  msgpack_object *obj = req->via.array.ptr
+    + (msgpack_rpc_is_notification(req) ? 1 : 2);
+  return obj->type == MSGPACK_OBJECT_STR || obj->type == MSGPACK_OBJECT_BIN ?
+    obj : NULL;
+}
+
+msgpack_object *msgpack_rpc_args(msgpack_object *req)
+{
+  msgpack_object *obj = req->via.array.ptr
+    + (msgpack_rpc_is_notification(req) ? 2 : 3);
+  return obj->type == MSGPACK_OBJECT_ARRAY ? obj : NULL;
+}
+
+static msgpack_object *msgpack_rpc_msg_id(msgpack_object *req)
+{
+  if (msgpack_rpc_is_notification(req)) {
+    return NULL;
+  }
+  msgpack_object *obj = &req->via.array.ptr[1];
+  return obj->type == MSGPACK_OBJECT_POSITIVE_INTEGER ? obj : NULL;
+}
+
 void msgpack_rpc_validate(uint64_t *response_id,
                           msgpack_object *req,
                           Error *err)
 {
   // response id not known yet
 
-  *response_id = 0;
+  *response_id = NO_RESPONSE;
   // Validate the basic structure of the msgpack-rpc payload
   if (req->type != MSGPACK_OBJECT_ARRAY) {
-    api_set_error(err, Validation, _("Request is not an array"));
+    api_set_error(err, Validation, _("Message is not an array"));
     return;
   }
 
-  if (req->via.array.size != 4) {
-    api_set_error(err, Validation, _("Request array size should be 4"));
+  if (req->via.array.size == 0) {
+    api_set_error(err, Validation, _("Message is empty"));
     return;
   }
 
-  if (req->via.array.ptr[1].type != MSGPACK_OBJECT_POSITIVE_INTEGER) {
-    api_set_error(err, Validation, _("Id must be a positive integer"));
-    return;
-  }
-
-  // Set the response id, which is the same as the request
-  *response_id = req->via.array.ptr[1].via.u64;
-
   if (req->via.array.ptr[0].type != MSGPACK_OBJECT_POSITIVE_INTEGER) {
     api_set_error(err, Validation, _("Message type must be an integer"));
     return;
   }
 
-  if (req->via.array.ptr[0].via.u64 != 0) {
-    api_set_error(err, Validation, _("Message type must be 0"));
+  uint64_t type = req->via.array.ptr[0].via.u64;
+  if (type != kMessageTypeRequest && type != kMessageTypeNotification) {
+    api_set_error(err, Validation, _("Unknown message type"));
     return;
   }
 
-  if (req->via.array.ptr[2].type != MSGPACK_OBJECT_BIN
-    && req->via.array.ptr[2].type != MSGPACK_OBJECT_STR) {
+  if ((type == kMessageTypeRequest && req->via.array.size != 4) ||
+      (type == kMessageTypeNotification && req->via.array.size != 3)) {
+    api_set_error(err, Validation, _("Request array size should be 4 (request) "
+                                     "or 3 (notification)"));
+    return;
+  }
+
+  if (type == kMessageTypeRequest) {
+    msgpack_object *id_obj = msgpack_rpc_msg_id(req);
+    if (!id_obj) {
+      api_set_error(err, Validation, _("ID must be a positive integer"));
+      return;
+    }
+    *response_id = id_obj->via.u64;
+  }
+
+  if (!msgpack_rpc_method(req)) {
     api_set_error(err, Validation, _("Method must be a string"));
     return;
   }
 
-  if (req->via.array.ptr[3].type != MSGPACK_OBJECT_ARRAY) {
+  if (!msgpack_rpc_args(req)) {
     api_set_error(err, Validation, _("Parameters must be an array"));
     return;
   }

test/functional/api/server_requests_spec.lua

@@ -3,8 +3,8 @@
 -- be running.
 local helpers = require('test.functional.helpers')
 local clear, nvim, eval = helpers.clear, helpers.nvim, helpers.eval
-local eq, run, stop = helpers.eq, helpers.run, helpers.stop
-
+local eq, neq, run, stop = helpers.eq, helpers.neq, helpers.run, helpers.stop
+local nvim_prog = helpers.nvim_prog
 
 
 describe('server -> client', function()
@@ -115,4 +115,38 @@ describe('server -> client', function()
       eq(expected, notified)
     end)
   end)
+
+  describe('when the client is a recursive vim instance', function()
+    before_each(function()
+      nvim('command', "let vim = rpcstart('"..nvim_prog.."', ['--embed'])")
+      neq(0, eval('vim'))
+    end)
+
+    after_each(function() nvim('command', 'call rpcstop(vim)') end)
+
+    it('can send/recieve notifications and make requests', function()
+      nvim('command', "call rpcnotify(vim, 'vim_set_current_line', 'SOME TEXT')")
+
+      -- Wait for the notification to complete.
+      nvim('command', "call rpcrequest(vim, 'vim_eval', '0')")
+
+      eq('SOME TEXT', eval("rpcrequest(vim, 'vim_get_current_line')"))
+    end)
+
+    it('can communicate buffers, tabpages, and windows', function()
+      eq({3}, eval("rpcrequest(vim, 'vim_get_tabpages')"))
+      eq({1}, eval("rpcrequest(vim, 'vim_get_windows')"))
+
+      local buf = eval("rpcrequest(vim, 'vim_get_buffers')")[1]
+      eq(2, buf)
+
+      eval("rpcnotify(vim, 'buffer_set_line', "..buf..", 0, 'SOME TEXT')")
+      nvim('command', "call rpcrequest(vim, 'vim_eval', '0')")  -- wait
+
+      eq('SOME TEXT', eval("rpcrequest(vim, 'buffer_get_line', "..buf..", 0)"))
+
+      -- Call get_line_slice(buf, range [0,0], includes start, includes end)
+      eq({'SOME TEXT'}, eval("rpcrequest(vim, 'buffer_get_line_slice', "..buf..", 0, 0, 1, 1)"))
+    end)
+  end)
 end)