4 files changed, 68 insertions, 0 deletions
diff --git a/src/nvim/os/channel.c b/src/nvim/os/channel.c
index c103a71f46..4981a268b2 100644
--- a/src/nvim/os/channel.c
+++ b/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);
diff --git a/src/nvim/os/channel.h b/src/nvim/os/channel.h
index 4a3962575d..543b91dd89 100644
--- a/src/nvim/os/channel.h
+++ b/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
 
diff --git a/src/nvim/os/msgpack_rpc.c b/src/nvim/os/msgpack_rpc.c
index 62d68ebdec..423c5d584d 100644
--- a/src/nvim/os/msgpack_rpc.c
+++ b/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);
diff --git a/src/nvim/os/msgpack_rpc.h b/src/nvim/os/msgpack_rpc.h
index bc216c0856..319a069df7 100644
--- a/src/nvim/os/msgpack_rpc.h
+++ b/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.