refactor: make LspApplyDocumentEditCommand take plain TextEdits (#2393)

Author: rchl

plugin/__init__.py

@@ -1,5 +1,6 @@
 from .core.collections import DottedDict
 from .core.css import css
+from .core.edit import apply_text_edits
 from .core.file_watcher import FileWatcher
 from .core.file_watcher import FileWatcherEvent
 from .core.file_watcher import FileWatcherEventType
@@ -22,12 +23,14 @@
 from .core.url import uri_to_filename  # deprecated
 from .core.version import __version__
 from .core.views import MarkdownLangMap
+from .core.views import uri_from_view
 from .core.workspace import WorkspaceFolder
 
 # This is the public API for LSP-* packages
 __all__ = [
     '__version__',
     'AbstractPlugin',
+    'apply_text_edits',
     'ClientConfig',
     'css',
     'DottedDict',
@@ -49,6 +52,7 @@
     'Session',
     'SessionBufferProtocol',
     'unregister_plugin',
+    'uri_from_view',
     'uri_to_filename',  # deprecated
     'WorkspaceFolder',
 ]

plugin/color.py

@@ -1,4 +1,4 @@
-from .core.edit import parse_text_edit
+from .core.edit import apply_text_edits
 from .core.protocol import ColorInformation
 from .core.protocol import ColorPresentation
 from .core.protocol import ColorPresentationParams
@@ -58,4 +58,4 @@ def _on_select(self, index: int) -> None:
         if index > -1:
             color_pres = self._filtered_response[index]
             text_edit = color_pres.get('textEdit') or {'range': self._range, 'newText': color_pres['label']}
-            self.view.run_command('lsp_apply_document_edit', {'changes': [parse_text_edit(text_edit, self._version)]})
+            apply_text_edits(self.view, [text_edit], required_view_version=self._version)

plugin/completion.py

@@ -1,5 +1,5 @@
 from .core.constants import COMPLETION_KINDS
-from .core.edit import parse_text_edit
+from .core.edit import apply_text_edits
 from .core.logging import debug
 from .core.promise import Promise
 from .core.protocol import CompletionEditRange
@@ -371,8 +371,7 @@ def _on_resolved_async(self, session_name: str, item: CompletionItem) -> None:
     def _on_resolved(self, session_name: str, item: CompletionItem) -> None:
         additional_edits = item.get('additionalTextEdits')
         if additional_edits:
-            edits = [parse_text_edit(additional_edit) for additional_edit in additional_edits]
-            self.view.run_command("lsp_apply_document_edit", {'changes': edits})
+            apply_text_edits(self.view, additional_edits)
         command = item.get("command")
         if command:
             debug('Running server command "{}" for view {}'.format(command, self.view.id()))

plugin/core/edit.py

@@ -1,21 +1,17 @@
 from .logging import debug
-from .open import open_file
-from .promise import Promise
 from .protocol import Position
 from .protocol import TextEdit
 from .protocol import UINT_MAX
 from .protocol import WorkspaceEdit
 from .typing import List, Dict, Optional, Tuple
-from functools import partial
 import sublime
 
 
-# tuple of start, end, newText, version
-TextEditTuple = Tuple[Tuple[int, int], Tuple[int, int], str, Optional[int]]
+WorkspaceChanges = Dict[str, Tuple[List[TextEdit], Optional[int]]]
 
 
-def parse_workspace_edit(workspace_edit: WorkspaceEdit) -> Dict[str, List[TextEditTuple]]:
-    changes = {}  # type: Dict[str, List[TextEditTuple]]
+def parse_workspace_edit(workspace_edit: WorkspaceEdit) -> WorkspaceChanges:
+    changes = {}  # type: WorkspaceChanges
     document_changes = workspace_edit.get('documentChanges')
     if isinstance(document_changes, list):
         for document_change in document_changes:
@@ -26,38 +22,34 @@ def parse_workspace_edit(workspace_edit: WorkspaceEdit) -> Dict[str, List[TextEd
             text_document = document_change["textDocument"]
             uri = text_document['uri']
             version = text_document.get('version')
-            text_edit = list(parse_text_edit(change, version) for change in document_change.get('edits'))
-            changes.setdefault(uri, []).extend(text_edit)
+            edits = document_change.get('edits')
+            changes.setdefault(uri, ([], version))[0].extend(edits)
     else:
         raw_changes = workspace_edit.get('changes')
         if isinstance(raw_changes, dict):
-            for uri, uri_changes in raw_changes.items():
-                changes[uri] = list(parse_text_edit(change) for change in uri_changes)
+            for uri, edits in raw_changes.items():
+                changes[uri] = (edits, None)
     return changes
 
 
 def parse_range(range: Position) -> Tuple[int, int]:
     return range['line'], min(UINT_MAX, range['character'])
 
 
-def parse_text_edit(text_edit: TextEdit, version: Optional[int] = None) -> TextEditTuple:
-    return (
-        parse_range(text_edit['range']['start']),
-        parse_range(text_edit['range']['end']),
-        # Strip away carriage returns -- SublimeText takes care of that.
-        text_edit.get('newText', '').replace("\r", ""),
-        version
+def apply_text_edits(
+    view: sublime.View,
+    edits: Optional[List[TextEdit]],
+    *,
+    process_placeholders: Optional[bool] = False,
+    required_view_version: Optional[int] = None
+) -> None:
+    if not edits:
+        return
+    view.run_command(
+        'lsp_apply_document_edit',
+        {
+            'changes': edits,
+            'process_placeholders': process_placeholders,
+            'required_view_version': required_view_version,
+        }
     )
-
-
-def apply_workspace_edit(window: sublime.Window, changes: Dict[str, List[TextEditTuple]]) -> Promise:
-    """
-    DEPRECATED: Use session.apply_workspace_edit_async instead.
-    """
-    return Promise.all([open_file(window, uri).then(partial(apply_edits, edits)) for uri, edits in changes.items()])
-
-
-def apply_edits(edits: List[TextEditTuple], view: Optional[sublime.View]) -> None:
-    if view and view.is_valid():
-        # Text commands run blocking. After this call has returned the changes are applied.
-        view.run_command("lsp_apply_document_edit", {"changes": edits})

plugin/core/sessions.py

@@ -1,9 +1,9 @@
 from .collections import DottedDict
 from .constants import SEMANTIC_TOKENS_MAP
 from .diagnostics_storage import DiagnosticsStorage
-from .edit import apply_edits
+from .edit import apply_text_edits
 from .edit import parse_workspace_edit
-from .edit import TextEditTuple
+from .edit import WorkspaceChanges
 from .file_watcher import DEFAULT_KIND
 from .file_watcher import file_watcher_event_type_to_lsp_file_change_type
 from .file_watcher import FileWatcher
@@ -67,6 +67,7 @@
 from .protocol import SymbolTag
 from .protocol import TextDocumentClientCapabilities
 from .protocol import TextDocumentSyncKind
+from .protocol import TextEdit
 from .protocol import TokenFormat
 from .protocol import UnregistrationParams
 from .protocol import WindowClientCapabilities
@@ -94,7 +95,7 @@
 from .types import SettingsRegistration
 from .types import sublime_pattern_to_glob
 from .types import WORKSPACE_DIAGNOSTICS_TIMEOUT
-from .typing import Callable, cast, Dict, Any, Optional, List, Tuple, Generator, Type, TypeGuard, Protocol, Mapping, Set, TypeVar, Union  # noqa: E501
+from .typing import Callable, cast, Dict, Any, Optional, List, Tuple, Generator, Type, TypeGuard, Protocol, Set, TypeVar, Union  # noqa: E501
 from .url import filename_to_uri
 from .url import parse_uri
 from .url import unparse_uri
@@ -976,7 +977,7 @@ def on_workspace_configuration(self, params: Dict, configuration: Any) -> Any:
         """
         return configuration
 
-    def on_pre_server_command(self, command: Mapping[str, Any], done_callback: Callable[[], None]) -> bool:
+    def on_pre_server_command(self, command: ExecuteCommandParams, done_callback: Callable[[], None]) -> bool:
         """
         Intercept a command that is about to be sent to the language server.
 
@@ -1768,12 +1769,22 @@ def apply_workspace_edit_async(self, edit: WorkspaceEdit) -> Promise[None]:
         """
         return self.apply_parsed_workspace_edits(parse_workspace_edit(edit))
 
-    def apply_parsed_workspace_edits(self, changes: Dict[str, List[TextEditTuple]]) -> Promise[None]:
+    def apply_parsed_workspace_edits(self, changes: WorkspaceChanges) -> Promise[None]:
         promises = []  # type: List[Promise[None]]
-        for uri, edits in changes.items():
-            promises.append(self.open_uri_async(uri).then(functools.partial(apply_edits, edits)))
+        for uri, (edits, view_version) in changes.items():
+            promises.append(
+                self.open_uri_async(uri).then(functools.partial(self._apply_text_edits, edits, view_version, uri))
+            )
         return Promise.all(promises).then(lambda _: None)
 
+    def _apply_text_edits(
+        self, edits: List[TextEdit], view_version: Optional[int], uri: str, view: Optional[sublime.View]
+    ) -> None:
+        if view is None or not view.is_valid():
+            print('LSP: ignoring edits due to no view for uri: {}'.format(uri))
+            return
+        apply_text_edits(view, edits, required_view_version=view_version)
+
     def decode_semantic_token(
             self, token_type_encoded: int, token_modifiers_encoded: int) -> Tuple[str, List[str], Optional[str]]:
         types_legend = tuple(cast(List[str], self.get_capability('semanticTokensProvider.legend.tokenTypes')))

plugin/edit.py

@@ -1,5 +1,5 @@
-from .core.edit import TextEditTuple
-from .core.logging import debug
+from .core.edit import parse_range
+from .core.protocol import TextEdit
 from .core.typing import List, Optional, Any, Generator, Iterable, Tuple
 from contextlib import contextmanager
 import operator
@@ -8,6 +8,9 @@
 import sublime_plugin
 
 
+TextEditTuple = Tuple[Tuple[int, int], Tuple[int, int], str]
+
+
 @contextmanager
 def temporary_setting(settings: sublime.Settings, key: str, val: Any) -> Generator[None, None, None]:
     prev_val = None
@@ -25,20 +28,25 @@ class LspApplyDocumentEditCommand(sublime_plugin.TextCommand):
     re_placeholder = re.compile(r'\$(0|\{0:([^}]*)\})')
 
     def run(
-        self, edit: sublime.Edit, changes: Optional[List[TextEditTuple]] = None, process_placeholders: bool = False
+        self,
+        edit: sublime.Edit,
+        changes: List[TextEdit],
+        required_view_version: Optional[int] = None,
+        process_placeholders: bool = False,
     ) -> None:
         # Apply the changes in reverse, so that we don't invalidate the range
         # of any change that we haven't applied yet.
         if not changes:
             return
+        view_version = self.view.change_count()
+        if required_view_version is not None and required_view_version != view_version:
+            print('LSP: ignoring edit due to non-matching document version')
+            return
+        edits = [_parse_text_edit(change) for change in changes or []]
         with temporary_setting(self.view.settings(), "translate_tabs_to_spaces", False):
-            view_version = self.view.change_count()
             last_row, _ = self.view.rowcol_utf16(self.view.size())
             placeholder_region_count = 0
-            for start, end, replacement, version in reversed(_sort_by_application_order(changes)):
-                if version is not None and version != view_version:
-                    debug('ignoring edit due to non-matching document version')
-                    continue
+            for start, end, replacement in reversed(_sort_by_application_order(edits)):
                 placeholder_region = None  # type: Optional[Tuple[Tuple[int, int], Tuple[int, int]]]
                 if process_placeholders and replacement:
                     parsed = self.parse_snippet(replacement)
@@ -96,6 +104,15 @@ def parse_snippet(self, replacement: str) -> Optional[Tuple[str, Tuple[int, int]
         return (new_replacement, placeholder_start_and_length)
 
 
+def _parse_text_edit(text_edit: TextEdit) -> TextEditTuple:
+    return (
+        parse_range(text_edit['range']['start']),
+        parse_range(text_edit['range']['end']),
+        # Strip away carriage returns -- SublimeText takes care of that.
+        text_edit.get('newText', '').replace("\r", "")
+    )
+
+
 def _sort_by_application_order(changes: Iterable[TextEditTuple]) -> List[TextEditTuple]:
     # The spec reads:
     # > However, it is possible that multiple edits have the same start position: multiple

plugin/formatting.py

@@ -1,5 +1,5 @@
 from .core.collections import DottedDict
-from .core.edit import parse_text_edit
+from .core.edit import apply_text_edits
 from .core.promise import Promise
 from .core.protocol import Error
 from .core.protocol import TextDocumentSaveReason
@@ -8,7 +8,7 @@
 from .core.registry import windows
 from .core.sessions import Session
 from .core.settings import userprefs
-from .core.typing import Any, Callable, List, Optional, Iterator, Union
+from .core.typing import Callable, List, Optional, Iterator, Union
 from .core.views import entire_content_region
 from .core.views import first_selection_region
 from .core.views import has_single_nonempty_selection
@@ -50,13 +50,6 @@ def format_document(text_command: LspTextCommand, formatter: Optional[str] = Non
     return Promise.resolve(None)
 
 
-def apply_text_edits_to_view(
-    response: Optional[List[TextEdit]], view: sublime.View, *, process_placeholders: bool = False
-) -> None:
-    edits = list(parse_text_edit(change) for change in response) if response else []
-    view.run_command('lsp_apply_document_edit', {'changes': edits, 'process_placeholders': process_placeholders})
-
-
 class WillSaveWaitTask(SaveTask):
     @classmethod
     def is_applicable(cls, view: sublime.View) -> bool:
@@ -85,9 +78,9 @@ def _will_save_wait_until_async(self, session: Session) -> None:
             self._on_response,
             lambda error: self._on_response(None))
 
-    def _on_response(self, response: Any) -> None:
-        if response and not self._cancelled:
-            apply_text_edits_to_view(response, self._task_runner.view)
+    def _on_response(self, response: FormatResponse) -> None:
+        if response and not isinstance(response, Error) and not self._cancelled:
+            apply_text_edits(self._task_runner.view, response)
         sublime.set_timeout_async(self._handle_next_session_async)
 
 
@@ -108,7 +101,7 @@ def run_async(self) -> None:
 
     def _on_response(self, response: FormatResponse) -> None:
         if response and not isinstance(response, Error) and not self._cancelled:
-            apply_text_edits_to_view(response, self._task_runner.view)
+            apply_text_edits(self._task_runner.view, response)
         sublime.set_timeout_async(self._on_complete)
 
 
@@ -143,7 +136,7 @@ def run(self, edit: sublime.Edit, event: Optional[dict] = None, select: bool = F
 
     def on_result(self, result: FormatResponse) -> None:
         if result and not isinstance(result, Error):
-            apply_text_edits_to_view(result, self.view)
+            apply_text_edits(self.view, result)
 
     def select_formatter(self, base_scope: str, session_names: List[str]) -> None:
         window = self.view.window()
@@ -194,12 +187,12 @@ def run(self, edit: sublime.Edit, event: Optional[dict] = None) -> None:
             selection = first_selection_region(self.view)
             if session and selection is not None:
                 req = text_document_range_formatting(self.view, selection)
-                session.send_request(req, lambda response: apply_text_edits_to_view(response, self.view))
+                session.send_request(req, lambda response: apply_text_edits(self.view, response))
         elif self.view.has_non_empty_selection_region():
             session = self.best_session('documentRangeFormattingProvider.rangesSupport')
             if session:
                 req = text_document_ranges_formatting(self.view)
-                session.send_request(req, lambda response: apply_text_edits_to_view(response, self.view))
+                session.send_request(req, lambda response: apply_text_edits(self.view, response))
 
 
 class LspFormatCommand(LspTextCommand):

plugin/inlay_hint.py

@@ -1,4 +1,5 @@
 from .core.css import css
+from .core.edit import apply_text_edits
 from .core.protocol import InlayHint
 from .core.protocol import InlayHintLabelPart
 from .core.protocol import MarkupContent
@@ -9,7 +10,6 @@
 from .core.settings import userprefs
 from .core.typing import cast, Optional, Union
 from .core.views import position_to_offset
-from .formatting import apply_text_edits_to_view
 import html
 import sublime
 import uuid
@@ -69,7 +69,7 @@ def handle_inlay_hint_text_edits(self, session_name: str, inlay_hint: InlayHint,
             return
         for sb in session.session_buffers_async():
             sb.remove_inlay_hint_phantom(phantom_uuid)
-        apply_text_edits_to_view(text_edits, self.view)
+        apply_text_edits(self.view, text_edits)
 
     def handle_label_part_command(self, session_name: str, label_part: Optional[InlayHintLabelPart] = None) -> None:
         if not label_part: