Tcp failures

CHANGELOG.md

@@ -6,6 +6,7 @@
   - implement jump target selector and jump to references ([#739])
   - implement settings UI using native JupyterLab 3.3 UI ([#778])
   - add option to show hover tooltip automatically ([#864], thanks @yamaton)
+  - add support for language servers that can (only) communicate through TCP rather than stdio (there is no support yet for servers already running on another machine and/or port) [(#636)]
 - bug fixes:
   - use correct websocket URL if configured as different from base URL ([#820], thanks @MikeSem)
   - clean up all completer styles when completer feature is disabled ([#829]).
@@ -61,6 +62,7 @@
 [#860]: https://github.com/jupyter-lsp/jupyterlab-lsp/pull/860
 [#864]: https://github.com/jupyter-lsp/jupyterlab-lsp/pull/864
 [#882]: https://github.com/jupyter-lsp/jupyterlab-lsp/pull/882
+[#636]: https://github.com/jupyter-lsp/jupyterlab-lsp/pull/636
 
 ### `@krassowski/jupyterlab-lsp 3.10.1` (2022-03-21)
 

CONTRIBUTING.md

@@ -388,8 +388,11 @@ otherwise an empty dictionary (`{}`) should be returned.
 ##### Common Concerns
 
 - some language servers need to have their connection mode specified
-  - the `stdio` interface is the only one supported by `jupyter_lsp`
-    - PRs welcome to support other modes!
+  - `jupyter_lsp` currently supports the `stdio` and `tcp` interface
+  - the mode used by `jupyter_lsp` to connect to the language server can be specified by including `mode="stdio"` or `mode="tcp"` in the language server `spec`-dictionary
+  - currently it is not possible to connect to externally running language servers via tcp, but only to servers spawned by `jupyter_lsp` as given by the `argv` specs entry
+    - PRs welcome to support externally running language servers!
+  - use the placeholder `{port}` within the `argv` entry to allow `jupyter_lsp` to specify the port on which to launch the language server
 - because of its VSCode heritage, many language servers use `nodejs`
   - `LanguageServerManager.nodejs` will provide the location of our best
     guess at where a user's `nodejs` might be found

atest/ports.py

@@ -9,7 +9,7 @@ def get_unused_port():
     Probably could introduce race conditions if inside a tight loop.
     """
     sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-    sock.bind(("localhost", 0))
+    sock.bind(("127.0.0.1", 0))
     sock.listen(1)
     port = sock.getsockname()[1]
     sock.close()

python_packages/jupyter_lsp/jupyter_lsp/connection.py

@@ -0,0 +1,169 @@
+""" Language Server readers and writers
+
+Parts of this code are derived from:
+
+> https://github.com/palantir/python-jsonrpc-server/blob/0.2.0/pyls_jsonrpc/streams.py#L83   # noqa
+> https://github.com/palantir/python-jsonrpc-server/blob/45ed1931e4b2e5100cc61b3992c16d6f68af2e80/pyls_jsonrpc/streams.py  # noqa
+> > MIT License   https://github.com/palantir/python-jsonrpc-server/blob/0.2.0/LICENSE
+> > Copyright 2018 Palantir Technologies, Inc.
+"""
+from abc import ABC, ABCMeta, abstractmethod
+from typing import Text
+
+# pylint: disable=broad-except
+import anyio
+from anyio.streams.buffered import BufferedByteReceiveStream
+from anyio.streams.stapled import StapledObjectStream
+from anyio.streams.text import TextSendStream
+from tornado.httputil import HTTPHeaders
+from traitlets import Instance, Int
+from traitlets.config import LoggingConfigurable
+from traitlets.traitlets import MetaHasTraits
+
+
+class LspStreamMeta(MetaHasTraits, ABCMeta):
+    pass
+
+
+class LspStreamBase(LoggingConfigurable, ABC, metaclass=LspStreamMeta):
+    """Non-blocking, queued base for communicating with Language Servers through anyio
+    streams
+    """
+
+    queue = Instance(StapledObjectStream, help="queue to get/put")
+
+    def __repr__(self):  # pragma: no cover
+        return "<{}(parent={})>".format(self.__class__.__name__, self.parent)
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self.log.debug("%s initialized", self)
+
+    @abstractmethod
+    async def close(self):
+        pass  # pragma: no cover
+
+
+class LspStreamReader(LspStreamBase):
+    """Language Server Reader"""
+
+    receive_max_bytes = Int(
+        65536,
+        help="the maximum size a header line send by the language server may have",
+    ).tag(config=True)
+
+    stream = Instance(  # type:ignore[assignment]
+        BufferedByteReceiveStream, help="the stream to read from"
+    )  # type: BufferedByteReceiveStream
+
+    def __init__(self, stream: anyio.abc.ByteReceiveStream, **kwargs):
+        super().__init__(**kwargs)
+        self.stream = BufferedByteReceiveStream(stream)
+
+    async def close(self):
+        await self.stream.aclose()
+        self.log.debug("%s closed", self)
+
+    async def read(self) -> None:
+        """Read from a Language Server until it is closed"""
+        while True:
+            message = None
+            try:
+                message = await self.read_one()
+                await self.queue.send(message)
+            except anyio.ClosedResourceError:
+                # stream was closed -> terminate
+                self.log.debug("Stream closed while a read was still in progress")
+                break
+            except Exception as e:  # pragma: no cover
+                self.log.exception(
+                    "%s couldn't enqueue message: %s (%s)", self, message, e
+                )
+
+    async def _read_content(self, length: int) -> bytes:
+        """Read the full length of the message.
+
+        Args:
+           - length: the content length
+        """
+        try:
+            return await self.stream.receive_exactly(length)
+        except anyio.IncompleteRead:  # pragma: no cover
+            # resource has been closed before the requested bytes could be retrieved
+            # -> signal recource closed
+            raise anyio.ClosedResourceError
+
+    async def read_one(self) -> Text:
+        """Read a single message"""
+        message = ""
+        headers = HTTPHeaders()
+
+        line = await self._readline()
+
+        if line:
+            while line and line.strip():
+                headers.parse_line(line)
+                line = await self._readline()
+
+            content_length = int(headers.get("content-length", "0"))
+
+            if content_length:
+                raw = await self._read_content(length=content_length)
+                message = raw.decode("utf-8").strip()
+
+        return message
+
+    async def _readline(self) -> Text:
+        """Read a line"""
+        try:
+            # use same max_bytes as is default for receive for now. It seems there is no
+            # way of getting the bytes read until max_bytes is reached, so we cannot
+            # iterate the receive_until call with smaller max_bytes values
+            line = await self.stream.receive_until(b"\r\n", self.receive_max_bytes)
+            return line.decode("utf-8").strip()
+        except anyio.IncompleteRead:
+            # resource has been closed before the requested bytes could be retrieved
+            # -> signal recource closed
+            raise anyio.ClosedResourceError
+        except anyio.DelimiterNotFound:  # pragma: no cover
+            self.log.error(
+                "Readline hit max_bytes before newline character was encountered"
+            )
+            return ""
+
+
+class LspStreamWriter(LspStreamBase):
+    """Language Server Writer"""
+
+    stream = Instance(  # type:ignore[assignment]
+        TextSendStream, help="the stream to write to"
+    )  # type: TextSendStream
+
+    def __init__(self, stream: anyio.abc.ByteSendStream, **kwargs):
+        super().__init__(**kwargs)
+        self.stream = TextSendStream(stream, encoding="utf-8")
+
+    async def close(self):
+        await self.stream.aclose()
+        self.log.debug("%s closed", self)
+
+    async def write(self) -> None:
+        """Write to a Language Server until it closes"""
+        while True:
+            message = await self.queue.receive()
+            try:
+                n_bytes = len(message.encode("utf-8"))
+                response = "Content-Length: {}\r\n\r\n{}".format(n_bytes, message)
+                await self._write_one(response)
+            except (
+                anyio.ClosedResourceError,
+                anyio.BrokenResourceError,
+            ):  # pragma: no cover
+                # stream was closed -> terminate
+                self.log.debug("Stream closed while a write was still in progress")
+                break
+            except Exception:  # pragma: no cover
+                self.log.exception("%s couldn't write message: %s", self, response)
+
+    async def _write_one(self, message) -> None:
+        await self.stream.send(message)

python_packages/jupyter_lsp/jupyter_lsp/manager.py

@@ -1,4 +1,4 @@
-""" A configurable frontend for stdio-based Language Servers
+""" A configurable frontend for stream-based Language Servers
 """
 import asyncio
 import os
@@ -35,7 +35,11 @@
     EP_SPEC_V1,
 )
 from .schema import LANGUAGE_SERVER_SPEC_MAP
-from .session import LanguageServerSession
+from .session import (
+    LanguageServerSessionBase,
+    LanguageServerSessionStdio,
+    LanguageServerSessionTCP,
+)
 from .trait_types import LoadableCallable, Schema
 from .types import (
     KeyedLanguageServerSpecs,
@@ -68,10 +72,10 @@ class LanguageServerManager(LanguageServerManagerAPI):
     )  # type: bool
 
     sessions = Dict_(  # type:ignore[assignment]
-        trait=Instance(LanguageServerSession),
+        trait=Instance(LanguageServerSessionBase),
         default_value={},
         help="sessions keyed by language server name",
-    )  # type: Dict[Tuple[Text], LanguageServerSession]
+    )  # type: Dict[Tuple[Text], LanguageServerSessionBase]
 
     virtual_documents_dir = Unicode(
         help="""Path to virtual documents relative to the content manager root
@@ -160,9 +164,21 @@ def init_sessions(self):
         """create, but do not initialize all sessions"""
         sessions = {}
         for language_server, spec in self.language_servers.items():
-            sessions[language_server] = LanguageServerSession(
-                language_server=language_server, spec=spec, parent=self
-            )
+            mode = spec.get("mode", "stdio")
+            if mode == "stdio":
+                sessions[language_server] = LanguageServerSessionStdio(
+                    language_server=language_server, spec=spec, parent=self
+                )
+            elif mode == "tcp":
+                sessions[language_server] = LanguageServerSessionTCP(
+                    language_server=language_server, spec=spec, parent=self
+                )
+            else:  # pragma: no cover
+                raise ValueError(
+                    "Unknown session mode {} for language server '{}'".format(
+                        mode, language_server
+                    )
+                )
         self.sessions = sessions
 
     def init_listeners(self):

python_packages/jupyter_lsp/jupyter_lsp/schema/schema.json

@@ -139,6 +139,13 @@
           "description": "list of MIME types supported by the language server",
           "title": "MIME Types"
         },
+        "mode": {
+          "description": "connection mode used, e.g. stdio (default), tcp",
+          "title": "Mode",
+          "type": "string",
+          "enum": ["stdio", "tcp"],
+          "default": "stdio"
+        },
         "troubleshoot": {
           "type": "string",
           "description": "information on troubleshooting the installation or auto-detection of the language server",