API: Events: Implement channel_send_event and vimscript wrapper

This function can be used to send arbitrary objects via the API channel back to
connected clients, identified by channel id.

src/nvim/eval.c

@@ -12,7 +12,6 @@
 
 #include <string.h>
 #include <stdlib.h>
-
 #include "nvim/vim.h"
 #include "nvim/eval.h"
 #include "nvim/buffer.h"
@@ -67,6 +66,7 @@
 #include "nvim/os/rstream.h"
 #include "nvim/os/rstream_defs.h"
 #include "nvim/os/time.h"
+#include "nvim/os/channel.h"
 
 #if defined(FEAT_FLOAT)
 # include <math.h>
@@ -695,6 +695,7 @@ static void f_searchdecl(typval_T *argvars, typval_T *rettv);
 static void f_searchpair(typval_T *argvars, typval_T *rettv);
 static void f_searchpairpos(typval_T *argvars, typval_T *rettv);
 static void f_searchpos(typval_T *argvars, typval_T *rettv);
+static void f_send_event(typval_T *argvars, typval_T *rettv);
 static void f_setbufvar(typval_T *argvars, typval_T *rettv);
 static void f_setcmdpos(typval_T *argvars, typval_T *rettv);
 static void f_setline(typval_T *argvars, typval_T *rettv);
@@ -6944,6 +6945,7 @@ static struct fst {
   {"searchpair",      3, 7, f_searchpair},
   {"searchpairpos",   3, 7, f_searchpairpos},
   {"searchpos",       1, 4, f_searchpos},
+  {"send_event",      3, 3, f_send_event},
   {"setbufvar",       3, 3, f_setbufvar},
   {"setcmdpos",       1, 1, f_setcmdpos},
   {"setline",         2, 2, f_setline},
@@ -13057,6 +13059,36 @@ do_searchpair (
   return retval;
 }
 
+// "send_event()" function
+static void f_send_event(typval_T *argvars, typval_T *rettv)
+{
+  rettv->v_type = VAR_NUMBER;
+  rettv->vval.v_number = 0;
+
+  if (check_restricted() || check_secure()) {
+    return;
+  }
+
+  if (argvars[0].v_type != VAR_NUMBER || argvars[0].vval.v_number <= 0) {
+    EMSG2(_(e_invarg2), "Channel id must be a positive integer");
+    return;
+  }
+
+  if (argvars[1].v_type != VAR_STRING) {
+    EMSG2(_(e_invarg2), "Event type must be a string");
+    return;
+  }
+
+  if (!channel_send_event((uint64_t)argvars[0].vval.v_number,
+                          (char *)argvars[1].vval.v_string,
+                          &argvars[2])) {
+    EMSG2(_(e_invarg2), "Channel doesn't exist");
+    return;
+  }
+
+  rettv->vval.v_number = 1;
+}
+
 /*
  * "searchpos()" function
  */

src/nvim/os/channel.c

@@ -3,6 +3,7 @@
 #include <uv.h>
 #include <msgpack.h>
 
+#include "nvim/api/private/helpers.h"
 #include "nvim/os/channel.h"
 #include "nvim/os/channel_defs.h"
 #include "nvim/os/rstream.h"
@@ -38,16 +39,19 @@ typedef struct {
 
 static uint64_t next_id = 1;
 static Map(uint64_t) *channels = NULL;
+static msgpack_sbuffer msgpack_event_buffer;
 
 static void on_job_stdout(RStream *rstream, void *data, bool eof);
 static void on_job_stderr(RStream *rstream, void *data, bool eof);
 static void parse_msgpack(RStream *rstream, void *data, bool eof);
+static void send_msgpack(Channel *channel, String type, Object data);
 static void close_channel(Channel *channel);
 static void close_cb(uv_handle_t *handle);
 
 void channel_init()
 {
   channels = map_new(uint64_t)();
+  msgpack_sbuffer_init(&msgpack_event_buffer);
 }
 
 void channel_teardown()
@@ -117,6 +121,30 @@ void channel_from_stream(uv_stream_t *stream, ChannelProtocol prot)
   map_put(uint64_t)(channels, channel->id, channel);
 }
 
+bool channel_send_event(uint64_t id, char *type, typval_T *data)
+{
+  Channel *channel = map_get(uint64_t)(channels, id);
+
+  if (!channel) {
+    return false;
+  }
+
+  String event_type = {.size = strnlen(type, 1024), .data = type};
+  Object event_data = vim_to_object(data);
+
+  switch (channel->protocol) {
+    case kChannelProtocolMsgpack:
+      send_msgpack(channel, event_type, event_data);
+      break;
+    default:
+      abort();
+  }
+
+  msgpack_rpc_free_object(event_data);
+
+  return true;
+}
+
 static void on_job_stdout(RStream *rstream, void *data, bool eof)
 {
   Job *job = data;
@@ -169,6 +197,21 @@ static void parse_msgpack(RStream *rstream, void *data, bool eof)
   }
 }
 
+static void send_msgpack(Channel *channel, String type, Object data)
+{
+  msgpack_packer packer;
+  msgpack_packer_init(&packer, &msgpack_event_buffer, msgpack_sbuffer_write);
+  msgpack_rpc_notification(type, data, &packer);
+  char *bytes = xmemdup(msgpack_event_buffer.data, msgpack_event_buffer.size);
+
+  wstream_write(channel->data.streams.write,
+      bytes,
+      msgpack_event_buffer.size,
+      true);
+
+  msgpack_sbuffer_clear(&msgpack_event_buffer);
+}
+
 static void close_channel(Channel *channel)
 {
   map_del(uint64_t)(channels, channel->id);

src/nvim/os/channel.h

@@ -3,6 +3,7 @@
 
 #include <uv.h>
 
+#include "nvim/vim.h"
 #include "nvim/os/channel_defs.h"
 
 /// Initializes the module
@@ -25,5 +26,13 @@ void channel_from_stream(uv_stream_t *stream, ChannelProtocol prot);
 /// @param prot The rpc protocol used
 void channel_from_job(char **argv, ChannelProtocol prot);
 
+/// Sends event/data to channel
+///
+/// @param id The channel id
+/// @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);
+
 #endif  // NVIM_OS_CHANNEL_H
 

src/nvim/os/msgpack_rpc.c

@@ -113,6 +113,15 @@ 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)
+{
+  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);
+}
+
 void msgpack_rpc_error(char *msg, msgpack_packer *res)
 {
   size_t len = strlen(msg);

src/nvim/os/msgpack_rpc.h

@@ -16,6 +16,13 @@
 /// @param res A packer that contains the response
 void msgpack_rpc_call(uint64_t id, msgpack_object *req, msgpack_packer *res);
 
+/// Packs a notification message
+///
+/// @param type The message type, an arbitrary string
+/// @param data The notification data
+/// @param packer Where the notification will be packed to
+void msgpack_rpc_notification(String type, Object data, msgpack_packer *pac);
+
 /// Dispatches to the actual API function after basic payload validation by
 /// `msgpack_rpc_call`. It is responsible for validating/converting arguments
 /// to C types, and converting the return value back to msgpack types.