Merge remote-tracking branch 'upstream/master' into usermarksusermarks

1 files changed, 431 insertions, 310 deletions

diff --git a/runtime/lua/vim/lsp/rpc.lua b/runtime/lua/vim/lsp/rpc.lua
index 0926912066..f1492601ff 100644
--- a/runtime/lua/vim/lsp/rpc.lua
+++ b/runtime/lua/vim/lsp/rpc.lua
@@ -1,4 +1,3 @@
-local vim = vim
 local uv = vim.loop
 local log = require('vim.lsp.log')
 local protocol = require('vim.lsp.protocol')
@@ -241,247 +240,162 @@ function default_dispatchers.on_error(code, err)
   local _ = log.error() and log.error('client_error:', client_errors[code], err)
 end
 
---- Starts an LSP server process and create an LSP RPC client object to
---- interact with it. Communication with the server is currently limited to stdio.
----
----@param cmd (string) Command to start the LSP server.
----@param cmd_args (table) List of additional string arguments to pass to {cmd}.
----@param dispatchers table|nil Dispatchers for LSP message types. Valid
----dispatcher names are:
---- - `"notification"`
---- - `"server_request"`
---- - `"on_error"`
---- - `"on_exit"`
----@param extra_spawn_params table|nil Additional context for the LSP
---- server process. May contain:
---- - {cwd} (string) Working directory for the LSP server process
---- - {env} (table) Additional environment variables for LSP server process
----@returns Client RPC object.
----
----@returns Methods:
---- - `notify()` |vim.lsp.rpc.notify()|
---- - `request()` |vim.lsp.rpc.request()|
----
----@returns Members:
---- - {pid} (number) The LSP server's PID.
---- - {handle} A handle for low-level interaction with the LSP server process
----   |vim.loop|.
-local function start(cmd, cmd_args, dispatchers, extra_spawn_params)
-  local _ = log.info()
-    and log.info('Starting RPC client', { cmd = cmd, args = cmd_args, extra = extra_spawn_params })
-  validate({
-    cmd = { cmd, 's' },
-    cmd_args = { cmd_args, 't' },
-    dispatchers = { dispatchers, 't', true },
-  })
-
-  if extra_spawn_params and extra_spawn_params.cwd then
-    assert(is_dir(extra_spawn_params.cwd), 'cwd must be a directory')
-  end
-  if dispatchers then
-    local user_dispatchers = dispatchers
-    dispatchers = {}
-    for dispatch_name, default_dispatch in pairs(default_dispatchers) do
-      local user_dispatcher = user_dispatchers[dispatch_name]
-      if user_dispatcher then
-        if type(user_dispatcher) ~= 'function' then
-          error(string.format('dispatcher.%s must be a function', dispatch_name))
-        end
-        -- server_request is wrapped elsewhere.
-        if
-          not (dispatch_name == 'server_request' or dispatch_name == 'on_exit') -- TODO this blocks the loop exiting for some reason.
-        then
-          user_dispatcher = schedule_wrap(user_dispatcher)
-        end
-        dispatchers[dispatch_name] = user_dispatcher
-      else
-        dispatchers[dispatch_name] = default_dispatch
-      end
+---@private
+local function create_read_loop(handle_body, on_no_chunk, on_error)
+  local parse_chunk = coroutine.wrap(request_parser_loop)
+  parse_chunk()
+  return function(err, chunk)
+    if err then
+      on_error(err)
+      return
     end
-  else
-    dispatchers = default_dispatchers
-  end
-
-  local stdin = uv.new_pipe(false)
-  local stdout = uv.new_pipe(false)
-  local stderr = uv.new_pipe(false)
-
-  local message_index = 0
-  local message_callbacks = {}
-  local notify_reply_callbacks = {}
 
-  local handle, pid
-  do
-    ---@private
-    --- Callback for |vim.loop.spawn()| Closes all streams and runs the `on_exit` dispatcher.
-    ---@param code (number) Exit code
-    ---@param signal (number) Signal that was used to terminate (if any)
-    local function onexit(code, signal)
-      stdin:close()
-      stdout:close()
-      stderr:close()
-      handle:close()
-      -- Make sure that message_callbacks/notify_reply_callbacks can be gc'd.
-      message_callbacks = nil
-      notify_reply_callbacks = nil
-      dispatchers.on_exit(code, signal)
-    end
-    local spawn_params = {
-      args = cmd_args,
-      stdio = { stdin, stdout, stderr },
-      detached = not is_win,
-    }
-    if extra_spawn_params then
-      spawn_params.cwd = extra_spawn_params.cwd
-      spawn_params.env = env_merge(extra_spawn_params.env)
-      if extra_spawn_params.detached ~= nil then
-        spawn_params.detached = extra_spawn_params.detached
+    if not chunk then
+      if on_no_chunk then
+        on_no_chunk()
       end
+      return
     end
-    handle, pid = uv.spawn(cmd, spawn_params, onexit)
-    if handle == nil then
-      stdin:close()
-      stdout:close()
-      stderr:close()
-      local msg = string.format('Spawning language server with cmd: `%s` failed', cmd)
-      if string.match(pid, 'ENOENT') then
-        msg = msg
-          .. '. The language server is either not installed, missing from PATH, or not executable.'
+
+    while true do
+      local headers, body = parse_chunk(chunk)
+      if headers then
+        handle_body(body)
+        chunk = ''
       else
-        msg = msg .. string.format(' with error message: %s', pid)
+        break
       end
-      vim.notify(msg, vim.log.levels.WARN)
-      return
     end
   end
+end
 
-  ---@private
-  --- Encodes {payload} into a JSON-RPC message and sends it to the remote
-  --- process.
-  ---
-  ---@param payload table
-  ---@returns true if the payload could be scheduled, false if the main event-loop is in the process of closing.
-  local function encode_and_send(payload)
-    local _ = log.debug() and log.debug('rpc.send', payload)
-    if handle == nil or handle:is_closing() then
-      return false
-    end
-    local encoded = vim.json.encode(payload)
-    stdin:write(format_message_with_content_length(encoded))
-    return true
-  end
+---@class RpcClient
+---@field message_index number
+---@field message_callbacks table
+---@field notify_reply_callbacks table
+---@field transport table
+---@field dispatchers table
 
-  -- FIXME: DOC: Should be placed on the RPC client object returned by
-  -- `start()`
-  --
-  --- Sends a notification to the LSP server.
-  ---@param method (string) The invoked LSP method
-  ---@param params (table): Parameters for the invoked LSP method
-  ---@returns (bool) `true` if notification could be sent, `false` if not
-  local function notify(method, params)
-    return encode_and_send({
-      jsonrpc = '2.0',
-      method = method,
-      params = params,
-    })
-  end
+---@class RpcClient
+local Client = {}
 
-  ---@private
-  --- sends an error object to the remote LSP process.
-  local function send_response(request_id, err, result)
-    return encode_and_send({
-      id = request_id,
-      jsonrpc = '2.0',
-      error = err,
-      result = result,
-    })
+---@private
+function Client:encode_and_send(payload)
+  local _ = log.debug() and log.debug('rpc.send', payload)
+  if self.transport.is_closing() then
+    return false
   end
+  local encoded = vim.json.encode(payload)
+  self.transport.write(format_message_with_content_length(encoded))
+  return true
+end
 
-  -- FIXME: DOC: Should be placed on the RPC client object returned by
-  -- `start()`
-  --
-  --- Sends a request to the LSP server and runs {callback} upon response.
-  ---
-  ---@param method (string) The invoked LSP method
-  ---@param params (table) Parameters for the invoked LSP method
-  ---@param callback (function) Callback to invoke
-  ---@param notify_reply_callback (function|nil) Callback to invoke as soon as a request is no longer pending
-  ---@returns (bool, number) `(true, message_id)` if request could be sent, `false` if not
-  local function request(method, params, callback, notify_reply_callback)
-    validate({
-      callback = { callback, 'f' },
-      notify_reply_callback = { notify_reply_callback, 'f', true },
-    })
-    message_index = message_index + 1
-    local message_id = message_index
-    local result = encode_and_send({
-      id = message_id,
-      jsonrpc = '2.0',
-      method = method,
-      params = params,
-    })
-    if result then
-      if message_callbacks then
-        message_callbacks[message_id] = schedule_wrap(callback)
-      else
-        return false
-      end
-      if notify_reply_callback and notify_reply_callbacks then
-        notify_reply_callbacks[message_id] = schedule_wrap(notify_reply_callback)
-      end
-      return result, message_id
+---@private
+--- Sends a notification to the LSP server.
+---@param method (string) The invoked LSP method
+---@param params (any): Parameters for the invoked LSP method
+---@returns (bool) `true` if notification could be sent, `false` if not
+function Client:notify(method, params)
+  return self:encode_and_send({
+    jsonrpc = '2.0',
+    method = method,
+    params = params,
+  })
+end
+
+---@private
+--- sends an error object to the remote LSP process.
+function Client:send_response(request_id, err, result)
+  return self:encode_and_send({
+    id = request_id,
+    jsonrpc = '2.0',
+    error = err,
+    result = result,
+  })
+end
+
+---@private
+--- Sends a request to the LSP server and runs {callback} upon response.
+---
+---@param method (string) The invoked LSP method
+---@param params (table|nil) Parameters for the invoked LSP method
+---@param callback (function) Callback to invoke
+---@param notify_reply_callback (function|nil) Callback to invoke as soon as a request is no longer pending
+---@returns (bool, number) `(true, message_id)` if request could be sent, `false` if not
+function Client:request(method, params, callback, notify_reply_callback)
+  validate({
+    callback = { callback, 'f' },
+    notify_reply_callback = { notify_reply_callback, 'f', true },
+  })
+  self.message_index = self.message_index + 1
+  local message_id = self.message_index
+  local result = self:encode_and_send({
+    id = message_id,
+    jsonrpc = '2.0',
+    method = method,
+    params = params,
+  })
+  local message_callbacks = self.message_callbacks
+  local notify_reply_callbacks = self.notify_reply_callbacks
+  if result then
+    if message_callbacks then
+      message_callbacks[message_id] = schedule_wrap(callback)
     else
       return false
     end
+    if notify_reply_callback and notify_reply_callbacks then
+      notify_reply_callbacks[message_id] = schedule_wrap(notify_reply_callback)
+    end
+    return result, message_id
+  else
+    return false
   end
+end
 
-  stderr:read_start(function(_, chunk)
-    if chunk then
-      local _ = log.error() and log.error('rpc', cmd, 'stderr', chunk)
-    end
-  end)
+---@private
+function Client:on_error(errkind, ...)
+  assert(client_errors[errkind])
+  -- TODO what to do if this fails?
+  pcall(self.dispatchers.on_error, errkind, ...)
+end
 
-  ---@private
-  local function on_error(errkind, ...)
-    assert(client_errors[errkind])
-    -- TODO what to do if this fails?
-    pcall(dispatchers.on_error, errkind, ...)
-  end
-  ---@private
-  local function pcall_handler(errkind, status, head, ...)
-    if not status then
-      on_error(errkind, head, ...)
-      return status, head
-    end
-    return status, head, ...
-  end
-  ---@private
-  local function try_call(errkind, fn, ...)
-    return pcall_handler(errkind, pcall(fn, ...))
+---@private
+function Client:pcall_handler(errkind, status, head, ...)
+  if not status then
+    self:on_error(errkind, head, ...)
+    return status, head
   end
+  return status, head, ...
+end
 
-  -- TODO periodically check message_callbacks for old requests past a certain
-  -- time and log them. This would require storing the timestamp. I could call
-  -- them with an error then, perhaps.
+---@private
+function Client:try_call(errkind, fn, ...)
+  return self:pcall_handler(errkind, pcall(fn, ...))
+end
 
-  ---@private
-  local function handle_body(body)
-    local ok, decoded = pcall(vim.json.decode, body, { luanil = { object = true } })
-    if not ok then
-      on_error(client_errors.INVALID_SERVER_JSON, decoded)
-      return
-    end
-    local _ = log.debug() and log.debug('rpc.receive', decoded)
+-- TODO periodically check message_callbacks for old requests past a certain
+-- time and log them. This would require storing the timestamp. I could call
+-- them with an error then, perhaps.
 
-    if type(decoded.method) == 'string' and decoded.id then
-      local err
-      -- Schedule here so that the users functions don't trigger an error and
-      -- we can still use the result.
-      schedule(function()
+---@private
+function Client:handle_body(body)
+  local ok, decoded = pcall(vim.json.decode, body, { luanil = { object = true } })
+  if not ok then
+    self:on_error(client_errors.INVALID_SERVER_JSON, decoded)
+    return
+  end
+  local _ = log.debug() and log.debug('rpc.receive', decoded)
+
+  if type(decoded.method) == 'string' and decoded.id then
+    local err
+    -- Schedule here so that the users functions don't trigger an error and
+    -- we can still use the result.
+    schedule(function()
+      coroutine.wrap(function()
         local status, result
-        status, result, err = try_call(
+        status, result, err = self:try_call(
           client_errors.SERVER_REQUEST_HANDLER_ERROR,
-          dispatchers.server_request,
+          self.dispatchers.server_request,
           decoded.method,
           decoded.params
         )
@@ -491,8 +405,7 @@ local function start(cmd, cmd_args, dispatchers, extra_spawn_params)
             { status = status, result = result, err = err }
           )
         if status then
-          if not (result or err) then
-            -- TODO this can be a problem if `null` is sent for result. needs vim.NIL
+          if result == nil and err == nil then
             error(
               string.format(
                 'method %q: either a result or an error must be sent to the server in response',
@@ -516,120 +429,328 @@ local function start(cmd, cmd_args, dispatchers, extra_spawn_params)
           err = rpc_response_error(protocol.ErrorCodes.InternalError, result)
           result = nil
         end
-        send_response(decoded.id, err, result)
-      end)
-      -- This works because we are expecting vim.NIL here
-    elseif decoded.id and (decoded.result ~= vim.NIL or decoded.error ~= vim.NIL) then
-      -- We sent a number, so we expect a number.
-      local result_id = assert(tonumber(decoded.id), 'response id must be a number')
-
-      -- Notify the user that a response was received for the request
-      local notify_reply_callback = notify_reply_callbacks and notify_reply_callbacks[result_id]
-      if notify_reply_callback then
-        validate({
-          notify_reply_callback = { notify_reply_callback, 'f' },
-        })
-        notify_reply_callback(result_id)
-        notify_reply_callbacks[result_id] = nil
-      end
+        self:send_response(decoded.id, err, result)
+      end)()
+    end)
+    -- This works because we are expecting vim.NIL here
+  elseif decoded.id and (decoded.result ~= vim.NIL or decoded.error ~= vim.NIL) then
+    -- We sent a number, so we expect a number.
+    local result_id = assert(tonumber(decoded.id), 'response id must be a number')
+
+    -- Notify the user that a response was received for the request
+    local notify_reply_callbacks = self.notify_reply_callbacks
+    local notify_reply_callback = notify_reply_callbacks and notify_reply_callbacks[result_id]
+    if notify_reply_callback then
+      validate({
+        notify_reply_callback = { notify_reply_callback, 'f' },
+      })
+      notify_reply_callback(result_id)
+      notify_reply_callbacks[result_id] = nil
+    end
 
-      -- Do not surface RequestCancelled to users, it is RPC-internal.
-      if decoded.error then
-        local mute_error = false
-        if decoded.error.code == protocol.ErrorCodes.RequestCancelled then
-          local _ = log.debug() and log.debug('Received cancellation ack', decoded)
-          mute_error = true
-        end
+    local message_callbacks = self.message_callbacks
 
-        if mute_error then
-          -- Clear any callback since this is cancelled now.
-          -- This is safe to do assuming that these conditions hold:
-          -- - The server will not send a result callback after this cancellation.
-          -- - If the server sent this cancellation ACK after sending the result, the user of this RPC
-          -- client will ignore the result themselves.
-          if result_id and message_callbacks then
-            message_callbacks[result_id] = nil
-          end
-          return
+    -- Do not surface RequestCancelled to users, it is RPC-internal.
+    if decoded.error then
+      local mute_error = false
+      if decoded.error.code == protocol.ErrorCodes.RequestCancelled then
+        local _ = log.debug() and log.debug('Received cancellation ack', decoded)
+        mute_error = true
+      end
+
+      if mute_error then
+        -- Clear any callback since this is cancelled now.
+        -- This is safe to do assuming that these conditions hold:
+        -- - The server will not send a result callback after this cancellation.
+        -- - If the server sent this cancellation ACK after sending the result, the user of this RPC
+        -- client will ignore the result themselves.
+        if result_id and message_callbacks then
+          message_callbacks[result_id] = nil
         end
+        return
       end
+    end
 
-      local callback = message_callbacks and message_callbacks[result_id]
-      if callback then
-        message_callbacks[result_id] = nil
-        validate({
-          callback = { callback, 'f' },
+    local callback = message_callbacks and message_callbacks[result_id]
+    if callback then
+      message_callbacks[result_id] = nil
+      validate({
+        callback = { callback, 'f' },
+      })
+      if decoded.error then
+        decoded.error = setmetatable(decoded.error, {
+          __tostring = format_rpc_error,
         })
-        if decoded.error then
-          decoded.error = setmetatable(decoded.error, {
-            __tostring = format_rpc_error,
-          })
-        end
-        try_call(
-          client_errors.SERVER_RESULT_CALLBACK_ERROR,
-          callback,
-          decoded.error,
-          decoded.result
-        )
-      else
-        on_error(client_errors.NO_RESULT_CALLBACK_FOUND, decoded)
-        local _ = log.error()
-          and log.error('No callback found for server response id ' .. result_id)
       end
-    elseif type(decoded.method) == 'string' then
-      -- Notification
-      try_call(
-        client_errors.NOTIFICATION_HANDLER_ERROR,
-        dispatchers.notification,
-        decoded.method,
-        decoded.params
+      self:try_call(
+        client_errors.SERVER_RESULT_CALLBACK_ERROR,
+        callback,
+        decoded.error,
+        decoded.result
       )
     else
-      -- Invalid server message
-      on_error(client_errors.INVALID_SERVER_MESSAGE, decoded)
+      self:on_error(client_errors.NO_RESULT_CALLBACK_FOUND, decoded)
+      local _ = log.error() and log.error('No callback found for server response id ' .. result_id)
     end
+  elseif type(decoded.method) == 'string' then
+    -- Notification
+    self:try_call(
+      client_errors.NOTIFICATION_HANDLER_ERROR,
+      self.dispatchers.notification,
+      decoded.method,
+      decoded.params
+    )
+  else
+    -- Invalid server message
+    self:on_error(client_errors.INVALID_SERVER_MESSAGE, decoded)
   end
+end
 
-  local request_parser = coroutine.wrap(request_parser_loop)
-  request_parser()
-  stdout:read_start(function(err, chunk)
-    if err then
-      -- TODO better handling. Can these be intermittent errors?
-      on_error(client_errors.READ_ERROR, err)
-      return
-    end
-    -- This should signal that we are done reading from the client.
-    if not chunk then
-      return
-    end
-    -- Flush anything in the parser by looping until we don't get a result
-    -- anymore.
-    while true do
-      local headers, body = request_parser(chunk)
-      -- If we successfully parsed, then handle the response.
-      if headers then
-        handle_body(body)
-        -- Set chunk to empty so that we can call request_parser to get
-        -- anything existing in the parser to flush.
-        chunk = ''
+---@private
+---@return RpcClient
+local function new_client(dispatchers, transport)
+  local state = {
+    message_index = 0,
+    message_callbacks = {},
+    notify_reply_callbacks = {},
+    transport = transport,
+    dispatchers = dispatchers,
+  }
+  return setmetatable(state, { __index = Client })
+end
+
+---@private
+---@param client RpcClient
+local function public_client(client)
+  local result = {}
+
+  ---@private
+  function result.is_closing()
+    return client.transport.is_closing()
+  end
+
+  ---@private
+  function result.terminate()
+    client.transport.terminate()
+  end
+
+  --- Sends a request to the LSP server and runs {callback} upon response.
+  ---
+  ---@param method (string) The invoked LSP method
+  ---@param params (table|nil) Parameters for the invoked LSP method
+  ---@param callback (function) Callback to invoke
+  ---@param notify_reply_callback (function|nil) Callback to invoke as soon as a request is no longer pending
+  ---@returns (bool, number) `(true, message_id)` if request could be sent, `false` if not
+  function result.request(method, params, callback, notify_reply_callback)
+    return client:request(method, params, callback, notify_reply_callback)
+  end
+
+  --- Sends a notification to the LSP server.
+  ---@param method (string) The invoked LSP method
+  ---@param params (table|nil): Parameters for the invoked LSP method
+  ---@returns (bool) `true` if notification could be sent, `false` if not
+  function result.notify(method, params)
+    return client:notify(method, params)
+  end
+
+  return result
+end
+
+---@private
+local function merge_dispatchers(dispatchers)
+  if dispatchers then
+    local user_dispatchers = dispatchers
+    dispatchers = {}
+    for dispatch_name, default_dispatch in pairs(default_dispatchers) do
+      local user_dispatcher = user_dispatchers[dispatch_name]
+      if user_dispatcher then
+        if type(user_dispatcher) ~= 'function' then
+          error(string.format('dispatcher.%s must be a function', dispatch_name))
+        end
+        -- server_request is wrapped elsewhere.
+        if
+          not (dispatch_name == 'server_request' or dispatch_name == 'on_exit') -- TODO this blocks the loop exiting for some reason.
+        then
+          user_dispatcher = schedule_wrap(user_dispatcher)
+        end
+        dispatchers[dispatch_name] = user_dispatcher
       else
-        break
+        dispatchers[dispatch_name] = default_dispatch
       end
     end
-  end)
+  else
+    dispatchers = default_dispatchers
+  end
+  return dispatchers
+end
+
+--- Create a LSP RPC client factory that connects via TCP to the given host
+--- and port
+---
+---@param host string
+---@param port number
+---@return function
+local function connect(host, port)
+  return function(dispatchers)
+    dispatchers = merge_dispatchers(dispatchers)
+    local tcp = uv.new_tcp()
+    local closing = false
+    local transport = {
+      write = function(msg)
+        tcp:write(msg)
+      end,
+      is_closing = function()
+        return closing
+      end,
+      terminate = function()
+        if not closing then
+          closing = true
+          tcp:shutdown()
+          tcp:close()
+          dispatchers.on_exit(0, 0)
+        end
+      end,
+    }
+    local client = new_client(dispatchers, transport)
+    tcp:connect(host, port, function(err)
+      if err then
+        vim.schedule(function()
+          vim.notify(
+            string.format('Could not connect to %s:%s, reason: %s', host, port, vim.inspect(err)),
+            vim.log.levels.WARN
+          )
+        end)
+        return
+      end
+      local handle_body = function(body)
+        client:handle_body(body)
+      end
+      tcp:read_start(create_read_loop(handle_body, transport.terminate, function(read_err)
+        client:on_error(client_errors.READ_ERROR, read_err)
+      end))
+    end)
 
-  return {
-    pid = pid,
-    handle = handle,
-    request = request,
-    notify = notify,
+    return public_client(client)
+  end
+end
+
+--- Starts an LSP server process and create an LSP RPC client object to
+--- interact with it. Communication with the spawned process happens via stdio. For
+--- communication via TCP, spawn a process manually and use |vim.lsp.rpc.connect()|
+---
+---@param cmd (string) Command to start the LSP server.
+---@param cmd_args (table) List of additional string arguments to pass to {cmd}.
+---@param dispatchers table|nil Dispatchers for LSP message types. Valid
+---dispatcher names are:
+--- - `"notification"`
+--- - `"server_request"`
+--- - `"on_error"`
+--- - `"on_exit"`
+---@param extra_spawn_params table|nil Additional context for the LSP
+--- server process. May contain:
+--- - {cwd} (string) Working directory for the LSP server process
+--- - {env} (table) Additional environment variables for LSP server process
+---@returns Client RPC object.
+---
+---@returns Methods:
+--- - `notify()` |vim.lsp.rpc.notify()|
+--- - `request()` |vim.lsp.rpc.request()|
+--- - `is_closing()` returns a boolean indicating if the RPC is closing.
+--- - `terminate()` terminates the RPC client.
+local function start(cmd, cmd_args, dispatchers, extra_spawn_params)
+  local _ = log.info()
+    and log.info('Starting RPC client', { cmd = cmd, args = cmd_args, extra = extra_spawn_params })
+  validate({
+    cmd = { cmd, 's' },
+    cmd_args = { cmd_args, 't' },
+    dispatchers = { dispatchers, 't', true },
+  })
+
+  if extra_spawn_params and extra_spawn_params.cwd then
+    assert(is_dir(extra_spawn_params.cwd), 'cwd must be a directory')
+  end
+
+  dispatchers = merge_dispatchers(dispatchers)
+  local stdin = uv.new_pipe(false)
+  local stdout = uv.new_pipe(false)
+  local stderr = uv.new_pipe(false)
+  local handle, pid
+
+  local client = new_client(dispatchers, {
+    write = function(msg)
+      stdin:write(msg)
+    end,
+    is_closing = function()
+      return handle == nil or handle:is_closing()
+    end,
+    terminate = function()
+      if handle then
+        handle:kill(15)
+      end
+    end,
+  })
+
+  ---@private
+  --- Callback for |vim.loop.spawn()| Closes all streams and runs the `on_exit` dispatcher.
+  ---@param code (number) Exit code
+  ---@param signal (number) Signal that was used to terminate (if any)
+  local function onexit(code, signal)
+    stdin:close()
+    stdout:close()
+    stderr:close()
+    handle:close()
+    dispatchers.on_exit(code, signal)
+  end
+  local spawn_params = {
+    args = cmd_args,
+    stdio = { stdin, stdout, stderr },
+    detached = not is_win,
   }
+  if extra_spawn_params then
+    spawn_params.cwd = extra_spawn_params.cwd
+    spawn_params.env = env_merge(extra_spawn_params.env)
+    if extra_spawn_params.detached ~= nil then
+      spawn_params.detached = extra_spawn_params.detached
+    end
+  end
+  handle, pid = uv.spawn(cmd, spawn_params, onexit)
+  if handle == nil then
+    stdin:close()
+    stdout:close()
+    stderr:close()
+    local msg = string.format('Spawning language server with cmd: `%s` failed', cmd)
+    if string.match(pid, 'ENOENT') then
+      msg = msg
+        .. '. The language server is either not installed, missing from PATH, or not executable.'
+    else
+      msg = msg .. string.format(' with error message: %s', pid)
+    end
+    vim.notify(msg, vim.log.levels.WARN)
+    return
+  end
+
+  stderr:read_start(function(_, chunk)
+    if chunk then
+      local _ = log.error() and log.error('rpc', cmd, 'stderr', chunk)
+    end
+  end)
+
+  local handle_body = function(body)
+    client:handle_body(body)
+  end
+  stdout:read_start(create_read_loop(handle_body, nil, function(err)
+    client:on_error(client_errors.READ_ERROR, err)
+  end))
+
+  return public_client(client)
 end
 
 return {
   start = start,
+  connect = connect,
   rpc_response_error = rpc_response_error,
   format_rpc_error = format_rpc_error,
   client_errors = client_errors,
+  create_read_loop = create_read_loop,
 }
 -- vim:sw=2 ts=2 et