21 files changed, 274 insertions, 315 deletions

diff --git a/src/nvim/event/defs.h b/src/nvim/event/defs.h
index 571f61dfdb..9b7d8708be 100644
--- a/src/nvim/event/defs.h
+++ b/src/nvim/event/defs.h
@@ -2,34 +2,155 @@
 
 #include <assert.h>
 #include <stdarg.h>
+#include <stdbool.h>
+#include <uv.h>
 
-#define EVENT_HANDLER_MAX_ARGC 10
+#include "nvim/eval/typval_defs.h"
+#include "nvim/rbuffer_defs.h"
+#include "nvim/types_defs.h"
+
+enum { EVENT_HANDLER_MAX_ARGC = 10, };
 
 typedef void (*argv_callback)(void **argv);
-typedef struct message {
+typedef struct {
   argv_callback handler;
   void *argv[EVENT_HANDLER_MAX_ARGC];
 } Event;
-typedef void (*event_scheduler)(Event event, void *data);
-
-#define VA_EVENT_INIT(event, h, a) \
-  do { \
-    assert(a <= EVENT_HANDLER_MAX_ARGC); \
-    (event)->handler = h; \
-    if (a) { \
-      va_list args; \
-      va_start(args, a); \
-      for (int i = 0; i < a; i++) { \
-        (event)->argv[i] = va_arg(args, void *); \
-      } \
-      va_end(args); \
-    } \
-  } while (0)
-
-static inline Event event_create(argv_callback cb, int argc, ...)
-{
-  assert(argc <= EVENT_HANDLER_MAX_ARGC);
-  Event event;
-  VA_EVENT_INIT(&event, cb, argc);
-  return event;
-}
+
+#define event_create(cb, ...) ((Event){ .handler = cb, .argv = { __VA_ARGS__ } })
+
+typedef struct multiqueue MultiQueue;
+typedef void (*PutCallback)(MultiQueue *multiq, void *data);
+
+typedef struct signal_watcher SignalWatcher;
+typedef void (*signal_cb)(SignalWatcher *watcher, int signum, void *data);
+typedef void (*signal_close_cb)(SignalWatcher *watcher, void *data);
+
+struct signal_watcher {
+  uv_signal_t uv;
+  void *data;
+  signal_cb cb;
+  signal_close_cb close_cb;
+  MultiQueue *events;
+};
+
+typedef struct time_watcher TimeWatcher;
+typedef void (*time_cb)(TimeWatcher *watcher, void *data);
+
+struct time_watcher {
+  uv_timer_t uv;
+  void *data;
+  time_cb cb, close_cb;
+  MultiQueue *events;
+  bool blockable;
+};
+
+typedef struct wbuffer WBuffer;
+typedef void (*wbuffer_data_finalizer)(void *data);
+
+struct wbuffer {
+  size_t size, refcount;
+  char *data;
+  wbuffer_data_finalizer cb;
+};
+
+typedef struct stream Stream;
+/// Type of function called when the Stream buffer is filled with data
+///
+/// @param stream The Stream instance
+/// @param buf The associated RBuffer instance
+/// @param count Number of bytes that was read.
+/// @param data User-defined data
+/// @param eof If the stream reached EOF.
+typedef void (*stream_read_cb)(Stream *stream, RBuffer *buf, size_t count, void *data, bool eof);
+
+/// Type of function called when the Stream has information about a write
+/// request.
+///
+/// @param stream The Stream instance
+/// @param data User-defined data
+/// @param status 0 on success, anything else indicates failure
+typedef void (*stream_write_cb)(Stream *stream, void *data, int status);
+typedef void (*stream_close_cb)(Stream *stream, void *data);
+
+struct stream {
+  bool closed;
+  bool did_eof;
+  union {
+    uv_pipe_t pipe;
+    uv_tcp_t tcp;
+    uv_idle_t idle;
+#ifdef MSWIN
+    uv_tty_t tty;
+#endif
+  } uv;
+  uv_stream_t *uvstream;
+  uv_buf_t uvbuf;
+  RBuffer *buffer;
+  uv_file fd;
+  stream_read_cb read_cb;
+  stream_write_cb write_cb;
+  void *cb_data;
+  stream_close_cb close_cb, internal_close_cb;
+  void *close_cb_data, *internal_data;
+  size_t fpos;
+  size_t curmem;
+  size_t maxmem;
+  size_t pending_reqs;
+  size_t num_bytes;
+  MultiQueue *events;
+};
+
+#define ADDRESS_MAX_SIZE 256
+
+typedef struct socket_watcher SocketWatcher;
+typedef void (*socket_cb)(SocketWatcher *watcher, int result, void *data);
+typedef void (*socket_close_cb)(SocketWatcher *watcher, void *data);
+
+struct socket_watcher {
+  // Pipe/socket path, or TCP address string
+  char addr[ADDRESS_MAX_SIZE];
+  // TCP server or unix socket (named pipe on Windows)
+  union {
+    struct {
+      uv_tcp_t handle;
+      struct addrinfo *addrinfo;
+    } tcp;
+    struct {
+      uv_pipe_t handle;
+    } pipe;
+  } uv;
+  uv_stream_t *stream;
+  void *data;
+  socket_cb cb;
+  socket_close_cb close_cb;
+  MultiQueue *events;
+};
+
+typedef enum {
+  kProcessTypeUv,
+  kProcessTypePty,
+} ProcessType;
+
+typedef struct process Process;
+typedef void (*process_exit_cb)(Process *proc, int status, void *data);
+typedef void (*internal_process_cb)(Process *proc);
+
+struct process {
+  ProcessType type;
+  Loop *loop;
+  void *data;
+  int pid, status, refcount;
+  uint8_t exit_signal;  // Signal used when killing (on Windows).
+  uint64_t stopped_time;  // process_stop() timestamp
+  const char *cwd;
+  char **argv;
+  const char *exepath;
+  dict_T *env;
+  Stream in, out, err;
+  /// Exit handler. If set, user must call process_free().
+  process_exit_cb cb;
+  internal_process_cb internal_exit_cb, internal_close_cb;
+  bool closed, detach, overlapped, fwd_err;
+  MultiQueue *events;
+};
diff --git a/src/nvim/event/libuv_process.c b/src/nvim/event/libuv_process.c
index be48b39af1..a7966994e0 100644
--- a/src/nvim/event/libuv_process.c
+++ b/src/nvim/event/libuv_process.c
@@ -4,12 +4,14 @@
 #include <uv.h>
 
 #include "nvim/eval/typval.h"
+#include "nvim/event/defs.h"
 #include "nvim/event/libuv_process.h"
+#include "nvim/event/loop.h"
 #include "nvim/event/process.h"
-#include "nvim/event/stream.h"
-#include "nvim/func_attr.h"
 #include "nvim/log.h"
 #include "nvim/os/os.h"
+#include "nvim/os/os_defs.h"
+#include "nvim/types_defs.h"
 #include "nvim/ui_client.h"
 
 #ifdef INCLUDE_GENERATED_DECLARATIONS
@@ -127,3 +129,11 @@ static void exit_cb(uv_process_t *handle, int64_t status, int term_signal)
   proc->status = term_signal ? 128 + term_signal : (int)status;
   proc->internal_exit_cb(proc);
 }
+
+LibuvProcess libuv_process_init(Loop *loop, void *data)
+{
+  LibuvProcess rv = {
+    .process = process_init(loop, kProcessTypeUv, data)
+  };
+  return rv;
+}
diff --git a/src/nvim/event/libuv_process.h b/src/nvim/event/libuv_process.h
index e3e2bfeb76..12401dbb35 100644
--- a/src/nvim/event/libuv_process.h
+++ b/src/nvim/event/libuv_process.h
@@ -2,24 +2,15 @@
 
 #include <uv.h>
 
-#include "nvim/event/loop.h"
-#include "nvim/event/process.h"
+#include "nvim/event/defs.h"
 
-typedef struct libuv_process {
+typedef struct {
   Process process;
   uv_process_t uv;
   uv_process_options_t uvopts;
   uv_stdio_container_t uvstdio[4];
 } LibuvProcess;
 
-static inline LibuvProcess libuv_process_init(Loop *loop, void *data)
-{
-  LibuvProcess rv = {
-    .process = process_init(loop, kProcessTypeUv, data)
-  };
-  return rv;
-}
-
 #ifdef INCLUDE_GENERATED_DECLARATIONS
 # include "event/libuv_process.h.generated.h"
 #endif
diff --git a/src/nvim/event/loop.c b/src/nvim/event/loop.c
index d61666e6d4..93948d3eaa 100644
--- a/src/nvim/event/loop.c
+++ b/src/nvim/event/loop.c
@@ -3,11 +3,12 @@
 #include <stdlib.h>
 #include <uv.h>
 
-#include "nvim/event/defs.h"
 #include "nvim/event/loop.h"
+#include "nvim/event/multiqueue.h"
 #include "nvim/log.h"
 #include "nvim/memory.h"
 #include "nvim/os/time.h"
+#include "nvim/types_defs.h"
 
 #ifdef INCLUDE_GENERATED_DECLARATIONS
 # include "event/loop.c.generated.h"
@@ -111,7 +112,7 @@ void loop_schedule_deferred(Loop *loop, Event event)
 {
   Event *eventp = xmalloc(sizeof(*eventp));
   *eventp = event;
-  loop_schedule_fast(loop, event_create(loop_deferred_event, 2, loop, eventp));
+  loop_schedule_fast(loop, event_create(loop_deferred_event, loop, eventp));
 }
 static void loop_deferred_event(void **argv)
 {
diff --git a/src/nvim/event/loop.h b/src/nvim/event/loop.h
index 5665332e95..6ecc7cb781 100644
--- a/src/nvim/event/loop.h
+++ b/src/nvim/event/loop.h
@@ -1,19 +1,18 @@
 #pragma once
 
 #include <stdbool.h>
-#include <stdint.h>
 #include <uv.h>
 
 #include "klib/klist.h"
-#include "nvim/event/multiqueue.h"
-#include "nvim/os/time.h"
+#include "nvim/event/defs.h"  // IWYU pragma: keep
+#include "nvim/types_defs.h"  // IWYU pragma: keep
 
 typedef void *WatcherPtr;
 
 #define _NOOP(x)
 KLIST_INIT(WatcherPtr, WatcherPtr, _NOOP)
 
-typedef struct loop {
+struct loop {
   uv_loop_t uv;
   MultiQueue *events;
   MultiQueue *thread_events;
@@ -42,46 +41,7 @@ typedef struct loop {
   uv_mutex_t mutex;
   int recursive;
   bool closing;  ///< Set to true if loop_close() has been called
-} Loop;
-
-#define CREATE_EVENT(multiqueue, handler, argc, ...) \
-  do { \
-    if (multiqueue) { \
-      multiqueue_put((multiqueue), (handler), argc, __VA_ARGS__); \
-    } else { \
-      void *argv[argc] = { __VA_ARGS__ }; \
-      (handler)(argv); \
-    } \
-  } while (0)
-
-// Poll for events until a condition or timeout
-#define LOOP_PROCESS_EVENTS_UNTIL(loop, multiqueue, timeout, condition) \
-  do { \
-    int64_t remaining = timeout; \
-    uint64_t before = (remaining > 0) ? os_hrtime() : 0; \
-    while (!(condition)) { \
-      LOOP_PROCESS_EVENTS(loop, multiqueue, remaining); \
-      if (remaining == 0) { \
-        break; \
-      } else if (remaining > 0) { \
-        uint64_t now = os_hrtime(); \
-        remaining -= (int64_t)((now - before) / 1000000); \
-        before = now; \
-        if (remaining <= 0) { \
-          break; \
-        } \
-      } \
-    } \
-  } while (0)
-
-#define LOOP_PROCESS_EVENTS(loop, multiqueue, timeout) \
-  do { \
-    if (multiqueue && !multiqueue_empty(multiqueue)) { \
-      multiqueue_process_events(multiqueue); \
-    } else { \
-      loop_poll_events(loop, timeout); \
-    } \
-  } while (0)
+};
 
 #ifdef INCLUDE_GENERATED_DECLARATIONS
 # include "event/loop.h.generated.h"
diff --git a/src/nvim/event/multiqueue.c b/src/nvim/event/multiqueue.c
index 3ab41bd299..8646173776 100644
--- a/src/nvim/event/multiqueue.c
+++ b/src/nvim/event/multiqueue.c
@@ -48,8 +48,7 @@
 
 #include "nvim/event/defs.h"
 #include "nvim/event/multiqueue.h"
-#include "nvim/func_attr.h"
-#include "nvim/lib/queue.h"
+#include "nvim/lib/queue_defs.h"
 #include "nvim/memory.h"
 
 typedef struct multiqueue_item MultiQueueItem;
@@ -157,7 +156,7 @@ void multiqueue_purge_events(MultiQueue *self)
 {
   assert(self);
   while (!multiqueue_empty(self)) {
-    (void)multiqueue_remove(self);
+    multiqueue_remove(self);
   }
 }
 
@@ -261,7 +260,7 @@ Event event_create_oneshot(Event ev, int num)
   data->event = ev;
   data->fired = false;
   data->refcount = num;
-  return event_create(multiqueue_oneshot_event, 1, data);
+  return event_create(multiqueue_oneshot_event, data);
 }
 static void multiqueue_oneshot_event(void **argv)
 {
diff --git a/src/nvim/event/multiqueue.h b/src/nvim/event/multiqueue.h
index e01ee1e710..24aaa34ef7 100644
--- a/src/nvim/event/multiqueue.h
+++ b/src/nvim/event/multiqueue.h
@@ -1,16 +1,54 @@
 #pragma once
 
-#include <uv.h>
+#include <stddef.h>  // IWYU pragma: keep
 
-#include "nvim/event/defs.h"
-#include "nvim/lib/queue.h"
-
-typedef struct multiqueue MultiQueue;
-typedef void (*PutCallback)(MultiQueue *multiq, void *data);
-
-#define multiqueue_put(q, h, ...) \
-  multiqueue_put_event(q, event_create(h, __VA_ARGS__));
+#include "nvim/event/defs.h"  // IWYU pragma: keep
+#include "nvim/os/time.h"  // IWYU pragma: keep
 
 #ifdef INCLUDE_GENERATED_DECLARATIONS
 # include "event/multiqueue.h.generated.h"
 #endif
+
+#define multiqueue_put(q, h, ...) \
+  do { \
+    multiqueue_put_event(q, event_create(h, __VA_ARGS__)); \
+  } while (0)
+
+#define CREATE_EVENT(multiqueue, handler, ...) \
+  do { \
+    if (multiqueue) { \
+      multiqueue_put((multiqueue), (handler), __VA_ARGS__); \
+    } else { \
+      void *argv[] = { __VA_ARGS__ }; \
+      (handler)(argv); \
+    } \
+  } while (0)
+
+// Poll for events until a condition or timeout
+#define LOOP_PROCESS_EVENTS_UNTIL(loop, multiqueue, timeout, condition) \
+  do { \
+    int64_t remaining = timeout; \
+    uint64_t before = (remaining > 0) ? os_hrtime() : 0; \
+    while (!(condition)) { \
+      LOOP_PROCESS_EVENTS(loop, multiqueue, remaining); \
+      if (remaining == 0) { \
+        break; \
+      } else if (remaining > 0) { \
+        uint64_t now = os_hrtime(); \
+        remaining -= (int64_t)((now - before) / 1000000); \
+        before = now; \
+        if (remaining <= 0) { \
+          break; \
+        } \
+      } \
+    } \
+  } while (0)
+
+#define LOOP_PROCESS_EVENTS(loop, multiqueue, timeout) \
+  do { \
+    if (multiqueue && !multiqueue_empty(multiqueue)) { \
+      multiqueue_process_events(multiqueue); \
+    } else { \
+      loop_poll_events(loop, timeout); \
+    } \
+  } while (0)
diff --git a/src/nvim/event/process.c b/src/nvim/event/process.c
index 864fc2c1d8..7460e92766 100644
--- a/src/nvim/event/process.c
+++ b/src/nvim/event/process.c
@@ -6,8 +6,9 @@
 #include "klib/klist.h"
 #include "nvim/event/libuv_process.h"
 #include "nvim/event/loop.h"
+#include "nvim/event/multiqueue.h"
 #include "nvim/event/process.h"
-#include "nvim/func_attr.h"
+#include "nvim/event/stream.h"
 #include "nvim/globals.h"
 #include "nvim/log.h"
 #include "nvim/main.h"
@@ -15,7 +16,7 @@
 #include "nvim/os/pty_process.h"
 #include "nvim/os/shell.h"
 #include "nvim/os/time.h"
-#include "nvim/rbuffer.h"
+#include "nvim/rbuffer_defs.h"
 #include "nvim/ui_client.h"
 
 #ifdef INCLUDE_GENERATED_DECLARATIONS
@@ -133,7 +134,7 @@ void process_teardown(Loop *loop) FUNC_ATTR_NONNULL_ALL
     Process *proc = (*current)->data;
     if (proc->detach || proc->type == kProcessTypePty) {
       // Close handles to process without killing it.
-      CREATE_EVENT(loop->events, process_close_handles, 1, proc);
+      CREATE_EVENT(loop->events, process_close_handles, proc);
     } else {
       process_stop(proc);
     }
@@ -301,7 +302,7 @@ static void decref(Process *proc)
   }
   assert(node);
   kl_shift_at(WatcherPtr, loop->children, node);
-  CREATE_EVENT(proc->events, process_close_event, 1, proc);
+  CREATE_EVENT(proc->events, process_close_event, proc);
 }
 
 static void process_close(Process *proc)
@@ -396,7 +397,7 @@ static void process_close_handles(void **argv)
 static void exit_delay_cb(uv_timer_t *handle)
 {
   uv_timer_stop(&main_loop.exit_delay_timer);
-  multiqueue_put(main_loop.fast_events, exit_event, 1, main_loop.exit_delay_timer.data);
+  multiqueue_put(main_loop.fast_events, exit_event, main_loop.exit_delay_timer.data);
 }
 
 static void exit_event(void **argv)
@@ -421,7 +422,7 @@ static void exit_event(void **argv)
 
 void exit_from_channel(int status)
 {
-  multiqueue_put(main_loop.fast_events, exit_event, 1, status);
+  multiqueue_put(main_loop.fast_events, exit_event, (void *)(intptr_t)status);
 }
 
 static void on_process_exit(Process *proc)
@@ -439,7 +440,7 @@ static void on_process_exit(Process *proc)
   // more data directly. Instead delay the reading after the libuv loop by
   // queueing process_close_handles() as an event.
   MultiQueue *queue = proc->events ? proc->events : loop->events;
-  CREATE_EVENT(queue, process_close_handles, 1, proc);
+  CREATE_EVENT(queue, process_close_handles, proc);
 }
 
 static void on_process_stream_close(Stream *stream, void *data)
diff --git a/src/nvim/event/process.h b/src/nvim/event/process.h
index 234fc815af..421a470244 100644
--- a/src/nvim/event/process.h
+++ b/src/nvim/event/process.h
@@ -2,44 +2,9 @@
 
 #include <stdbool.h>
 #include <stddef.h>
-#include <stdint.h>
 
-#include "nvim/eval/typval_defs.h"
-#include "nvim/event/loop.h"
-#include "nvim/event/multiqueue.h"
-#include "nvim/event/rstream.h"
-#include "nvim/event/stream.h"
-#include "nvim/event/wstream.h"
-
-struct process;
-
-typedef enum {
-  kProcessTypeUv,
-  kProcessTypePty,
-} ProcessType;
-
-typedef struct process Process;
-typedef void (*process_exit_cb)(Process *proc, int status, void *data);
-typedef void (*internal_process_cb)(Process *proc);
-
-struct process {
-  ProcessType type;
-  Loop *loop;
-  void *data;
-  int pid, status, refcount;
-  uint8_t exit_signal;  // Signal used when killing (on Windows).
-  uint64_t stopped_time;  // process_stop() timestamp
-  const char *cwd;
-  char **argv;
-  const char *exepath;
-  dict_T *env;
-  Stream in, out, err;
-  /// Exit handler. If set, user must call process_free().
-  process_exit_cb cb;
-  internal_process_cb internal_exit_cb, internal_close_cb;
-  bool closed, detach, overlapped, fwd_err;
-  MultiQueue *events;
-};
+#include "nvim/event/defs.h"  // IWYU pragma: keep
+#include "nvim/types_defs.h"
 
 static inline Process process_init(Loop *loop, ProcessType type, void *data)
 {
diff --git a/src/nvim/event/rstream.c b/src/nvim/event/rstream.c
index 73828a2271..f50c8a0e5a 100644
--- a/src/nvim/event/rstream.c
+++ b/src/nvim/event/rstream.c
@@ -2,18 +2,18 @@
 #include <stdbool.h>
 #include <stddef.h>
 #include <stdint.h>
-#include <sys/types.h>
 #include <uv.h>
 
-#include "nvim/event/loop.h"
+#include "nvim/event/multiqueue.h"
 #include "nvim/event/rstream.h"
 #include "nvim/event/stream.h"
-#include "nvim/func_attr.h"
 #include "nvim/log.h"
 #include "nvim/macros_defs.h"
 #include "nvim/main.h"
 #include "nvim/os/os_defs.h"
 #include "nvim/rbuffer.h"
+#include "nvim/rbuffer_defs.h"
+#include "nvim/types_defs.h"
 
 #ifdef INCLUDE_GENERATED_DECLARATIONS
 # include "event/rstream.c.generated.h"
@@ -201,10 +201,6 @@ static void invoke_read_cb(Stream *stream, size_t count, bool eof)
   // Don't let the stream be closed before the event is processed.
   stream->pending_reqs++;
 
-  CREATE_EVENT(stream->events,
-               read_event,
-               3,
-               stream,
-               (void *)(uintptr_t *)count,
-               (void *)(uintptr_t)eof);
+  CREATE_EVENT(stream->events, read_event,
+               stream, (void *)(uintptr_t *)count, (void *)(uintptr_t)eof);
 }
diff --git a/src/nvim/event/rstream.h b/src/nvim/event/rstream.h
index b2a62acf83..4e2893bf88 100644
--- a/src/nvim/event/rstream.h
+++ b/src/nvim/event/rstream.h
@@ -1,11 +1,7 @@
 #pragma once
 
-#include <stdbool.h>
-#include <stddef.h>
-#include <uv.h>
-
-#include "nvim/event/loop.h"
-#include "nvim/event/stream.h"
+#include "nvim/event/defs.h"  // IWYU pragma: keep
+#include "nvim/types_defs.h"  // IWYU pragma: keep
 
 #ifdef INCLUDE_GENERATED_DECLARATIONS
 # include "event/rstream.h.generated.h"
diff --git a/src/nvim/event/signal.c b/src/nvim/event/signal.c
index e64d526856..57241e79b2 100644
--- a/src/nvim/event/signal.c
+++ b/src/nvim/event/signal.c
@@ -1,9 +1,11 @@
 #include <stddef.h>
 #include <uv.h>
 
+#include "nvim/event/defs.h"
 #include "nvim/event/loop.h"
+#include "nvim/event/multiqueue.h"
 #include "nvim/event/signal.h"
-#include "nvim/func_attr.h"
+#include "nvim/types_defs.h"
 
 #ifdef INCLUDE_GENERATED_DECLARATIONS
 # include "event/signal.c.generated.h"
@@ -48,7 +50,7 @@ static void signal_event(void **argv)
 static void signal_watcher_cb(uv_signal_t *handle, int signum)
 {
   SignalWatcher *watcher = handle->data;
-  CREATE_EVENT(watcher->events, signal_event, 1, watcher);
+  CREATE_EVENT(watcher->events, signal_event, watcher);
 }
 
 static void close_cb(uv_handle_t *handle)
diff --git a/src/nvim/event/signal.h b/src/nvim/event/signal.h
index 946de1b4f0..16cd2be951 100644
--- a/src/nvim/event/signal.h
+++ b/src/nvim/event/signal.h
@@ -1,23 +1,7 @@
 #pragma once
 
-#include <uv.h>
-
-#include "nvim/event/loop.h"
-#include "nvim/event/multiqueue.h"
-
-struct signal_watcher;
-
-typedef struct signal_watcher SignalWatcher;
-typedef void (*signal_cb)(SignalWatcher *watcher, int signum, void *data);
-typedef void (*signal_close_cb)(SignalWatcher *watcher, void *data);
-
-struct signal_watcher {
-  uv_signal_t uv;
-  void *data;
-  signal_cb cb;
-  signal_close_cb close_cb;
-  MultiQueue *events;
-};
+#include "nvim/event/defs.h"  // IWYU pragma: keep
+#include "nvim/types_defs.h"  // IWYU pragma: keep
 
 #ifdef INCLUDE_GENERATED_DECLARATIONS
 # include "event/signal.h.generated.h"
diff --git a/src/nvim/event/socket.c b/src/nvim/event/socket.c
index e787e023f0..4e878a2ecf 100644
--- a/src/nvim/event/socket.c
+++ b/src/nvim/event/socket.c
@@ -7,17 +7,19 @@
 
 #include "nvim/ascii_defs.h"
 #include "nvim/charset.h"
+#include "nvim/event/defs.h"
 #include "nvim/event/loop.h"
+#include "nvim/event/multiqueue.h"
 #include "nvim/event/socket.h"
 #include "nvim/event/stream.h"
-#include "nvim/func_attr.h"
-#include "nvim/gettext.h"
+#include "nvim/gettext_defs.h"
 #include "nvim/log.h"
 #include "nvim/main.h"
 #include "nvim/memory.h"
 #include "nvim/os/fs.h"
-#include "nvim/os/os.h"
+#include "nvim/os/os_defs.h"
 #include "nvim/path.h"
+#include "nvim/types_defs.h"
 
 #ifdef INCLUDE_GENERATED_DECLARATIONS
 # include "event/socket.c.generated.h"
@@ -175,8 +177,7 @@ static void connection_event(void **argv)
 static void connection_cb(uv_stream_t *handle, int status)
 {
   SocketWatcher *watcher = handle->data;
-  CREATE_EVENT(watcher->events, connection_event, 2, watcher,
-               (void *)(uintptr_t)status);
+  CREATE_EVENT(watcher->events, connection_event, watcher, (void *)(uintptr_t)status);
 }
 
 static void close_cb(uv_handle_t *handle)
diff --git a/src/nvim/event/socket.h b/src/nvim/event/socket.h
index 504af3c7a8..64a77a9a06 100644
--- a/src/nvim/event/socket.h
+++ b/src/nvim/event/socket.h
@@ -1,39 +1,7 @@
 #pragma once
 
-#include <uv.h>
-
-#include "nvim/event/loop.h"
-#include "nvim/event/multiqueue.h"
-#include "nvim/event/rstream.h"
-#include "nvim/event/wstream.h"
-
-struct socket_watcher;
-
-#define ADDRESS_MAX_SIZE 256
-
-typedef struct socket_watcher SocketWatcher;
-typedef void (*socket_cb)(SocketWatcher *watcher, int result, void *data);
-typedef void (*socket_close_cb)(SocketWatcher *watcher, void *data);
-
-struct socket_watcher {
-  // Pipe/socket path, or TCP address string
-  char addr[ADDRESS_MAX_SIZE];
-  // TCP server or unix socket (named pipe on Windows)
-  union {
-    struct {
-      uv_tcp_t handle;
-      struct addrinfo *addrinfo;
-    } tcp;
-    struct {
-      uv_pipe_t handle;
-    } pipe;
-  } uv;
-  uv_stream_t *stream;
-  void *data;
-  socket_cb cb;
-  socket_close_cb close_cb;
-  MultiQueue *events;
-};
+#include "nvim/event/defs.h"  // IWYU pragma: keep
+#include "nvim/types_defs.h"  // IWYU pragma: keep
 
 #ifdef INCLUDE_GENERATED_DECLARATIONS
 # include "event/socket.h.generated.h"
diff --git a/src/nvim/event/stream.c b/src/nvim/event/stream.c
index aff116bad9..0b9ed4f25b 100644
--- a/src/nvim/event/stream.c
+++ b/src/nvim/event/stream.c
@@ -4,11 +4,12 @@
 #include <uv.h>
 #include <uv/version.h>
 
+#include "nvim/event/defs.h"
 #include "nvim/event/loop.h"
 #include "nvim/event/stream.h"
-#include "nvim/func_attr.h"
 #include "nvim/log.h"
 #include "nvim/rbuffer.h"
+#include "nvim/types_defs.h"
 #ifdef MSWIN
 # include "nvim/os/os_win_console.h"
 #endif
@@ -129,15 +130,22 @@ void stream_may_close(Stream *stream)
 void stream_close_handle(Stream *stream)
   FUNC_ATTR_NONNULL_ALL
 {
+  uv_handle_t *handle = NULL;
   if (stream->uvstream) {
     if (uv_stream_get_write_queue_size(stream->uvstream) > 0) {
       WLOG("closed Stream (%p) with %zu unwritten bytes",
            (void *)stream,
            uv_stream_get_write_queue_size(stream->uvstream));
     }
-    uv_close((uv_handle_t *)stream->uvstream, close_cb);
+    handle = (uv_handle_t *)stream->uvstream;
   } else {
-    uv_close((uv_handle_t *)&stream->uv.idle, close_cb);
+    handle = (uv_handle_t *)&stream->uv.idle;
+  }
+
+  assert(handle != NULL);
+
+  if (!uv_is_closing(handle)) {
+    uv_close(handle, close_cb);
   }
 }
 
diff --git a/src/nvim/event/stream.h b/src/nvim/event/stream.h
index d02707dc45..9bdfc421d6 100644
--- a/src/nvim/event/stream.h
+++ b/src/nvim/event/stream.h
@@ -1,61 +1,7 @@
 #pragma once
 
-#include <stdbool.h>
-#include <stddef.h>
-#include <uv.h>
-
-#include "nvim/event/loop.h"
-#include "nvim/event/multiqueue.h"
-#include "nvim/rbuffer.h"
-
-struct stream;
-
-typedef struct stream Stream;
-/// Type of function called when the Stream buffer is filled with data
-///
-/// @param stream The Stream instance
-/// @param buf The associated RBuffer instance
-/// @param count Number of bytes that was read.
-/// @param data User-defined data
-/// @param eof If the stream reached EOF.
-typedef void (*stream_read_cb)(Stream *stream, RBuffer *buf, size_t count, void *data, bool eof);
-
-/// Type of function called when the Stream has information about a write
-/// request.
-///
-/// @param stream The Stream instance
-/// @param data User-defined data
-/// @param status 0 on success, anything else indicates failure
-typedef void (*stream_write_cb)(Stream *stream, void *data, int status);
-typedef void (*stream_close_cb)(Stream *stream, void *data);
-
-struct stream {
-  bool closed;
-  bool did_eof;
-  union {
-    uv_pipe_t pipe;
-    uv_tcp_t tcp;
-    uv_idle_t idle;
-#ifdef MSWIN
-    uv_tty_t tty;
-#endif
-  } uv;
-  uv_stream_t *uvstream;
-  uv_buf_t uvbuf;
-  RBuffer *buffer;
-  uv_file fd;
-  stream_read_cb read_cb;
-  stream_write_cb write_cb;
-  void *cb_data;
-  stream_close_cb close_cb, internal_close_cb;
-  void *close_cb_data, *internal_data;
-  size_t fpos;
-  size_t curmem;
-  size_t maxmem;
-  size_t pending_reqs;
-  size_t num_bytes;
-  MultiQueue *events;
-};
+#include "nvim/event/defs.h"  // IWYU pragma: keep
+#include "nvim/types_defs.h"  // IWYU pragma: keep
 
 #ifdef INCLUDE_GENERATED_DECLARATIONS
 # include "event/stream.h.generated.h"
diff --git a/src/nvim/event/time.c b/src/nvim/event/time.c
index f678f25f3f..861b15f6dd 100644
--- a/src/nvim/event/time.c
+++ b/src/nvim/event/time.c
@@ -1,9 +1,12 @@
+#include <stdbool.h>
 #include <stdint.h>
 #include <uv.h>
 
+#include "nvim/event/defs.h"
 #include "nvim/event/loop.h"
+#include "nvim/event/multiqueue.h"
 #include "nvim/event/time.h"
-#include "nvim/func_attr.h"
+#include "nvim/types_defs.h"
 
 #ifdef INCLUDE_GENERATED_DECLARATIONS
 # include "event/time.c.generated.h"
@@ -53,7 +56,7 @@ static void time_watcher_cb(uv_timer_t *handle)
     // the timer blocked and there already is an unprocessed event waiting
     return;
   }
-  CREATE_EVENT(watcher->events, time_event, 1, watcher);
+  CREATE_EVENT(watcher->events, time_event, watcher);
 }
 
 static void close_event(void **argv)
@@ -67,6 +70,6 @@ static void close_cb(uv_handle_t *handle)
 {
   TimeWatcher *watcher = handle->data;
   if (watcher->close_cb) {
-    CREATE_EVENT(watcher->events, close_event, 1, watcher);
+    CREATE_EVENT(watcher->events, close_event, watcher);
   }
 }
diff --git a/src/nvim/event/time.h b/src/nvim/event/time.h
index 3514566901..d1f92a3c0e 100644
--- a/src/nvim/event/time.h
+++ b/src/nvim/event/time.h
@@ -1,23 +1,7 @@
 #pragma once
 
-#include <stdbool.h>
-#include <uv.h>
-
-#include "nvim/event/loop.h"
-#include "nvim/event/multiqueue.h"
-
-struct time_watcher;
-
-typedef struct time_watcher TimeWatcher;
-typedef void (*time_cb)(TimeWatcher *watcher, void *data);
-
-struct time_watcher {
-  uv_timer_t uv;
-  void *data;
-  time_cb cb, close_cb;
-  MultiQueue *events;
-  bool blockable;
-};
+#include "nvim/event/defs.h"  // IWYU pragma: keep
+#include "nvim/types_defs.h"  // IWYU pragma: keep
 
 #ifdef INCLUDE_GENERATED_DECLARATIONS
 # include "event/time.h.generated.h"
diff --git a/src/nvim/event/wstream.c b/src/nvim/event/wstream.c
index e8f757874b..406ff1620d 100644
--- a/src/nvim/event/wstream.c
+++ b/src/nvim/event/wstream.c
@@ -1,13 +1,14 @@
 #include <assert.h>
 #include <stdbool.h>
+#include <stddef.h>
 #include <uv.h>
 
-#include "nvim/event/loop.h"
+#include "nvim/event/defs.h"
 #include "nvim/event/stream.h"
 #include "nvim/event/wstream.h"
-#include "nvim/func_attr.h"
 #include "nvim/macros_defs.h"
 #include "nvim/memory.h"
+#include "nvim/types_defs.h"
 
 #define DEFAULT_MAXMEM 1024 * 1024 * 2000
 
diff --git a/src/nvim/event/wstream.h b/src/nvim/event/wstream.h
index 4cba7bde8f..a431f940d2 100644
--- a/src/nvim/event/wstream.h
+++ b/src/nvim/event/wstream.h
@@ -1,23 +1,7 @@
 #pragma once
 
-#include <stdbool.h>
-#include <stddef.h>
-#include <stdint.h>
-#include <uv.h>
-
-#include "nvim/event/loop.h"
-#include "nvim/event/stream.h"
-
-struct wbuffer;
-
-typedef struct wbuffer WBuffer;
-typedef void (*wbuffer_data_finalizer)(void *data);
-
-struct wbuffer {
-  size_t size, refcount;
-  char *data;
-  wbuffer_data_finalizer cb;
-};
+#include "nvim/event/defs.h"  // IWYU pragma: keep
+#include "nvim/types_defs.h"  // IWYU pragma: keep
 
 #ifdef INCLUDE_GENERATED_DECLARATIONS
 # include "event/wstream.h.generated.h"