1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
|
import logging
from tornado import web, ioloop, websocket
from pylsp_jsonrpc import dispatchers, endpoint
try:
import ujson as json
except Exception: # pylint: disable=broad-except
import json
log = logging.getLogger(__name__)
class LanguageServer(dispatchers.MethodDispatcher):
"""Implement a JSON RPC method dispatcher for the language server protocol."""
def __init__(self):
# Endpoint is lazily set after construction
self.endpoint = None
def m_initialize(self, rootUri=None, **kwargs):
log.info("Got initialize params: %s", kwargs)
return {"capabilities": {
"textDocumentSync": {
"openClose": True,
}
}}
def m_text_document__did_open(self, textDocument=None, **_kwargs):
log.info("Opened text document %s", textDocument)
self.endpoint.notify('textDocument/publishDiagnostics', {
'uri': textDocument['uri'],
'diagnostics': [{
'range': {
'start': {'line': 0, 'character': 0},
'end': {'line': 1, 'character': 0},
},
'message': 'Some very bad Python code',
'severity': 1 # DiagnosticSeverity.Error
}]
})
class LanguageServerWebSocketHandler(websocket.WebSocketHandler):
"""Setup tornado websocket handler to host language server."""
def __init__(self, *args, **kwargs):
# Create an instance of the language server used to dispatch JSON RPC methods
langserver = LanguageServer()
# Setup an endpoint that dispatches to the ls, and writes server->client messages
# back to the client websocket
self.endpoint = endpoint.Endpoint(langserver, lambda msg: self.write_message(json.dumps(msg)))
# Give the language server a handle to the endpoint so it can send JSON RPC
# notifications and requests.
langserver.endpoint = self.endpoint
super(LanguageServerWebSocketHandler, self).__init__(*args, **kwargs)
def on_message(self, message):
"""Forward client->server messages to the endpoint."""
self.endpoint.consume(json.loads(message))
def check_origin(self, origin):
return True
if __name__ == "__main__":
app = web.Application([
(r"/python", LanguageServerWebSocketHandler),
])
app.listen(3000, address='127.0.0.1')
ioloop.IOLoop.current().start()
|
