refactor(io): separate types for read and write streams

This is a structural refactor with no logical changes, yet. Done in
preparation for simplifying rstream/rbuffer which will require more
state inline in RStream.

The initial idea was to have RStream and WStream as sub-types
symetrically but that doesn't work, as sockets are both reading and
writing. Also there is very little write-specific state to start with,
so the benefit of a separate WStream struct is a lot smaller. Just
document what fields in `Stream` are write specific.

8 files changed, 115 insertions, 89 deletions

diff --git a/src/nvim/event/defs.h b/src/nvim/event/defs.h
index 9b7d8708be..8563006159 100644
--- a/src/nvim/event/defs.h
+++ b/src/nvim/event/defs.h
@@ -55,14 +55,15 @@ struct wbuffer {
 };
 
 typedef struct stream Stream;
-/// Type of function called when the Stream buffer is filled with data
+typedef struct rstream RStream;
+/// Type of function called when the RStream 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);
+typedef void (*stream_read_cb)(RStream *stream, RBuffer *buf, size_t count, void *data, bool eof);
 
 /// Type of function called when the Stream has information about a write
 /// request.
@@ -71,11 +72,11 @@ typedef void (*stream_read_cb)(Stream *stream, RBuffer *buf, size_t count, void
 /// @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;
@@ -85,20 +86,27 @@ struct stream {
 #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 pending_reqs;
+  MultiQueue *events;
+
+  // only used for writing:
+  stream_write_cb write_cb;
   size_t curmem;
   size_t maxmem;
-  size_t pending_reqs;
+};
+
+struct rstream {
+  Stream s;
+  bool did_eof;
+  RBuffer *buffer;
+  uv_buf_t uvbuf;
+  stream_read_cb read_cb;
   size_t num_bytes;
-  MultiQueue *events;
+  size_t fpos;
 };
 
 #define ADDRESS_MAX_SIZE 256
@@ -147,7 +155,8 @@ struct process {
   char **argv;
   const char *exepath;
   dict_T *env;
-  Stream in, out, err;
+  Stream in;
+  RStream out, err;
   /// Exit handler. If set, user must call process_free().
   process_exit_cb cb;
   internal_process_cb internal_exit_cb, internal_close_cb;
diff --git a/src/nvim/event/libuv_process.c b/src/nvim/event/libuv_process.c
index f77d686c10..0dead1f9b4 100644
--- a/src/nvim/event/libuv_process.c
+++ b/src/nvim/event/libuv_process.c
@@ -70,19 +70,19 @@ int libuv_process_spawn(LibuvProcess *uvproc)
     uvproc->uvstdio[0].data.stream = (uv_stream_t *)(&proc->in.uv.pipe);
   }
 
-  if (!proc->out.closed) {
+  if (!proc->out.s.closed) {
     uvproc->uvstdio[1].flags = UV_CREATE_PIPE | UV_WRITABLE_PIPE;
 #ifdef MSWIN
     // pipe must be readable for IOCP to work on Windows.
     uvproc->uvstdio[1].flags |= proc->overlapped
                                 ? (UV_READABLE_PIPE | UV_OVERLAPPED_PIPE) : 0;
 #endif
-    uvproc->uvstdio[1].data.stream = (uv_stream_t *)(&proc->out.uv.pipe);
+    uvproc->uvstdio[1].data.stream = (uv_stream_t *)(&proc->out.s.uv.pipe);
   }
 
-  if (!proc->err.closed) {
+  if (!proc->err.s.closed) {
     uvproc->uvstdio[2].flags = UV_CREATE_PIPE | UV_WRITABLE_PIPE;
-    uvproc->uvstdio[2].data.stream = (uv_stream_t *)(&proc->err.uv.pipe);
+    uvproc->uvstdio[2].data.stream = (uv_stream_t *)(&proc->err.s.uv.pipe);
   } else if (proc->fwd_err) {
     uvproc->uvstdio[2].flags = UV_INHERIT_FD;
     uvproc->uvstdio[2].data.fd = STDERR_FILENO;
diff --git a/src/nvim/event/process.c b/src/nvim/event/process.c
index 7460e92766..710376cd62 100644
--- a/src/nvim/event/process.c
+++ b/src/nvim/event/process.c
@@ -8,7 +8,9 @@
 #include "nvim/event/loop.h"
 #include "nvim/event/multiqueue.h"
 #include "nvim/event/process.h"
+#include "nvim/event/rstream.h"
 #include "nvim/event/stream.h"
+#include "nvim/event/wstream.h"
 #include "nvim/globals.h"
 #include "nvim/log.h"
 #include "nvim/main.h"
@@ -51,15 +53,15 @@ int process_spawn(Process *proc, bool in, bool out, bool err)
   }
 
   if (out) {
-    uv_pipe_init(&proc->loop->uv, &proc->out.uv.pipe, 0);
+    uv_pipe_init(&proc->loop->uv, &proc->out.s.uv.pipe, 0);
   } else {
-    proc->out.closed = true;
+    proc->out.s.closed = true;
   }
 
   if (err) {
-    uv_pipe_init(&proc->loop->uv, &proc->err.uv.pipe, 0);
+    uv_pipe_init(&proc->loop->uv, &proc->err.s.uv.pipe, 0);
   } else {
-    proc->err.closed = true;
+    proc->err.s.closed = true;
   }
 
 #ifdef USE_GCOV
@@ -82,10 +84,10 @@ int process_spawn(Process *proc, bool in, bool out, bool err)
       uv_close((uv_handle_t *)&proc->in.uv.pipe, NULL);
     }
     if (out) {
-      uv_close((uv_handle_t *)&proc->out.uv.pipe, NULL);
+      uv_close((uv_handle_t *)&proc->out.s.uv.pipe, NULL);
     }
     if (err) {
-      uv_close((uv_handle_t *)&proc->err.uv.pipe, NULL);
+      uv_close((uv_handle_t *)&proc->err.s.uv.pipe, NULL);
     }
 
     if (proc->type == kProcessTypeUv) {
@@ -106,16 +108,16 @@ int process_spawn(Process *proc, bool in, bool out, bool err)
   }
 
   if (out) {
-    stream_init(NULL, &proc->out, -1, (uv_stream_t *)&proc->out.uv.pipe);
-    proc->out.internal_data = proc;
-    proc->out.internal_close_cb = on_process_stream_close;
+    stream_init(NULL, &proc->out.s, -1, (uv_stream_t *)&proc->out.s.uv.pipe);
+    proc->out.s.internal_data = proc;
+    proc->out.s.internal_close_cb = on_process_stream_close;
     proc->refcount++;
   }
 
   if (err) {
-    stream_init(NULL, &proc->err, -1, (uv_stream_t *)&proc->err.uv.pipe);
-    proc->err.internal_data = proc;
-    proc->err.internal_close_cb = on_process_stream_close;
+    stream_init(NULL, &proc->err.s, -1, (uv_stream_t *)&proc->err.s.uv.pipe);
+    proc->err.s.internal_data = proc;
+    proc->err.s.internal_close_cb = on_process_stream_close;
     proc->refcount++;
   }
 
@@ -148,9 +150,9 @@ void process_teardown(Loop *loop) FUNC_ATTR_NONNULL_ALL
 
 void process_close_streams(Process *proc) FUNC_ATTR_NONNULL_ALL
 {
-  stream_may_close(&proc->in);
-  stream_may_close(&proc->out);
-  stream_may_close(&proc->err);
+  wstream_may_close(&proc->in);
+  rstream_may_close(&proc->out);
+  rstream_may_close(&proc->err);
 }
 
 /// Synchronously wait for a process to finish
@@ -337,10 +339,10 @@ static void process_close(Process *proc)
 ///
 /// @param proc     Process, for which an output stream should be flushed.
 /// @param stream   Stream to flush.
-static void flush_stream(Process *proc, Stream *stream)
+static void flush_stream(Process *proc, RStream *stream)
   FUNC_ATTR_NONNULL_ARG(1)
 {
-  if (!stream || stream->closed) {
+  if (!stream || stream->s.closed) {
     return;
   }
 
@@ -350,7 +352,7 @@ static void flush_stream(Process *proc, Stream *stream)
   // keeps sending data, we only accept as much data as the system buffer size.
   // Otherwise this would block cleanup/teardown.
   int system_buffer_size = 0;
-  int err = uv_recv_buffer_size((uv_handle_t *)&stream->uv.pipe,
+  int err = uv_recv_buffer_size((uv_handle_t *)&stream->s.uv.pipe,
                                 &system_buffer_size);
   if (err) {
     system_buffer_size = (int)rbuffer_capacity(stream->buffer);
@@ -359,14 +361,14 @@ static void flush_stream(Process *proc, Stream *stream)
   size_t max_bytes = stream->num_bytes + (size_t)system_buffer_size;
 
   // Read remaining data.
-  while (!stream->closed && stream->num_bytes < max_bytes) {
+  while (!stream->s.closed && stream->num_bytes < max_bytes) {
     // Remember number of bytes before polling
     size_t num_bytes = stream->num_bytes;
 
     // Poll for data and process the generated events.
     loop_poll_events(proc->loop, 0);
-    if (stream->events) {
-      multiqueue_process_events(stream->events);
+    if (stream->s.events) {
+      multiqueue_process_events(stream->s.events);
     }
 
     // Stream can be closed if it is empty.
@@ -374,7 +376,7 @@ static void flush_stream(Process *proc, Stream *stream)
       if (stream->read_cb && !stream->did_eof) {
         // Stream callback could miss EOF handling if a child keeps the stream
         // open. But only send EOF if we haven't already.
-        stream->read_cb(stream, stream->buffer, 0, stream->cb_data, true);
+        stream->read_cb(stream, stream->buffer, 0, stream->s.cb_data, true);
       }
       break;
     }
diff --git a/src/nvim/event/process.h b/src/nvim/event/process.h
index 421a470244..74b52cbbb1 100644
--- a/src/nvim/event/process.h
+++ b/src/nvim/event/process.h
@@ -21,8 +21,8 @@ static inline Process process_init(Loop *loop, ProcessType type, void *data)
     .argv = NULL,
     .exepath = NULL,
     .in = { .closed = false },
-    .out = { .closed = false },
-    .err = { .closed = false },
+    .out = { .s.closed = false },
+    .err = { .s.closed = false },
     .cb = NULL,
     .closed = false,
     .internal_close_cb = NULL,
diff --git a/src/nvim/event/rstream.c b/src/nvim/event/rstream.c
index 6b4ab472e4..6c7fa20bd8 100644
--- a/src/nvim/event/rstream.c
+++ b/src/nvim/event/rstream.c
@@ -19,23 +19,26 @@
 # include "event/rstream.c.generated.h"
 #endif
 
-void rstream_init_fd(Loop *loop, Stream *stream, int fd, size_t bufsize)
+void rstream_init_fd(Loop *loop, RStream *stream, int fd, size_t bufsize)
   FUNC_ATTR_NONNULL_ARG(1, 2)
 {
-  stream_init(loop, stream, fd, NULL);
+  stream_init(loop, &stream->s, fd, NULL);
   rstream_init(stream, bufsize);
 }
 
-void rstream_init_stream(Stream *stream, uv_stream_t *uvstream, size_t bufsize)
+void rstream_init_stream(RStream *stream, uv_stream_t *uvstream, size_t bufsize)
   FUNC_ATTR_NONNULL_ARG(1, 2)
 {
-  stream_init(NULL, stream, -1, uvstream);
+  stream_init(NULL, &stream->s, -1, uvstream);
   rstream_init(stream, bufsize);
 }
 
-void rstream_init(Stream *stream, size_t bufsize)
+void rstream_init(RStream *stream, size_t bufsize)
   FUNC_ATTR_NONNULL_ARG(1)
 {
+  stream->fpos = 0;
+  stream->read_cb = NULL;
+  stream->num_bytes = 0;
   stream->buffer = rbuffer_new(bufsize);
   stream->buffer->data = stream;
   stream->buffer->full_cb = on_rbuffer_full;
@@ -45,28 +48,28 @@ void rstream_init(Stream *stream, size_t bufsize)
 /// Starts watching for events from a `Stream` instance.
 ///
 /// @param stream The `Stream` instance
-void rstream_start(Stream *stream, stream_read_cb cb, void *data)
+void rstream_start(RStream *stream, stream_read_cb cb, void *data)
   FUNC_ATTR_NONNULL_ARG(1)
 {
   stream->read_cb = cb;
-  stream->cb_data = data;
-  if (stream->uvstream) {
-    uv_read_start(stream->uvstream, alloc_cb, read_cb);
+  stream->s.cb_data = data;
+  if (stream->s.uvstream) {
+    uv_read_start(stream->s.uvstream, alloc_cb, read_cb);
   } else {
-    uv_idle_start(&stream->uv.idle, fread_idle_cb);
+    uv_idle_start(&stream->s.uv.idle, fread_idle_cb);
   }
 }
 
 /// Stops watching for events from a `Stream` instance.
 ///
 /// @param stream The `Stream` instance
-void rstream_stop(Stream *stream)
+void rstream_stop(RStream *stream)
   FUNC_ATTR_NONNULL_ALL
 {
-  if (stream->uvstream) {
-    uv_read_stop(stream->uvstream);
+  if (stream->s.uvstream) {
+    uv_read_stop(stream->s.uvstream);
   } else {
-    uv_idle_stop(&stream->uv.idle);
+    uv_idle_stop(&stream->s.uv.idle);
   }
 }
 
@@ -77,9 +80,9 @@ static void on_rbuffer_full(RBuffer *buf, void *data)
 
 static void on_rbuffer_nonfull(RBuffer *buf, void *data)
 {
-  Stream *stream = data;
+  RStream *stream = data;
   assert(stream->read_cb);
-  rstream_start(stream, stream->read_cb, stream->cb_data);
+  rstream_start(stream, stream->read_cb, stream->s.cb_data);
 }
 
 // Callbacks used by libuv
@@ -87,7 +90,7 @@ static void on_rbuffer_nonfull(RBuffer *buf, void *data)
 /// Called by libuv to allocate memory for reading.
 static void alloc_cb(uv_handle_t *handle, size_t suggested, uv_buf_t *buf)
 {
-  Stream *stream = handle->data;
+  RStream *stream = handle->data;
   // `uv_buf_t.len` happens to have different size on Windows.
   size_t write_count;
   buf->base = rbuffer_write_ptr(stream->buffer, &write_count);
@@ -99,7 +102,7 @@ static void alloc_cb(uv_handle_t *handle, size_t suggested, uv_buf_t *buf)
 /// 0-length buffer.
 static void read_cb(uv_stream_t *uvstream, ssize_t cnt, const uv_buf_t *buf)
 {
-  Stream *stream = uvstream->data;
+  RStream *stream = uvstream->data;
 
   if (cnt <= 0) {
     // cnt == 0 means libuv asked for a buffer and decided it wasn't needed:
@@ -141,7 +144,7 @@ static void read_cb(uv_stream_t *uvstream, ssize_t cnt, const uv_buf_t *buf)
 static void fread_idle_cb(uv_idle_t *handle)
 {
   uv_fs_t req;
-  Stream *stream = handle->data;
+  RStream *stream = handle->data;
 
   // `uv_buf_t.len` happens to have different size on Windows.
   size_t write_count;
@@ -160,7 +163,7 @@ static void fread_idle_cb(uv_idle_t *handle)
   // Synchronous read
   uv_fs_read(handle->loop,
              &req,
-             stream->fd,
+             stream->s.fd,
              &stream->uvbuf,
              1,
              (int64_t)stream->fpos,
@@ -169,7 +172,7 @@ static void fread_idle_cb(uv_idle_t *handle)
   uv_fs_req_cleanup(&req);
 
   if (req.result <= 0) {
-    uv_idle_stop(&stream->uv.idle);
+    uv_idle_stop(&stream->s.uv.idle);
     invoke_read_cb(stream, 0, true);
     return;
   }
@@ -183,24 +186,29 @@ static void fread_idle_cb(uv_idle_t *handle)
 
 static void read_event(void **argv)
 {
-  Stream *stream = argv[0];
+  RStream *stream = argv[0];
   if (stream->read_cb) {
     size_t count = (uintptr_t)argv[1];
     bool eof = (uintptr_t)argv[2];
     stream->did_eof = eof;
-    stream->read_cb(stream, stream->buffer, count, stream->cb_data, eof);
+    stream->read_cb(stream, stream->buffer, count, stream->s.cb_data, eof);
   }
-  stream->pending_reqs--;
-  if (stream->closed && !stream->pending_reqs) {
-    stream_close_handle(stream);
+  stream->s.pending_reqs--;
+  if (stream->s.closed && !stream->s.pending_reqs) {
+    stream_close_handle(&stream->s, true);
   }
 }
 
-static void invoke_read_cb(Stream *stream, size_t count, bool eof)
+static void invoke_read_cb(RStream *stream, size_t count, bool eof)
 {
   // Don't let the stream be closed before the event is processed.
-  stream->pending_reqs++;
+  stream->s.pending_reqs++;
 
-  CREATE_EVENT(stream->events, read_event,
+  CREATE_EVENT(stream->s.events, read_event,
                stream, (void *)(uintptr_t *)count, (void *)(uintptr_t)eof);
 }
+
+void rstream_may_close(RStream *stream)
+{
+  stream_may_close(&stream->s, true);
+}
diff --git a/src/nvim/event/socket.c b/src/nvim/event/socket.c
index 4e878a2ecf..017f159fa1 100644
--- a/src/nvim/event/socket.c
+++ b/src/nvim/event/socket.c
@@ -135,17 +135,17 @@ int socket_watcher_start(SocketWatcher *watcher, int backlog, socket_cb cb)
   return 0;
 }
 
-int socket_watcher_accept(SocketWatcher *watcher, Stream *stream)
+int socket_watcher_accept(SocketWatcher *watcher, RStream *stream)
   FUNC_ATTR_NONNULL_ARG(1) FUNC_ATTR_NONNULL_ARG(2)
 {
   uv_stream_t *client;
 
   if (watcher->stream->type == UV_TCP) {
-    client = (uv_stream_t *)(&stream->uv.tcp);
+    client = (uv_stream_t *)(&stream->s.uv.tcp);
     uv_tcp_init(watcher->uv.tcp.handle.loop, (uv_tcp_t *)client);
     uv_tcp_nodelay((uv_tcp_t *)client, true);
   } else {
-    client = (uv_stream_t *)&stream->uv.pipe;
+    client = (uv_stream_t *)&stream->s.uv.pipe;
     uv_pipe_init(watcher->uv.pipe.handle.loop, (uv_pipe_t *)client, 0);
   }
 
@@ -156,7 +156,7 @@ int socket_watcher_accept(SocketWatcher *watcher, Stream *stream)
     return result;
   }
 
-  stream_init(NULL, stream, -1, client);
+  stream_init(NULL, &stream->s, -1, client);
   return 0;
 }
 
@@ -197,7 +197,7 @@ static void connect_cb(uv_connect_t *req, int status)
   }
 }
 
-bool socket_connect(Loop *loop, Stream *stream, bool is_tcp, const char *address, int timeout,
+bool socket_connect(Loop *loop, RStream *stream, bool is_tcp, const char *address, int timeout,
                     const char **error)
 {
   bool success = false;
@@ -206,7 +206,7 @@ bool socket_connect(Loop *loop, Stream *stream, bool is_tcp, const char *address
   req.data = &status;
   uv_stream_t *uv_stream;
 
-  uv_tcp_t *tcp = &stream->uv.tcp;
+  uv_tcp_t *tcp = &stream->s.uv.tcp;
   uv_getaddrinfo_t addr_req;
   addr_req.addrinfo = NULL;
   const struct addrinfo *addrinfo = NULL;
@@ -237,7 +237,7 @@ tcp_retry:
     uv_tcp_connect(&req,  tcp, addrinfo->ai_addr, connect_cb);
     uv_stream = (uv_stream_t *)tcp;
   } else {
-    uv_pipe_t *pipe = &stream->uv.pipe;
+    uv_pipe_t *pipe = &stream->s.uv.pipe;
     uv_pipe_init(&loop->uv, pipe, 0);
     uv_pipe_connect(&req,  pipe, address, connect_cb);
     uv_stream = (uv_stream_t *)pipe;
@@ -245,7 +245,7 @@ tcp_retry:
   status = 1;
   LOOP_PROCESS_EVENTS_UNTIL(&main_loop, NULL, timeout, status != 1);
   if (status == 0) {
-    stream_init(NULL, stream, -1, uv_stream);
+    stream_init(NULL, &stream->s, -1, uv_stream);
     success = true;
   } else if (is_tcp && addrinfo->ai_next) {
     addrinfo = addrinfo->ai_next;
diff --git a/src/nvim/event/stream.c b/src/nvim/event/stream.c
index 0b9ed4f25b..3d26dd868f 100644
--- a/src/nvim/event/stream.c
+++ b/src/nvim/event/stream.c
@@ -85,21 +85,17 @@ void stream_init(Loop *loop, Stream *stream, int fd, uv_stream_t *uvstream)
   }
 
   stream->internal_data = NULL;
-  stream->fpos = 0;
   stream->curmem = 0;
   stream->maxmem = 0;
   stream->pending_reqs = 0;
-  stream->read_cb = NULL;
   stream->write_cb = NULL;
   stream->close_cb = NULL;
   stream->internal_close_cb = NULL;
   stream->closed = false;
-  stream->buffer = NULL;
   stream->events = NULL;
-  stream->num_bytes = 0;
 }
 
-void stream_close(Stream *stream, stream_close_cb on_stream_close, void *data)
+void stream_close(Stream *stream, stream_close_cb on_stream_close, void *data, bool rstream)
   FUNC_ATTR_NONNULL_ARG(1)
 {
   assert(!stream->closed);
@@ -116,18 +112,18 @@ void stream_close(Stream *stream, stream_close_cb on_stream_close, void *data)
 #endif
 
   if (!stream->pending_reqs) {
-    stream_close_handle(stream);
+    stream_close_handle(stream, rstream);
   }
 }
 
-void stream_may_close(Stream *stream)
+void stream_may_close(Stream *stream, bool rstream)
 {
   if (!stream->closed) {
-    stream_close(stream, NULL, NULL);
+    stream_close(stream, NULL, NULL, rstream);
   }
 }
 
-void stream_close_handle(Stream *stream)
+void stream_close_handle(Stream *stream, bool rstream)
   FUNC_ATTR_NONNULL_ALL
 {
   uv_handle_t *handle = NULL;
@@ -145,16 +141,22 @@ void stream_close_handle(Stream *stream)
   assert(handle != NULL);
 
   if (!uv_is_closing(handle)) {
-    uv_close(handle, close_cb);
+    uv_close(handle, rstream ? rstream_close_cb : close_cb);
   }
 }
 
-static void close_cb(uv_handle_t *handle)
+static void rstream_close_cb(uv_handle_t *handle)
 {
-  Stream *stream = handle->data;
+  RStream *stream = handle->data;
   if (stream->buffer) {
     rbuffer_free(stream->buffer);
   }
+  close_cb(handle);
+}
+
+static void close_cb(uv_handle_t *handle)
+{
+  Stream *stream = handle->data;
   if (stream->close_cb) {
     stream->close_cb(stream, stream->close_cb_data);
   }
diff --git a/src/nvim/event/wstream.c b/src/nvim/event/wstream.c
index c67a9b96ed..07aab87e4d 100644
--- a/src/nvim/event/wstream.c
+++ b/src/nvim/event/wstream.c
@@ -141,7 +141,7 @@ static void write_cb(uv_write_t *req, int status)
 
   if (data->stream->closed && data->stream->pending_reqs == 0) {
     // Last pending write, free the stream;
-    stream_close_handle(data->stream);
+    stream_close_handle(data->stream, false);
   }
 
   xfree(data);
@@ -158,3 +158,8 @@ void wstream_release_wbuffer(WBuffer *buffer)
     xfree(buffer);
   }
 }
+
+void wstream_may_close(Stream *stream)
+{
+  stream_may_close(stream, false);
+}