Make the single instance data type checked

This commit is contained in:
Kovid Goyal 2023-01-23 17:45:11 +05:30
parent 60791bb57b
commit d8bbb16d5e
No known key found for this signature in database
GPG Key ID: 06BC317B515ACE7C
3 changed files with 23 additions and 11 deletions

View File

@ -122,7 +122,7 @@ from .prewarm import PrewarmProcess
from .rgb import color_from_int from .rgb import color_from_int
from .session import Session, create_sessions, get_os_window_sizing_data from .session import Session, create_sessions, get_os_window_sizing_data
from .tabs import SpecialWindow, SpecialWindowInstance, Tab, TabDict, TabManager from .tabs import SpecialWindow, SpecialWindowInstance, Tab, TabDict, TabManager
from .types import _T, AsyncResponse, WindowSystemMouseEvent, ac from .types import _T, AsyncResponse, SingleInstanceData, WindowSystemMouseEvent, ac
from .typing import PopenType, TypedDict from .typing import PopenType, TypedDict
from .utils import ( from .utils import (
cleanup_ssh_control_masters, cleanup_ssh_control_masters,
@ -748,12 +748,12 @@ class Boss:
from kitty.remote_control import encode_response_for_peer from kitty.remote_control import encode_response_for_peer
return encode_response_for_peer(response) return encode_response_for_peer(response)
data = json.loads(msg_bytes.decode('utf-8')) data:SingleInstanceData = json.loads(msg_bytes.decode('utf-8'))
if isinstance(data, dict) and data.get('cmd') == 'new_instance': if isinstance(data, dict) and data.get('cmd') == 'new_instance':
from .cli_stub import CLIOptions from .cli_stub import CLIOptions
startup_id = data.get('startup_id') startup_id = data['environ'].get('DESKTOP_STARTUP_ID', '')
activation_token = data.get('activation_token', '') activation_token = data['environ'].get('XDG_ACTIVATION_TOKEN', '')
args, rest = parse_args(data['args'][1:], result_class=CLIOptions) args, rest = parse_args(list(data['args'][1:]), result_class=CLIOptions)
cmdline_args_for_open = data.get('cmdline_args_for_open') cmdline_args_for_open = data.get('cmdline_args_for_open')
if cmdline_args_for_open: if cmdline_args_for_open:
self.launch_urls(*cmdline_args_for_open, no_replace_window=True) self.launch_urls(*cmdline_args_for_open, no_replace_window=True)
@ -779,9 +779,9 @@ class Boss:
if data.get('notify_on_os_window_death'): if data.get('notify_on_os_window_death'):
self.os_window_death_actions[os_window_id] = partial(self.notify_on_os_window_death, data['notify_on_os_window_death']) self.os_window_death_actions[os_window_id] = partial(self.notify_on_os_window_death, data['notify_on_os_window_death'])
if focused_os_window > 0: if focused_os_window > 0:
focus_os_window(focused_os_window, True, activation_token or '') focus_os_window(focused_os_window, True, activation_token)
elif activation_token and is_wayland() and os_window_id: elif activation_token and is_wayland() and os_window_id:
focus_os_window(os_window_id, True, activation_token or '') focus_os_window(os_window_id, True, activation_token)
else: else:
log_error('Unknown message received from peer, ignoring') log_error('Unknown message received from peer, ignoring')
return None return None

View File

@ -50,6 +50,7 @@ from .options.utils import DELETE_ENV_VAR
from .os_window_size import initial_window_size_func from .os_window_size import initial_window_size_func
from .prewarm import PrewarmProcess, fork_prewarm_process from .prewarm import PrewarmProcess, fork_prewarm_process
from .session import create_sessions, get_os_window_sizing_data from .session import create_sessions, get_os_window_sizing_data
from .types import SingleInstanceData
from .utils import ( from .utils import (
cleanup_ssh_control_masters, cleanup_ssh_control_masters,
detach, detach,
@ -87,9 +88,10 @@ def talk_to_instance(args: CLIOptions) -> None:
with open(args.session) as f: with open(args.session) as f:
session_data = f.read() session_data = f.read()
data = {'cmd': 'new_instance', 'args': tuple(sys.argv), 'cmdline_args_for_open': getattr(sys, 'cmdline_args_for_open', []), data: SingleInstanceData = {
'startup_id': os.environ.get('DESKTOP_STARTUP_ID'), 'activation_token': os.environ.get('XDG_ACTIVATION_TOKEN'), 'cmd': 'new_instance', 'args': tuple(sys.argv), 'cmdline_args_for_open': getattr(sys, 'cmdline_args_for_open', ()),
'cwd': os.getcwd(), 'session_data': session_data, 'environ': dict(os.environ)} 'cwd': os.getcwd(), 'session_data': session_data, 'environ': dict(os.environ), 'notify_on_os_window_death': None
}
notify_socket = None notify_socket = None
if args.wait_for_single_instance_window_close: if args.wait_for_single_instance_window_close:
address = f'\0{appname}-os-window-close-notify-{os.getpid()}-{os.geteuid()}' address = f'\0{appname}-os-window-close-notify-{os.getpid()}-{os.geteuid()}'

View File

@ -3,7 +3,7 @@
from enum import Enum from enum import Enum
from functools import update_wrapper from functools import update_wrapper
from typing import TYPE_CHECKING, Any, Callable, Dict, Generic, Iterator, NamedTuple, Tuple, TypeVar, Union from typing import TYPE_CHECKING, Any, Callable, Dict, Generic, Iterator, Mapping, NamedTuple, Optional, Sequence, Tuple, TypedDict, TypeVar, Union
if TYPE_CHECKING: if TYPE_CHECKING:
from kitty.fast_data_types import SingleKey from kitty.fast_data_types import SingleKey
@ -11,6 +11,16 @@ if TYPE_CHECKING:
_T = TypeVar('_T') _T = TypeVar('_T')
class SingleInstanceData(TypedDict):
cmd: str
args: Sequence[str]
cmdline_args_for_open: Sequence[str]
cwd: str
session_data: str
environ: Mapping[str, str]
notify_on_os_window_death: Optional[str]
class OverlayType(Enum): class OverlayType(Enum):
transient: str = 'transient' transient: str = 'transient'
main: str = 'main' main: str = 'main'