We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Race condition in JsonMessageChannel
```py
    def send_request(self, command, arguments=None):
        d = {'command': command}
        if arguments is not None:
            d['arguments'] = arguments
        seq = self._send_message('request', d)
        request = Request(self, seq)
        with self._lock:
            self._requests[seq] = request
        return request
```
Note that the message is sent before the requests dict is updated. If it goes fast enough, the response handler will receive a response to an "unknown" request.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `ptvsd/messaging.py`
Content:
```
1 # Copyright (c) Microsoft Corporation. All rights reserved.
2 # Licensed under the MIT License. See LICENSE in the project root
3 # for license information.
4 
5 from __future__ import print_function, with_statement, absolute_import
6 
7 import collections
8 import itertools
9 import json
10 import sys
11 import threading
12 
13 
14 class JsonIOStream(object):
15     """Implements a JSON value stream over two byte streams (input and output).
16 
17     Each value is encoded as a packet consisting of a header and a body, as defined by the
18     Debug Adapter Protocol (https://microsoft.github.io/debug-adapter-protocol/overview).
19     """
20 
21     MAX_BODY_SIZE = 0xFFFFFF
22 
23     @classmethod
24     def from_stdio(cls):
25         if sys.version_info >= (3,):
26             stdin = sys.stdin.buffer
27             stdout = sys.stdout.buffer
28         else:
29             stdin = sys.stdin
30             stdout = sys.stdout
31             if sys.platform == 'win32':
32                 import os, msvcrt
33                 msvcrt.setmode(stdin.fileno(), os.O_BINARY)
34                 msvcrt.setmode(stdout.fileno(), os.O_BINARY)
35         return cls(stdin, stdout)
36 
37     @classmethod
38     def from_socket(cls, socket):
39         if socket.gettimeout() is not None:
40             raise ValueError('Socket must be in blocking mode')
41         socket_io = socket.makefile('rwb', 0)
42         return cls(socket_io, socket_io)
43 
44     def __init__(self, reader, writer):
45         """Creates a new JsonIOStream.
46 
47         reader is a BytesIO-like object from which incoming messages are read;
48         reader.readline() must treat '\n' as the line terminator, and must leave
49         '\r' as is (i.e. it must not translate '\r\n' to just plain '\n'!).
50 
51         writer is a BytesIO-like object to which outgoing messages are written.
52         """
53         self._reader = reader
54         self._writer = writer
55         self._is_closing = False
56 
57     def close(self):
58         self._is_closing = True
59         self._reader.close()
60         self._writer.close()
61 
62     def _read_line(self):
63         line = b''
64         while True:
65             line += self._reader.readline()
66             if not line:
67                 raise EOFError
68             if line.endswith(b'\r\n'):
69                 line = line[0:-2]
70                 return line
71 
72     def read_json(self):
73         """Read a single JSON value from reader.
74 
75         Returns JSON value as parsed by json.loads(), or raises EOFError
76         if there are no more objects to be read.
77         """
78 
79         headers = {}
80         while True:
81             try:
82                 line = self._read_line()
83             except Exception:
84                 if self._is_closing:
85                     raise EOFError
86                 else:
87                     raise
88 
89             if line == b'':
90                 break
91             key, _, value = line.partition(b':')
92             headers[key] = value
93 
94         try:
95             length = int(headers[b'Content-Length'])
96             if not (0 <= length <= self.MAX_BODY_SIZE):
97                 raise ValueError
98         except (KeyError, ValueError):
99             raise IOError('Content-Length is missing or invalid')
100 
101         try:
102             body = self._reader.read(length)
103         except Exception:
104             if self._is_closing:
105                 raise EOFError
106             else:
107                 raise
108 
109         if isinstance(body, bytes):
110             body = body.decode('utf-8')
111         return json.loads(body)
112 
113     def write_json(self, value):
114         """Write a single JSON object to writer.
115 
116         object must be in the format suitable for json.dump().
117         """
118 
119         body = json.dumps(value, sort_keys=True)
120         if not isinstance(body, bytes):
121             body = body.encode('utf-8')
122 
123         header = 'Content-Length: %d\r\n\r\n' % len(body)
124         if not isinstance(header, bytes):
125             header = header.encode('ascii')
126 
127         self._writer.write(header)
128         self._writer.write(body)
129 
130 
131 Response = collections.namedtuple('Response', ('success', 'command', 'error_message', 'body'))
132 Response.__new__.__defaults__ = (None, None)
133 class Response(Response):
134     """Represents a received response to a Request."""
135 
136 
137 class RequestFailure(Exception):
138     def __init__(self, message):
139         self.message = message
140 
141 
142 class Request(object):
143     """Represents a request that was sent to the other party, and is awaiting or has
144     already received a response.
145     """
146 
147     def __init__(self, channel, seq):
148         self.channel = channel
149         self.seq = seq
150         self.response = None
151         self._lock = threading.Lock()
152         self._got_response = threading.Event()
153         self._callback = lambda _: None
154 
155     def _handle_response(self, success, command, error_message=None, body=None):
156         assert self.response is None
157         with self._lock:
158             response = Response(success, command, error_message, body)
159             self.response = response
160             callback = self._callback
161         callback(response)
162         self._got_response.set()
163 
164     def wait_for_response(self, raise_if_failed=True):
165         """Waits until a response is received for this request, records that
166         response as a new Response object accessible via self.response,
167         and returns self.response.body.
168 
169         If raise_if_failed is True, and the received response does not indicate
170         success, raises RequestFailure. Otherwise, self.response.success has to
171         be inspected to determine whether the request failed or succeeded, since
172         self.response.body can be None in either case.
173         """
174 
175         self._got_response.wait()
176         if raise_if_failed and not self.response.success:
177             raise RequestFailure(self.response.error_message)
178         return self.response
179 
180     def on_response(self, callback):
181         """Registers a callback to invoke when a response is received for this
182         request. If response was already received, invokes callback immediately.
183         Callback is invoked with Response object as the sole argument.
184 
185         The callback is invoked on an unspecified background thread that performs
186         processing of incoming messages; therefore, no further message processing
187         occurs until the callback returns.
188         """
189 
190         with self._lock:
191             response = self.response
192             if response is None:
193                 self._callback = callback
194                 return
195         callback(response)
196 
197 
198 class JsonMessageChannel(object):
199     """Implements a JSON message channel on top of a JSON stream, with
200     support for generic Request, Response and Event messages as defined by the
201     Debug Adapter Protocol (https://microsoft.github.io/debug-adapter-protocol/overview).
202     """
203 
204     def __init__(self, stream, handlers=None):
205         self.stream = stream
206         self.send_callback = lambda channel, message: None
207         self.receive_callback = lambda channel, message: None
208         self._lock = threading.Lock()
209         self._stop = threading.Event()
210         self._seq_iter = itertools.count(1)
211         self._requests = {}
212         self._handlers = handlers
213         self._worker = threading.Thread(target=self._process_incoming_messages)
214         self._worker.daemon = True
215 
216     def close(self):
217         self.stream.close()
218 
219     def start(self):
220         self._worker.start()
221 
222     def wait(self):
223         self._worker.join()
224 
225     def _send_message(self, type, rest={}):
226         with self._lock:
227             seq = next(self._seq_iter)
228         message = {
229             'seq': seq,
230             'type': type,
231         }
232         message.update(rest)
233         with self._lock:
234             self.stream.write_json(message)
235         self.send_callback(self, message)
236         return seq
237 
238     def send_request(self, command, arguments=None):
239         d = {'command': command}
240         if arguments is not None:
241             d['arguments'] = arguments
242         seq = self._send_message('request', d)
243         request = Request(self, seq)
244         with self._lock:
245             self._requests[seq] = request
246         return request
247 
248     def send_event(self, event, body=None):
249         d = {'event': event}
250         if body is not None:
251             d['body'] = body
252         self._send_message('event', d)
253 
254     def send_response(self, request_seq, success, command, error_message=None, body=None):
255         d = {
256             'request_seq': request_seq,
257             'success': success,
258             'command': command,
259         }
260         if success:
261             if body is not None:
262                 d['body'] = body
263         else:
264             if error_message is not None:
265                 d['message'] = error_message
266         self._send_message('response', d)
267 
268     def on_message(self, message):
269         self.receive_callback(self, message)
270         seq = message['seq']
271         typ = message['type']
272         if typ == 'request':
273             command = message['command']
274             arguments = message.get('arguments', None)
275             self.on_request(seq, command, arguments)
276         elif typ == 'event':
277             event = message['event']
278             body = message.get('body', None)
279             self.on_event(seq, event, body)
280         elif typ == 'response':
281             request_seq = message['request_seq']
282             success = message['success']
283             command = message['command']
284             error_message = message.get('message', None)
285             body = message.get('body', None)
286             self.on_response(seq, request_seq, success, command, error_message, body)
287         else:
288             raise IOError('Incoming message has invalid "type":\n%r' % message)
289 
290     def on_request(self, seq, command, arguments):
291         handler_name = '%s_request' % command
292         specific_handler = getattr(self._handlers, handler_name, None)
293         if specific_handler is not None:
294             handler = lambda: specific_handler(self, arguments)
295         else:
296             generic_handler = getattr(self._handlers, 'request')
297             handler = lambda: generic_handler(self, command, arguments)
298         try:
299             response_body = handler()
300         except Exception as ex:
301             self.send_response(seq, False, command, str(ex))
302         else:
303             self.send_response(seq, True, command, None, response_body)
304 
305     def on_event(self, seq, event, body):
306         handler_name = '%s_event' % event
307         specific_handler = getattr(self._handlers, handler_name, None)
308         if specific_handler is not None:
309             handler = lambda: specific_handler(self, body)
310         else:
311             generic_handler = getattr(self._handlers, 'event')
312             handler = lambda: generic_handler(self, event, body)
313         handler()
314 
315     def on_response(self, seq, request_seq, success, command, error_message, body):
316         try:
317             with self._lock:
318                 request = self._requests.pop(request_seq)
319         except KeyError:
320             raise KeyError('Received response to unknown request %d', request_seq)
321         return request._handle_response(success, command, error_message, body)
322 
323     def _process_incoming_messages(self):
324         while True:
325             try:
326                 message = self.stream.read_json()
327             except EOFError:
328                 break
329             try:
330                 self.on_message(message)
331             except Exception:
332                 print('Error while processing message:\n%r\n\n' % message, file=sys.__stderr__)
333                 raise
334 
335 
336 class MessageHandlers(object):
337     """A simple delegating message handlers object for use with JsonMessageChannel.
338     For every argument provided, the object has an attribute with the corresponding
339     name and value. Example:
340     """
341 
342     def __init__(self, **kwargs):
343         for name, func in kwargs.items():
344             setattr(self, name, func)
```
--- 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/ptvsd/messaging.py b/ptvsd/messaging.py
--- a/ptvsd/messaging.py
+++ b/ptvsd/messaging.py
@@ -5,6 +5,7 @@
 from __future__ import print_function, with_statement, absolute_import
 
 import collections
+import contextlib
 import itertools
 import json
 import sys
@@ -222,6 +223,7 @@
     def wait(self):
         self._worker.join()
 
+    @contextlib.contextmanager
     def _send_message(self, type, rest={}):
         with self._lock:
             seq = next(self._seq_iter)
@@ -231,17 +233,16 @@
         }
         message.update(rest)
         with self._lock:
+            yield seq
             self.stream.write_json(message)
         self.send_callback(self, message)
-        return seq
 
     def send_request(self, command, arguments=None):
         d = {'command': command}
         if arguments is not None:
             d['arguments'] = arguments
-        seq = self._send_message('request', d)
-        request = Request(self, seq)
-        with self._lock:
+        with self._send_message('request', d) as seq:
+            request = Request(self, seq)
             self._requests[seq] = request
         return request
 
@@ -249,7 +250,8 @@
         d = {'event': event}
         if body is not None:
             d['body'] = body
-        self._send_message('event', d)
+        with self._send_message('event', d):
+            pass
 
     def send_response(self, request_seq, success, command, error_message=None, body=None):
         d = {
@@ -263,7 +265,8 @@
         else:
             if error_message is not None:
                 d['message'] = error_message
-        self._send_message('response', d)
+        with self._send_message('response', d):
+            pass
 
     def on_message(self, message):
         self.receive_callback(self, message)

{"golden_diff": "diff --git a/ptvsd/messaging.py b/ptvsd/messaging.py\n--- a/ptvsd/messaging.py\n+++ b/ptvsd/messaging.py\n@@ -5,6 +5,7 @@\n from __future__ import print_function, with_statement, absolute_import\n \n import collections\n+import contextlib\n import itertools\n import json\n import sys\n@@ -222,6 +223,7 @@\n     def wait(self):\n         self._worker.join()\n \n+    @contextlib.contextmanager\n     def _send_message(self, type, rest={}):\n         with self._lock:\n             seq = next(self._seq_iter)\n@@ -231,17 +233,16 @@\n         }\n         message.update(rest)\n         with self._lock:\n+            yield seq\n             self.stream.write_json(message)\n         self.send_callback(self, message)\n-        return seq\n \n     def send_request(self, command, arguments=None):\n         d = {'command': command}\n         if arguments is not None:\n             d['arguments'] = arguments\n-        seq = self._send_message('request', d)\n-        request = Request(self, seq)\n-        with self._lock:\n+        with self._send_message('request', d) as seq:\n+            request = Request(self, seq)\n             self._requests[seq] = request\n         return request\n \n@@ -249,7 +250,8 @@\n         d = {'event': event}\n         if body is not None:\n             d['body'] = body\n-        self._send_message('event', d)\n+        with self._send_message('event', d):\n+            pass\n \n     def send_response(self, request_seq, success, command, error_message=None, body=None):\n         d = {\n@@ -263,7 +265,8 @@\n         else:\n             if error_message is not None:\n                 d['message'] = error_message\n-        self._send_message('response', d)\n+        with self._send_message('response', d):\n+            pass\n \n     def on_message(self, message):\n         self.receive_callback(self, message)\n", "issue": "Race condition in JsonMessageChannel\n```py\r\n    def send_request(self, command, arguments=None):\r\n        d = {'command': command}\r\n        if arguments is not None:\r\n            d['arguments'] = arguments\r\n        seq = self._send_message('request', d)\r\n        request = Request(self, seq)\r\n        with self._lock:\r\n            self._requests[seq] = request\r\n        return request\r\n```\r\nNote that the message is sent before the requests dict is updated. If it goes fast enough, the response handler will receive a response to an \"unknown\" request.\n", "before_files": [{"content": "# Copyright (c) Microsoft Corporation. All rights reserved.\n# Licensed under the MIT License. See LICENSE in the project root\n# for license information.\n\nfrom __future__ import print_function, with_statement, absolute_import\n\nimport collections\nimport itertools\nimport json\nimport sys\nimport threading\n\n\nclass JsonIOStream(object):\n    \"\"\"Implements a JSON value stream over two byte streams (input and output).\n\n    Each value is encoded as a packet consisting of a header and a body, as defined by the\n    Debug Adapter Protocol (https://microsoft.github.io/debug-adapter-protocol/overview).\n    \"\"\"\n\n    MAX_BODY_SIZE = 0xFFFFFF\n\n    @classmethod\n    def from_stdio(cls):\n        if sys.version_info >= (3,):\n            stdin = sys.stdin.buffer\n            stdout = sys.stdout.buffer\n        else:\n            stdin = sys.stdin\n            stdout = sys.stdout\n            if sys.platform == 'win32':\n                import os, msvcrt\n                msvcrt.setmode(stdin.fileno(), os.O_BINARY)\n                msvcrt.setmode(stdout.fileno(), os.O_BINARY)\n        return cls(stdin, stdout)\n\n    @classmethod\n    def from_socket(cls, socket):\n        if socket.gettimeout() is not None:\n            raise ValueError('Socket must be in blocking mode')\n        socket_io = socket.makefile('rwb', 0)\n        return cls(socket_io, socket_io)\n\n    def __init__(self, reader, writer):\n        \"\"\"Creates a new JsonIOStream.\n\n        reader is a BytesIO-like object from which incoming messages are read;\n        reader.readline() must treat '\\n' as the line terminator, and must leave\n        '\\r' as is (i.e. it must not translate '\\r\\n' to just plain '\\n'!).\n\n        writer is a BytesIO-like object to which outgoing messages are written.\n        \"\"\"\n        self._reader = reader\n        self._writer = writer\n        self._is_closing = False\n\n    def close(self):\n        self._is_closing = True\n        self._reader.close()\n        self._writer.close()\n\n    def _read_line(self):\n        line = b''\n        while True:\n            line += self._reader.readline()\n            if not line:\n                raise EOFError\n            if line.endswith(b'\\r\\n'):\n                line = line[0:-2]\n                return line\n\n    def read_json(self):\n        \"\"\"Read a single JSON value from reader.\n\n        Returns JSON value as parsed by json.loads(), or raises EOFError\n        if there are no more objects to be read.\n        \"\"\"\n\n        headers = {}\n        while True:\n            try:\n                line = self._read_line()\n            except Exception:\n                if self._is_closing:\n                    raise EOFError\n                else:\n                    raise\n\n            if line == b'':\n                break\n            key, _, value = line.partition(b':')\n            headers[key] = value\n\n        try:\n            length = int(headers[b'Content-Length'])\n            if not (0 <= length <= self.MAX_BODY_SIZE):\n                raise ValueError\n        except (KeyError, ValueError):\n            raise IOError('Content-Length is missing or invalid')\n\n        try:\n            body = self._reader.read(length)\n        except Exception:\n            if self._is_closing:\n                raise EOFError\n            else:\n                raise\n\n        if isinstance(body, bytes):\n            body = body.decode('utf-8')\n        return json.loads(body)\n\n    def write_json(self, value):\n        \"\"\"Write a single JSON object to writer.\n\n        object must be in the format suitable for json.dump().\n        \"\"\"\n\n        body = json.dumps(value, sort_keys=True)\n        if not isinstance(body, bytes):\n            body = body.encode('utf-8')\n\n        header = 'Content-Length: %d\\r\\n\\r\\n' % len(body)\n        if not isinstance(header, bytes):\n            header = header.encode('ascii')\n\n        self._writer.write(header)\n        self._writer.write(body)\n\n\nResponse = collections.namedtuple('Response', ('success', 'command', 'error_message', 'body'))\nResponse.__new__.__defaults__ = (None, None)\nclass Response(Response):\n    \"\"\"Represents a received response to a Request.\"\"\"\n\n\nclass RequestFailure(Exception):\n    def __init__(self, message):\n        self.message = message\n\n\nclass Request(object):\n    \"\"\"Represents a request that was sent to the other party, and is awaiting or has\n    already received a response.\n    \"\"\"\n\n    def __init__(self, channel, seq):\n        self.channel = channel\n        self.seq = seq\n        self.response = None\n        self._lock = threading.Lock()\n        self._got_response = threading.Event()\n        self._callback = lambda _: None\n\n    def _handle_response(self, success, command, error_message=None, body=None):\n        assert self.response is None\n        with self._lock:\n            response = Response(success, command, error_message, body)\n            self.response = response\n            callback = self._callback\n        callback(response)\n        self._got_response.set()\n\n    def wait_for_response(self, raise_if_failed=True):\n        \"\"\"Waits until a response is received for this request, records that\n        response as a new Response object accessible via self.response,\n        and returns self.response.body.\n\n        If raise_if_failed is True, and the received response does not indicate\n        success, raises RequestFailure. Otherwise, self.response.success has to\n        be inspected to determine whether the request failed or succeeded, since\n        self.response.body can be None in either case.\n        \"\"\"\n\n        self._got_response.wait()\n        if raise_if_failed and not self.response.success:\n            raise RequestFailure(self.response.error_message)\n        return self.response\n\n    def on_response(self, callback):\n        \"\"\"Registers a callback to invoke when a response is received for this\n        request. If response was already received, invokes callback immediately.\n        Callback is invoked with Response object as the sole argument.\n\n        The callback is invoked on an unspecified background thread that performs\n        processing of incoming messages; therefore, no further message processing\n        occurs until the callback returns.\n        \"\"\"\n\n        with self._lock:\n            response = self.response\n            if response is None:\n                self._callback = callback\n                return\n        callback(response)\n\n\nclass JsonMessageChannel(object):\n    \"\"\"Implements a JSON message channel on top of a JSON stream, with\n    support for generic Request, Response and Event messages as defined by the\n    Debug Adapter Protocol (https://microsoft.github.io/debug-adapter-protocol/overview).\n    \"\"\"\n\n    def __init__(self, stream, handlers=None):\n        self.stream = stream\n        self.send_callback = lambda channel, message: None\n        self.receive_callback = lambda channel, message: None\n        self._lock = threading.Lock()\n        self._stop = threading.Event()\n        self._seq_iter = itertools.count(1)\n        self._requests = {}\n        self._handlers = handlers\n        self._worker = threading.Thread(target=self._process_incoming_messages)\n        self._worker.daemon = True\n\n    def close(self):\n        self.stream.close()\n\n    def start(self):\n        self._worker.start()\n\n    def wait(self):\n        self._worker.join()\n\n    def _send_message(self, type, rest={}):\n        with self._lock:\n            seq = next(self._seq_iter)\n        message = {\n            'seq': seq,\n            'type': type,\n        }\n        message.update(rest)\n        with self._lock:\n            self.stream.write_json(message)\n        self.send_callback(self, message)\n        return seq\n\n    def send_request(self, command, arguments=None):\n        d = {'command': command}\n        if arguments is not None:\n            d['arguments'] = arguments\n        seq = self._send_message('request', d)\n        request = Request(self, seq)\n        with self._lock:\n            self._requests[seq] = request\n        return request\n\n    def send_event(self, event, body=None):\n        d = {'event': event}\n        if body is not None:\n            d['body'] = body\n        self._send_message('event', d)\n\n    def send_response(self, request_seq, success, command, error_message=None, body=None):\n        d = {\n            'request_seq': request_seq,\n            'success': success,\n            'command': command,\n        }\n        if success:\n            if body is not None:\n                d['body'] = body\n        else:\n            if error_message is not None:\n                d['message'] = error_message\n        self._send_message('response', d)\n\n    def on_message(self, message):\n        self.receive_callback(self, message)\n        seq = message['seq']\n        typ = message['type']\n        if typ == 'request':\n            command = message['command']\n            arguments = message.get('arguments', None)\n            self.on_request(seq, command, arguments)\n        elif typ == 'event':\n            event = message['event']\n            body = message.get('body', None)\n            self.on_event(seq, event, body)\n        elif typ == 'response':\n            request_seq = message['request_seq']\n            success = message['success']\n            command = message['command']\n            error_message = message.get('message', None)\n            body = message.get('body', None)\n            self.on_response(seq, request_seq, success, command, error_message, body)\n        else:\n            raise IOError('Incoming message has invalid \"type\":\\n%r' % message)\n\n    def on_request(self, seq, command, arguments):\n        handler_name = '%s_request' % command\n        specific_handler = getattr(self._handlers, handler_name, None)\n        if specific_handler is not None:\n            handler = lambda: specific_handler(self, arguments)\n        else:\n            generic_handler = getattr(self._handlers, 'request')\n            handler = lambda: generic_handler(self, command, arguments)\n        try:\n            response_body = handler()\n        except Exception as ex:\n            self.send_response(seq, False, command, str(ex))\n        else:\n            self.send_response(seq, True, command, None, response_body)\n\n    def on_event(self, seq, event, body):\n        handler_name = '%s_event' % event\n        specific_handler = getattr(self._handlers, handler_name, None)\n        if specific_handler is not None:\n            handler = lambda: specific_handler(self, body)\n        else:\n            generic_handler = getattr(self._handlers, 'event')\n            handler = lambda: generic_handler(self, event, body)\n        handler()\n\n    def on_response(self, seq, request_seq, success, command, error_message, body):\n        try:\n            with self._lock:\n                request = self._requests.pop(request_seq)\n        except KeyError:\n            raise KeyError('Received response to unknown request %d', request_seq)\n        return request._handle_response(success, command, error_message, body)\n\n    def _process_incoming_messages(self):\n        while True:\n            try:\n                message = self.stream.read_json()\n            except EOFError:\n                break\n            try:\n                self.on_message(message)\n            except Exception:\n                print('Error while processing message:\\n%r\\n\\n' % message, file=sys.__stderr__)\n                raise\n\n\nclass MessageHandlers(object):\n    \"\"\"A simple delegating message handlers object for use with JsonMessageChannel.\n    For every argument provided, the object has an attribute with the corresponding\n    name and value. Example:\n    \"\"\"\n\n    def __init__(self, **kwargs):\n        for name, func in kwargs.items():\n            setattr(self, name, func)", "path": "ptvsd/messaging.py"}], "after_files": [{"content": "# Copyright (c) Microsoft Corporation. All rights reserved.\n# Licensed under the MIT License. See LICENSE in the project root\n# for license information.\n\nfrom __future__ import print_function, with_statement, absolute_import\n\nimport collections\nimport contextlib\nimport itertools\nimport json\nimport sys\nimport threading\n\n\nclass JsonIOStream(object):\n    \"\"\"Implements a JSON value stream over two byte streams (input and output).\n\n    Each value is encoded as a packet consisting of a header and a body, as defined by the\n    Debug Adapter Protocol (https://microsoft.github.io/debug-adapter-protocol/overview).\n    \"\"\"\n\n    MAX_BODY_SIZE = 0xFFFFFF\n\n    @classmethod\n    def from_stdio(cls):\n        if sys.version_info >= (3,):\n            stdin = sys.stdin.buffer\n            stdout = sys.stdout.buffer\n        else:\n            stdin = sys.stdin\n            stdout = sys.stdout\n            if sys.platform == 'win32':\n                import os, msvcrt\n                msvcrt.setmode(stdin.fileno(), os.O_BINARY)\n                msvcrt.setmode(stdout.fileno(), os.O_BINARY)\n        return cls(stdin, stdout)\n\n    @classmethod\n    def from_socket(cls, socket):\n        if socket.gettimeout() is not None:\n            raise ValueError('Socket must be in blocking mode')\n        socket_io = socket.makefile('rwb', 0)\n        return cls(socket_io, socket_io)\n\n    def __init__(self, reader, writer):\n        \"\"\"Creates a new JsonIOStream.\n\n        reader is a BytesIO-like object from which incoming messages are read;\n        reader.readline() must treat '\\n' as the line terminator, and must leave\n        '\\r' as is (i.e. it must not translate '\\r\\n' to just plain '\\n'!).\n\n        writer is a BytesIO-like object to which outgoing messages are written.\n        \"\"\"\n        self._reader = reader\n        self._writer = writer\n        self._is_closing = False\n\n    def close(self):\n        self._is_closing = True\n        self._reader.close()\n        self._writer.close()\n\n    def _read_line(self):\n        line = b''\n        while True:\n            line += self._reader.readline()\n            if not line:\n                raise EOFError\n            if line.endswith(b'\\r\\n'):\n                line = line[0:-2]\n                return line\n\n    def read_json(self):\n        \"\"\"Read a single JSON value from reader.\n\n        Returns JSON value as parsed by json.loads(), or raises EOFError\n        if there are no more objects to be read.\n        \"\"\"\n\n        headers = {}\n        while True:\n            try:\n                line = self._read_line()\n            except Exception:\n                if self._is_closing:\n                    raise EOFError\n                else:\n                    raise\n\n            if line == b'':\n                break\n            key, _, value = line.partition(b':')\n            headers[key] = value\n\n        try:\n            length = int(headers[b'Content-Length'])\n            if not (0 <= length <= self.MAX_BODY_SIZE):\n                raise ValueError\n        except (KeyError, ValueError):\n            raise IOError('Content-Length is missing or invalid')\n\n        try:\n            body = self._reader.read(length)\n        except Exception:\n            if self._is_closing:\n                raise EOFError\n            else:\n                raise\n\n        if isinstance(body, bytes):\n            body = body.decode('utf-8')\n        return json.loads(body)\n\n    def write_json(self, value):\n        \"\"\"Write a single JSON object to writer.\n\n        object must be in the format suitable for json.dump().\n        \"\"\"\n\n        body = json.dumps(value, sort_keys=True)\n        if not isinstance(body, bytes):\n            body = body.encode('utf-8')\n\n        header = 'Content-Length: %d\\r\\n\\r\\n' % len(body)\n        if not isinstance(header, bytes):\n            header = header.encode('ascii')\n\n        self._writer.write(header)\n        self._writer.write(body)\n\n\nResponse = collections.namedtuple('Response', ('success', 'command', 'error_message', 'body'))\nResponse.__new__.__defaults__ = (None, None)\nclass Response(Response):\n    \"\"\"Represents a received response to a Request.\"\"\"\n\n\nclass RequestFailure(Exception):\n    def __init__(self, message):\n        self.message = message\n\n\nclass Request(object):\n    \"\"\"Represents a request that was sent to the other party, and is awaiting or has\n    already received a response.\n    \"\"\"\n\n    def __init__(self, channel, seq):\n        self.channel = channel\n        self.seq = seq\n        self.response = None\n        self._lock = threading.Lock()\n        self._got_response = threading.Event()\n        self._callback = lambda _: None\n\n    def _handle_response(self, success, command, error_message=None, body=None):\n        assert self.response is None\n        with self._lock:\n            response = Response(success, command, error_message, body)\n            self.response = response\n            callback = self._callback\n        callback(response)\n        self._got_response.set()\n\n    def wait_for_response(self, raise_if_failed=True):\n        \"\"\"Waits until a response is received for this request, records that\n        response as a new Response object accessible via self.response,\n        and returns self.response.body.\n\n        If raise_if_failed is True, and the received response does not indicate\n        success, raises RequestFailure. Otherwise, self.response.success has to\n        be inspected to determine whether the request failed or succeeded, since\n        self.response.body can be None in either case.\n        \"\"\"\n\n        self._got_response.wait()\n        if raise_if_failed and not self.response.success:\n            raise RequestFailure(self.response.error_message)\n        return self.response\n\n    def on_response(self, callback):\n        \"\"\"Registers a callback to invoke when a response is received for this\n        request. If response was already received, invokes callback immediately.\n        Callback is invoked with Response object as the sole argument.\n\n        The callback is invoked on an unspecified background thread that performs\n        processing of incoming messages; therefore, no further message processing\n        occurs until the callback returns.\n        \"\"\"\n\n        with self._lock:\n            response = self.response\n            if response is None:\n                self._callback = callback\n                return\n        callback(response)\n\n\nclass JsonMessageChannel(object):\n    \"\"\"Implements a JSON message channel on top of a JSON stream, with\n    support for generic Request, Response and Event messages as defined by the\n    Debug Adapter Protocol (https://microsoft.github.io/debug-adapter-protocol/overview).\n    \"\"\"\n\n    def __init__(self, stream, handlers=None):\n        self.stream = stream\n        self.send_callback = lambda channel, message: None\n        self.receive_callback = lambda channel, message: None\n        self._lock = threading.Lock()\n        self._stop = threading.Event()\n        self._seq_iter = itertools.count(1)\n        self._requests = {}\n        self._handlers = handlers\n        self._worker = threading.Thread(target=self._process_incoming_messages)\n        self._worker.daemon = True\n\n    def close(self):\n        self.stream.close()\n\n    def start(self):\n        self._worker.start()\n\n    def wait(self):\n        self._worker.join()\n\n    @contextlib.contextmanager\n    def _send_message(self, type, rest={}):\n        with self._lock:\n            seq = next(self._seq_iter)\n        message = {\n            'seq': seq,\n            'type': type,\n        }\n        message.update(rest)\n        with self._lock:\n            yield seq\n            self.stream.write_json(message)\n        self.send_callback(self, message)\n\n    def send_request(self, command, arguments=None):\n        d = {'command': command}\n        if arguments is not None:\n            d['arguments'] = arguments\n        with self._send_message('request', d) as seq:\n            request = Request(self, seq)\n            self._requests[seq] = request\n        return request\n\n    def send_event(self, event, body=None):\n        d = {'event': event}\n        if body is not None:\n            d['body'] = body\n        with self._send_message('event', d):\n            pass\n\n    def send_response(self, request_seq, success, command, error_message=None, body=None):\n        d = {\n            'request_seq': request_seq,\n            'success': success,\n            'command': command,\n        }\n        if success:\n            if body is not None:\n                d['body'] = body\n        else:\n            if error_message is not None:\n                d['message'] = error_message\n        with self._send_message('response', d):\n            pass\n\n    def on_message(self, message):\n        self.receive_callback(self, message)\n        seq = message['seq']\n        typ = message['type']\n        if typ == 'request':\n            command = message['command']\n            arguments = message.get('arguments', None)\n            self.on_request(seq, command, arguments)\n        elif typ == 'event':\n            event = message['event']\n            body = message.get('body', None)\n            self.on_event(seq, event, body)\n        elif typ == 'response':\n            request_seq = message['request_seq']\n            success = message['success']\n            command = message['command']\n            error_message = message.get('message', None)\n            body = message.get('body', None)\n            self.on_response(seq, request_seq, success, command, error_message, body)\n        else:\n            raise IOError('Incoming message has invalid \"type\":\\n%r' % message)\n\n    def on_request(self, seq, command, arguments):\n        handler_name = '%s_request' % command\n        specific_handler = getattr(self._handlers, handler_name, None)\n        if specific_handler is not None:\n            handler = lambda: specific_handler(self, arguments)\n        else:\n            generic_handler = getattr(self._handlers, 'request')\n            handler = lambda: generic_handler(self, command, arguments)\n        try:\n            response_body = handler()\n        except Exception as ex:\n            self.send_response(seq, False, command, str(ex))\n        else:\n            self.send_response(seq, True, command, None, response_body)\n\n    def on_event(self, seq, event, body):\n        handler_name = '%s_event' % event\n        specific_handler = getattr(self._handlers, handler_name, None)\n        if specific_handler is not None:\n            handler = lambda: specific_handler(self, body)\n        else:\n            generic_handler = getattr(self._handlers, 'event')\n            handler = lambda: generic_handler(self, event, body)\n        handler()\n\n    def on_response(self, seq, request_seq, success, command, error_message, body):\n        try:\n            with self._lock:\n                request = self._requests.pop(request_seq)\n        except KeyError:\n            raise KeyError('Received response to unknown request %d', request_seq)\n        return request._handle_response(success, command, error_message, body)\n\n    def _process_incoming_messages(self):\n        while True:\n            try:\n                message = self.stream.read_json()\n            except EOFError:\n                break\n            try:\n                self.on_message(message)\n            except Exception:\n                print('Error while processing message:\\n%r\\n\\n' % message, file=sys.__stderr__)\n                raise\n\n\nclass MessageHandlers(object):\n    \"\"\"A simple delegating message handlers object for use with JsonMessageChannel.\n    For every argument provided, the object has an attribute with the corresponding\n    name and value. Example:\n    \"\"\"\n\n    def __init__(self, **kwargs):\n        for name, func in kwargs.items():\n            setattr(self, name, func)", "path": "ptvsd/messaging.py"}]}