API: Events: Add functions for {un}subscribing to broadcasted events

src/nvim/api/vim.c

@@ -7,6 +7,7 @@
 #include "nvim/api/private/helpers.h"
 #include "nvim/api/private/defs.h"
 #include "nvim/api/buffer.h"
+#include "nvim/os/channel.h"
 #include "nvim/vim.h"
 #include "nvim/buffer.h"
 #include "nvim/window.h"
@@ -327,6 +328,24 @@ void vim_set_current_tabpage(Tabpage tabpage, Error *err)
   try_end(err);
 }
 
+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];
+  memcpy(e, event.data, length);
+  e[length] = NUL;
+  channel_subscribe(channel_id, e);
+}
+
+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];
+  memcpy(e, event.data, length);
+  e[length] = NUL;
+  channel_unsubscribe(channel_id, e);
+}
+
 static void write_msg(String message, bool to_err)
 {
   static int pos = 0;

src/nvim/api/vim.h

@@ -155,5 +155,17 @@ Tabpage vim_get_current_tabpage(void);
 /// @param[out] err Details of an error that may have occurred
 void vim_set_current_tabpage(Tabpage tabpage, Error *err);
 
+/// Subscribes to event broadcasts
+///
+/// @param channel_id The channel id(passed automatically by the dispatcher)
+/// @param event The event type string
+void vim_subscribe(uint64_t channel_id, String event);
+
+/// Unsubscribes to event broadcasts
+///
+/// @param channel_id The channel id(passed automatically by the dispatcher)
+/// @param event The event type string
+void vim_unsubscribe(uint64_t channel_id, String event);
+
 #endif  // NVIM_API_VIM_H
 

src/nvim/os/channel.c

@@ -35,6 +35,7 @@ typedef struct {
 
 static uint64_t next_id = 1;
 static Map(uint64_t) *channels = NULL;
+static Map(cstr_t) *event_strings = NULL;
 static msgpack_sbuffer msgpack_event_buffer;
 
 static void on_job_stdout(RStream *rstream, void *data, bool eof);
@@ -42,12 +43,14 @@ static void on_job_stderr(RStream *rstream, void *data, bool eof);
 static void parse_msgpack(RStream *rstream, void *data, bool eof);
 static void send_event(Channel *channel, char *type, typval_T *data);
 static void broadcast_event(char *type, typval_T *data);
+static void unsubscribe(Channel *channel, char *event);
 static void close_channel(Channel *channel);
 static void close_cb(uv_handle_t *handle);
 
 void channel_init()
 {
   channels = map_new(uint64_t)();
+  event_strings = map_new(cstr_t)();
   msgpack_sbuffer_init(&msgpack_event_buffer);
 }
 
@@ -116,6 +119,35 @@ bool channel_send_event(uint64_t id, char *type, typval_T *data)
   return true;
 }
 
+void channel_subscribe(uint64_t id, char *event)
+{
+  Channel *channel;
+
+  if (!(channel = map_get(uint64_t)(channels, id))) {
+    return;
+  }
+
+  char *event_string = map_get(cstr_t)(event_strings, event);
+
+  if (!event_string) {
+    event_string = xstrdup(event);
+    map_put(cstr_t)(event_strings, event_string, event_string);
+  }
+
+  map_put(cstr_t)(channel->subscribed_events, event_string, event_string);
+}
+
+void channel_unsubscribe(uint64_t id, char *event)
+{
+  Channel *channel;
+
+  if (!(channel = map_get(uint64_t)(channels, id))) {
+    return;
+  }
+
+  unsubscribe(channel, event);
+}
+
 static void on_job_stdout(RStream *rstream, void *data, bool eof)
 {
   Job *job = data;
@@ -165,7 +197,7 @@ static void parse_msgpack(RStream *rstream, void *data, bool eof)
 
 static void send_event(Channel *channel, char *type, typval_T *data)
 {
-  String event_type = {.size = strnlen(type, 1024), .data = type};
+  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);
@@ -216,6 +248,22 @@ end:
   kv_destroy(subscribed);
 }
 
+static void unsubscribe(Channel *channel, char *event)
+{
+  char *event_string = map_get(cstr_t)(event_strings, event);
+  map_del(cstr_t)(channel->subscribed_events, event_string);
+
+  map_foreach_value(channels, channel, {
+    if (map_has(cstr_t)(channel->subscribed_events, event_string)) {
+      return;
+    }
+  });
+
+  // Since the string is no longer used by other channels, release it's memory
+  map_del(cstr_t)(event_strings, event_string);
+  free(event_string);
+}
+
 static void close_channel(Channel *channel)
 {
   map_del(uint64_t)(channels, channel->id);
@@ -230,6 +278,12 @@ static void close_channel(Channel *channel)
     uv_close((uv_handle_t *)channel->data.streams.uv, close_cb);
   }
 
+  // Unsubscribe from all events
+  char *event_string;
+  map_foreach_value(channel->subscribed_events, event_string, {
+    unsubscribe(channel, event_string);
+  });
+
   map_free(cstr_t)(channel->subscribed_events);
   free(channel);
 }

src/nvim/os/channel.h

@@ -5,6 +5,8 @@
 
 #include "nvim/vim.h"
 
+#define EVENT_MAXLEN 512
+
 /// Initializes the module
 void channel_init(void);
 
@@ -32,5 +34,17 @@ void channel_from_job(char **argv);
 /// @return True if the data was sent successfully, false otherwise.
 bool channel_send_event(uint64_t id, char *type, typval_T *data);
 
+/// Subscribes to event broadcasts
+///
+/// @param id The channel id
+/// @param event The event type string
+void channel_subscribe(uint64_t id, char *event);
+
+/// Unsubscribes to event broadcasts
+///
+/// @param id The channel id
+/// @param event The event type string
+void channel_unsubscribe(uint64_t id, char *event);
+
 #endif  // NVIM_OS_CHANNEL_H