Merge pull request #19916 from mfussenegger/lsp-tcp

Adds TCP support for lsp.
Usage example:

```
vim.lsp.start({ name = 'godot', cmd = vim.lsp.rpc.connect('127.0.0.1', 6008) })
```

runtime/doc/lsp.txt

@@ -825,7 +825,7 @@ start({config}, {opts})                                      *vim.lsp.start()*
                     re-uses a client if name and root_dir matches.
 
     Return: ~
-        (number) client_id
+        (number|nil) client_id
 
 start_client({config})                                *vim.lsp.start_client()*
     Starts and initializes a client with the given configuration.
@@ -835,9 +835,16 @@ start_client({config})                                *vim.lsp.start_client()*
     The following parameters describe fields in the {config} table.
 
     Parameters: ~
-        {cmd}                (required, string or list treated like
+        {cmd}                (table|string|fun(dispatchers: table):table)
-                             |jobstart()|) Base command that initiates the LSP
+                             command string or list treated like |jobstart|.
-                             client.
+                             The command must launch the language server
+                             process. `cmd` can also be a function that
+                             creates an RPC client. The function receives a
+                             dispatchers table and must return a table with
+                             the functions `request`, `notify`, `is_closing`
+                             and `terminate` See |vim.lsp.rpc.request| and
+                             |vim.lsp.rpc.notify| For TCP there is a built-in
+                             rpc client factory: |vim.lsp.rpc.connect|
         {cmd_cwd}            (string, default=|getcwd()|) Directory to launch
                              the `cmd` process. Not related to `root_dir`.
         {cmd_env}            (table) Environment flags to pass to the LSP on
@@ -1890,6 +1897,17 @@ should_log({level})                                 *vim.lsp.log.should_log()*
 ==============================================================================
 Lua module: vim.lsp.rpc                                              *lsp-rpc*
 
+connect({host}, {port})                                *vim.lsp.rpc.connect()*
+    Create a LSP RPC client factory that connects via TCP to the given host
+    and port
+
+    Parameters: ~
+        {host}  (string)
+        {port}  (number)
+
+    Return: ~
+        (function)
+
 format_rpc_error({err})                       *vim.lsp.rpc.format_rpc_error()*
     Constructs an error message from an LSP error object.
 
@@ -1904,7 +1922,7 @@ notify({method}, {params})                              *vim.lsp.rpc.notify()*
 
     Parameters: ~
         {method}  (string) The invoked LSP method
-        {params}  (table): Parameters for the invoked LSP method
+        {params}  (table|nil): Parameters for the invoked LSP method
 
     Return: ~
         (bool) `true` if notification could be sent, `false` if not
@@ -1915,7 +1933,8 @@ request({method}, {params}, {callback}, {notify_reply_callback})
 
     Parameters: ~
         {method}                 (string) The invoked LSP method
-        {params}                 (table) Parameters for the invoked LSP method
+        {params}                 (table|nil) Parameters for the invoked LSP
+                                 method
         {callback}               (function) Callback to invoke
         {notify_reply_callback}  (function|nil) Callback to invoke as soon as
                                  a request is no longer pending

runtime/lua/vim/lsp.lua

@@ -289,7 +289,12 @@ local function validate_client_config(config)
     'flags.debounce_text_changes must be a number with the debounce time in milliseconds'
   )
 
-  local cmd, cmd_args = lsp._cmd_parts(config.cmd)
+  local cmd, cmd_args
+  if type(config.cmd) == 'function' then
+    cmd = config.cmd
+  else
+    cmd, cmd_args = lsp._cmd_parts(config.cmd)
+  end
   local offset_encoding = valid_encodings.UTF16
   if config.offset_encoding then
     offset_encoding = validate_encoding(config.offset_encoding)
@@ -855,14 +860,17 @@ end
 ---                            Used on all running clients.
 ---                            The default implementation re-uses a client if name
 ---                            and root_dir matches.
----@return number client_id
+---@return number|nil client_id
 function lsp.start(config, opts)
   opts = opts or {}
   local reuse_client = opts.reuse_client
     or function(client, conf)
       return client.config.root_dir == conf.root_dir and client.name == conf.name
     end
-  config.name = config.name or (config.cmd[1] and vim.fs.basename(config.cmd[1])) or nil
+  config.name = config.name
+  if not config.name and type(config.cmd) == 'table' then
+    config.name = config.cmd[1] and vim.fs.basename(config.cmd[1]) or nil
+  end
   local bufnr = api.nvim_get_current_buf()
   for _, clients in ipairs({ uninitialized_clients, lsp.get_active_clients() }) do
     for _, client in pairs(clients) do
@@ -893,8 +901,13 @@ end
 --- The following parameters describe fields in the {config} table.
 ---
 ---
----@param cmd: (required, string or list treated like |jobstart()|) Base command
+---@param cmd: (table|string|fun(dispatchers: table):table) command string or
---- that initiates the LSP client.
+--- list treated like |jobstart|. The command must launch the language server
+--- process. `cmd` can also be a function that creates an RPC client.
+--- The function receives a dispatchers table and must return a table with the
+--- functions `request`, `notify`, `is_closing` and `terminate`
+--- See |vim.lsp.rpc.request| and |vim.lsp.rpc.notify|
+--- For TCP there is a built-in rpc client factory: |vim.lsp.rpc.connect|
 ---
 ---@param cmd_cwd: (string, default=|getcwd()|) Directory to launch
 --- the `cmd` process. Not related to `root_dir`.
@@ -1164,11 +1177,16 @@ function lsp.start_client(config)
   end
 
   -- Start the RPC client.
-  local rpc = lsp_rpc.start(cmd, cmd_args, dispatch, {
+  local rpc
+  if type(cmd) == 'function' then
+    rpc = cmd(dispatch)
+  else
+    rpc = lsp_rpc.start(cmd, cmd_args, dispatch, {
       cwd = config.cmd_cwd,
       env = config.cmd_env,
       detached = config.detached,
     })
+  end
 
   -- Return nil if client fails to start
   if not rpc then
@@ -1464,14 +1482,13 @@ function lsp.start_client(config)
   --- you request to stop a client which has previously been requested to
   --- shutdown, it will automatically escalate and force shutdown.
   ---
-  ---@param force (bool, optional)
+  ---@param force boolean|nil
   function client.stop(force)
-    local handle = rpc.handle
+    if rpc.is_closing() then
-    if handle:is_closing() then
       return
     end
     if force or not client.initialized or graceful_shutdown_failed then
-      handle:kill(15)
+      rpc.terminate()
       return
     end
     -- Sending a signal after a process has exited is acceptable.
@@ -1480,7 +1497,7 @@ function lsp.start_client(config)
         rpc.notify('exit')
       else
         -- If there was an error in the shutdown request, then term to be safe.
-        handle:kill(15)
+        rpc.terminate()
         graceful_shutdown_failed = true
       end
     end)
@@ -1492,7 +1509,7 @@ function lsp.start_client(config)
   ---@returns (bool) true if client is stopped or in the process of being
   ---stopped; false otherwise
   function client.is_stopped()
-    return rpc.handle:is_closing()
+    return rpc.is_closing()
   end
 
   ---@private

runtime/lua/vim/lsp/rpc.lua

@@ -241,6 +241,399 @@ function default_dispatchers.on_error(code, err)
   local _ = log.error() and log.error('client_error:', client_errors[code], err)
 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
+
+    if not chunk then
+      if on_no_chunk then
+        on_no_chunk()
+      end
+      return
+    end
+
+    while true do
+      local headers, body = parse_chunk(chunk)
+      if headers then
+        handle_body(body)
+        chunk = ''
+      else
+        break
+      end
+    end
+  end
+end
+
+---@class RpcClient
+---@field message_index number
+---@field message_callbacks table
+---@field notify_reply_callbacks table
+---@field transport table
+---@field dispatchers table
+
+---@class RpcClient
+local Client = {}
+
+---@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
+
+---@private
+--- 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 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
+
+---@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
+function Client:pcall_handler(errkind, status, head, ...)
+  if not status then
+    self:on_error(errkind, head, ...)
+    return status, head
+  end
+  return status, head, ...
+end
+
+---@private
+function Client:try_call(errkind, fn, ...)
+  return self:pcall_handler(errkind, pcall(fn, ...))
+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: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()
+      local status, result
+      status, result, err = self:try_call(
+        client_errors.SERVER_REQUEST_HANDLER_ERROR,
+        self.dispatchers.server_request,
+        decoded.method,
+        decoded.params
+      )
+      local _ = log.debug()
+        and log.debug(
+          'server_request: callback result',
+          { 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
+          error(
+            string.format(
+              'method %q: either a result or an error must be sent to the server in response',
+              decoded.method
+            )
+          )
+        end
+        if err then
+          assert(
+            type(err) == 'table',
+            'err must be a table. Use rpc_response_error to help format errors.'
+          )
+          local code_name = assert(
+            protocol.ErrorCodes[err.code],
+            'Errors must use protocol.ErrorCodes. Use rpc_response_error to help format errors.'
+          )
+          err.message = err.message or code_name
+        end
+      else
+        -- On an exception, result will contain the error message.
+        err = rpc_response_error(protocol.ErrorCodes.InternalError, result)
+        result = nil
+      end
+      self: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_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
+
+    local message_callbacks = self.message_callbacks
+
+    -- 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' },
+      })
+      if decoded.error then
+        decoded.error = setmetatable(decoded.error, {
+          __tostring = format_rpc_error,
+        })
+      end
+      self:try_call(
+        client_errors.SERVER_RESULT_CALLBACK_ERROR,
+        callback,
+        decoded.error,
+        decoded.result
+      )
+    else
+      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
+
+---@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
+        dispatchers[dispatch_name] = default_dispatch
+      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 public_client(client)
+  end
+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.
 ---
@@ -278,40 +671,27 @@ local function start(cmd, cmd_args, dispatchers, extra_spawn_params)
   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
-    end
-  else
-    dispatchers = default_dispatchers
-  end
 
+  dispatchers = merge_dispatchers(dispatchers)
   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
+
+  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
@@ -321,9 +701,6 @@ local function start(cmd, cmd_args, dispatchers, extra_spawn_params)
     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 = {
@@ -353,87 +730,6 @@ local function start(cmd, cmd_args, dispatchers, extra_spawn_params)
     vim.notify(msg, vim.log.levels.WARN)
     return
   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
-
-  -- 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
-
-  ---@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,
-    })
-  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
-    else
-      return false
-    end
-  end
 
   stderr:read_start(function(_, chunk)
     if chunk then
@@ -441,195 +737,22 @@ local function start(cmd, cmd_args, dispatchers, extra_spawn_params)
     end
   end)
 
-  ---@private
+  local handle_body = function(body)
-  local function on_error(errkind, ...)
+    client:handle_body(body)
-    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, ...))
   end
+  stdout:read_start(create_read_loop(handle_body, nil, function(err)
+    client:on_error(client_errors.READ_ERROR, err)
+  end))
 
-  -- TODO periodically check message_callbacks for old requests past a certain
+  return public_client(client)
-  -- time and log them. This would require storing the timestamp. I could call
-  -- them with an error then, perhaps.
-
-  ---@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)
-
-    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()
-        local status, result
-        status, result, err = try_call(
-          client_errors.SERVER_REQUEST_HANDLER_ERROR,
-          dispatchers.server_request,
-          decoded.method,
-          decoded.params
-        )
-        local _ = log.debug()
-          and log.debug(
-            'server_request: callback result',
-            { 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
-            error(
-              string.format(
-                'method %q: either a result or an error must be sent to the server in response',
-                decoded.method
-              )
-            )
-          end
-          if err then
-            assert(
-              type(err) == 'table',
-              'err must be a table. Use rpc_response_error to help format errors.'
-            )
-            local code_name = assert(
-              protocol.ErrorCodes[err.code],
-              'Errors must use protocol.ErrorCodes. Use rpc_response_error to help format errors.'
-            )
-            err.message = err.message or code_name
-          end
-        else
-          -- On an exception, result will contain the error message.
-          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
-
-      -- 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' },
-        })
-        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
-      )
-    else
-      -- Invalid server message
-      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 = ''
-      else
-        break
-      end
-    end
-  end)
-
-  return {
-    pid = pid,
-    handle = handle,
-    request = request,
-    notify = notify,
-  }
 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

test/functional/plugin/lsp_spec.lua

@@ -3181,4 +3181,32 @@ describe('LSP', function()
       }
     end)
   end)
+
+  describe('cmd', function()
+    it('can connect to lsp server via rpc.connect', function()
+      local result = exec_lua [[
+        local uv = vim.loop
+        local server = uv.new_tcp()
+        local init = nil
+        server:bind('127.0.0.1', 0)
+        server:listen(127, function(err)
+          assert(not err, err)
+          local socket = uv.new_tcp()
+          server:accept(socket)
+          socket:read_start(require('vim.lsp.rpc').create_read_loop(function(body)
+            init = body
+            socket:close()
+          end))
+        end)
+        local port = server:getsockname().port
+        vim.lsp.start({ name = 'dummy', cmd = vim.lsp.rpc.connect('127.0.0.1', port) })
+        vim.wait(1000, function() return init ~= nil end)
+        assert(init, "server must receive `initialize` request")
+        server:close()
+        server:shutdown()
+        return vim.json.decode(init)
+      ]]
+      eq(result.method, "initialize")
+    end)
+  end)
 end)