wip

pygls/client/_native.py

@@ -0,0 +1,90 @@
+############################################################################
+# Copyright(c) Open Law Library. All rights reserved.                      #
+# See ThirdPartyNotices.txt in the project root for additional notices.    #
+#                                                                          #
+# Licensed under the Apache License, Version 2.0 (the "License")           #
+# you may not use this file except in compliance with the License.         #
+# You may obtain a copy of the License at                                  #
+#                                                                          #
+#     http: // www.apache.org/licenses/LICENSE-2.0                         #
+#                                                                          #
+# Unless required by applicable law or agreed to in writing, software      #
+# distributed under the License is distributed on an "AS IS" BASIS,        #
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. #
+# See the License for the specific language governing permissions and      #
+# limitations under the License.                                           #
+############################################################################
+import asyncio
+from typing import Optional
+from typing import Type
+
+from pygls.handler._native import JsonRPCHandler
+from pygls.handler._native import aio_main
+from pygls.protocol.next import JsonRPCProtocol
+
+
+class JsonRPCClient(JsonRPCHandler):
+    """Base JSON-RPC client for "native" runtimes"""
+
+    def __init__(
+        self, *args, protocol_cls: Type[JsonRPCProtocol] = JsonRPCProtocol, **kwargs
+    ):
+        super().__init__(*args, protocol=protocol_cls(), **kwargs)
+
+        self._server: Optional[asyncio.subprocess.Process] = None
+
+    @property
+    def stopped(self) -> bool:
+        """Return ``True`` if the client has been stopped."""
+        return self._stop_event.is_set()
+
+    async def start_io(self, cmd: str, *args, **kwargs):
+        """Start the given server and communicate with it over stdio."""
+
+        self.logger.debug("Starting server process: %s", " ".join([cmd, *args]))
+        server = await asyncio.create_subprocess_exec(
+            cmd,
+            *args,
+            stdout=asyncio.subprocess.PIPE,
+            stdin=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+            **kwargs,
+        )
+
+        assert server.stdout is not None, "Missing server stdout"
+        assert server.stdin is not None, "Missing server stdin"
+
+        self._writer = server.stdin
+        self._tasks["<<client-server-connection>>"] = asyncio.create_task(
+            aio_main(
+                reader=server.stdout,
+                stop_event=self._stop_event,
+                message_handler=self,
+            )
+        )
+        self._tasks["<<exit-notificaiton>>"] = asyncio.create_task(self._server_exit())
+        self._server = server
+
+    async def _server_exit(self):
+        if self._server is not None:
+            await self._server.wait()
+            self.logger.debug(
+                "Server process %s exited with return code: %s",
+                self._server.pid,
+                self._server.returncode,
+            )
+            await self.server_exit(self._server)
+            self._stop_event.set()
+
+    async def server_exit(self, server: asyncio.subprocess.Process):
+        """Called when the server process exits."""
+
+    async def stop(self):
+        self._stop_event.set()
+
+        if self._server is not None and self._server.returncode is None:
+            self.logger.debug("Terminating server process: %s", self._server.pid)
+            self._server.terminate()
+
+        if len(self._tasks) > 0:
+            await asyncio.gather(*self._tasks.values())