channel/msgpack_rpc: Refactor msgpack_rpc_notification/serialize_event

- Generalize some argument names(event type -> event name,
                                 event data -> event arg)
- Rename serialize_event to serialize_message
- Rename msgpack_rpc_notification to msgpack_rpc_message
- Extract the message type out of msgpack_rpc_message
- Add 'id' parameter to msgpack_rpc_message/serialize_message to create messages
  that are not notifications

src/nvim/api/vim.c

@@ -424,8 +424,8 @@ void vim_set_current_tabpage(Tabpage tabpage, Error *err)
 /// @param event The event type string
 void vim_subscribe(uint64_t channel_id, String event)
 {
-  size_t length = (event.size < EVENT_MAXLEN ? event.size : EVENT_MAXLEN);
-  char e[EVENT_MAXLEN + 1];
+  size_t length = (event.size < METHOD_MAXLEN ? event.size : METHOD_MAXLEN);
+  char e[METHOD_MAXLEN + 1];
   memcpy(e, event.data, length);
   e[length] = NUL;
   channel_subscribe(channel_id, e);
@@ -437,8 +437,10 @@ void vim_subscribe(uint64_t channel_id, String event)
 /// @param event The event type string
 void vim_unsubscribe(uint64_t channel_id, String event)
 {
-  size_t length = (event.size < EVENT_MAXLEN ? event.size : EVENT_MAXLEN);
-  char e[EVENT_MAXLEN + 1];
+  size_t length = (event.size < METHOD_MAXLEN ?
+                   event.size :
+                   METHOD_MAXLEN);
+  char e[METHOD_MAXLEN + 1];
   memcpy(e, event.data, length);
   e[length] = NUL;
   channel_unsubscribe(channel_id, e);

src/nvim/os/channel.c

@@ -115,21 +115,21 @@ void channel_from_stream(uv_stream_t *stream)
 ///
 /// @param id The channel id. If 0, the event will be sent to all
 ///        channels that have subscribed to the event type
-/// @param type The event type, an arbitrary string
-/// @param obj The event data
+/// @param name The event name, an arbitrary string
+/// @param arg The event arg
 /// @return True if the data was sent successfully, false otherwise.
-bool channel_send_event(uint64_t id, char *type, Object data)
+bool channel_send_event(uint64_t id, char *name, Object arg)
 {
   Channel *channel = NULL;
 
   if (id > 0) {
     if (!(channel = pmap_get(uint64_t)(channels, id))) {
-      msgpack_rpc_free_object(data);
+      msgpack_rpc_free_object(arg);
       return false;
     }
-    send_event(channel, type, data);
+    send_message(channel, 2, 0, name, arg);
   } else {
-    broadcast_event(type, data);
+    broadcast_event(name, arg);
   }
 
   return true;
@@ -284,29 +284,33 @@ static bool channel_write(Channel *channel, WBuffer *buffer)
   return success;
 }
 
-static void send_event(Channel *channel, char *type, Object data)
+static void send_message(Channel *channel,
+                         int type,
+                         uint64_t id,
+                         char *name,
+                         Object arg)
 {
-  channel_write(channel, serialize_event(type, data));
+  channel_write(channel, serialize_message(type, id, name, arg));
 }
 
-static void broadcast_event(char *type, Object data)
+static void broadcast_event(char *name, Object arg)
 {
   kvec_t(Channel *) subscribed;
   kv_init(subscribed);
   Channel *channel;
 
   map_foreach_value(channels, channel, {
-    if (pmap_has(cstr_t)(channel->subscribed_events, type)) {
+    if (pmap_has(cstr_t)(channel->subscribed_events, name)) {
       kv_push(Channel *, subscribed, channel);
     }
   });
 
   if (!kv_size(subscribed)) {
-    msgpack_rpc_free_object(data);
+    msgpack_rpc_free_object(arg);
     goto end;
   }
 
-  WBuffer *buffer = serialize_event(type, data);
+  WBuffer *buffer = serialize_message(2, 0, name, arg);
 
   for (size_t i = 0; i < kv_size(subscribed); i++) {
     channel_write(kv_A(subscribed, i), buffer);
@@ -364,17 +368,20 @@ static void close_cb(uv_handle_t *handle)
   free(handle);
 }
 
-static WBuffer *serialize_event(char *type, Object data)
+static WBuffer *serialize_message(int type,
+                                  uint64_t id,
+                                  char *method,
+                                  Object arg)
 {
-  String event_type = {.size = strnlen(type, EVENT_MAXLEN), .data = type};
+  String method_str = {.size = strnlen(method, METHOD_MAXLEN), .data = method};
   msgpack_packer packer;
   msgpack_packer_init(&packer, &msgpack_event_buffer, msgpack_sbuffer_write);
-  msgpack_rpc_notification(event_type, data, &packer);
+  msgpack_rpc_message(type, id, method_str, arg, &packer);
   WBuffer *rv = wstream_new_buffer(xmemdup(msgpack_event_buffer.data,
                                            msgpack_event_buffer.size),
                                    msgpack_event_buffer.size,
                                    free);
-  msgpack_rpc_free_object(data);
+  msgpack_rpc_free_object(arg);
   msgpack_sbuffer_clear(&msgpack_event_buffer);
 
   return rv;
@@ -391,3 +398,4 @@ static Channel *register_channel()
   pmap_put(uint64_t)(channels, rv->id, rv);
   return rv;
 }
+

src/nvim/os/channel.h

@@ -6,7 +6,7 @@
 #include "nvim/api/private/defs.h"
 #include "nvim/vim.h"
 
-#define EVENT_MAXLEN 512
+#define METHOD_MAXLEN 512
 
 #ifdef INCLUDE_GENERATED_DECLARATIONS
 # include "os/channel.h.generated.h"

src/nvim/os/msgpack_rpc.c

@@ -115,13 +115,22 @@ void msgpack_rpc_call(uint64_t id, msgpack_object *req, msgpack_packer *res)
   msgpack_rpc_dispatch(id, req, res);
 }
 
-void msgpack_rpc_notification(String type, Object data, msgpack_packer *pac)
+void msgpack_rpc_message(int type,
+                         uint64_t id,
+                         String method,
+                         Object arg,
+                         msgpack_packer *pac)
 {
-  msgpack_pack_array(pac, 3);
-  msgpack_pack_int(pac, 2);
-  msgpack_pack_raw(pac, type.size);
-  msgpack_pack_raw_body(pac, type.data, type.size);
-  msgpack_rpc_from_object(data, pac);
+  msgpack_pack_array(pac, id ? 4 : 3);
+  msgpack_pack_int(pac, type);
+
+  if (id) {
+    msgpack_pack_uint64(pac, id);
+  }
+
+  msgpack_pack_raw(pac, method.size);
+  msgpack_pack_raw_body(pac, method.data, method.size);
+  msgpack_rpc_from_object(arg, pac);
 }
 
 void msgpack_rpc_error(char *msg, msgpack_packer *res)

src/nvim/os/msgpack_rpc.h

@@ -24,13 +24,20 @@ typedef enum {
 void msgpack_rpc_call(uint64_t id, msgpack_object *req, msgpack_packer *res)
   FUNC_ATTR_NONNULL_ARG(2) FUNC_ATTR_NONNULL_ARG(3);
 
-/// Packs a notification message
+/// Packs a message
 ///
-/// @param type The message type, an arbitrary string
-/// @param data The notification data
+/// @param type The message type
+/// @param id The message id, must be an unique integer > 0 or will be
+///        ignored(the message array will have 3 elements instead of 4).
+/// @param method The message name, an arbitrary string
+/// @param arg The message argument
 /// @param packer Where the notification will be packed to
-void msgpack_rpc_notification(String type, Object data, msgpack_packer *pac)
-  FUNC_ATTR_NONNULL_ARG(3);
+void msgpack_rpc_message(int type,
+                         uint64_t id,
+                         String method,
+                         Object arg,
+                         msgpack_packer *pac)
+  FUNC_ATTR_NONNULL_ARG(5);
 
 /// Dispatches to the actual API function after basic payload validation by
 /// `msgpack_rpc_call`. It is responsible for validating/converting arguments