channels: allow bytes sockets and stdio, and buffered bytes output

1 files changed, 8 insertions, 79 deletions

diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index 42b1f63830..56af4fa791 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -13,7 +13,6 @@
 #include "nvim/api/ui.h"
 #include "nvim/channel.h"
 #include "nvim/msgpack_rpc/channel.h"
-#include "nvim/msgpack_rpc/server.h"
 #include "nvim/event/loop.h"
 #include "nvim/event/libuv_process.h"
 #include "nvim/event/rstream.h"
@@ -30,12 +29,9 @@
 #include "nvim/map.h"
 #include "nvim/log.h"
 #include "nvim/misc1.h"
-#include "nvim/path.h"
 #include "nvim/lib/kvec.h"
 #include "nvim/os/input.h"
 
-#define CHANNEL_BUFFER_SIZE 0xffff
-
 #if MIN_LOG_LEVEL > DEBUG_LOG_LEVEL
 #define log_client_msg(...)
 #define log_server_msg(...)
@@ -66,57 +62,18 @@ void rpc_start(Channel *channel)
   rpc->next_request_id = 1;
   kv_init(rpc->call_stack);
 
-  Stream *in = channel_instream(channel);
-  Stream *out = channel_outstream(channel);
-
-  DLOG("rpc ch %" PRIu64 " in-stream=%p out-stream=%p", channel->id, in, out);
-
-  wstream_init(in, 0);
-  rstream_init(out, CHANNEL_BUFFER_SIZE);
-  rstream_start(out, receive_msgpack, channel);
-}
-
-/// Creates an API channel from a tcp/pipe socket connection
-///
-/// @param watcher The SocketWatcher ready to accept the connection
-void channel_from_connection(SocketWatcher *watcher)
-{
-  Channel *channel = channel_alloc(kChannelStreamSocket);
-  socket_watcher_accept(watcher, &channel->stream.socket);
-  channel_incref(channel);  // close channel only after the stream is closed
-  channel->stream.socket.internal_close_cb = close_cb;
-  channel->stream.socket.internal_data = channel;
-  rpc_start(channel);
-}
-
-/// TODO: move to eval.c, also support bytes
-uint64_t channel_connect(bool tcp, const char *address,
-                         int timeout, const char **error)
-{
-  if (!tcp) {
-    char *path = fix_fname(address);
-    if (server_owns_pipe_address(path)) {
-      // avoid deadlock
-      xfree(path);
-      return channel_create_internal();
-    }
-    xfree(path);
-  }
+  if (channel->streamtype != kChannelStreamInternal) {
+    Stream *out = channel_outstream(channel);
+#if MIN_LOG_LEVEL <= DEBUG_LOG_LEVEL
+    Stream *in = channel_instream(channel);
+    DLOG("rpc ch %" PRIu64 " in-stream=%p out-stream=%p", channel->id, in, out);
+#endif
 
-  Channel *channel = channel_alloc(kChannelStreamSocket);
-  if (!socket_connect(&main_loop, &channel->stream.socket,
-                      tcp, address, timeout, error)) {
-    channel_decref(channel);
-    return 0;
+    rstream_start(out, receive_msgpack, channel);
   }
-
-  channel_incref(channel);  // close channel only after the stream is closed
-  channel->stream.socket.internal_close_cb = close_cb;
-  channel->stream.socket.internal_data = channel;
-  rpc_start(channel);
-  return channel->id;
 }
 
+
 static Channel *find_rpc_channel(uint64_t id)
 {
   Channel *chan = find_channel(id);
@@ -263,29 +220,6 @@ bool channel_close(uint64_t id)
   return true;
 }
 
-/// Creates an API channel from stdin/stdout. This is used when embedding
-/// Neovim
-void channel_from_stdio(void)
-{
-  Channel *channel = channel_alloc(kChannelStreamStdio);
-  channel_incref(channel);  // stdio channels are only closed on exit
-  // read stream
-  rstream_init_fd(&main_loop, &channel->stream.stdio.in, 0, CHANNEL_BUFFER_SIZE);
-  wstream_init_fd(&main_loop, &channel->stream.stdio.out, 1, 0);
-
-  rpc_start(channel);
-}
-
-/// Creates a loopback channel. This is used to avoid deadlock
-/// when an instance connects to its own named pipe.
-uint64_t channel_create_internal(void)
-{
-  Channel *channel = channel_alloc(kChannelStreamInternal);
-  channel_incref(channel);  // internal channel lives until process exit
-  rpc_start(channel);
-  return channel->id;
-}
-
 void channel_process_exit(uint64_t id, int status)
 {
   Channel *channel = pmap_get(uint64_t)(channels, id);
@@ -663,11 +597,6 @@ void rpc_free(Channel *channel)
   kv_destroy(channel->rpc.call_stack);
 }
 
-static void close_cb(Stream *stream, void *data)
-{
-  channel_decref(data);
-}
-
 static bool is_rpc_response(msgpack_object *obj)
 {
   return obj->type == MSGPACK_OBJECT_ARRAY