hexedtech/codemp-sublime
3
0
Fork 0
mirror of https://github.com/hexedtech/codemp-sublime.git synced 2024-12-04 20:04:52 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge branch 'viewtag' into dev

Browse source
This commit is contained in:
cschen 2024-11-30 14:58:19 +01:00
commit d07d2f144e
21 changed files with 1399 additions and 1087 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.gitignore vendored
View file

@ -1,6 +1,6 @@
/target
test*
/lib
lib
# Byte-compiled / optimized / DLL files
__pycache__/

41
Codemp.sublime-commands
View file

@ -19,24 +19,38 @@
"file": "${packages}/CodempClient/README.md"
}
},
{
"caption": "Codemp: Browse Server",
"command": "codemp_browse_server",
"args": {}
},
{
"caption": "Codemp: Browse Workspace",
"command": "codemp_browse_workspace",
"args": {
}
},
{
// # on_window_command, does not trigger when called from the command palette
// # See: https://github.com/sublimehq/sublime_text/issues/2234
"caption": "Codemp: Connect",
"command": "codemp_connect",
"args": {
"server_host": "http://code.mp:50053",
"server_host": "code.mp",
"user_name" : "cschen@codemp.dev",
"password" : "***REMOVED***"
}
},
{
"caption": "Codemp: Disconnect Client",
"command": "codemp_disconnect",
"arg": {}
"args": {}
},
{
"caption": "Codemp: Join Workspace",
"command": "codemp_join_workspace",
"arg": {
"args": {
// 'workspace_id': 'asd'
// 'buffer_id': 'test'
},
@ -44,14 +58,14 @@
{
"caption": "Codemp: Leave Workspace",
"command": "codemp_leave_workspace",
"arg": {
"args": {
// "id": 'lmaaaao'
}
},
{
"caption": "Codemp: Invite To Workspace",
"command": "codemp_invite_to_workspace",
"arg": {
"args": {
// "id": 'lmaaaao'
// "user": 'lupo'
}
@ -59,21 +73,21 @@
{
"caption": "Codemp: Create Workspace",
"command": "codemp_create_workspace",
"arg": {
"args": {
// "id": 'lmaaaao'
}
},
{
"caption": "Codemp: Delete Workspace",
"command": "codemp_delete_workspace",
"arg": {
"args": {
// "id": 'lmaaaao'
}
},
{
"caption": "Codemp: Join Buffer",
"command": "codemp_join_buffer",
"arg": {
"args": {
// 'workspace_id': 'asd'
// 'buffer_id': 'test'
},
@ -81,7 +95,7 @@
{
"caption": "Codemp: Leave Buffer",
"command": "codemp_leave_buffer",
"arg": {
"args": {
// 'workspace_id': 'asd'
// 'buffer_id': 'test'
}
@ -89,7 +103,7 @@
{
"caption": "Codemp: Create Buffer",
"command": "codemp_create_buffer",
"arg": {
"args": {
// 'workspace_id': 'asd'
// 'buffer_id': 'test'
}
@ -97,9 +111,14 @@
{
"caption": "Codemp: Delete Buffer",
"command": "codemp_delete_buffer",
"arg": {
"args": {
// 'workspace_id': 'asd'
// 'buffer_id': 'test'
}
},
{
"caption": "Codemp: Sync",
"command": "codemp_sync_buffer",
"args": {}
}
]

159
input_handlers.py
View file

@ -1,159 +0,0 @@
import sublime_plugin
import logging
from typing import Tuple, Union, List
from .src.client import client
logger = logging.getLogger(__name__)
# Input handlers
############################################################
class SimpleTextInput(sublime_plugin.TextInputHandler):
def __init__(self, *args: Tuple[str, Union[str, List[str]]]):
logging.debug(f"why isn't the text input working? {args}")
self.argname = args[0][0]
self.default = args[0][1]
self.next_inputs = args[1:]
def initial_text(self):
if isinstance(self.default, str):
return self.default
else: return ""
def name(self):
return self.argname
def next_input(self, args):
if len(self.next_inputs) > 0:
if self.next_inputs[0][0] not in args:
if isinstance(self.next_inputs[0][1], list):
return SimpleListInput(*self.next_inputs)
else:
return SimpleTextInput(*self.next_inputs)
class SimpleListInput(sublime_plugin.ListInputHandler):
def __init__(self, *args: Tuple[str, Union["list[str]", str]]):
self.argname = args[0][0]
self.list = args[0][1]
self.next_inputs = args[1:]
def name(self):
return self.argname
def list_items(self):
if isinstance(self.list, list):
return self.list
else:
return [self.list]
def next_input(self, args):
if len(self.next_inputs) > 0:
if self.next_inputs[0][0] not in args:
if isinstance(self.next_inputs[0][1], str):
return SimpleTextInput(*self.next_inputs)
else:
return SimpleListInput(*self.next_inputs)
class ActiveWorkspacesIdList(sublime_plugin.ListInputHandler):
def __init__(self, window=None, buffer_list=False, buffer_text=False):
self.window = window
self.buffer_list = buffer_list
self.buffer_text = buffer_text
def name(self):
return "workspace_id"
def list_items(self):
return [vws.id for vws in client.all_workspaces(self.window)]
def next_input(self, args):
if self.buffer_list:
return BufferIdList(args["workspace_id"])
elif self.buffer_text:
return SimpleTextInput(("buffer_id", "new buffer"))
# To allow for having a selection and choosing non existing workspaces
# we do a little dance: We pass this list input handler to a TextInputHandler
# when we select "Create New..." which adds his result to the list of possible
# workspaces and pop itself off the stack to go back to the list handler.
class WorkspaceIdList(sublime_plugin.ListInputHandler):
def __init__(self):
assert client.codemp is not None # the command should not be available
# at the moment, the client can't give us a full list of existing workspaces
# so a textinputhandler would be more appropriate. but we keep this for the future
self.add_entry_text = "* add entry..."
self.list = client.codemp.list_workspaces(True, True).wait()
self.list.sort()
self.list.append(self.add_entry_text)
self.preselected = None
def name(self):
return "workspace_id"
def placeholder(self):
return "Workspace"
def list_items(self):
if self.preselected is not None:
return (self.list, self.preselected)
else:
return self.list
def next_input(self, args):
if args["workspace_id"] == self.add_entry_text:
return AddListEntry(self)
class BufferIdList(sublime_plugin.ListInputHandler):
def __init__(self, workspace_id):
vws = client.workspace_from_id(workspace_id)
self.add_entry_text = "* create new..."
self.list = vws.codemp.filetree(None)
self.list.sort()
self.list.append(self.add_entry_text)
self.preselected = None
def name(self):
return "buffer_id"
def placeholder(self):
return "Buffer Id"
def list_items(self):
if self.preselected is not None:
return (self.list, self.preselected)
else:
return self.list
def next_input(self, args):
if args["buffer_id"] == self.add_entry_text:
return AddListEntry(self)
class AddListEntry(sublime_plugin.TextInputHandler):
# this class works when the list input handler
# added appended a new element to it's list that will need to be
# replaced with the entry added from here!
def __init__(self, list_input_handler):
self.parent = list_input_handler
def name(self):
return ""
def validate(self, text: str) -> bool:
return not len(text) == 0
def confirm(self, text: str):
self.parent.list.pop() # removes the add_entry_text
self.parent.list.insert(0, text)
self.parent.preselected = 0
def next_input(self, args):
return sublime_plugin.BackInputHandler()

117
listeners.py
View file

@ -1,117 +0,0 @@
import sublime
import sublime_plugin
import logging
from .src.client import client
from .src.utils import safe_listener_attach
from .src.utils import safe_listener_detach
from .src import globals as g
logger = logging.getLogger(__name__)
# Listeners
##############################################################################
class EventListener(sublime_plugin.EventListener):
def is_enabled(self):
return client.codemp is not None
def on_exit(self):
client.disconnect()
if client.driver is not None:
client.driver.stop()
def on_pre_close_window(self, window):
assert client.codemp is not None
for vws in client.all_workspaces(window):
client.codemp.leave_workspace(vws.id)
client.uninstall_workspace(vws)
def on_text_command(self, view, command_name, args):
if command_name == "codemp_replace_text":
logger.info("got a codemp_replace_text command!")
def on_post_text_command(self, view, command_name, args):
if command_name == "codemp_replace_text":
logger.info("got a codemp_replace_text command!")
class CodempClientViewEventListener(sublime_plugin.ViewEventListener):
@classmethod
def is_applicable(cls, settings):
return settings.get(g.CODEMP_BUFFER_TAG) is not None
@classmethod
def applies_to_primary_view_only(cls):
return False
def on_selection_modified_async(self):
region = self.view.sel()[0]
start = self.view.rowcol(region.begin())
end = self.view.rowcol(region.end())
vws = client.workspace_from_view(self.view)
vbuff = client.buffer_from_view(self.view)
if vws is None or vbuff is None:
logger.error("we couldn't find the matching buffer or workspace!")
return
logger.debug(f"selection modified! {vws.id}, {vbuff.id} - {start}, {end}")
vws.send_cursor(vbuff.id, start, end)
def on_activated(self):
global TEXT_LISTENER
logger.debug(f"'{self.view}' view activated!")
safe_listener_attach(TEXT_LISTENER, self.view.buffer()) # pyright: ignore
def on_deactivated(self):
global TEXT_LISTENER
logger.debug(f"'{self.view}' view deactivated!")
safe_listener_detach(TEXT_LISTENER) # pyright: ignore
def on_pre_close(self):
if self.view == sublime.active_window().active_view():
logger.debug("closing active view")
global TEXT_LISTENER
safe_listener_detach(TEXT_LISTENER) # pyright: ignore
vws = client.workspace_from_view(self.view)
vbuff = client.buffer_from_view(self.view)
if vws is None or vbuff is None:
logger.debug("no matching workspace or buffer.")
return
client.unregister_buffer(vbuff)
vws.uninstall_buffer(vbuff)
def on_text_command(self, command_name, args):
if command_name == "codemp_replace_text":
logger.info("got a codemp_replace_text command! but in the view listener")
def on_post_text_command(self, command_name, args):
if command_name == "codemp_replace_text":
logger.info("got a codemp_replace_text command! but in the view listener")
class CodempClientTextChangeListener(sublime_plugin.TextChangeListener):
@classmethod
def is_applicable(cls, buffer): # pyright: ignore
# don't attach this event listener automatically
# we'll do it by hand with .attach(buffer).
return False
def on_text_changed(self, changes):
s = self.buffer.primary_view().settings()
if s.get(g.CODEMP_IGNORE_NEXT_TEXT_CHANGE, False):
logger.debug("Ignoring echoing back the change.")
s[g.CODEMP_IGNORE_NEXT_TEXT_CHANGE] = False
return
vbuff = client.buffer_from_view(self.buffer.primary_view())
if vbuff is not None:
logger.debug(f"local buffer change! {vbuff.id}")
vbuff.send_buffer_change(changes)
TEXT_LISTENER = CodempClientTextChangeListener()

221
main.py Normal file
View file

@ -0,0 +1,221 @@
# pyright: reportIncompatibleMethodOverride=false
import sublime
import sublime_plugin
import logging
import codemp
from .plugin.utils import safe_listener_detach
from .plugin.utils import safe_listener_attach
from .plugin.utils import some
from .plugin.core.session import session
from .plugin.core.workspace import workspaces
from .plugin.core.buffers import buffers
from .plugin.text_listener import TEXT_LISTENER
from .plugin import globals as g
# We import these just to showcase the commands available.
from .plugin.commands.client import CodempConnectCommand
from .plugin.commands.client import CodempDisconnectCommand
from .plugin.commands.client import CodempCreateWorkspaceCommand
from .plugin.commands.client import CodempDeleteWorkspaceCommand
from .plugin.commands.client import CodempJoinWorkspaceCommand
from .plugin.commands.client import CodempLeaveWorkspaceCommand
from .plugin.commands.client import CodempInviteToWorkspaceCommand
from .plugin.commands.workspace import CodempCreateBufferCommand
from .plugin.commands.workspace import CodempDeleteBufferCommand
from .plugin.commands.workspace import CodempJoinBufferCommand
from .plugin.commands.workspace import CodempLeaveBufferCommand
from .plugin.quickpanel.qpbrowser import QPServerBrowser
from .plugin.quickpanel.qpbrowser import QPWorkspaceBrowser
LOG_LEVEL = logging.DEBUG
handler = logging.StreamHandler()
handler.setFormatter(
logging.Formatter(
fmt="<{thread}/{threadName}> {levelname} [{name} :: {funcName}] {message}",
style="{",
)
)
package_logger = logging.getLogger(__package__)
package_logger.setLevel(LOG_LEVEL)
package_logger.propagate = False
logger = logging.getLogger(__name__)
# Initialisation and Deinitialisation
##############################################################################
def plugin_loaded():
package_logger.addHandler(handler)
logger.debug("plugin loaded")
def plugin_unloaded():
logger.debug("unloading")
safe_listener_detach(TEXT_LISTENER)
package_logger.removeHandler(handler)
def kill_all():
for ws in workspaces.lookup():
session.client.leave_workspace(ws.id)
workspaces.remove(ws)
session.stop()
def objects_from_view(view):
assert view.settings().get(g.CODEMP_VIEW_TAG, False)
buffid = str(view.settings().get(g.CODEMP_BUFFER_ID))
vbuff = buffers.lookupId(buffid)
vws = buffers.lookupParent(vbuff)
win = workspaces.lookupParent(vws)
return win, vws, vbuff
class CodempBrowseWorkspaceCommand(sublime_plugin.WindowCommand):
def is_enabled(self) -> bool:
return session.is_active()
def run(self, workspace_id):
wks = workspaces.lookupId(workspace_id)
buffers = wks.handle.fetch_buffers()
QPWorkspaceBrowser(self.window, workspace_id, buffers.wait()).run()
class CodempBrowseServerCommand(sublime_plugin.WindowCommand):
def is_enabled(self) -> bool:
return session.is_active()
def run(self):
wks = session.get_workspaces()
QPServerBrowser(self.window, session.config.host, wks).run()
class CodempReplaceTextCommand(sublime_plugin.TextCommand):
def run(self, edit, start, end, content, change_id = None):
# we modify the region to account for any change that happened in the mean time
region = sublime.Region(start, end)
if change_id:
region = self.view.transform_region_from(sublime.Region(start, end), change_id)
self.view.replace(edit, region, content)
class CodempSyncBuffer(sublime_plugin.TextCommand):
def run(self, edit):
buff = buffers.lookupId(self.view.settings().get(g.CODEMP_BUFFER_ID))
buff.sync(TEXT_LISTENER)
class EventListener(sublime_plugin.EventListener):
def is_enabled(self):
return session.is_active()
def on_exit(self):
kill_all()
# client.disconnect()
# if client.driver is not None:
# client.driver.stop()
def on_pre_close_window(self, window):
for vws in workspaces.lookup(window):
sublime.run_command("codemp_leave_workspace", {
"workspace_id": vws.id
})
def on_text_command(self, view, command_name, args):
if command_name == "codemp_replace_text":
logger.info("got a codemp_replace_text command!")
def on_post_text_command(self, view, command_name, args):
if command_name == "codemp_replace_text":
logger.info("got a codemp_replace_text command!")
class CodempClientViewEventListener(sublime_plugin.ViewEventListener):
@classmethod
def is_applicable(cls, settings):
return settings.get(g.CODEMP_VIEW_TAG) is not None
@classmethod
def applies_to_primary_view_only(cls):
return False
def on_selection_modified_async(self):
region = self.view.sel()[0]
start = self.view.rowcol(region.begin())
end = self.view.rowcol(region.end())
try:
_, vws, vbuff = objects_from_view(self.view)
except KeyError:
logger.error(f"Could not find buffers associated with the view {self.view}.\
Removing the tag to disable event listener. Reattach.")
# delete the tag so we disable this event listener on the view
del self.view.settings()[g.CODEMP_VIEW_TAG]
return
vws.send_cursor(vbuff.id, start, end)
# logger.debug(f"selection modified! {vws.id}, {vbuff.id} - {start}, {end}")
def on_activated(self):
logger.debug(f"'{self.view}' view activated!")
safe_listener_attach(TEXT_LISTENER, self.view.buffer()) # pyright: ignore
def on_deactivated(self):
logger.debug(f"'{self.view}' view deactivated!")
safe_listener_detach(TEXT_LISTENER) # pyright: ignore
def on_pre_close(self):
if self.view == sublime.active_window().active_view():
logger.debug("closing active view")
safe_listener_detach(TEXT_LISTENER) # pyright: ignore
try:
bid = str(self.view.settings().get(g.CODEMP_BUFFER_ID))
vws = buffers.lookupParent(bid)
some(self.view.window()).run_command(
"codemp_leave_buffer",
{"workspace_id": vws.id, "buffer_id": bid})
except KeyError:
return
def on_text_command(self, command_name, args):
if command_name == "codemp_replace_text":
logger.info("got a codemp_replace_text command! but in the view listener")
def on_post_text_command(self, command_name, args):
if command_name == "codemp_replace_text":
logger.info("got a codemp_replace_text command! but in the view listener")
# Next TODO:
# Server configurations:
# - where do we store it?
# - TOML? yes probably toml
# * Quickpanel for connecting with stuff.
# * Quickpanel for browsing the servers
# * Move all "server actions" like, create, delete, rename etc. as quickpanel actions. (See SFTP plugin.)
# * make panel for notifications!
# * make panel for errors and logging!
# Proxy Commands ( NOT USED, left just in case we need it again. )
#############################################################################
# class ProxyCodempShareCommand(sublime_plugin.WindowCommand):
# # on_window_command, does not trigger when called from the command palette
# # See: https://github.com/sublimehq/sublime_text/issues/2234
# def run(self, **kwargs):
# self.window.run_command("codemp_share", kwargs)
#
# def input(self, args):
# if 'sublime_buffer' not in args:
# return SublimeBufferPathInputHandler()
#
# def input_description(self):
# return 'Share Buffer:'

72
plugin.py
View file

@ -1,72 +0,0 @@
# pyright: reportIncompatibleMethodOverride=false
import sublime
import sublime_plugin
import logging
from .src.utils import safe_listener_detach
from listeners import TEXT_LISTENER
from client_commands import CodempConnectCommand
from client_commands import CodempDisconnectCommand
from client_commands import CodempCreateWorkspaceCommand
from client_commands import CodempDeleteWorkspaceCommand
from client_commands import CodempJoinWorkspaceCommand
from client_commands import CodempLeaveWorkspaceCommand
from client_commands import CodempInviteToWorkspaceCommand
from workspace_commands import CodempCreateBufferCommand
from workspace_commands import CodempDeleteBufferCommand
from workspace_commands import CodempJoinBufferCommand
from workspace_commands import CodempLeaveBufferCommand
LOG_LEVEL = logging.DEBUG
handler = logging.StreamHandler()
handler.setFormatter(
logging.Formatter(
fmt="<{thread}/{threadName}> {levelname} [{name} :: {funcName}] {message}",
style="{",
)
)
package_logger = logging.getLogger(__package__)
package_logger.addHandler(handler)
package_logger.setLevel(LOG_LEVEL)
package_logger.propagate = False
logger = logging.getLogger(__name__)
# Initialisation and Deinitialisation
##############################################################################
def plugin_loaded():
logger.debug("plugin loaded")
def plugin_unloaded():
logger.debug("unloading")
safe_listener_detach(TEXT_LISTENER)
package_logger.removeHandler(handler)
# client.disconnect()
# Text Change Command
#############################################################################
class CodempReplaceTextCommand(sublime_plugin.TextCommand):
def run(self, edit, start, end, content, change_id):
# we modify the region to account for any change that happened in the mean time
region = self.view.transform_region_from(sublime.Region(start, end), change_id)
self.view.replace(edit, region, content)
# Proxy Commands ( NOT USED, left just in case we need it again. )
#############################################################################
# class ProxyCodempShareCommand(sublime_plugin.WindowCommand):
# # on_window_command, does not trigger when called from the command palette
# # See: https://github.com/sublimehq/sublime_text/issues/2234
# def run(self, **kwargs):
# self.window.run_command("codemp_share", kwargs)
#
# def input(self, args):
# if 'sublime_buffer' not in args:
# return SublimeBufferPathInputHandler()
#
# def input_description(self):
# return 'Share Buffer:'

202
client_commands.py → plugin/commands/client.py
View file

@ -1,40 +1,18 @@
# pyright: ignore[reportIncompatibleMethodOverride]
import sublime
import sublime_plugin
import logging
import random
from .src.client import client
from input_handlers import SimpleTextInput
from input_handlers import SimpleListInput
from input_handlers import ActiveWorkspacesIdList
import codemp
from ..core.session import session
from ..core.workspace import workspaces
from ..input_handlers import SimpleTextInput
from ..input_handlers import SimpleListInput
logger = logging.getLogger(__name__)
# Client Commands
#############################################################################
# Connect Command
class CodempConnectCommand(sublime_plugin.WindowCommand):
def is_enabled(self) -> bool:
return client.codemp is None
def run(self, server_host, user_name, password): # pyright: ignore[reportIncompatibleMethodOverride]
logger.info(f"Connecting to {server_host} with user {user_name}...")
def _():
try:
client.connect(server_host, user_name, password)
except Exception as e:
sublime.error_message(
"Could not connect:\n Make sure the server is up\n\
and your credentials are correct."
)
sublime.set_timeout_async(_)
def input_description(self):
return "Server host:"
def input(self, args):
if "server_host" not in args:
return SimpleTextInput(
@ -54,135 +32,157 @@ class CodempConnectCommand(sublime_plugin.WindowCommand):
("password", "password?"),
)
def input_description(self):
return "Server host:"
def run(self, server_host, user_name, password): # pyright: ignore[reportIncompatibleMethodOverride]
def _():
try:
config = codemp.Config(
username = user_name,
password = password,
host = server_host,
port=50053,
tls=False)
session.connect(config)
except Exception as e:
logger.error(e)
sublime.error_message(
"Could not connect:\n Make sure the server is up\n\
and your credentials are correct."
)
sublime.set_timeout_async(_)
# Disconnect Command
class CodempDisconnectCommand(sublime_plugin.WindowCommand):
def is_enabled(self):
return client.codemp is not None
return session.is_active()
def run(self):
client.disconnect()
cli = session.client
for ws in workspaces.lookup():
if cli.leave_workspace(ws.id):
workspaces.remove(ws)
session.drop_client()
logger.info(f"disconnected from server '{session.config.host}'!")
# Join Workspace Command
class CodempJoinWorkspaceCommand(sublime_plugin.WindowCommand):
def is_enabled(self) -> bool:
return client.codemp is not None
def run(self, workspace_id): # pyright: ignore[reportIncompatibleMethodOverride]
assert client.codemp is not None
if workspace_id is None:
return
logger.info(f"Joining workspace: '{workspace_id}'...")
promise = client.codemp.join_workspace(workspace_id)
active_window = sublime.active_window()
def _():
try:
workspace = promise.wait()
except Exception as e:
logger.error(
f"Could not join workspace '{workspace_id}'.\n\nerror: {e}"
)
sublime.error_message(f"Could not join workspace '{workspace_id}'")
return
client.install_workspace(workspace, active_window)
sublime.set_timeout_async(_)
# the else shouldn't really happen, and if it does, it should already be instantiated.
# ignore.
return session.is_active()
def input_description(self):
return "Join:"
def input(self, args):
assert client.codemp is not None
if "workspace_id" not in args:
list = client.codemp.list_workspaces(True, True)
wslist = session.get_workspaces()
return SimpleListInput(
("workspace_id", list.wait()),
("workspace_id", wslist),
)
def run(self, workspace_id): # pyright: ignore[reportIncompatibleMethodOverride]
if workspace_id is None:
return
logger.info(f"Joining workspace: '{workspace_id}'...")
try:
ws = session.client.attach_workspace(workspace_id).wait()
except Exception as e:
logger.error(f"Could not join workspace '{workspace_id}': {e}")
sublime.error_message(f"Could not join workspace '{workspace_id}'")
return
logger.debug("Joined! Adding workspace to registry")
workspaces.add(ws)
# Leave Workspace Command
class CodempLeaveWorkspaceCommand(sublime_plugin.WindowCommand):
def is_enabled(self):
return client.codemp is not None and len(client.all_workspaces(self.window)) > 0
def run(self, workspace_id: str): # pyright: ignore[reportIncompatibleMethodOverride]
assert client.codemp is not None
if client.codemp.leave_workspace(workspace_id):
vws = client.workspace_from_id(workspace_id)
if vws is not None:
client.uninstall_workspace(vws)
else:
logger.error(f"could not leave the workspace '{workspace_id}'")
return session.is_active() and \
len(workspaces.lookup(self.window)) > 0
def input(self, args):
if "workspace_id" not in args:
return ActiveWorkspacesIdList()
wslist = session.client.active_workspaces()
return SimpleListInput(
("workspace_id", wslist),
)
def run(self, workspace_id: str): # pyright: ignore[reportIncompatibleMethodOverride]
try:
workspaces.remove(workspace_id)
finally:
if not session.client.leave_workspace(workspace_id):
logger.error(f"could not leave the workspace '{workspace_id}'")
class CodempInviteToWorkspaceCommand(sublime_plugin.WindowCommand):
def is_enabled(self) -> bool:
return client.codemp is not None and len(client.all_workspaces(self.window)) > 0
def run(self, workspace_id: str, user: str): # pyright: ignore[reportIncompatibleMethodOverride]
assert client.codemp is not None
client.codemp.invite_to_workspace(workspace_id, user)
logger.debug(f"invite sent to user {user} for workspace {workspace_id}.")
return session.is_active() and len(workspaces.lookup(self.window)) > 0
def input(self, args):
assert client.codemp is not None
if "workspace_id" not in args:
wslist = client.codemp.list_workspaces(True, False)
wslist = session.get_workspaces(owned=True, invited=False)
return SimpleListInput(
("workspace_id", wslist.wait()), ("user", "invitee's username")
("workspace_id", wslist), ("user", "invitee's username")
)
if "user" not in args:
return SimpleTextInput(("user", "invitee's username"))
def run(self, workspace_id: str, user: str): # pyright: ignore[reportIncompatibleMethodOverride]
try:
session.client.invite_to_workspace(workspace_id, user)
logger.debug(f"invite sent to user {user} for workspace {workspace_id}.")
except Exception as e:
logger.error(f"Could not invite to workspace: {e}")
class CodempCreateWorkspaceCommand(sublime_plugin.WindowCommand):
def is_enabled(self):
return client.codemp is not None
def run(self, workspace_id: str): # pyright: ignore[reportIncompatibleMethodOverride]
assert client.codemp is not None
client.codemp.create_workspace(workspace_id)
return session.is_active()
def input(self, args):
if "workspace_id" not in args:
return SimpleTextInput(("workspace_id", "new workspace"))
return SimpleTextInput(("workspace_id", "new workspace name"))
def run(self, workspace_id: str): # pyright: ignore[reportIncompatibleMethodOverride]
try:
session.client.create_workspace(workspace_id)
except Exception as e:
logger.error(f"Could not create workspace: {e}")
class CodempDeleteWorkspaceCommand(sublime_plugin.WindowCommand):
def is_enabled(self):
return client.codemp is not None
return session.is_active()
def input(self, args):
workspaces = session.get_workspaces(owned=True, invited=False) # noqa: F841
if "workspace_id" not in args:
return SimpleListInput(("workspace_id", workspaces))
def run(self, workspace_id: str): # pyright: ignore[reportIncompatibleMethodOverride]
assert client.codemp is not None
vws = client.workspace_from_id(workspace_id)
if vws is not None:
try:
vws = workspaces.lookupId(workspace_id)
if not sublime.ok_cancel_dialog(
"You are currently attached to '{workspace_id}'.\n\
Do you want to detach and delete it?",
ok_title="yes",
title="Delete Workspace?",
ok_title="yes", title="Delete Workspace?",
):
return
if not client.codemp.leave_workspace(workspace_id):
logger.debug("error while leaving the workspace:")
return
client.uninstall_workspace(vws)
self.window.run_command(
"codemp_leave_workspace",
{"workspace_id": workspace_id})
client.codemp.delete_workspace(workspace_id)
except KeyError: pass
finally:
session.client.delete_workspace(workspace_id)
def input(self, args):
assert client.codemp is not None
workspaces = client.codemp.list_workspaces(True, False) # noqa: F841
if "workspace_id" not in args:
return SimpleListInput(("workspace_id", workspaces.wait()))

224
plugin/commands/workspace.py Normal file
View file

@ -0,0 +1,224 @@
import sublime
import sublime_plugin
import logging
from ..core.session import session
from ..core.workspace import workspaces
from ..core.buffers import buffers
from ..text_listener import TEXT_LISTENER
from ..utils import safe_listener_attach, safe_listener_detach, populate_view
from ..input_handlers import SimpleListInput, SimpleTextInput
logger = logging.getLogger(__name__)
# Join Buffer Command
class CodempJoinBufferCommand(sublime_plugin.WindowCommand):
def is_enabled(self):
return len(workspaces.lookup(self.window)) > 0
def input_description(self) -> str:
return "Attach: "
def input(self, args):
if "workspace_id" not in args:
wslist = session.get_workspaces(owned=True, invited=True)
return SimpleListInput(
("workspace_id", wslist),
)
if "buffer_id" not in args:
try: ws = workspaces.lookupId(args["workspace_id"])
except KeyError:
sublime.error_message("Workspace does not exists or is not active.")
return None
bflist = ws.handle.fetch_buffers().wait()
return SimpleListInput(
("buffer_id", bflist),
)
def run(self, workspace_id, buffer_id): # pyright: ignore[reportIncompatibleMethodOverride]
try: vws = workspaces.lookupId(workspace_id)
except KeyError:
logger.error(f"Can't create buffer: '{workspace_id}' does not exists or is not active.")
return
try: # if it exists already, focus and listen
buff = buffers.lookupId(buffer_id)
safe_listener_detach(TEXT_LISTENER)
safe_listener_attach(TEXT_LISTENER, buff.view.buffer())
self.window.focus_view(buff.view)
return
except KeyError:
pass
# # if doesn't exist in the workspace, ask for creation.
# if vws.handle.get_buffer(buffer_id) is None:
# if sublime.ok_cancel_dialog(
# f"There is no buffer named '{buffer_id}' in the workspace '{workspace_id}'.\n\
# Do you want to create it?",
# ok_title="yes", title="Create Buffer?",
# ):
# sublime.run_command("codemp_create_buffer", {
# "workspace_id": workspace_id,
# "buffer_id": buffer_id
# })
# now we can defer the attaching process
logger.debug(f"attempting to attach to {buffer_id}...")
ctl_promise = vws.handle.attach_buffer(buffer_id)
def _():
try:
buff_ctl = ctl_promise.wait()
logger.debug("attach successfull!")
except Exception as e:
logging.error(f"error when attaching to buffer '{id}':\n\n {e}")
sublime.error_message(f"Could not attach to buffer '{buffer_id}'")
return
safe_listener_detach(TEXT_LISTENER)
content_promise = buff_ctl.content()
vbuff = buffers.add(buff_ctl, vws)
content = content_promise.wait()
populate_view(vbuff.view, content)
if self.window.active_view() == vbuff.view:
# if view is already active focusing it won't trigger `on_activate`.
safe_listener_attach(TEXT_LISTENER, vbuff.view.buffer())
else:
self.window.focus_view(vbuff.view)
sublime.set_timeout_async(_)
# Leave Buffer Comand
class CodempLeaveBufferCommand(sublime_plugin.WindowCommand):
def is_enabled(self):
return len(buffers.lookup()) > 0
def input_description(self) -> str:
return "Leave: "
def input(self, args):
if "workspace_id" not in args:
wslist = session.client.active_workspaces()
return SimpleListInput(
("workspace_id", wslist),
)
if "buffer_id" not in args:
bflist = [bf.id for bf in buffers.lookup(args["workspace_id"])]
return SimpleListInput(
("buffer_id", bflist)
)
def run(self, workspace_id, buffer_id): # pyright: ignore[reportIncompatibleMethodOverride]
try:
buffers.lookupId(buffer_id)
vws = workspaces.lookupId(workspace_id)
except KeyError:
sublime.error_message(f"You are not attached to the buffer '{buffer_id}'")
logging.warning(f"You are not attached to the buffer '{buffer_id}'")
return
if not vws.handle.get_buffer(buffer_id):
logging.error("The desired buffer is not managed by the workspace.")
return
def _():
try:
buffers.remove(buffer_id)
finally:
if not vws.handle.detach_buffer(buffer_id):
logger.error(f"could not leave the buffer {buffer_id}.")
sublime.set_timeout_async(_)
# Leave Buffer Comand
class CodempCreateBufferCommand(sublime_plugin.WindowCommand):
def is_enabled(self):
return len(workspaces.lookup()) > 0
def input_description(self) -> str:
return "Create Buffer: "
def input(self, args):
if "workspace_id" not in args:
wslist = session.client.active_workspaces()
return SimpleListInput(
("workspace_id", wslist),
)
if "buffer_id" not in args:
return SimpleTextInput(
("buffer_id", "new buffer name"),
)
def run(self, workspace_id, buffer_id):# pyright: ignore[reportIncompatibleMethodOverride]
try: vws = workspaces.lookupId(workspace_id)
except KeyError:
sublime.error_message(
f"You are not attached to the workspace '{workspace_id}'"
)
logging.warning(f"You are not attached to the workspace '{workspace_id}'")
return
vws.handle.create_buffer(buffer_id)
logging.info(
"created buffer '{buffer_id}' in the workspace '{workspace_id}'.\n\
To interact with it you need to attach to it with Codemp: Attach."
)
class CodempDeleteBufferCommand(sublime_plugin.WindowCommand):
def is_enabled(self):
return len(workspaces.lookup()) > 0
def input_description(self) -> str:
return "Delete buffer: "
def input(self, args):
if "workspace_id" not in args:
wslist = session.get_workspaces(owned=True, invited=False)
return SimpleListInput(
("workspace_id", wslist),
)
if "buffer_id" not in args:
try: ws = workspaces.lookupId(args["workspace_id"])
except KeyError:
sublime.error_message("Workspace does not exists or is not attached.")
return sublime_plugin.BackInputHandler()
bflist = ws.handle.fetch_buffers().wait()
return SimpleListInput(
("buffer_id", bflist),
)
def run(self, workspace_id, buffer_id):# pyright: ignore[reportIncompatibleMethodOverride]
try: vws = workspaces.lookupId(workspace_id)
except KeyError:
sublime.error_message(
f"You are not attached to the workspace '{workspace_id}'"
)
logging.warning(f"You are not attached to the workspace '{workspace_id}'")
return
if not sublime.ok_cancel_dialog(
f"Confirm you want to delete the buffer '{buffer_id}'",
ok_title="delete", title="Delete Buffer?",
): return
try:
buffers.lookupId(buffer_id)
if not sublime.ok_cancel_dialog(
"You are currently attached to '{buffer_id}'.\n\
Do you want to detach and delete it?",
ok_title="yes", title="Delete Buffer?",
):
return
self.window.run_command(
"codemp_leave_buffer",
{ "workspace_id": workspace_id, "buffer_id": buffer_id })
except KeyError: pass
finally:
vws.handle.delete_buffer(buffer_id).wait()

166
plugin/core/buffers.py Normal file
View file

@ -0,0 +1,166 @@
from __future__ import annotations
from typing import Optional, TYPE_CHECKING
if TYPE_CHECKING:
from .workspace import WorkspaceManager
import codemp
import sublime
import os
import logging
import threading
from codemp import TextChange
from .. import globals as g
from ..utils import populate_view
from ..utils import get_contents
from ..utils import safe_listener_attach
from ..utils import safe_listener_detach
from ..utils import bidict
logger = logging.getLogger(__name__)
def bind_callback(v: sublime.View):
# we need this lock to prevent multiple instance of try_recv() to spin up
# which would cause out of order insertion of changes.
multi_tryrecv_lock = threading.Lock()
def _callback(bufctl: codemp.BufferController):
def _():
try:
# change_id = v.change_id()
change_id = None
while buffup := bufctl.try_recv().wait():
logger.debug("received remote buffer change!")
if buffup is None:
break
if buffup.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 v == sublime.active_window().active_view():
v.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.
change = buffup.change
v.run_command(
"codemp_replace_text",
{
"start": change.start_idx,
"end": change.end_idx,
"content": change.content,
"change_id": change_id,
}, # pyright: ignore
)
bufctl.ack(buffup.version)
except Exception as e:
raise e
finally:
logger.debug("releasing lock")
multi_tryrecv_lock.release()
if multi_tryrecv_lock.acquire(blocking=False):
logger.debug("acquiring lock")
sublime.set_timeout(_)
return _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
self.handle.callback(bind_callback(self.view))
def __del__(self):
logger.debug(f"dropping buffer {self.id}")
self.handle.clear_callback()
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(TextChange(start=region.begin(), end=region.end(), content=change.str))
def sync(self, text_listener):
promise = self.handle.content()
def _():
content = promise.wait()
current_contents = get_contents(self.view)
if content == current_contents:
return
safe_listener_detach(text_listener)
populate_view(self.view, content)
safe_listener_attach(text_listener, self.view.buffer())
sublime.status_message("Syncd contents.")
sublime.set_timeout_async(_)
class BufferRegistry():
def __init__(self):
self._buffers: bidict[BufferManager, WorkspaceManager] = bidict()
def lookup(self, ws: Optional[WorkspaceManager] = None) -> list[BufferManager]:
if not ws:
return list(self._buffers.keys())
bfs = self._buffers.inverse.get(ws)
return bfs if bfs else []
def lookupParent(self, bf: BufferManager | str) -> WorkspaceManager:
if isinstance(bf, str):
bf = self.lookupId(bf)
return self._buffers[bf]
def lookupId(self, bid: str) -> BufferManager:
bfm = next((bf for bf in self._buffers if bf.id == bid), None)
if not bfm: raise KeyError
return bfm
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]")
tmpfile = "DISABLE"
bfm = BufferManager(bhandle, view, tmpfile)
self._buffers[bfm] = wsm
return bfm
def remove(self, bf: BufferManager | str):
if isinstance(bf, str):
bf = self.lookupId(bf)
del self._buffers[bf]
bf.view.close()
buffers = BufferRegistry()

66
plugin/core/session.py Normal file
View file

@ -0,0 +1,66 @@
import logging
import codemp
from ..utils import some
logger = logging.getLogger(__name__)
class SessionManager():
def __init__(self) -> None:
self._running = False
self._driver = None
self._client = None
def is_init(self):
return self._running and self._driver is not None
def is_active(self):
return self.is_init() and self._client is not None
@property
def client(self):
return some(self._client)
def get_or_init(self) -> codemp.Driver:
if self._driver:
return self._driver
self._driver = codemp.init()
logger.debug("registering logger callback...")
if not codemp.set_logger(lambda msg: logger.debug(msg), False):
logger.debug(
"could not register the logger... \
If reconnecting it's ok, \
the previous logger is still registered"
)
self._running = True
return self._driver
def stop(self):
if not self._driver:
return
self.drop_client()
self._driver.stop()
self._running = False
self._driver = None
def connect(self, config: codemp.Config) -> codemp.Client:
if not self.is_init():
self.get_or_init()
self._client = codemp.connect(config).wait()
self.config = config
logger.debug(f"Connected to '{self.config.host}' as user {self._client.current_user().name} (id: {self._client.current_user().id})")
return self._client
def get_workspaces(self, owned: bool = True, invited: bool = True):
owned_wss = self.client.fetch_owned_workspaces().wait() if owned else []
invited_wss = self.client.fetch_joined_workspaces().wait() if invited else []
return owned_wss + invited_wss
def drop_client(self):
self._client = None
session = SessionManager()

146
plugin/core/workspace.py Normal file
View file

@ -0,0 +1,146 @@
from __future__ import annotations
from typing import Optional, Tuple
from typing import TYPE_CHECKING
if TYPE_CHECKING:
import codemp
import sublime
import shutil
import tempfile
import logging
import gc
from codemp import Selection
from .. import globals as g
from ..utils import draw_cursor_region
from ..utils import bidict
from .buffers import buffers
logger = logging.getLogger(__name__)
def add_project_folder(w: sublime.Window, folder: str, name: str = ""):
proj = w.project_data()
if not isinstance(proj, dict):
proj = {"folders": []}
if name == "":
entry = {"path": folder}
else:
entry = {"name": name, "path": folder}
proj["folders"].append(entry)
w.set_project_data(proj)
def remove_project_folder(w: sublime.Window, filterstr: str):
proj: dict = w.project_data() # type:ignore
if proj is None:
return
clean_proj_folders = list(
filter(
lambda f: f.get("name", "") != filterstr,
proj["folders"],
)
)
proj["folders"] = clean_proj_folders
w.set_project_data(proj)
def cursor_callback(ctl: codemp.CursorController):
def _():
while event := ctl.try_recv().wait():
if event is None: break
try: bfm = buffers.lookupId(event.sel.buffer)
except KeyError: continue
region_start = (event.sel.start_row, event.sel.start_col)
region_end = (event.sel.end_row, event.sel.end_col)
draw_cursor_region(bfm.view, region_start, region_end, event.user)
sublime.set_timeout_async(_)
class WorkspaceManager():
def __init__(self, handle: codemp.Workspace, window: sublime.Window, rootdir: str) -> None:
self.handle: codemp.Workspace = handle
self.window: sublime.Window = window
self.curctl: codemp.CursorController = self.handle.cursor()
self.rootdir: str = rootdir
self.id: str = self.handle.id()
self.curctl.callback(cursor_callback)
def __del__(self):
logger.debug(f"dropping workspace {self.id}")
self.curctl.clear_callback()
for buff in self.handle.active_buffers():
if not self.handle.detach_buffer(buff):
logger.warning(
f"could not detach from '{buff}' for workspace '{self.id}'."
)
for bfm in buffers.lookup(self):
buffers.remove(bfm)
def send_cursor(self, id: str, start: Tuple[int, int], end: Tuple[int, int]):
# we can safely ignore the promise, we don't really care if everything
# is ok for now with the cursor.
sel = Selection(
start_row=start[0],
start_col=start[1],
end_row=end[0],
end_col=end[1],
buffer=id
)
self.curctl.send(sel)
class WorkspaceRegistry():
def __init__(self) -> None:
self._workspaces: bidict[WorkspaceManager, sublime.Window] = bidict()
def lookup(self, w: Optional[sublime.Window] = None) -> list[WorkspaceManager]:
if not w:
return list(self._workspaces.keys())
ws = self._workspaces.inverse.get(w)
return ws if ws else []
def lookupParent(self, ws: WorkspaceManager | str) -> sublime.Window:
if isinstance(ws, str):
wsm = self.lookupId(ws)
return self._workspaces[ws]
def lookupId(self, wid: str) -> WorkspaceManager:
wsm = next((ws for ws in self._workspaces if ws.id == wid), None)
if not wsm: raise KeyError
return wsm
def add(self, wshandle: codemp.Workspace) -> WorkspaceManager:
win = sublime.active_window()
# tmpdir = tempfile.mkdtemp(prefix="codemp_")
# add_project_folder(win, tmpdir, f"{g.WORKSPACE_FOLDER_PREFIX}{wshandle.id()}")
tmpdir = "DISABLED"
wm = WorkspaceManager(wshandle, win, tmpdir)
self._workspaces[wm] = win
return wm
def remove(self, ws: WorkspaceManager | str):
if isinstance(ws, str):
ws = self.lookupId(ws)
# remove_project_folder(ws.window, f"{g.WORKSPACE_FOLDER_PREFIX}{ws.id}")
# shutil.rmtree(ws.rootdir, ignore_errors=True)
del self._workspaces[ws]
workspaces = WorkspaceRegistry()

8
src/globals.py → plugin/globals.py
View file

@ -1,16 +1,14 @@
BUFFCTL_TASK_PREFIX = "buffer-ctl"
CURCTL_TASK_PREFIX = "cursor-ctl"
CODEMP_BUFFER_TAG = "codemp-buffer"
CODEMP_REMOTE_ID = "codemp-buffer-id"
CODEMP_VIEW_TAG = "codemp-buffer"
CODEMP_BUFFER_ID = "codemp-buffer-id"
CODEMP_WORKSPACE_ID = "codemp-workspace-id"
CODEMP_WINDOW_TAG = "codemp-window"
CODEMP_WINDOW_WORKSPACES = "codemp-workspaces"
WORKSPACE_FOLDER_PREFIX = "CODEMP::"
SUBLIME_REGIONS_PREFIX = "codemp-cursors"
SUBLIME_STATUS_ID = "z_codemp_buffer"
CODEMP_IGNORE_NEXT_TEXT_CHANGE = "codemp-skip-change-event"
ACTIVE_CODEMP_VIEW = None

154
plugin/input_handlers.py Normal file
View file

@ -0,0 +1,154 @@
import sublime_plugin
import logging
from typing import Tuple, Union, List
# Input handlers
############################################################
class SimpleTextInput(sublime_plugin.TextInputHandler):
def __init__(self, *args: Tuple[str, Union[str, List[str]]]):
self.input, *self.next_inputs = args
self.argname = self.input[0]
self.default = self.input[1]
def initial_text(self):
if isinstance(self.default, str):
return self.default
else:
return ""
def name(self):
return self.argname
def next_input(self, args):
if len(self.next_inputs) > 0:
if self.next_inputs[0][0] not in args:
if isinstance(self.next_inputs[0][1], list):
return SimpleListInput(*self.next_inputs)
else:
return SimpleTextInput(*self.next_inputs)
class SimpleListInput(sublime_plugin.ListInputHandler):
def __init__(self, *args: Tuple[str, Union["list[str]", str]]):
self.input, *self.next_inputs = args
self.argname = self.input[0]
self.list = self.input[1]
def name(self):
return self.argname
def list_items(self):
if isinstance(self.list, list):
return self.list
else:
return [self.list]
def next_input(self, args):
if len(self.next_inputs) > 0:
if self.next_inputs[0][0] not in args:
if isinstance(self.next_inputs[0][1], str):
return SimpleTextInput(*self.next_inputs)
else:
return SimpleListInput(*self.next_inputs)
# class ActiveWorkspacesIdList(sublime_plugin.ListInputHandler):
# def __init__(self, window=None, buffer_list=False, buffer_text=False):
# self.window = window
# self.buffer_list = buffer_list
# self.buffer_text = buffer_text
# def name(self):
# return "workspace_id"
# def list_items(self):
# return [vws.id for vws in client.all_workspaces(self.window)]
# def next_input(self, args):
# if self.buffer_list:
# return BufferIdList(args["workspace_id"])
# elif self.buffer_text:
# return SimpleTextInput(("buffer_id", "new buffer"))
# # To allow for having a selection and choosing non existing workspaces
# # we do a little dance: We pass this list input handler to a TextInputHandler
# # when we select "Create New..." which adds his result to the list of possible
# # workspaces and pop itself off the stack to go back to the list handler.
# class WorkspaceIdList(sublime_plugin.ListInputHandler):
# def __init__(self):
# assert client.codemp is not None # the command should not be available
# # at the moment, the client can't give us a full list of existing workspaces
# # so a textinputhandler would be more appropriate. but we keep this for the future
# self.add_entry_text = "* add entry..."
# self.list = client.codemp.list_workspaces(True, True).wait()
# self.list.sort()
# self.list.append(self.add_entry_text)
# self.preselected = None
# def name(self):
# return "workspace_id"
# def placeholder(self):
# return "Workspace"
# def list_items(self):
# if self.preselected is not None:
# return (self.list, self.preselected)
# else:
# return self.list
# def next_input(self, args):
# if args["workspace_id"] == self.add_entry_text:
# return AddListEntry(self)
# class BufferIdList(sublime_plugin.ListInputHandler):
# def __init__(self, workspace_id):
# vws = client.workspace_from_id(workspace_id)
# self.add_entry_text = "* create new..."
# self.list = vws.codemp.filetree(None)
# self.list.sort()
# self.list.append(self.add_entry_text)
# self.preselected = None
# def name(self):
# return "buffer_id"
# def placeholder(self):
# return "Buffer Id"
# def list_items(self):
# if self.preselected is not None:
# return (self.list, self.preselected)
# else:
# return self.list
# def next_input(self, args):
# if args["buffer_id"] == self.add_entry_text:
# return AddListEntry(self)
# class AddListEntry(sublime_plugin.TextInputHandler):
# # this class works when the list input handler
# # added appended a new element to it's list that will need to be
# # replaced with the entry added from here!
# def __init__(self, list_input_handler):
# self.parent = list_input_handler
# def name(self):
# return ""
# def validate(self, text: str) -> bool:
# return not len(text) == 0
# def confirm(self, text: str):
# self.parent.list.pop() # removes the add_entry_text
# self.parent.list.insert(0, text)
# self.parent.preselected = 0
# def next_input(self, args):
# return sublime_plugin.BackInputHandler()

24
plugin/quickpanel/qp_globals.py Normal file
View file

@ -0,0 +1,24 @@
import sublime
QP_COLOR_NONE = sublime.KIND_ID_AMBIGUOUS
QP_COLOR_REDISH = sublime.KIND_ID_COLOR_REDISH
QP_COLOR_ORANGISH = sublime.KIND_ID_COLOR_ORANGISH
QP_COLOR_YELLOWISH = sublime.KIND_ID_COLOR_YELLOWISH
QP_COLOR_GREENISH = sublime.KIND_ID_COLOR_GREENISH
QP_COLOR_CYANISH = sublime.KIND_ID_COLOR_CYANISH
QP_COLOR_BLUISH = sublime.KIND_ID_COLOR_BLUISH
QP_COLOR_PURPLISH = sublime.KIND_ID_COLOR_PURPLISH
QP_COLOR_PINKISH = sublime.KIND_ID_COLOR_PINKISH
QP_COLOR_DARK = sublime.KIND_ID_COLOR_DARK
QP_COLOR_LIGHT = sublime.KIND_ID_COLOR_LIGHT
QP_YES = "✓"
QP_NO = "✗"
QP_ADD = "+"
QP_FORWARD = "❯"
QP_BACK = "❮"
QP_DETAILS = "⋯"
QP_RENAME = "*"
QP_CHMOD = "7"
QP_DOWNLOAD = "⏬"
QP_EDIT = "a"

221
plugin/quickpanel/qpbrowser.py Normal file
View file

@ -0,0 +1,221 @@
import sublime
import logging
from . import qp_globals as qpg
from ..core.workspace import workspaces
from ..core.buffers import buffers
logger = logging.getLogger(__name__)
def qpi(text, details="", color=qpg.QP_COLOR_NONE, letter="", name="", hint="", prefix=""):
return sublime.QuickPanelItem(text, details, annotation=hint, kind=(color, letter, name))
def show_qp(window, choices, on_done, placeholder=''):
def _():
flags = sublime.KEEP_OPEN_ON_FOCUS_LOST
window.show_quick_panel(choices, on_done, flags, placeholder=placeholder)
sublime.set_timeout(_, 10)
class QPServerBrowser():
def __init__(self, window, host, raw_input_items):
self.window = window
self.host = host
self.raw_input_items = raw_input_items
def make_entry(self, wsid):
return qpi(wsid, letter="w", color=qpg.QP_COLOR_BLUISH, hint="Workspace", prefix=" ")
def qp_placeholder(self):
return f"Browsing workspaces on host: {self.host}"
def run(self):
self.current_wid_selection = None
self.entries = []
for item in self.raw_input_items:
self.entries.append(self.make_entry(item))
self.entries.insert(0, qpi("Server Actions",
color=qpg.QP_COLOR_CYANISH,
letter=qpg.QP_DETAILS,
hint="Submenu",
prefix=" • "))
show_qp(self.window, self.entries, self.server_actions, self.qp_placeholder())
def server_actions(self, index):
if index == -1:
return
elif index == 0:
self.edit_server()
return
wid = self.entries[index].trigger
self.current_wid_selection = wid
# self.select_workspace()
def _():
self.window.run_command(
"codemp_join_workspace",
{"workspace_id": self.current_wid_selection})
ws = workspaces.lookupId(wid)
buffers = ws.handle.fetch_buffers()
QPWorkspaceBrowser(self.window, wid, buffers.wait()).run()
sublime.set_timeout(_)
logger.debug("exiting the server_broswer.")
def select_workspace(self):
assert self.current_wid_selection
actions = [
qpi("Join", details=self.current_wid_selection, color=qpg.QP_COLOR_BLUISH, letter=qpg.QP_FORWARD),
# qpi("Join and open all",
# details="opens all buffer in the workspace",
# color=qpg.QP_COLOR_PINKISH, letter=qpg.QP_DETAILS),
qpi("Back", color=qpg.QP_COLOR_BLUISH, letter=qpg.QP_BACK)
]
show_qp(self.window, actions, self.select_workspace_actions, self.qp_placeholder())
def select_workspace_actions(self, index):
if index == -1:
return
elif index == 0:
self.window.run_command(
"codemp_join_workspace",
{"workspace_id": self.current_wid_selection})
elif index == 1:
self.run()
def edit_server(self):
actions = [
qpi("Back", color=qpg.QP_COLOR_CYANISH, letter=qpg.QP_BACK),
qpi("New Workspace", color=qpg.QP_COLOR_GREENISH, letter=qpg.QP_ADD),
qpi("Delete Workspace", color=qpg.QP_COLOR_REDISH, letter=qpg.QP_NO)
]
show_qp(self.window, actions, self.edit_server_actions, self.qp_placeholder())
def edit_server_actions(self, index):
if index == -1:
return
if index == 0:
self.run()
if index == 1:
def create_workspace(name):
self.window.run_command(
"codemp_create_workspace",
{"workspace_id": name})
self.window.show_input_panel("New Workspace Name", "", create_workspace, None, self.edit_server)
if index == 2:
def delete_workspace(index):
if index == -1 or index == 0:
self.edit_server()
# we must be careful here. here with index 1 we are selecting the correct
# workspace, because the index zero in the entries is the workspace action submenu.
# which is occupied by the back action.
# if we add extra non workspace entries, then we must shift the index accordingly.
# Do this differently?
selected = self.entries[index]
self.window.run_command(
"codemp_delete_workspace",
{"workspace_id": selected.trigger})
show_qp(self.window, self.entries, delete_workspace, self.qp_placeholder())
class QPWorkspaceBrowser():
def __init__(self, window, workspace_id, raw_input_items):
self.window = window
self.workspace_id = workspace_id
self.raw_input_items = raw_input_items
def qp_placeholder(self):
return f"Browsing buffers in {self.workspace_id}"
def make_entry(self, item):
return qpi(item, letter="b", color=qpg.QP_COLOR_BLUISH, hint="Buffer", prefix=" ")
def run(self):
self.entries = []
for buffer in self.raw_input_items:
self.entries.append(self.make_entry(buffer))
self.entries.insert(0, qpi("Workspace Actions",
color=qpg.QP_COLOR_CYANISH,
letter=qpg.QP_DETAILS,
hint="Submenu",
prefix=" • "))
show_qp(self.window, self.entries, self.workspace_actions, self.qp_placeholder())
def workspace_actions(self, index):
if index == -1:
return
elif index == 0:
self.edit_workspace()
return
bid = self.entries[index].trigger
self.window.run_command(
"codemp_join_buffer",
{
"workspace_id": self.workspace_id,
"buffer_id": bid
})
def edit_workspace(self):
actions = [
qpi("Back", color=qpg.QP_COLOR_CYANISH, letter=qpg.QP_BACK),
qpi("Leave Workspace", color=qpg.QP_COLOR_ORANGISH, letter=qpg.QP_BACK),
qpi("Invite User", color=qpg.QP_COLOR_PINKISH, letter=qpg.QP_FORWARD),
qpi("Create Buffer", color=qpg.QP_COLOR_GREENISH, letter=qpg.QP_ADD),
qpi("Delete Buffer", color=qpg.QP_COLOR_REDISH, letter=qpg.QP_NO),
qpi("Rename Buffer", color=qpg.QP_COLOR_ORANGISH, letter=qpg.QP_RENAME),
]
show_qp(self.window, actions, self.edit_workspace_actions, self.qp_placeholder())
def edit_workspace_actions(self, index):
if index == -1 or index == 0:
self.edit_workspace()
elif index == 1:
self.window.run_command(
"codemp_leave_workspace",
{"workspace_id": self.workspace_id})
self.window.run_command(
"codemp_browse_server", {})
elif index == 2:
self.window.run_command(
"codemp_invite_to_workspace",
{"workspace_id": self.workspace_id})
elif index == 3:
def create_buffer(name):
self.window.run_command(
"codemp_create_buffer",
{
"workspace_id": self.workspace_id,
"buffer_id": name
})
self.window.show_input_panel("New Buffer Name", "", create_buffer, None, self.edit_workspace)
elif index == 4:
def delete_buffer(index):
if index == -1 or index == 0:
self.edit_workspace()
# same warning as the server browser. Check your indexed 3 times
selected = self.entries[index]
self.window.run_command(
"codemp_delete_buffer",
{
"workspace_id": self.workspace_id,
"buffer_id": selected.trigger
})
show_qp(self.window, self.entries, delete_buffer, self.qp_placeholder())
elif index == 5:
sublime.message_dialog("renaming is not yet implemented.")
self.edit_workspace()

33
plugin/text_listener.py Normal file
View file

@ -0,0 +1,33 @@
import sublime
import sublime_plugin
import logging
from .core.buffers import buffers
from . import globals as g
logger = logging.getLogger(__name__)
class CodempClientTextChangeListener(sublime_plugin.TextChangeListener):
@classmethod
def is_applicable(cls, _): # pyright: ignore
# don't attach this event listener automatically
# we'll do it by hand with .attach(buffer).
return False
def on_text_changed(self, changes):
s = self.buffer.primary_view().settings()
if s.get(g.CODEMP_IGNORE_NEXT_TEXT_CHANGE, False):
logger.debug("Ignoring echoing back the change.")
s[g.CODEMP_IGNORE_NEXT_TEXT_CHANGE] = False
return
bid = str(s.get(g.CODEMP_BUFFER_ID))
try:
vbuff = buffers.lookupId(bid)
logger.debug(f"local buffer change! {vbuff.id}")
vbuff.send_change(changes)
except KeyError:
logger.error(f"could not find registered buffer with id {bid}")
pass
TEXT_LISTENER = CodempClientTextChangeListener()

12
src/utils.py → plugin/utils.py
View file

@ -1,6 +1,6 @@
import sublime
import sublime_plugin
from typing import Dict, Generic, TypeVar
from typing import Dict, Generic, TypeVar, Optional
from . import globals as g
# bidirectional dictionary so that we can have bidirectional
@ -13,7 +13,6 @@ D = TypeVar("D", Dict, dict)
K = TypeVar("K")
V = TypeVar("V")
# using del bd.inverse[key] doesn't work since it can't be intercepted.
# the only way is to iterate:
# for key in bd.inverse[inverse_key]
@ -89,12 +88,12 @@ def populate_view(view, content):
"start": 0,
"end": view.size(),
"content": content,
"change_id": view.change_id(),
"change_id": None,
},
)
def get_view_from_local_path(path):
def view_from_local_path(path):
for window in sublime.windows():
for view in window.views():
if view.file_name() == path:
@ -115,3 +114,8 @@ def draw_cursor_region(view, start, end, user):
annotations=[user], # pyright: ignore
annotation_color=g.PALETTE[user_hash % len(g.PALETTE)],
)
T = TypeVar("T")
def some(x: Optional[T]) -> T:
assert x is not None
return x

119
src/buffers.py
View file

@ -1,119 +0,0 @@
from __future__ import annotations
import sublime
import os
import logging
from . import globals as g
from .utils import populate_view, safe_listener_attach, safe_listener_detach
import codemp
logger = logging.getLogger(__name__)
def make_bufferchange_cb(buff: VirtualBuffer):
def __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 __callback
class VirtualBuffer:
def __init__(
self,
buffctl: codemp.BufferController,
view: sublime.View,
rootdir: str,
):
self.buffctl = buffctl
self.view = view
self.id = self.buffctl.path()
self.tmpfile = os.path.join(rootdir, self.id)
open(self.tmpfile, "a").close()
self.view.set_scratch(True)
self.view.set_name(self.id)
self.view.retarget(self.tmpfile)
self.view.settings().set(g.CODEMP_BUFFER_TAG, True)
self.view.set_status(g.SUBLIME_STATUS_ID, "[Codemp]")
logger.info(f"registering a callback for buffer: {self.id}")
self.buffctl.callback(make_bufferchange_cb(self))
self.isactive = True
def __del__(self):
logger.debug("__del__ buffer called.")
def __hash__(self) -> int:
return hash(self.id)
def uninstall(self):
logger.info(f"clearing a callback for buffer: {self.id}")
self.buffctl.clear_callback()
self.buffctl.stop()
self.isactive = False
os.remove(self.tmpfile)
def onclose(did_close):
if did_close:
logger.info(f"'{self.id}' closed successfully")
else:
logger.info(f"failed to close the view for '{self.id}'")
self.view.close(onclose)
def sync(self, text_listener):
promise = self.buffctl.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(_)
def send_buffer_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.buffctl.send(region.begin(), region.end(), change.str).wait()

156
src/client.py
View file

@ -1,156 +0,0 @@
from __future__ import annotations
from typing import Optional
import sublime
import logging
import codemp
from .workspace import VirtualWorkspace
from .buffers import VirtualBuffer
from .utils import bidict
logger = logging.getLogger(__name__)
# the client will be responsible to keep track of everything!
# it will need 3 bidirectional dictionaries and 2 normal ones
# normal: workspace_id -> VirtualWorkspaces
# normal: buffer_id -> VirtualBuffer
# bidir: VirtualBuffer <-> VirtualWorkspace
# bidir: VirtualBuffer <-> Sublime.View
# bidir: VirtualWorkspace <-> Sublime.Window
class VirtualClient:
def __init__(self):
self.codemp: Optional[codemp.Client] = None
self.driver: Optional[codemp.Driver] = None
# bookkeeping corner
self._id2buffer: dict[str, VirtualBuffer] = {}
self._id2workspace: dict[str, VirtualWorkspace] = {}
self._view2buff: dict[sublime.View, VirtualBuffer] = {}
self._buff2workspace: bidict[VirtualBuffer, VirtualWorkspace] = bidict()
self._workspace2window: dict[VirtualWorkspace, sublime.Window] = {}
def all_workspaces(
self, window: Optional[sublime.Window] = None
) -> list[VirtualWorkspace]:
if window is None:
return list(self._workspace2window.keys())
else:
return [
ws
for ws in self._workspace2window
if self._workspace2window[ws] == window
]
def workspace_from_view(self, view: sublime.View) -> Optional[VirtualWorkspace]:
buff = self._view2buff.get(view, None)
return self.workspace_from_buffer(buff) if buff is not None else None
def workspace_from_buffer(self, vbuff: VirtualBuffer) -> Optional[VirtualWorkspace]:
return self._buff2workspace.get(vbuff, None)
def workspace_from_id(self, id: str) -> Optional[VirtualWorkspace]:
return self._id2workspace.get(id)
def all_buffers(
self, workspace: Optional[VirtualWorkspace | str] = None
) -> list[VirtualBuffer]:
if workspace is None:
return list(self._id2buffer.values())
elif isinstance(workspace, str):
workspace = client._id2workspace[workspace]
return self._buff2workspace.inverse.get(workspace, [])
else:
return self._buff2workspace.inverse.get(workspace, [])
def buffer_from_view(self, view: sublime.View) -> Optional[VirtualBuffer]:
return self._view2buff.get(view)
def buffer_from_id(self, id: str) -> Optional[VirtualBuffer]:
return self._id2buffer.get(id)
def view_from_buffer(self, buff: VirtualBuffer) -> sublime.View:
return buff.view
def register_buffer(self, workspace: VirtualWorkspace, buffer: VirtualBuffer):
self._buff2workspace[buffer] = workspace
self._id2buffer[buffer.id] = buffer
self._view2buff[buffer.view] = buffer
def disconnect(self):
if self.codemp is None:
return
logger.info("disconnecting from the current client")
# for each workspace tell it to clean up after itself.
for vws in self.all_workspaces():
self.uninstall_workspace(vws)
self.codemp.leave_workspace(vws.id)
self._id2workspace.clear()
self._id2buffer.clear()
self._buff2workspace.clear()
self._view2buff.clear()
self._workspace2window.clear()
if self.driver is not None:
self.driver.stop()
self.driver = None
self.codemp = None
def connect(self, host: str, user: str, password: str):
if self.codemp is not None:
logger.info("Disconnecting from previous client.")
return self.disconnect()
if self.driver is None:
self.driver = codemp.init()
logger.debug("registering logger callback...")
if not codemp.set_logger(lambda msg: logger.debug(msg), False):
logger.debug(
"could not register the logger... If reconnecting it's ok, the previous logger is still registered"
)
config = codemp.get_default_config()
config.username = user
config.host = host
config.password = password
self.codemp = codemp.connect(config).wait()
id = self.codemp.user_id()
logger.debug(f"Connected to '{host}' as user {user} (id: {id})")
def install_workspace(self, workspace: codemp.Workspace, window: sublime.Window):
vws = VirtualWorkspace(workspace, window)
self._workspace2window[vws] = window
self._id2workspace[vws.id] = vws
def uninstall_workspace(self, vws: VirtualWorkspace):
# we aim at dropping all references to the workspace
# as well as all the buffers associated with it.
# if we did a good job the dunder del method will kick
# and continue with the cleanup.
logger.info(f"Uninstalling workspace '{vws.id}'...")
del self._workspace2window[vws]
del self._id2workspace[vws.id]
for vbuff in self.all_buffers(vws):
self.unregister_buffer(vbuff)
vws.uninstall()
def unregister_buffer(self, buffer: VirtualBuffer):
del self._buff2workspace[buffer]
del self._id2buffer[buffer.id]
del self._view2buff[buffer.view]
def workspaces_in_server(self):
return self.codemp.active_workspaces() if self.codemp else []
def user_id(self):
return self.codemp.user_id() if self.codemp else None
client = VirtualClient()

133
src/workspace.py
View file

@ -1,133 +0,0 @@
from __future__ import annotations
from typing import Optional, Tuple
from ..listeners import CodempClientTextChangeListener
import sublime
import shutil
import tempfile
import logging
import codemp
from . import globals as g
from .buffers import VirtualBuffer
from .utils import draw_cursor_region
logger = logging.getLogger(__name__)
def make_cursor_callback(workspace: VirtualWorkspace):
def _callback(ctl: codemp.CursorController):
def _():
while event := ctl.try_recv().wait():
logger.debug("received remote cursor movement!")
if event is None:
break
vbuff = workspace.buff_by_id(event.buffer)
if vbuff is None:
logger.warning(
f"{workspace.id} received a cursor event for a buffer that wasn't saved internally."
)
continue
draw_cursor_region(vbuff.view, event.start, event.end, event.user)
sublime.set_timeout_async(_)
return _callback
# A virtual workspace is a bridge class that aims to translate
# events that happen to the codemp workspaces into sublime actions
class VirtualWorkspace:
def __init__(self, handle: codemp.Workspace, window: sublime.Window):
self.codemp: codemp.Workspace = handle
self.window: sublime.Window = window
self.curctl: codemp.CursorController = self.codemp.cursor()
self.id: str = self.codemp.id()
self.codemp.fetch_buffers()
self.codemp.fetch_users()
self._id2buff: dict[str, VirtualBuffer] = {}
tmpdir = tempfile.mkdtemp(prefix="codemp_")
self.rootdir = tmpdir
proj: dict = self.window.project_data() # pyright: ignore
if proj is None:
proj = {"folders": []} # pyright: ignore, Value can be None
proj["folders"].append(
{"name": f"{g.WORKSPACE_FOLDER_PREFIX}{self.id}", "path": self.rootdir}
)
self.window.set_project_data(proj)
self.curctl.callback(make_cursor_callback(self))
self.isactive = True
def __del__(self):
logger.debug("workspace destroyed!")
def __hash__(self) -> int:
# so we can use these as dict keys!
return hash(self.id)
def uninstall(self):
self.curctl.clear_callback()
self.isactive = False
self.curctl.stop()
for vbuff in self._id2buff.values():
vbuff.uninstall()
if not self.codemp.detach(vbuff.id):
logger.warning(
f"could not detach from '{vbuff.id}' for workspace '{self.id}'."
)
self._id2buff.clear()
proj: dict = self.window.project_data() # type:ignore
if proj is None:
raise
clean_proj_folders = list(
filter(
lambda f: f.get("name", "") != f"{g.WORKSPACE_FOLDER_PREFIX}{self.id}",
proj["folders"],
)
)
proj["folders"] = clean_proj_folders
self.window.set_project_data(proj)
logger.info(f"cleaning up virtual workspace '{self.id}'")
shutil.rmtree(self.rootdir, ignore_errors=True)
def all_buffers(self) -> list[VirtualBuffer]:
return list(self._id2buff.values())
def buff_by_id(self, id: str) -> Optional[VirtualBuffer]:
return self._id2buff.get(id)
def install_buffer(
self, buff: codemp.BufferController, listener: CodempClientTextChangeListener
) -> VirtualBuffer:
logger.debug(f"installing buffer {buff.path()}")
view = self.window.new_file()
vbuff = VirtualBuffer(buff, view, self.rootdir)
self._id2buff[vbuff.id] = vbuff
vbuff.sync(listener)
return vbuff
def uninstall_buffer(self, vbuff: VirtualBuffer):
del self._id2buff[vbuff.id]
self.codemp.detach(vbuff.id)
vbuff.uninstall()
def send_cursor(self, id: str, start: Tuple[int, int], end: Tuple[int, int]):
# we can safely ignore the promise, we don't really care if everything
# is ok for now with the cursor.
self.curctl.send(id, start, end)

208
workspace_commands.py
View file

@ -1,208 +0,0 @@
import sublime
import sublime_plugin
import logging
from .src.client import client
from listeners import TEXT_LISTENER
from input_handlers import SimpleTextInput
from input_handlers import ActiveWorkspacesIdList
from input_handlers import BufferIdList
logger = logging.getLogger(__name__)
# Join Buffer Command
class CodempJoinBufferCommand(sublime_plugin.WindowCommand):
def is_enabled(self):
available_workspaces = client.all_workspaces(self.window)
return len(available_workspaces) > 0
def run(self, workspace_id, buffer_id): # pyright: ignore[reportIncompatibleMethodOverride]
# A workspace has some Buffers inside of it (filetree)
# some of those you are already attached to
# If already attached to it return the same alredy existing bufferctl
# if existing but not attached (attach)
# if not existing ask for creation (create + attach)
vws = client.workspace_from_id(workspace_id)
assert vws is not None
# is the buffer already installed?
if buffer_id in vws.codemp.buffer_list():
logger.info("buffer already installed!")
return # do nothing.
if buffer_id not in vws.codemp.filetree(filter=buffer_id):
create = sublime.ok_cancel_dialog(
"There is no buffer named '{buffer_id}' in the workspace '{workspace_id}'.\n\
Do you want to create it?",
ok_title="yes",
title="Create Buffer?",
)
if create:
try:
create_promise = vws.codemp.create(buffer_id)
except Exception as e:
logging.error(f"could not create buffer:\n\n {e}")
return
create_promise.wait()
# now we can defer the attaching process
logger.debug(f"attempting to attach to {buffer_id}...")
promise = vws.codemp.attach(buffer_id)
def deferred_attach(promise):
try:
buff_ctl = promise.wait()
logger.debug("attach successfull!")
except Exception as e:
logging.error(f"error when attaching to buffer '{id}':\n\n {e}")
sublime.error_message(f"Could not attach to buffer '{buffer_id}'")
return
vbuff = vws.install_buffer(buff_ctl, TEXT_LISTENER)
client.register_buffer(vws, vbuff) # we need to keep track of it.
# TODO! if the view is already active calling focus_view()
# will not trigger the on_activate
self.window.focus_view(vbuff.view)
sublime.set_timeout_async(lambda: deferred_attach(promise))
def input_description(self) -> str:
return "Attach: "
def input(self, args):
if "workspace_id" not in args:
return ActiveWorkspacesIdList(self.window, buffer_list=True)
if "buffer_id" not in args:
return BufferIdList(args["workspace_id"])
# Leave Buffer Comand
class CodempLeaveBufferCommand(sublime_plugin.WindowCommand):
def is_enabled(self):
return len(client.all_buffers()) > 0
def run(self, workspace_id, buffer_id): # pyright: ignore[reportIncompatibleMethodOverride]
vbuff = client.buffer_from_id(buffer_id)
vws = client.workspace_from_id(workspace_id)
if vbuff is None or vws is None:
sublime.error_message(f"You are not attached to the buffer '{id}'")
logging.warning(f"You are not attached to the buffer '{id}'")
return
def defer_detach():
if vws.codemp.detach(buffer_id):
vws.uninstall_buffer(vbuff)
client.unregister_buffer(vbuff)
sublime.set_timeout_async(defer_detach)
def input_description(self) -> str:
return "Leave: "
def input(self, args):
if "workspace_id" not in args:
return ActiveWorkspacesIdList(self.window, buffer_list=True)
if "buffer_id" not in args:
return BufferIdList(args["workspace_id"])
# Leave Buffer Comand
class CodempCreateBufferCommand(sublime_plugin.WindowCommand):
def is_enabled(self):
return len(client.all_workspaces(self.window)) > 0
def run(self, workspace_id, buffer_id):# pyright: ignore[reportIncompatibleMethodOverride]
vws = client.workspace_from_id(workspace_id)
if vws is None:
sublime.error_message(
f"You are not attached to the workspace '{workspace_id}'"
)
logging.warning(f"You are not attached to the workspace '{workspace_id}'")
return
vws.codemp.create(buffer_id)
logging.info(
"created buffer '{buffer_id}' in the workspace '{workspace_id}'.\n\
To interact with it you need to attach to it with Codemp: Attach."
)
def input_description(self) -> str:
return "Create Buffer: "
def input(self, args):
if "workspace_id" not in args:
return ActiveWorkspacesIdList(self.window, buffer_text=True)
if "buffer_id" not in args:
return SimpleTextInput(
(("buffer_id", "new buffer")),
)
class CodempDeleteBufferCommand(sublime_plugin.WindowCommand):
def is_enabled(self):
return client.codemp is not None and len(client.codemp.active_workspaces()) > 0
def run(self, workspace_id, buffer_id):# pyright: ignore[reportIncompatibleMethodOverride]
vws = client.workspace_from_id(workspace_id)
if vws is None:
sublime.error_message(
f"You are not attached to the workspace '{workspace_id}'"
)
logging.warning(f"You are not attached to the workspace '{workspace_id}'")
return
fetch_promise = vws.codemp.fetch_buffers()
delete = sublime.ok_cancel_dialog(
f"Confirm you want to delete the buffer '{buffer_id}'",
ok_title="delete",
title="Delete Buffer?",
)
if not delete:
return
fetch_promise.wait()
existing = vws.codemp.filetree(buffer_id)
if len(existing) == 0:
sublime.error_message(
f"The buffer '{buffer_id}' does not exists in the workspace."
)
logging.info(f"The buffer '{buffer_id}' does not exists in the workspace.")
return
def deferred_delete():
try:
vws.codemp.delete(buffer_id).wait()
except Exception as e:
logging.error(
f"error when deleting the buffer '{buffer_id}':\n\n {e}", True
)
return
vbuff = client.buffer_from_id(buffer_id)
if vbuff is None:
# we are not attached to it!
sublime.set_timeout_async(deferred_delete)
else:
if vws.codemp.detach(buffer_id):
vws.uninstall_buffer(vbuff)
sublime.set_timeout_async(deferred_delete)
else:
logging.error(
f"error while detaching from buffer '{buffer_id}', aborting the delete."
)
return
def input_description(self) -> str:
return "Delete buffer: "
def input(self, args):
if "workspace_id" not in args:
return ActiveWorkspacesIdList(self.window, buffer_list=True)
if "buffer_id" not in args:
return BufferIdList(args["workspace_id"])