<gh_stars>0
"""This module handle client"""

import datetime
import io
import json
import logging
import os
import re
import socket
import subprocess
import time
from collections import OrderedDict
from collections.abc import Mapping, MutableMapping
from itertools import chain

from . import errors

LOGGER = logging.getLogger(__name__)
# LOGGER.setLevel(logging.DEBUG)
STREAM_HANDLER = logging.StreamHandler()
LOG_TEMPLATE = "%(levelname)s %(asctime)s %(filename)s:%(lineno)s  %(message)s"
STREAM_HANDLER.setFormatter(logging.Formatter(LOG_TEMPLATE))
LOGGER.addHandler(STREAM_HANDLER)


class CaseInsensitiveDict(MutableMapping):
    """A case-insensitive ``dict``-like object"""

    def __init__(self, data=None, **kwargs):
        self._store = OrderedDict()
        if data is None:
            data = {}
        self.update(data, **kwargs)

    def __setitem__(self, key, value):
        # Use the lowercased key for lookups, but store the actual
        # key alongside the value.
        self._store[key.lower()] = (key, value)

    def __getitem__(self, key):
        return self._store[key.lower()][1]

    def __delitem__(self, key):
        del self._store[key.lower()]

    def __iter__(self):
        return (casedkey for casedkey, mappedvalue in self._store.values())

    def __len__(self):
        return len(self._store)

    def lower_items(self):
        """Like iteritems(), but with all lowercase keys."""
        return ((lowerkey, keyval[1]) for (lowerkey, keyval) in self._store.items())

    def __eq__(self, other):
        if isinstance(other, Mapping):
            other = CaseInsensitiveDict(other)
        else:
            return NotImplemented
        # Compare insensitively
        return dict(self.lower_items()) == dict(other.lower_items())

    # Copy is required
    def copy(self):
        return CaseInsensitiveDict(self._store.values())

    def __repr__(self):
        return str(dict(self.items()))


class Headers(CaseInsensitiveDict):
    """handle headers item"""

    item_separator = b"\r\n"
    item_pattern = re.compile(r"^(.+): (.+)$")
    encoding = "ascii"

    def to_bytes(self):
        """convert to bytes"""

        lines = ("%s: %s" % (key, value) for key, value in self.items())
        return str.encode("\r\n".join(lines), self.encoding)

    @classmethod
    def from_bytes(cls, header: bytes):
        """load header from bytes"""

        decoded_header = bytes.decode(header, cls.encoding)
        items = {}
        for line in decoded_header.splitlines():
            match = cls.item_pattern.match(line)
            if match:
                items[match.group(1)] = match.group(2)

        return cls(items)


class Message:
    """message interface"""

    encoding = "utf8"
    body_separator = b"\r\n\r\n"

    @property
    def message_body(self):
        """interface get message body"""
        return ""

    def to_bytes(self):
        """convert to bytes"""
        encoded = self.message_body.encode(self.encoding)
        header = Headers({"Content-Length": len(encoded), "Encoding": self.encoding})
        return self.body_separator.join((header.to_bytes(), encoded))

    @staticmethod
    def build_message(obj, params: dict) -> "Message":
        """build message body implementation"""
        raise NotImplementedError("build_message not implemented")

    @classmethod
    def from_bytes(cls, message: bytes) -> "Message":
        """build message from bytes"""
        try:
            header, content = message.split(cls.body_separator)
            parsed_header = Headers.from_bytes(header)

            size = int(parsed_header["Content-Length"])
            if len(content) != size:
                raise errors.InvalidSize("want %d, expected %d" % (size, len(content)))

            encoding = parsed_header.get("encoding", cls.encoding)
            decoded = json.loads(content.decode(encoding))
            message = cls.build_message(cls, decoded)

        except (ValueError, KeyError) as err:
            raise errors.InvalidMessage(err) from err

        else:
            return message


class RequestMessage(Message):
    """RequestMessage"""

    def __init__(self, id_: str, method: str, params: dict = None):
        self.id_ = id_
        self.method = method
        self.params = params

    @property
    def _message(self):
        return {"id": self.id_, "method": self.method, "params": self.params}

    def __repr__(self):
        return str(self._message)

    @property
    def message_body(self):
        return json.dumps(self._message)

    @staticmethod
    def build_message(obj, params: dict):
        """build message body implementation"""
        return obj(params["id"], params["method"], params["params"])


class ResponseError(dict):
    """Response Error"""

    @classmethod
    def new(cls, code: str, message: str, **kwargs):
        """new response error"""
        return cls(code=code, message=message, **kwargs)

    @property
    def message(self):
        """message property"""
        return str(self["message"])

    @classmethod
    def from_rpc(cls, params: dict):
        """from rpc"""

        if not params:
            return None

        code = params["code"]
        message = params["message"]
        options = {k: v for (k, v) in params.items() if k not in ("code", "message")}

        return cls(code=code, message=message, **options)


class ResponseMessage(Message):
    """ResponseMessage"""

    def __init__(self, id_: str, results: object = None, error: ResponseError = None):
        self.id_ = id_
        self.results = results
        self.error = error

    @property
    def _message(self):
        return {"id": self.id_, "results": self.results, "error": self.error}

    def __repr__(self):
        return str(self._message)

    @property
    def message_body(self):
        return json.dumps(self._message)

    @staticmethod
    def build_message(obj, params: dict):
        """build message body implementation"""
        return obj(params["id"], params["results"], params["error"])


class Params(dict):
    """Params interface"""


class Results(dict):
    """Results interface"""

    @classmethod
    def from_rpc(cls, results: dict):
        """load from rpc

        Return:
            Results or None

        Raises:
            InvalidMessage caused by KeyError
        """
        if not results:
            return None
        return cls(results)


class RequestConfig:
    """request config options"""

    def __init__(self, timeout=30, buf_size=2048, n_retry=1):
        self.timeout = timeout
        self.buf_size = buf_size
        self.n_retry = n_retry


class Service:
    """Client service"""

    def __init__(self, address: tuple = None):
        if not address:
            address = ("localhost", 8088)

        # server address
        self.address = address

        # indicate if server error
        self.server_error = False
        # pending used to cancel request until pending elapsed
        self.abort_timeout = None
        # server feature capability map
        self.server_capability = {}
        self.workdir = ""
        self.server_running = False

    def reset_state(self):
        """reset state"""

        self.server_error = False
        self.abort_timeout = None
        self.server_capability = {}
        self.workdir = ""
        self.server_running = False

    def info(self):
        """service info"""

        return json.dumps(
            {
                # server address
                "address": self.address,
                # indicate if server error
                "server_error": self.server_error,
                # pending used to cancel request until pending elapsed
                "abort_timeout": str(self.abort_timeout),
                # server feature capability map
                "server_capability": self.server_capability,
                "workdir": self.workdir,
                "server_running": self.server_running,
                "initialized": self.initialized,
            },
            indent="\t",
        )

    @property
    def initialized(self):
        return bool(self.server_capability)

    @staticmethod
    def request_id():
        """generate request id"""
        return str(time.time())

    @staticmethod
    def check_response(req: RequestMessage, resp: ResponseMessage):
        """check if response id is equal to request id"""

        if req.id_ != resp.id_:
            raise errors.InvalidMessage(
                "want response id : '%s', expected : '%s'" % (req.id_, resp.id_)
            )

    def request(
        self, request_message: bytes, address: tuple, config: RequestConfig = None
    ) -> bytes:
        """request to server"""

        if self.abort_timeout:
            now = datetime.datetime.now()
            if now < self.abort_timeout:
                raise errors.RequestAborted(
                    "abort request until %s" % self.abort_timeout
                )
            self.abort_timeout = None

        if not config:
            config = RequestConfig()

        timeout = config.timeout
        buf_size = config.buf_size
        downloaded = io.BytesIO()

        # retry for n-retry times
        for _ in range(config.n_retry):
            try:
                with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as conn:
                    conn.settimeout(timeout)
                    conn.connect(address)
                    conn.sendall(request_message)

                    while True:
                        n = downloaded.write(conn.recv(buf_size))
                        if n < buf_size:
                            break

            except socket.timeout as err:
                raise TimeoutError(err) from err
            except ConnectionError:
                continue
            else:
                return downloaded.getvalue()

        # abort all request to next 5 minutes
        self.abort_timeout = datetime.datetime.now() + datetime.timedelta(minutes=5)

        self.server_running = False
        LOGGER.debug("abort_timeout: %s", str(self.abort_timeout))
        raise TimeoutError("timed out, server may be offline")

    @staticmethod
    def _run_server(executable: list, workdir: str) -> None:
        """run server"""

        try:
            if os.name == "nt":
                # if on Windows, hide process window
                startupinfo = subprocess.STARTUPINFO()
                startupinfo.dwFlags |= (
                    subprocess.SW_HIDE | subprocess.STARTF_USESHOWWINDOW
                )
            else:
                startupinfo = None

            server_proc = subprocess.Popen(
                executable,
                shell=True,
                cwd=workdir,
                # env=env,
                startupinfo=startupinfo,
            )

            # wait 5 second to make sure if server is running
            time.sleep(5)
            poll = server_proc.poll()

            if poll:
                LOGGER.debug("server terminated, exit code: %d", poll)

                if poll == 123:
                    raise errors.PortUsed("port used")

                raise errors.ServerError("server error, exit code: %d" % poll)

        except FileNotFoundError as err:
            raise errors.ServerError(err) from err

    def run_server(
        self, server_path: str, env_activate: list = None, workdir: str = None
    ):
        """run server

        workdir : relative to server path
        """

        if self.initialized:
            self.server_running = True
            return

        if self.server_error:
            raise errors.ServerError("server error")

        if self.server_running:
            raise errors.ServerRunning("server already running")

        executable = ["python", server_path]
        if env_activate:
            executable = chain(env_activate, ["&&"], executable)

        executable = tuple(executable)
        LOGGER.debug("executable: %s", executable)

        try:
            self._run_server(executable, workdir)

        except errors.ServerError as err:
            self.reset_state()
            self.server_error = True
            LOGGER.debug("server error: %s", err)

        except errors.PortUsed as err:
            self.server_running = True
            LOGGER.debug(err)

        else:
            self.reset_state()
            self.server_running = True
            LOGGER.info("server running")

    def ping(self, *args) -> object:
        """ping server"""

        req_message = RequestMessage(self.request_id(), "ping", args)
        response = self.request(req_message.to_bytes(), self.address)
        resp_message = ResponseMessage.from_bytes(response)
        LOGGER.debug("response message: %s", resp_message)

        err = ResponseError.from_rpc(resp_message.error)
        if err:
            raise errors.ResultError(err.message)

        self.check_response(req_message, resp_message)
        return resp_message.results

    def initialize(self, retry: bool = True) -> None:
        """initialize server

        * retry: retry request if connection error raised
        """

        config = RequestConfig()
        if retry:
            config.n_retry = 10

        req_message = RequestMessage(self.request_id(), "initialize")
        response = self.request(req_message.to_bytes(), self.address, config=config)
        resp_message = ResponseMessage.from_bytes(response)
        LOGGER.debug("response message: %s", resp_message)

        err = ResponseError.from_rpc(resp_message.error)
        if err:
            raise errors.ResultError(err.message)

        self.check_response(req_message, resp_message)
        self.server_capability = {
            "completion": resp_message.results["completion"],
            "hover": resp_message.results["hover"],
            "document_format": resp_message.results["document_format"],
            "diagnostic": resp_message.results["diagnostic"],
            "rename": resp_message.results["rename"],
        }
        LOGGER.debug("server capability: %s", self.server_capability)

    def shutdown(self) -> None:
        """shutdown server"""

        req_message = RequestMessage(self.request_id(), "exit")
        response = self.request(req_message.to_bytes(), self.address)
        resp_message = ResponseMessage.from_bytes(response)
        LOGGER.debug("response message: %s", resp_message)

        err = ResponseError.from_rpc(resp_message.error)
        if err:
            raise errors.ResultError(err.message)

        self.check_response(req_message, resp_message)
        self.reset_state()
        LOGGER.debug("server info:\n%s", self.info())

    def change_workspace(self, path: str) -> str:
        """change workspace

        Raises:
            InvalidMessage
        """

        if path == self.workdir:
            raise errors.ProjectNotChanged("project no changed: %s" % path)

        req_message = RequestMessage(
            self.request_id(), "document.changeWorkspace", {"uri": path},
        )

        response = self.request(req_message.to_bytes(), self.address)
        resp_message = ResponseMessage.from_bytes(response)
        LOGGER.debug("response message: %s", resp_message)

        err = ResponseError.from_rpc(resp_message.error)
        if err:
            raise errors.ResultError(err.message)

        self.check_response(req_message, resp_message)
        self.workdir = path
        LOGGER.debug("workspace path: %s", self.workdir)
        return path

    def is_server_capable(self, name: str):
        """check if server can run command"""
        return self.server_capability.get(name, False)

    def command_complete(self, params: Params) -> Results:
        """on get completion

        Return:
            Results

        Raises:
            InvalidMessage
            FeatureDisabled
        """

        if not self.is_server_capable("completion"):
            raise errors.FeatureDisabled("completion")

        req_message = RequestMessage(
            self.request_id(), "textDocument.completion", params
        )
        response = self.request(req_message.to_bytes(), self.address)
        resp_message = ResponseMessage.from_bytes(response)
        LOGGER.debug("response message: %s", resp_message)

        err = ResponseError.from_rpc(resp_message.error)
        if err:
            raise errors.ResultError(err.message)

        self.check_response(req_message, resp_message)
        return resp_message.results

    def command_hover(self, params: Params) -> Results:
        """on hover

        Return:
            Results

        Raises:
            InvalidMessage
            FeatureDisabled
        """

        if not self.is_server_capable("hover"):
            raise errors.FeatureDisabled("hover")

        req_message = RequestMessage(self.request_id(), "textDocument.hover", params)
        response = self.request(req_message.to_bytes(), self.address)
        resp_message = ResponseMessage.from_bytes(response)
        LOGGER.debug("response message: %s", resp_message)

        err = ResponseError.from_rpc(resp_message.error)
        if err:
            raise errors.ResultError(err.message)

        self.check_response(req_message, resp_message)
        return resp_message.results

    def command_format_document(self, params: Params) -> Results:
        """on format document

        Return:
            Results

        Raises:
            InvalidMessage
            FeatureDisabled
        """

        if not self.is_server_capable("document_format"):
            raise errors.FeatureDisabled("document_format")

        req_message = RequestMessage(
            self.request_id(), "textDocument.formatting", params
        )
        response = self.request(req_message.to_bytes(), self.address)
        resp_message = ResponseMessage.from_bytes(response)
        LOGGER.debug("response message: %s", resp_message)

        err = ResponseError.from_rpc(resp_message.error)
        if err:
            raise errors.ResultError(err.message)

        self.check_response(req_message, resp_message)
        return resp_message.results

    def command_get_diagnostic(self, params: Params) -> Results:
        """on get diagnostic

        Return:
            Results

        Raises:
            InvalidMessage
            FeatureDisabled
        """

        if not self.is_server_capable("diagnostic"):
            raise errors.FeatureDisabled("diagnostic")

        req_message = RequestMessage(
            self.request_id(), "textDocument.getDiagnostic", params
        )
        response = self.request(req_message.to_bytes(), self.address)
        resp_message = ResponseMessage.from_bytes(response)
        LOGGER.debug("response message: %s", resp_message)

        err = ResponseError.from_rpc(resp_message.error)
        if err:
            raise errors.ResultError(err.message)

        self.check_response(req_message, resp_message)
        return resp_message.results

    def command_rename(self, params: Params) -> Results:
        """on get diagnostic

        Return:
            Results

        Raises:
            InvalidMessage
            FeatureDisabled
        """

        if not self.is_server_capable("rename"):
            raise errors.FeatureDisabled("rename")

        req_message = RequestMessage(self.request_id(), "document.rename", params)
        response = self.request(req_message.to_bytes(), self.address)
        resp_message = ResponseMessage.from_bytes(response)
        LOGGER.debug("response message: %s", resp_message)

        err = ResponseError.from_rpc(resp_message.error)
        if err:
            raise errors.ResultError(err.message)

        self.check_response(req_message, resp_message)
        return resp_message.results

"""This module handle client"""

import datetime
import io
import json
import logging
import os
import re
import socket
import subprocess
import time
from collections import OrderedDict
from collections.abc import Mapping, MutableMapping
from itertools import chain

from . import errors

LOGGER = logging.getLogger(__name__)
# LOGGER.setLevel(logging.DEBUG)
STREAM_HANDLER = logging.StreamHandler()
LOG_TEMPLATE = "%(levelname)s %(asctime)s %(filename)s:%(lineno)s  %(message)s"
STREAM_HANDLER.setFormatter(logging.Formatter(LOG_TEMPLATE))
LOGGER.addHandler(STREAM_HANDLER)


class CaseInsensitiveDict(MutableMapping):
    """A case-insensitive ``dict``-like object"""

    def __init__(self, data=None, **kwargs):
        self._store = OrderedDict()
        if data is None:
            data = {}
        self.update(data, **kwargs)

    def __setitem__(self, key, value):
        # Use the lowercased key for lookups, but store the actual
        # key alongside the value.
        self._store[key.lower()] = (key, value)

    def __getitem__(self, key):
        return self._store[key.lower()][1]

    def __delitem__(self, key):
        del self._store[key.lower()]

    def __iter__(self):
        return (casedkey for casedkey, mappedvalue in self._store.values())

    def __len__(self):
        return len(self._store)

    def lower_items(self):
        """Like iteritems(), but with all lowercase keys."""
        return ((lowerkey, keyval[1]) for (lowerkey, keyval) in self._store.items())

    def __eq__(self, other):
        if isinstance(other, Mapping):
            other = CaseInsensitiveDict(other)
        else:
            return NotImplemented
        # Compare insensitively
        return dict(self.lower_items()) == dict(other.lower_items())

    # Copy is required
    def copy(self):
        return CaseInsensitiveDict(self._store.values())

    def __repr__(self):
        return str(dict(self.items()))


class Headers(CaseInsensitiveDict):
    """handle headers item"""

    item_separator = b"\r\n"
    item_pattern = re.compile(r"^(.+): (.+)$")
    encoding = "ascii"

    def to_bytes(self):
        """convert to bytes"""

        lines = ("%s: %s" % (key, value) for key, value in self.items())
        return str.encode("\r\n".join(lines), self.encoding)

    @classmethod
    def from_bytes(cls, header: bytes):
        """load header from bytes"""

        decoded_header = bytes.decode(header, cls.encoding)
        items = {}
        for line in decoded_header.splitlines():
            match = cls.item_pattern.match(line)
            if match:
                items[match.group(1)] = match.group(2)

        return cls(items)


class Message:
    """message interface"""

    encoding = "utf8"
    body_separator = b"\r\n\r\n"

    @property
    def message_body(self):
        """interface get message body"""
        return ""

    def to_bytes(self):
        """convert to bytes"""
        encoded = self.message_body.encode(self.encoding)
        header = Headers({"Content-Length": len(encoded), "Encoding": self.encoding})
        return self.body_separator.join((header.to_bytes(), encoded))

    @staticmethod
    def build_message(obj, params: dict) -> "Message":
        """build message body implementation"""
        raise NotImplementedError("build_message not implemented")

    @classmethod
    def from_bytes(cls, message: bytes) -> "Message":
        """build message from bytes"""
        try:
            header, content = message.split(cls.body_separator)
            parsed_header = Headers.from_bytes(header)

            size = int(parsed_header["Content-Length"])
            if len(content) != size:
                raise errors.InvalidSize("want %d, expected %d" % (size, len(content)))

            encoding = parsed_header.get("encoding", cls.encoding)
            decoded = json.loads(content.decode(encoding))
            message = cls.build_message(cls, decoded)

        except (ValueError, KeyError) as err:
            raise errors.InvalidMessage(err) from err

        else:
            return message


class RequestMessage(Message):
    """RequestMessage"""

    def __init__(self, id_: str, method: str, params: dict = None):
        self.id_ = id_
        self.method = method
        self.params = params

    @property
    def _message(self):
        return {"id": self.id_, "method": self.method, "params": self.params}

    def __repr__(self):
        return str(self._message)

    @property
    def message_body(self):
        return json.dumps(self._message)

    @staticmethod
    def build_message(obj, params: dict):
        """build message body implementation"""
        return obj(params["id"], params["method"], params["params"])


class ResponseError(dict):
    """Response Error"""

    @classmethod
    def new(cls, code: str, message: str, **kwargs):
        """new response error"""
        return cls(code=code, message=message, **kwargs)

    @property
    def message(self):
        """message property"""
        return str(self["message"])

    @classmethod
    def from_rpc(cls, params: dict):
        """from rpc"""

        if not params:
            return None

        code = params["code"]
        message = params["message"]
        options = {k: v for (k, v) in params.items() if k not in ("code", "message")}

        return cls(code=code, message=message, **options)


class ResponseMessage(Message):
    """ResponseMessage"""

    def __init__(self, id_: str, results: object = None, error: ResponseError = None):
        self.id_ = id_
        self.results = results
        self.error = error

    @property
    def _message(self):
        return {"id": self.id_, "results": self.results, "error": self.error}

    def __repr__(self):
        return str(self._message)

    @property
    def message_body(self):
        return json.dumps(self._message)

    @staticmethod
    def build_message(obj, params: dict):
        """build message body implementation"""
        return obj(params["id"], params["results"], params["error"])


class Params(dict):
    """Params interface"""


class Results(dict):
    """Results interface"""

    @classmethod
    def from_rpc(cls, results: dict):
        """load from rpc

        Return:
            Results or None

        Raises:
            InvalidMessage caused by KeyError
        """
        if not results:
            return None
        return cls(results)


class RequestConfig:
    """request config options"""

    def __init__(self, timeout=30, buf_size=2048, n_retry=1):
        self.timeout = timeout
        self.buf_size = buf_size
        self.n_retry = n_retry


class Service:
    """Client service"""

    def __init__(self, address: tuple = None):
        if not address:
            address = ("localhost", 8088)

        # server address
        self.address = address

        # indicate if server error
        self.server_error = False
        # pending used to cancel request until pending elapsed
        self.abort_timeout = None
        # server feature capability map
        self.server_capability = {}
        self.workdir = ""
        self.server_running = False

    def reset_state(self):
        """reset state"""

        self.server_error = False
        self.abort_timeout = None
        self.server_capability = {}
        self.workdir = ""
        self.server_running = False

    def info(self):
        """service info"""

        return json.dumps(
            {
                # server address
                "address": self.address,
                # indicate if server error
                "server_error": self.server_error,
                # pending used to cancel request until pending elapsed
                "abort_timeout": str(self.abort_timeout),
                # server feature capability map
                "server_capability": self.server_capability,
                "workdir": self.workdir,
                "server_running": self.server_running,
                "initialized": self.initialized,
            },
            indent="\t",
        )

    @property
    def initialized(self):
        return bool(self.server_capability)

    @staticmethod
    def request_id():
        """generate request id"""
        return str(time.time())

    @staticmethod
    def check_response(req: RequestMessage, resp: ResponseMessage):
        """check if response id is equal to request id"""

        if req.id_ != resp.id_:
            raise errors.InvalidMessage(
                "want response id : '%s', expected : '%s'" % (req.id_, resp.id_)
            )

    def request(
        self, request_message: bytes, address: tuple, config: RequestConfig = None
    ) -> bytes:
        """request to server"""

        if self.abort_timeout:
            now = datetime.datetime.now()
            if now < self.abort_timeout:
                raise errors.RequestAborted(
                    "abort request until %s" % self.abort_timeout
                )
            self.abort_timeout = None

        if not config:
            config = RequestConfig()

        timeout = config.timeout
        buf_size = config.buf_size
        downloaded = io.BytesIO()

        # retry for n-retry times
        for _ in range(config.n_retry):
            try:
                with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as conn:
                    conn.settimeout(timeout)
                    conn.connect(address)
                    conn.sendall(request_message)

                    while True:
                        n = downloaded.write(conn.recv(buf_size))
                        if n < buf_size:
                            break

            except socket.timeout as err:
                raise TimeoutError(err) from err
            except ConnectionError:
                continue
            else:
                return downloaded.getvalue()

        # abort all request to next 5 minutes
        self.abort_timeout = datetime.datetime.now() + datetime.timedelta(minutes=5)

        self.server_running = False
        LOGGER.debug("abort_timeout: %s", str(self.abort_timeout))
        raise TimeoutError("timed out, server may be offline")

    @staticmethod
    def _run_server(executable: list, workdir: str) -> None:
        """run server"""

        try:
            if os.name == "nt":
                # if on Windows, hide process window
                startupinfo = subprocess.STARTUPINFO()
                startupinfo.dwFlags |= (
                    subprocess.SW_HIDE | subprocess.STARTF_USESHOWWINDOW
                )
            else:
                startupinfo = None

            server_proc = subprocess.Popen(
                executable,
                shell=True,
                cwd=workdir,
                # env=env,
                startupinfo=startupinfo,
            )

            # wait 5 second to make sure if server is running
            time.sleep(5)
            poll = server_proc.poll()

            if poll:
                LOGGER.debug("server terminated, exit code: %d", poll)

                if poll == 123:
                    raise errors.PortUsed("port used")

                raise errors.ServerError("server error, exit code: %d" % poll)

        except FileNotFoundError as err:
            raise errors.ServerError(err) from err

    def run_server(
        self, server_path: str, env_activate: list = None, workdir: str = None
    ):
        """run server

        workdir : relative to server path
        """

        if self.initialized:
            self.server_running = True
            return

        if self.server_error:
            raise errors.ServerError("server error")

        if self.server_running:
            raise errors.ServerRunning("server already running")

        executable = ["python", server_path]
        if env_activate:
            executable = chain(env_activate, ["&&"], executable)

        executable = tuple(executable)
        LOGGER.debug("executable: %s", executable)

        try:
            self._run_server(executable, workdir)

        except errors.ServerError as err:
            self.reset_state()
            self.server_error = True
            LOGGER.debug("server error: %s", err)

        except errors.PortUsed as err:
            self.server_running = True
            LOGGER.debug(err)

        else:
            self.reset_state()
            self.server_running = True
            LOGGER.info("server running")

    def ping(self, *args) -> object:
        """ping server"""

        req_message = RequestMessage(self.request_id(), "ping", args)
        response = self.request(req_message.to_bytes(), self.address)
        resp_message = ResponseMessage.from_bytes(response)
        LOGGER.debug("response message: %s", resp_message)

        err = ResponseError.from_rpc(resp_message.error)
        if err:
            raise errors.ResultError(err.message)

        self.check_response(req_message, resp_message)
        return resp_message.results

    def initialize(self, retry: bool = True) -> None:
        """initialize server

        * retry: retry request if connection error raised
        """

        config = RequestConfig()
        if retry:
            config.n_retry = 10

        req_message = RequestMessage(self.request_id(), "initialize")
        response = self.request(req_message.to_bytes(), self.address, config=config)
        resp_message = ResponseMessage.from_bytes(response)
        LOGGER.debug("response message: %s", resp_message)

        err = ResponseError.from_rpc(resp_message.error)
        if err:
            raise errors.ResultError(err.message)

        self.check_response(req_message, resp_message)
        self.server_capability = {
            "completion": resp_message.results["completion"],
            "hover": resp_message.results["hover"],
            "document_format": resp_message.results["document_format"],
            "diagnostic": resp_message.results["diagnostic"],
            "rename": resp_message.results["rename"],
        }
        LOGGER.debug("server capability: %s", self.server_capability)

    def shutdown(self) -> None:
        """shutdown server"""

        req_message = RequestMessage(self.request_id(), "exit")
        response = self.request(req_message.to_bytes(), self.address)
        resp_message = ResponseMessage.from_bytes(response)
        LOGGER.debug("response message: %s", resp_message)

        err = ResponseError.from_rpc(resp_message.error)
        if err:
            raise errors.ResultError(err.message)

        self.check_response(req_message, resp_message)
        self.reset_state()
        LOGGER.debug("server info:\n%s", self.info())

    def change_workspace(self, path: str) -> str:
        """change workspace

        Raises:
            InvalidMessage
        """

        if path == self.workdir:
            raise errors.ProjectNotChanged("project no changed: %s" % path)

        req_message = RequestMessage(
            self.request_id(), "document.changeWorkspace", {"uri": path},
        )

        response = self.request(req_message.to_bytes(), self.address)
        resp_message = ResponseMessage.from_bytes(response)
        LOGGER.debug("response message: %s", resp_message)

        err = ResponseError.from_rpc(resp_message.error)
        if err:
            raise errors.ResultError(err.message)

        self.check_response(req_message, resp_message)
        self.workdir = path
        LOGGER.debug("workspace path: %s", self.workdir)
        return path

    def is_server_capable(self, name: str):
        """check if server can run command"""
        return self.server_capability.get(name, False)

    def command_complete(self, params: Params) -> Results:
        """on get completion

        Return:
            Results

        Raises:
            InvalidMessage
            FeatureDisabled
        """

        if not self.is_server_capable("completion"):
            raise errors.FeatureDisabled("completion")

        req_message = RequestMessage(
            self.request_id(), "textDocument.completion", params
        )
        response = self.request(req_message.to_bytes(), self.address)
        resp_message = ResponseMessage.from_bytes(response)
        LOGGER.debug("response message: %s", resp_message)

        err = ResponseError.from_rpc(resp_message.error)
        if err:
            raise errors.ResultError(err.message)

        self.check_response(req_message, resp_message)
        return resp_message.results

    def command_hover(self, params: Params) -> Results:
        """on hover

        Return:
            Results

        Raises:
            InvalidMessage
            FeatureDisabled
        """

        if not self.is_server_capable("hover"):
            raise errors.FeatureDisabled("hover")

        req_message = RequestMessage(self.request_id(), "textDocument.hover", params)
        response = self.request(req_message.to_bytes(), self.address)
        resp_message = ResponseMessage.from_bytes(response)
        LOGGER.debug("response message: %s", resp_message)

        err = ResponseError.from_rpc(resp_message.error)
        if err:
            raise errors.ResultError(err.message)

        self.check_response(req_message, resp_message)
        return resp_message.results

    def command_format_document(self, params: Params) -> Results:
        """on format document

        Return:
            Results

        Raises:
            InvalidMessage
            FeatureDisabled
        """

        if not self.is_server_capable("document_format"):
            raise errors.FeatureDisabled("document_format")

        req_message = RequestMessage(
            self.request_id(), "textDocument.formatting", params
        )
        response = self.request(req_message.to_bytes(), self.address)
        resp_message = ResponseMessage.from_bytes(response)
        LOGGER.debug("response message: %s", resp_message)

        err = ResponseError.from_rpc(resp_message.error)
        if err:
            raise errors.ResultError(err.message)

        self.check_response(req_message, resp_message)
        return resp_message.results

    def command_get_diagnostic(self, params: Params) -> Results:
        """on get diagnostic

        Return:
            Results

        Raises:
            InvalidMessage
            FeatureDisabled
        """

        if not self.is_server_capable("diagnostic"):
            raise errors.FeatureDisabled("diagnostic")

        req_message = RequestMessage(
            self.request_id(), "textDocument.getDiagnostic", params
        )
        response = self.request(req_message.to_bytes(), self.address)
        resp_message = ResponseMessage.from_bytes(response)
        LOGGER.debug("response message: %s", resp_message)

        err = ResponseError.from_rpc(resp_message.error)
        if err:
            raise errors.ResultError(err.message)

        self.check_response(req_message, resp_message)
        return resp_message.results

    def command_rename(self, params: Params) -> Results:
        """on get diagnostic

        Return:
            Results

        Raises:
            InvalidMessage
            FeatureDisabled
        """

        if not self.is_server_capable("rename"):
            raise errors.FeatureDisabled("rename")

        req_message = RequestMessage(self.request_id(), "document.rename", params)
        response = self.request(req_message.to_bytes(), self.address)
        resp_message = ResponseMessage.from_bytes(response)
        LOGGER.debug("response message: %s", resp_message)

        err = ResponseError.from_rpc(resp_message.error)
        if err:
            raise errors.ResultError(err.message)

        self.check_response(req_message, resp_message)
        return resp_message.results
This module handle client # LOGGER.setLevel(logging.DEBUG) A case-insensitive ``dict``-like object # Use the lowercased key for lookups, but store the actual # key alongside the value. Like iteritems(), but with all lowercase keys. # Compare insensitively # Copy is required handle headers item convert to bytes load header from bytes message interface interface get message body convert to bytes build message body implementation build message from bytes RequestMessage build message body implementation Response Error new response error message property from rpc ResponseMessage build message body implementation Params interface Results interface load from rpc          Return:             Results or None          Raises:             InvalidMessage caused by KeyError request config options Client service # server address # indicate if server error # pending used to cancel request until pending elapsed # server feature capability map reset state service info # server address # indicate if server error # pending used to cancel request until pending elapsed # server feature capability map generate request id check if response id is equal to request id request to server # retry for n-retry times # abort all request to next 5 minutes run server # if on Windows, hide process window # env=env, # wait 5 second to make sure if server is running run server          workdir : relative to server path ping server initialize server          * retry: retry request if connection error raised shutdown server change workspace          Raises:             InvalidMessage check if server can run command on get completion          Return:             Results          Raises:             InvalidMessage             FeatureDisabled on hover          Return:             Results          Raises:             InvalidMessage             FeatureDisabled on format document          Return:             Results          Raises:             InvalidMessage             FeatureDisabled on get diagnostic          Return:             Results          Raises:             InvalidMessage             FeatureDisabled on get diagnostic          Return:             Results          Raises:             InvalidMessage             FeatureDisabled