We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Server is being shutdown on server sending empty stderr line
When server triggers stderr output that is an empty string (or becomes an empty string after `rstrip` then LSP closes the transports and thus the server.

Issue found when analyzing https://github.com/sublimelsp/LSP-angular/issues/1

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `plugin/core/transports.py`
Content:
```
1 from .logging import exception_log, debug
2 from .types import ClientConfig
3 from .typing import Dict, Any, Optional, IO, Protocol
4 from abc import ABCMeta, abstractmethod
5 from contextlib import closing
6 from queue import Queue
7 import json
8 import os
9 import shutil
10 import socket
11 import sublime
12 import subprocess
13 import threading
14 import time
15 import weakref
16 
17 
18 TCP_CONNECT_TIMEOUT = 5
19 
20 
21 class Transport(metaclass=ABCMeta):
22 
23     @abstractmethod
24     def send(self, payload: Dict[str, Any]) -> None:
25         pass
26 
27     @abstractmethod
28     def close(self) -> None:
29         pass
30 
31 
32 class TransportCallbacks(Protocol):
33 
34     def on_transport_close(self, exit_code: int, exception: Optional[Exception]) -> None:
35         ...
36 
37     def on_payload(self, payload: Dict[str, Any]) -> None:
38         ...
39 
40     def on_stderr_message(self, message: str) -> None:
41         ...
42 
43 
44 class JsonRpcTransport(Transport):
45 
46     def __init__(self, name: str, process: subprocess.Popen, socket: Optional[socket.socket], reader: IO[bytes],
47                  writer: IO[bytes], stderr: Optional[IO[bytes]], callback_object: TransportCallbacks) -> None:
48         self._process = process
49         self._socket = socket
50         self._reader = reader
51         self._writer = writer
52         self._stderr = stderr
53         self._reader_thread = threading.Thread(target=self._read_loop, name='{}-reader'.format(name))
54         self._writer_thread = threading.Thread(target=self._write_loop, name='{}-writer'.format(name))
55         self._stderr_thread = threading.Thread(target=self._stderr_loop, name='{}-stderr'.format(name))
56         self._callback_object = weakref.ref(callback_object)
57         self._send_queue = Queue(0)  # type: Queue[Optional[Dict[str, Any]]]
58         self._reader_thread.start()
59         self._writer_thread.start()
60         self._stderr_thread.start()
61         self._closed = False
62 
63     def send(self, payload: Dict[str, Any]) -> None:
64         self._send_queue.put_nowait(payload)
65 
66     def close(self) -> None:
67         if not self._closed:
68             self._send_queue.put_nowait(None)
69             if self._socket:
70                 self._socket.close()
71             self._closed = True
72 
73     def _join_thread(self, t: threading.Thread) -> None:
74         if t.ident == threading.current_thread().ident:
75             return
76         try:
77             t.join(2)
78         except TimeoutError as ex:
79             exception_log("failed to join {} thread".format(t.name), ex)
80 
81     def __del__(self) -> None:
82         self.close()
83         self._join_thread(self._writer_thread)
84         self._join_thread(self._reader_thread)
85         self._join_thread(self._stderr_thread)
86 
87     def _read_loop(self) -> None:
88         try:
89             while self._reader:
90                 line = self._reader.readline()
91                 if not line:
92                     break
93                 try:
94                     num_bytes = _content_length(line)
95                 except ValueError:
96                     continue
97                 if num_bytes is None:
98                     continue
99                 while line and line.strip():
100                     line = self._reader.readline()
101                 if not line:
102                     continue
103                 body = self._reader.read(num_bytes)
104                 callback_object = self._callback_object()
105                 if callback_object:
106                     try:
107                         callback_object.on_payload(_decode(body))
108                     except Exception as ex:
109                         exception_log("Error handling payload", ex)
110                 else:
111                     break
112         except (AttributeError, BrokenPipeError):
113             pass
114         except Exception as ex:
115             exception_log("Unexpected exception", ex)
116         self._send_queue.put_nowait(None)
117 
118     def _end(self, exception: Optional[Exception]) -> None:
119         exit_code = 0
120         if not exception:
121             try:
122                 # Allow the process to stop itself.
123                 exit_code = self._process.wait(1)
124             except (AttributeError, ProcessLookupError, subprocess.TimeoutExpired):
125                 pass
126         if self._process:
127             try:
128                 # The process didn't stop itself. Terminate!
129                 self._process.kill()
130                 # still wait for the process to die, or zombie processes might be the result
131                 # Ignore the exit code in this case, it's going to be something non-zero because we sent SIGKILL.
132                 self._process.wait()
133             except (AttributeError, ProcessLookupError):
134                 pass
135             except Exception as ex:
136                 exception = ex  # TODO: Old captured exception is overwritten
137         callback_object = self._callback_object()
138         if callback_object:
139             callback_object.on_transport_close(exit_code, exception)
140 
141     def _write_loop(self) -> None:
142         exception = None  # type: Optional[Exception]
143         try:
144             while self._writer:
145                 d = self._send_queue.get()
146                 if d is None:
147                     break
148                 body = _encode(d)
149                 self._writer.writelines(("Content-Length: {}\r\n\r\n".format(len(body)).encode('ascii'), body))
150                 self._writer.flush()
151         except (BrokenPipeError, AttributeError):
152             pass
153         except Exception as ex:
154             exception = ex
155         self._end(exception)
156 
157     def _stderr_loop(self) -> None:
158         try:
159             while self._stderr:
160                 message = self._stderr.readline().decode('utf-8', 'replace').rstrip()
161                 if not message:
162                     break
163                 callback_object = self._callback_object()
164                 if callback_object:
165                     callback_object.on_stderr_message(message)
166                 else:
167                     break
168         except (BrokenPipeError, AttributeError):
169             pass
170         except Exception as ex:
171             exception_log('unexpected exception type in stderr loop', ex)
172         self._send_queue.put_nowait(None)
173 
174 
175 def create_transport(config: ClientConfig, cwd: Optional[str], window: sublime.Window,
176                      callback_object: TransportCallbacks, variables: Dict[str, str]) -> JsonRpcTransport:
177     tcp_port = None  # type: Optional[int]
178     if config.tcp_port is not None:
179         tcp_port = _find_free_port() if config.tcp_port == 0 else config.tcp_port
180     if tcp_port is not None:
181         variables["port"] = str(tcp_port)
182     args = sublime.expand_variables(config.binary_args, variables)
183     args = [os.path.expanduser(arg) for arg in args]
184     if tcp_port is not None:
185         # DEPRECATED -- replace {port} with $port or ${port} in your client config
186         args = [a.replace('{port}', str(tcp_port)) for a in args]
187     env = os.environ.copy()
188     for var, value in config.env.items():
189         env[var] = sublime.expand_variables(value, variables)
190     if tcp_port is not None:
191         stdout = subprocess.DEVNULL
192         stdin = subprocess.DEVNULL
193     else:
194         stdout = subprocess.PIPE
195         stdin = subprocess.PIPE
196     if sublime.platform() == "windows":
197         startupinfo = subprocess.STARTUPINFO()  # type: ignore
198         startupinfo.dwFlags |= subprocess.SW_HIDE | subprocess.STARTF_USESHOWWINDOW  # type: ignore
199         executable_arg = args[0]
200         fname, ext = os.path.splitext(executable_arg)
201         if len(ext) < 1:
202             path_to_executable = shutil.which(executable_arg)
203             # what extensions should we append so CreateProcess can find it?
204             # node has .cmd
205             # dart has .bat
206             # python has .exe wrappers - not needed
207             for extension in ['.cmd', '.bat']:
208                 if path_to_executable and path_to_executable.lower().endswith(extension):
209                     args[0] = executable_arg + extension
210                     break
211     else:
212         startupinfo = None
213     debug("starting {} in {}".format(args, cwd if cwd else os.getcwd()))
214     process = subprocess.Popen(
215         args=args,
216         stdin=stdin,
217         stdout=stdout,
218         stderr=subprocess.PIPE,
219         startupinfo=startupinfo,
220         env=env,
221         cwd=cwd)
222     _subprocesses.add(process)
223     sock = None  # type: Optional[socket.socket]
224     if tcp_port:
225         sock = _connect_tcp(tcp_port)
226         if sock is None:
227             raise RuntimeError("Failed to connect on port {}".format(config.tcp_port))
228         reader = sock.makefile('rwb')  # type: IO[bytes]
229         writer = reader
230     else:
231         reader = process.stdout  # type: ignore
232         writer = process.stdin  # type: ignore
233     return JsonRpcTransport(config.name, process, sock, reader, writer, process.stderr, callback_object)
234 
235 
236 _subprocesses = weakref.WeakSet()  # type: weakref.WeakSet[subprocess.Popen]
237 
238 
239 def kill_all_subprocesses() -> None:
240     global _subprocesses
241     subprocesses = list(_subprocesses)
242     for p in subprocesses:
243         try:
244             p.kill()
245         except Exception:
246             pass
247     for p in subprocesses:
248         try:
249             p.wait()
250         except Exception:
251             pass
252 
253 
254 def _connect_tcp(port: int) -> Optional[socket.socket]:
255     start_time = time.time()
256     while time.time() - start_time < TCP_CONNECT_TIMEOUT:
257         try:
258             return socket.create_connection(('localhost', port))
259         except ConnectionRefusedError:
260             pass
261     return None
262 
263 
264 def _find_free_port() -> int:
265     with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
266         s.bind(('', 0))
267         s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
268         return s.getsockname()[1]
269 
270 
271 def _encode(d: Dict[str, Any]) -> bytes:
272     return json.dumps(d, sort_keys=False, check_circular=False, separators=(',', ':')).encode('utf-8')
273 
274 
275 def _decode(message: bytes) -> Dict[str, Any]:
276     return json.loads(message.decode('utf-8'))
277 
278 
279 def _content_length(line: bytes) -> Optional[int]:
280     if line.startswith(b'Content-Length: '):
281         _, value = line.split(b'Content-Length: ')
282         value = value.strip()
283         try:
284             return int(value)
285         except ValueError as ex:
286             raise ValueError("Invalid Content-Length header: {}".format(value.decode('ascii'))) from ex
287     return None
288
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/plugin/core/transports.py b/plugin/core/transports.py
--- a/plugin/core/transports.py
+++ b/plugin/core/transports.py
@@ -158,8 +158,6 @@
         try:
             while self._stderr:
                 message = self._stderr.readline().decode('utf-8', 'replace').rstrip()
-                if not message:
-                    break
                 callback_object = self._callback_object()
                 if callback_object:
                     callback_object.on_stderr_message(message)

{"golden_diff": "diff --git a/plugin/core/transports.py b/plugin/core/transports.py\n--- a/plugin/core/transports.py\n+++ b/plugin/core/transports.py\n@@ -158,8 +158,6 @@\n         try:\n             while self._stderr:\n                 message = self._stderr.readline().decode('utf-8', 'replace').rstrip()\n-                if not message:\n-                    break\n                 callback_object = self._callback_object()\n                 if callback_object:\n                     callback_object.on_stderr_message(message)\n", "issue": "Server is being shutdown on server sending empty stderr line\nWhen server triggers stderr output that is an empty string (or becomes an empty string after `rstrip` then LSP closes the transports and thus the server.\r\n\r\nIssue found when analyzing https://github.com/sublimelsp/LSP-angular/issues/1\n", "before_files": [{"content": "from .logging import exception_log, debug\nfrom .types import ClientConfig\nfrom .typing import Dict, Any, Optional, IO, Protocol\nfrom abc import ABCMeta, abstractmethod\nfrom contextlib import closing\nfrom queue import Queue\nimport json\nimport os\nimport shutil\nimport socket\nimport sublime\nimport subprocess\nimport threading\nimport time\nimport weakref\n\n\nTCP_CONNECT_TIMEOUT = 5\n\n\nclass Transport(metaclass=ABCMeta):\n\n    @abstractmethod\n    def send(self, payload: Dict[str, Any]) -> None:\n        pass\n\n    @abstractmethod\n    def close(self) -> None:\n        pass\n\n\nclass TransportCallbacks(Protocol):\n\n    def on_transport_close(self, exit_code: int, exception: Optional[Exception]) -> None:\n        ...\n\n    def on_payload(self, payload: Dict[str, Any]) -> None:\n        ...\n\n    def on_stderr_message(self, message: str) -> None:\n        ...\n\n\nclass JsonRpcTransport(Transport):\n\n    def __init__(self, name: str, process: subprocess.Popen, socket: Optional[socket.socket], reader: IO[bytes],\n                 writer: IO[bytes], stderr: Optional[IO[bytes]], callback_object: TransportCallbacks) -> None:\n        self._process = process\n        self._socket = socket\n        self._reader = reader\n        self._writer = writer\n        self._stderr = stderr\n        self._reader_thread = threading.Thread(target=self._read_loop, name='{}-reader'.format(name))\n        self._writer_thread = threading.Thread(target=self._write_loop, name='{}-writer'.format(name))\n        self._stderr_thread = threading.Thread(target=self._stderr_loop, name='{}-stderr'.format(name))\n        self._callback_object = weakref.ref(callback_object)\n        self._send_queue = Queue(0)  # type: Queue[Optional[Dict[str, Any]]]\n        self._reader_thread.start()\n        self._writer_thread.start()\n        self._stderr_thread.start()\n        self._closed = False\n\n    def send(self, payload: Dict[str, Any]) -> None:\n        self._send_queue.put_nowait(payload)\n\n    def close(self) -> None:\n        if not self._closed:\n            self._send_queue.put_nowait(None)\n            if self._socket:\n                self._socket.close()\n            self._closed = True\n\n    def _join_thread(self, t: threading.Thread) -> None:\n        if t.ident == threading.current_thread().ident:\n            return\n        try:\n            t.join(2)\n        except TimeoutError as ex:\n            exception_log(\"failed to join {} thread\".format(t.name), ex)\n\n    def __del__(self) -> None:\n        self.close()\n        self._join_thread(self._writer_thread)\n        self._join_thread(self._reader_thread)\n        self._join_thread(self._stderr_thread)\n\n    def _read_loop(self) -> None:\n        try:\n            while self._reader:\n                line = self._reader.readline()\n                if not line:\n                    break\n                try:\n                    num_bytes = _content_length(line)\n                except ValueError:\n                    continue\n                if num_bytes is None:\n                    continue\n                while line and line.strip():\n                    line = self._reader.readline()\n                if not line:\n                    continue\n                body = self._reader.read(num_bytes)\n                callback_object = self._callback_object()\n                if callback_object:\n                    try:\n                        callback_object.on_payload(_decode(body))\n                    except Exception as ex:\n                        exception_log(\"Error handling payload\", ex)\n                else:\n                    break\n        except (AttributeError, BrokenPipeError):\n            pass\n        except Exception as ex:\n            exception_log(\"Unexpected exception\", ex)\n        self._send_queue.put_nowait(None)\n\n    def _end(self, exception: Optional[Exception]) -> None:\n        exit_code = 0\n        if not exception:\n            try:\n                # Allow the process to stop itself.\n                exit_code = self._process.wait(1)\n            except (AttributeError, ProcessLookupError, subprocess.TimeoutExpired):\n                pass\n        if self._process:\n            try:\n                # The process didn't stop itself. Terminate!\n                self._process.kill()\n                # still wait for the process to die, or zombie processes might be the result\n                # Ignore the exit code in this case, it's going to be something non-zero because we sent SIGKILL.\n                self._process.wait()\n            except (AttributeError, ProcessLookupError):\n                pass\n            except Exception as ex:\n                exception = ex  # TODO: Old captured exception is overwritten\n        callback_object = self._callback_object()\n        if callback_object:\n            callback_object.on_transport_close(exit_code, exception)\n\n    def _write_loop(self) -> None:\n        exception = None  # type: Optional[Exception]\n        try:\n            while self._writer:\n                d = self._send_queue.get()\n                if d is None:\n                    break\n                body = _encode(d)\n                self._writer.writelines((\"Content-Length: {}\\r\\n\\r\\n\".format(len(body)).encode('ascii'), body))\n                self._writer.flush()\n        except (BrokenPipeError, AttributeError):\n            pass\n        except Exception as ex:\n            exception = ex\n        self._end(exception)\n\n    def _stderr_loop(self) -> None:\n        try:\n            while self._stderr:\n                message = self._stderr.readline().decode('utf-8', 'replace').rstrip()\n                if not message:\n                    break\n                callback_object = self._callback_object()\n                if callback_object:\n                    callback_object.on_stderr_message(message)\n                else:\n                    break\n        except (BrokenPipeError, AttributeError):\n            pass\n        except Exception as ex:\n            exception_log('unexpected exception type in stderr loop', ex)\n        self._send_queue.put_nowait(None)\n\n\ndef create_transport(config: ClientConfig, cwd: Optional[str], window: sublime.Window,\n                     callback_object: TransportCallbacks, variables: Dict[str, str]) -> JsonRpcTransport:\n    tcp_port = None  # type: Optional[int]\n    if config.tcp_port is not None:\n        tcp_port = _find_free_port() if config.tcp_port == 0 else config.tcp_port\n    if tcp_port is not None:\n        variables[\"port\"] = str(tcp_port)\n    args = sublime.expand_variables(config.binary_args, variables)\n    args = [os.path.expanduser(arg) for arg in args]\n    if tcp_port is not None:\n        # DEPRECATED -- replace {port} with $port or ${port} in your client config\n        args = [a.replace('{port}', str(tcp_port)) for a in args]\n    env = os.environ.copy()\n    for var, value in config.env.items():\n        env[var] = sublime.expand_variables(value, variables)\n    if tcp_port is not None:\n        stdout = subprocess.DEVNULL\n        stdin = subprocess.DEVNULL\n    else:\n        stdout = subprocess.PIPE\n        stdin = subprocess.PIPE\n    if sublime.platform() == \"windows\":\n        startupinfo = subprocess.STARTUPINFO()  # type: ignore\n        startupinfo.dwFlags |= subprocess.SW_HIDE | subprocess.STARTF_USESHOWWINDOW  # type: ignore\n        executable_arg = args[0]\n        fname, ext = os.path.splitext(executable_arg)\n        if len(ext) < 1:\n            path_to_executable = shutil.which(executable_arg)\n            # what extensions should we append so CreateProcess can find it?\n            # node has .cmd\n            # dart has .bat\n            # python has .exe wrappers - not needed\n            for extension in ['.cmd', '.bat']:\n                if path_to_executable and path_to_executable.lower().endswith(extension):\n                    args[0] = executable_arg + extension\n                    break\n    else:\n        startupinfo = None\n    debug(\"starting {} in {}\".format(args, cwd if cwd else os.getcwd()))\n    process = subprocess.Popen(\n        args=args,\n        stdin=stdin,\n        stdout=stdout,\n        stderr=subprocess.PIPE,\n        startupinfo=startupinfo,\n        env=env,\n        cwd=cwd)\n    _subprocesses.add(process)\n    sock = None  # type: Optional[socket.socket]\n    if tcp_port:\n        sock = _connect_tcp(tcp_port)\n        if sock is None:\n            raise RuntimeError(\"Failed to connect on port {}\".format(config.tcp_port))\n        reader = sock.makefile('rwb')  # type: IO[bytes]\n        writer = reader\n    else:\n        reader = process.stdout  # type: ignore\n        writer = process.stdin  # type: ignore\n    return JsonRpcTransport(config.name, process, sock, reader, writer, process.stderr, callback_object)\n\n\n_subprocesses = weakref.WeakSet()  # type: weakref.WeakSet[subprocess.Popen]\n\n\ndef kill_all_subprocesses() -> None:\n    global _subprocesses\n    subprocesses = list(_subprocesses)\n    for p in subprocesses:\n        try:\n            p.kill()\n        except Exception:\n            pass\n    for p in subprocesses:\n        try:\n            p.wait()\n        except Exception:\n            pass\n\n\ndef _connect_tcp(port: int) -> Optional[socket.socket]:\n    start_time = time.time()\n    while time.time() - start_time < TCP_CONNECT_TIMEOUT:\n        try:\n            return socket.create_connection(('localhost', port))\n        except ConnectionRefusedError:\n            pass\n    return None\n\n\ndef _find_free_port() -> int:\n    with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:\n        s.bind(('', 0))\n        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)\n        return s.getsockname()[1]\n\n\ndef _encode(d: Dict[str, Any]) -> bytes:\n    return json.dumps(d, sort_keys=False, check_circular=False, separators=(',', ':')).encode('utf-8')\n\n\ndef _decode(message: bytes) -> Dict[str, Any]:\n    return json.loads(message.decode('utf-8'))\n\n\ndef _content_length(line: bytes) -> Optional[int]:\n    if line.startswith(b'Content-Length: '):\n        _, value = line.split(b'Content-Length: ')\n        value = value.strip()\n        try:\n            return int(value)\n        except ValueError as ex:\n            raise ValueError(\"Invalid Content-Length header: {}\".format(value.decode('ascii'))) from ex\n    return None\n", "path": "plugin/core/transports.py"}], "after_files": [{"content": "from .logging import exception_log, debug\nfrom .types import ClientConfig\nfrom .typing import Dict, Any, Optional, IO, Protocol\nfrom abc import ABCMeta, abstractmethod\nfrom contextlib import closing\nfrom queue import Queue\nimport json\nimport os\nimport shutil\nimport socket\nimport sublime\nimport subprocess\nimport threading\nimport time\nimport weakref\n\n\nTCP_CONNECT_TIMEOUT = 5\n\n\nclass Transport(metaclass=ABCMeta):\n\n    @abstractmethod\n    def send(self, payload: Dict[str, Any]) -> None:\n        pass\n\n    @abstractmethod\n    def close(self) -> None:\n        pass\n\n\nclass TransportCallbacks(Protocol):\n\n    def on_transport_close(self, exit_code: int, exception: Optional[Exception]) -> None:\n        ...\n\n    def on_payload(self, payload: Dict[str, Any]) -> None:\n        ...\n\n    def on_stderr_message(self, message: str) -> None:\n        ...\n\n\nclass JsonRpcTransport(Transport):\n\n    def __init__(self, name: str, process: subprocess.Popen, socket: Optional[socket.socket], reader: IO[bytes],\n                 writer: IO[bytes], stderr: Optional[IO[bytes]], callback_object: TransportCallbacks) -> None:\n        self._process = process\n        self._socket = socket\n        self._reader = reader\n        self._writer = writer\n        self._stderr = stderr\n        self._reader_thread = threading.Thread(target=self._read_loop, name='{}-reader'.format(name))\n        self._writer_thread = threading.Thread(target=self._write_loop, name='{}-writer'.format(name))\n        self._stderr_thread = threading.Thread(target=self._stderr_loop, name='{}-stderr'.format(name))\n        self._callback_object = weakref.ref(callback_object)\n        self._send_queue = Queue(0)  # type: Queue[Optional[Dict[str, Any]]]\n        self._reader_thread.start()\n        self._writer_thread.start()\n        self._stderr_thread.start()\n        self._closed = False\n\n    def send(self, payload: Dict[str, Any]) -> None:\n        self._send_queue.put_nowait(payload)\n\n    def close(self) -> None:\n        if not self._closed:\n            self._send_queue.put_nowait(None)\n            if self._socket:\n                self._socket.close()\n            self._closed = True\n\n    def _join_thread(self, t: threading.Thread) -> None:\n        if t.ident == threading.current_thread().ident:\n            return\n        try:\n            t.join(2)\n        except TimeoutError as ex:\n            exception_log(\"failed to join {} thread\".format(t.name), ex)\n\n    def __del__(self) -> None:\n        self.close()\n        self._join_thread(self._writer_thread)\n        self._join_thread(self._reader_thread)\n        self._join_thread(self._stderr_thread)\n\n    def _read_loop(self) -> None:\n        try:\n            while self._reader:\n                line = self._reader.readline()\n                if not line:\n                    break\n                try:\n                    num_bytes = _content_length(line)\n                except ValueError:\n                    continue\n                if num_bytes is None:\n                    continue\n                while line and line.strip():\n                    line = self._reader.readline()\n                if not line:\n                    continue\n                body = self._reader.read(num_bytes)\n                callback_object = self._callback_object()\n                if callback_object:\n                    try:\n                        callback_object.on_payload(_decode(body))\n                    except Exception as ex:\n                        exception_log(\"Error handling payload\", ex)\n                else:\n                    break\n        except (AttributeError, BrokenPipeError):\n            pass\n        except Exception as ex:\n            exception_log(\"Unexpected exception\", ex)\n        self._send_queue.put_nowait(None)\n\n    def _end(self, exception: Optional[Exception]) -> None:\n        exit_code = 0\n        if not exception:\n            try:\n                # Allow the process to stop itself.\n                exit_code = self._process.wait(1)\n            except (AttributeError, ProcessLookupError, subprocess.TimeoutExpired):\n                pass\n        if self._process:\n            try:\n                # The process didn't stop itself. Terminate!\n                self._process.kill()\n                # still wait for the process to die, or zombie processes might be the result\n                # Ignore the exit code in this case, it's going to be something non-zero because we sent SIGKILL.\n                self._process.wait()\n            except (AttributeError, ProcessLookupError):\n                pass\n            except Exception as ex:\n                exception = ex  # TODO: Old captured exception is overwritten\n        callback_object = self._callback_object()\n        if callback_object:\n            callback_object.on_transport_close(exit_code, exception)\n\n    def _write_loop(self) -> None:\n        exception = None  # type: Optional[Exception]\n        try:\n            while self._writer:\n                d = self._send_queue.get()\n                if d is None:\n                    break\n                body = _encode(d)\n                self._writer.writelines((\"Content-Length: {}\\r\\n\\r\\n\".format(len(body)).encode('ascii'), body))\n                self._writer.flush()\n        except (BrokenPipeError, AttributeError):\n            pass\n        except Exception as ex:\n            exception = ex\n        self._end(exception)\n\n    def _stderr_loop(self) -> None:\n        try:\n            while self._stderr:\n                message = self._stderr.readline().decode('utf-8', 'replace').rstrip()\n                callback_object = self._callback_object()\n                if callback_object:\n                    callback_object.on_stderr_message(message)\n                else:\n                    break\n        except (BrokenPipeError, AttributeError):\n            pass\n        except Exception as ex:\n            exception_log('unexpected exception type in stderr loop', ex)\n        self._send_queue.put_nowait(None)\n\n\ndef create_transport(config: ClientConfig, cwd: Optional[str], window: sublime.Window,\n                     callback_object: TransportCallbacks, variables: Dict[str, str]) -> JsonRpcTransport:\n    tcp_port = None  # type: Optional[int]\n    if config.tcp_port is not None:\n        tcp_port = _find_free_port() if config.tcp_port == 0 else config.tcp_port\n    if tcp_port is not None:\n        variables[\"port\"] = str(tcp_port)\n    args = sublime.expand_variables(config.binary_args, variables)\n    args = [os.path.expanduser(arg) for arg in args]\n    if tcp_port is not None:\n        # DEPRECATED -- replace {port} with $port or ${port} in your client config\n        args = [a.replace('{port}', str(tcp_port)) for a in args]\n    env = os.environ.copy()\n    for var, value in config.env.items():\n        env[var] = sublime.expand_variables(value, variables)\n    if tcp_port is not None:\n        stdout = subprocess.DEVNULL\n        stdin = subprocess.DEVNULL\n    else:\n        stdout = subprocess.PIPE\n        stdin = subprocess.PIPE\n    if sublime.platform() == \"windows\":\n        startupinfo = subprocess.STARTUPINFO()  # type: ignore\n        startupinfo.dwFlags |= subprocess.SW_HIDE | subprocess.STARTF_USESHOWWINDOW  # type: ignore\n        executable_arg = args[0]\n        fname, ext = os.path.splitext(executable_arg)\n        if len(ext) < 1:\n            path_to_executable = shutil.which(executable_arg)\n            # what extensions should we append so CreateProcess can find it?\n            # node has .cmd\n            # dart has .bat\n            # python has .exe wrappers - not needed\n            for extension in ['.cmd', '.bat']:\n                if path_to_executable and path_to_executable.lower().endswith(extension):\n                    args[0] = executable_arg + extension\n                    break\n    else:\n        startupinfo = None\n    debug(\"starting {} in {}\".format(args, cwd if cwd else os.getcwd()))\n    process = subprocess.Popen(\n        args=args,\n        stdin=stdin,\n        stdout=stdout,\n        stderr=subprocess.PIPE,\n        startupinfo=startupinfo,\n        env=env,\n        cwd=cwd)\n    _subprocesses.add(process)\n    sock = None  # type: Optional[socket.socket]\n    if tcp_port:\n        sock = _connect_tcp(tcp_port)\n        if sock is None:\n            raise RuntimeError(\"Failed to connect on port {}\".format(config.tcp_port))\n        reader = sock.makefile('rwb')  # type: IO[bytes]\n        writer = reader\n    else:\n        reader = process.stdout  # type: ignore\n        writer = process.stdin  # type: ignore\n    return JsonRpcTransport(config.name, process, sock, reader, writer, process.stderr, callback_object)\n\n\n_subprocesses = weakref.WeakSet()  # type: weakref.WeakSet[subprocess.Popen]\n\n\ndef kill_all_subprocesses() -> None:\n    global _subprocesses\n    subprocesses = list(_subprocesses)\n    for p in subprocesses:\n        try:\n            p.kill()\n        except Exception:\n            pass\n    for p in subprocesses:\n        try:\n            p.wait()\n        except Exception:\n            pass\n\n\ndef _connect_tcp(port: int) -> Optional[socket.socket]:\n    start_time = time.time()\n    while time.time() - start_time < TCP_CONNECT_TIMEOUT:\n        try:\n            return socket.create_connection(('localhost', port))\n        except ConnectionRefusedError:\n            pass\n    return None\n\n\ndef _find_free_port() -> int:\n    with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:\n        s.bind(('', 0))\n        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)\n        return s.getsockname()[1]\n\n\ndef _encode(d: Dict[str, Any]) -> bytes:\n    return json.dumps(d, sort_keys=False, check_circular=False, separators=(',', ':')).encode('utf-8')\n\n\ndef _decode(message: bytes) -> Dict[str, Any]:\n    return json.loads(message.decode('utf-8'))\n\n\ndef _content_length(line: bytes) -> Optional[int]:\n    if line.startswith(b'Content-Length: '):\n        _, value = line.split(b'Content-Length: ')\n        value = value.strip()\n        try:\n            return int(value)\n        except ValueError as ex:\n            raise ValueError(\"Invalid Content-Length header: {}\".format(value.decode('ascii'))) from ex\n    return None\n", "path": "plugin/core/transports.py"}]}