mirror of
https://github.com/hexedtech/codemp-sublime.git
synced 2024-12-23 13:14:52 +01:00
132 lines
4.5 KiB
Python
132 lines
4.5 KiB
Python
from __future__ import annotations
|
|
from typing import TYPE_CHECKING
|
|
if TYPE_CHECKING:
|
|
from .workspace import WorkspaceManager
|
|
from ...lib import codemp
|
|
|
|
import sublime
|
|
import os
|
|
import logging
|
|
|
|
from .. import globals as g
|
|
from ..utils import populate_view
|
|
from ..utils import safe_listener_attach
|
|
from ..utils import safe_listener_detach
|
|
from ..utils import bidict
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def bind_buffer_manager(buff: BufferManager):
|
|
def text_callback(bufctl: codemp.BufferController):
|
|
def _():
|
|
change_id = buff.view.change_id()
|
|
while change := bufctl.try_recv().wait():
|
|
logger.debug("received remote buffer change!")
|
|
if change is None:
|
|
break
|
|
|
|
if change.is_empty():
|
|
logger.debug("change is empty. skipping.")
|
|
continue
|
|
|
|
# In case a change arrives to a background buffer, just apply it.
|
|
# We are not listening on it. Otherwise, interrupt the listening
|
|
# to avoid echoing back the change just received.
|
|
if buff.view.id() == g.ACTIVE_CODEMP_VIEW:
|
|
buff.view.settings()[g.CODEMP_IGNORE_NEXT_TEXT_CHANGE] = True
|
|
|
|
# we need to go through a sublime text command, since the method,
|
|
# view.replace needs an edit token, that is obtained only when calling
|
|
# a textcommand associated with a view.
|
|
buff.view.run_command(
|
|
"codemp_replace_text",
|
|
{
|
|
"start": change.start,
|
|
"end": change.end,
|
|
"content": change.content,
|
|
"change_id": change_id,
|
|
}, # pyright: ignore
|
|
)
|
|
sublime.set_timeout(_)
|
|
return text_callback
|
|
|
|
class BufferManager():
|
|
def __init__(self, handle: codemp.BufferController, v: sublime.View, filename: str):
|
|
self.handle: codemp.BufferController = handle
|
|
self.view: sublime.View = v
|
|
self.id = self.handle.path()
|
|
self.filename = filename
|
|
|
|
def __del__(self):
|
|
logger.debug(f"dropping buffer {self.id}")
|
|
self.handle.clear_callback()
|
|
self.handle.stop()
|
|
|
|
def __hash__(self):
|
|
return hash(self.id)
|
|
|
|
def send_change(self, changes):
|
|
# we do not do any index checking, and trust sublime with providing the correct
|
|
# sequential indexing, assuming the changes are applied in the order they are received.
|
|
for change in changes:
|
|
region = sublime.Region(change.a.pt, change.b.pt)
|
|
logger.debug(
|
|
"sending txt change: Reg({} {}) -> '{}'".format(
|
|
region.begin(), region.end(), change.str
|
|
)
|
|
)
|
|
# we must block and wait the send request to make sure the change went through ok
|
|
self.handle.send(region.begin(), region.end(), change.str).wait()
|
|
|
|
def sync(self, text_listener):
|
|
promise = self.handle.content()
|
|
|
|
def _():
|
|
content = promise.wait()
|
|
safe_listener_detach(text_listener)
|
|
populate_view(self.view, content)
|
|
safe_listener_attach(text_listener, self.view.buffer())
|
|
sublime.set_timeout_async(_)
|
|
|
|
class BufferRegistry():
|
|
def __init__(self):
|
|
self._buffers: bidict[BufferManager, WorkspaceManager] = bidict()
|
|
|
|
def lookup(self, ws: WorkspaceManager | None = None) -> list[BufferManager]:
|
|
if not ws:
|
|
return list(self._buffers.keys())
|
|
bf = self._buffers.inverse.get(ws)
|
|
return bf if bf else []
|
|
|
|
def lookupId(self, bid: str) -> BufferManager | None:
|
|
return next((bf for bf in self._buffers if bf.id == bid), None)
|
|
|
|
def add(self, bhandle: codemp.BufferController, wsm: WorkspaceManager):
|
|
bid = bhandle.path()
|
|
tmpfile = os.path.join(wsm.rootdir, bid)
|
|
open(tmpfile, "a").close()
|
|
|
|
win = sublime.active_window()
|
|
view = win.open_file(bid)
|
|
view.set_scratch(True)
|
|
view.retarget(tmpfile)
|
|
view.settings().set(g.CODEMP_VIEW_TAG, True)
|
|
view.settings().set(g.CODEMP_BUFFER_ID, bid)
|
|
view.set_status(g.SUBLIME_STATUS_ID, "[Codemp]")
|
|
|
|
bfm = BufferManager(bhandle, view, tmpfile)
|
|
|
|
def remove(self, bf: BufferManager | str | None):
|
|
if isinstance(bf, str):
|
|
bf = self.lookupId(bf)
|
|
|
|
if not bf: return
|
|
|
|
del self._buffers[bf]
|
|
bf.view.close()
|
|
|
|
|
|
|
|
|
|
|
|
|