sankalp/Modus
1
0 Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

feat(src/utils): centralize SESSION bus and shrink DBusMenuItem with __slots__

Browse Source 
Prevent duplicate bus connections and cut per-instance memory usage.
This commit is contained in:
sankalp committed 2026-06-23 14:11:07 +05:45
1 parent d89b392a0e
commit 01fed4c52f
2 files changed
+206 -160
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/utils/dbusmenu.py
+86 -82
View File
@@ -7,50 +7,94 @@ Supports:
"""
import threading
from dataclasses import dataclass, field
from typing import List, Optional
from fabric.utils import Any, Gio, GLib, logger
# SHARED BUS — one connection for the entire process
_SESSION_BUS: Optional[Gio.DBusConnection] = None
_BUS_LOCK = threading.Lock()
@dataclass
def _get_bus() -> Gio.DBusConnection:
global _SESSION_BUS
if _SESSION_BUS is None:
with _BUS_LOCK:
if _SESSION_BUS is None:
_SESSION_BUS = Gio.bus_get_sync(Gio.BusType.SESSION, None)
return _SESSION_BUS
# DATA MODEL — __slots__ cuts ~200 bytes per instance
class DBusMenuItem:
id: int
label: str = ""
enabled: bool = True
visible: bool = True
type: str = "standard"
shortcut: str = ""
icon_name: str = ""
has_submenu: bool = False
children: List["DBusMenuItem"] = field(default_factory=list)
__slots__ = (
"id",
"label",
"enabled",
"visible",
"type",
"shortcut",
"icon_name",
"has_submenu",
"children",
"action_name",
"action_target",
)
# GtkMenu specific fields
action_name: str = ""
action_target: Optional[Any] = None # FIX #1: any → Any
def __init__(
self,
id: int,
label: str = "",
enabled: bool = True,
visible: bool = True,
type: str = "standard",
shortcut: str = "",
icon_name: str = "",
has_submenu: bool = False,
children: Optional[List["DBusMenuItem"]] = None,
action_name: str = "",
action_target: Optional[Any] = None,
):
self.id = id
self.label = label
self.enabled = enabled
self.visible = visible
self.type = type
self.shortcut = shortcut
self.icon_name = icon_name
self.has_submenu = has_submenu
self.children: List["DBusMenuItem"] = children if children is not None else []
self.action_name = action_name
self.action_target = action_target
# MAIN CLIENT
class DBusMenuClient:
__slots__ = (
"service_name",
"object_path",
"_cache",
"_cache_valid",
"_hash_cache",
"_lock",
"_fetching",
)
def __init__(self, service_name: str, object_path: str):
self.service_name = service_name
self.object_path = object_path
self.bus = Gio.bus_get_sync(Gio.BusType.SESSION, None)
self._pid: Optional[int] = None
self._cache: Optional[List[DBusMenuItem]] = None
self._cache_valid = False
self._hash_cache: Optional[str] = None
# concurrency guard
self._lock = threading.Lock()
self._fetching = False
def set_pid(self, pid: int):
self._pid = pid
# DBUS CALL — uses shared session bus
def _call(self, method, params, reply_type=None):
try:
return self.bus.call_sync(
return _get_bus().call_sync(
self.service_name,
self.object_path,
"com.canonical.dbusmenu",
@@ -64,6 +108,7 @@ class DBusMenuClient:
except Exception:
return None
# PUBLIC ENTRY
def get_layout(self, force_refresh=False):
if self._cache and self._cache_valid and not force_refresh:
return self._cache
@@ -79,6 +124,7 @@ class DBusMenuClient:
with self._lock:
self._fetching = False
# CORE FETCH LOGIC
def _fetch(self):
params = GLib.Variant("(iias)", (0, -1, []))
res = self._call("GetLayout", params)
@@ -94,13 +140,11 @@ class DBusMenuClient:
try:
layout = res.get_child_value(1)
while layout.is_of_type(GLib.VariantType("v")):
layout = layout.get_variant()
parsed = self._parse_dbusmenu(layout).children
# diff detection — FIX #5: use fast string hash instead of MD5
h = self._hash(parsed)
if h == self._hash_cache:
return self._cache or parsed
@@ -108,15 +152,15 @@ class DBusMenuClient:
self._hash_cache = h
self._cache = parsed
self._cache_valid = True
return parsed
except Exception as e:
logger.error(f"[Menu] parse error: {e}")
return []
def _hash(self, items) -> str:
flat = []
# HASH DIFF SYSTEM
def _hash(self, items):
flat: List[str] = []
def walk(nodes):
for n in nodes:
@@ -126,7 +170,7 @@ class DBusMenuClient:
walk(items)
return str(hash("".join(flat)))
# DBUSMENU PARSER (KDE / XFCE / Unity)
# DBUSMENU PARSER
def _parse_dbusmenu(self, node):
item = DBusMenuItem(id=0)
@@ -136,7 +180,6 @@ class DBusMenuClient:
item.id = node.get_child_value(0).get_int32()
# props
try:
props = node.get_child_value(1).unpack()
except Exception:
@@ -151,40 +194,31 @@ class DBusMenuClient:
if props.get("children-display") == "submenu":
item.has_submenu = True
# shortcut
sc = props.get("shortcut")
if sc:
try:
parts = []
if isinstance(sc, list):
flat = sc[0] if isinstance(sc[0], list) else sc
mapping = {
"Control": "Ctrl",
"Shift": "Shift",
"Alt": "Alt",
"Meta": "Super",
}
for s in flat:
parts.append(mapping.get(s, str(s).upper()))
item.shortcut = "+".join(parts)
except Exception:
item.shortcut = ""
# children
children = node.get_child_value(2)
for i in range(children.n_children()):
c = children.get_child_value(i)
if c.is_of_type(GLib.VariantType("v")):
c = c.get_variant()
item.children.append(self._parse_dbusmenu(c))
# lazy GTK submenu trigger
if item.has_submenu and not item.children:
try:
self.about_to_show(item.id)
@@ -198,64 +232,46 @@ class DBusMenuClient:
# GTK FALLBACK (GMenuModel-style)
def _try_gtk_menu(self):
# Passing None caused silent failure. Start with group 0.
try:
res = self.bus.call_sync(
res = _get_bus().call_sync(
self.service_name,
self.object_path,
"org.gtk.Menus",
"Get",
GLib.Variant("(au)", ([0],)), # FIX #2: correct parameter
"Start",
GLib.Variant("(au)", ([0],)),
None,
Gio.DBusCallFlags.NONE,
1000,
None,
)
return res.get_child_value(0) if res else None
except Exception:
return None
def _parse_gtk_menu(self, variant):
items = []
try:
if not variant:
return []
if variant.is_container():
for i in range(variant.n_children()):
c = variant.get_child_value(i)
item = DBusMenuItem(
id=i,
label="",
enabled=True,
visible=True,
)
item = DBusMenuItem(id=i)
try:
if c.n_children() > 0:
item.label = c.get_child_value(0).get_string()
except Exception:
pass
# Use n_children() - 1 explicitly.
n = c.n_children()
if n > 1:
sub = c.get_child_value(n - 1)
if c.n_children() > 1:
sub = c.get_child_value(c.n_children() - 1)
item.children = self._parse_gtk_menu(sub)
item.has_submenu = len(item.children) > 0
item.has_submenu = bool(item.children)
items.append(item)
except Exception:
pass
return items
# GTK / DBUS ACTIONS
# ACTIONS
def about_to_show(self, item_id):
try:
res = self._call(
@@ -271,34 +287,22 @@ class DBusMenuClient:
try:
self._call(
"Event",
GLib.Variant(
"(isvu)",
(item_id, "clicked", GLib.Variant("s", ""), 0),
),
GLib.Variant("(isvu)", (item_id, "clicked", GLib.Variant("s", ""), 0)),
)
self.invalidate()
except Exception:
pass
# CACHE CONTROL
def invalidate(self):
self._cache_valid = False
def prefetch_async(self):
# If already fetching, skip — get_layout's _fetching flag handles dedup.
with self._lock:
if self._fetching:
return
# Mark fetching now so a second call before the thread starts also bails out.
self._fetching = True
def worker():
try:
self._fetch()
finally:
with self._lock:
self._fetching = False
threading.Thread(target=worker, daemon=True).start()
if self._fetching:
return
threading.Thread(
target=self.get_layout, kwargs={"force_refresh": True}, daemon=True
).start()
def on_focus(self):
self.prefetch_async()
src/utils/gtkmenu.py
+120 -78
View File
@@ -1,36 +1,30 @@
import re
from typing import List, Optional
from gi.repository import Gio, GLib
from fabric.utils import logger
from utils.dbusmenu import DBusMenuItem
from fabric.utils import Gio, GLib, logger
from utils.dbusmenu import DBusMenuItem, _get_bus
_SUBMENU_RE = re.compile(r"submenu")
class GtkMenuClient:
__slots__ = ("service_name", "object_path", "_action_path", "_cached_items")
def __init__(self, service_name: str, object_path: str):
self.service_name = service_name
self.object_path = object_path
self.bus = Gio.bus_get_sync(Gio.BusType.SESSION, None)
self._action_group_path = self._guess_action_group_path(object_path)
# Actions live at the parent of /menus/...
self._action_path = (
object_path.split("/menus/")[0] if "/menus/" in object_path else object_path
)
self._cached_items: List[DBusMenuItem] = []
def _guess_action_group_path(self, menu_path: str) -> str:
# Menus are usually /org/appmenu/gtk/window/menus/menubar/10
# Actions are usually /org/appmenu/gtk/window/10
if "/menus/" in menu_path:
return menu_path.split("/menus/")[0]
# Some apps export actions at the exact same path
return menu_path
def _call_method(
self,
interface: str,
method: str,
params: Optional[GLib.Variant],
reply_type: Optional[GLib.VariantType],
path_override: Optional[str] = None,
) -> Optional[GLib.Variant]:
def _call(self, interface, method, params, reply_type=None, path=None):
try:
return self.bus.call_sync(
return _get_bus().call_sync(
self.service_name,
path_override or self.object_path,
path or self.object_path,
interface,
method,
params,
@@ -40,93 +34,141 @@ class GtkMenuClient:
None,
)
except Exception as e:
logger.debug(f"[GtkMenuClient] Call to {method} failed: {e}")
logger.debug(f"[GtkMenuClient] {method} failed: {e}")
return None
def get_layout(self, parent_id: int = 0) -> List[DBusMenuItem]:
return self._fetch_menu(parent_id)
items = self._fetch_menu(parent_id)
self._cached_items = items
return items
def _fetch_menu(self, menu_id: int) -> List[DBusMenuItem]:
params = GLib.Variant("(au)", ([menu_id],))
res = self._call_method(
"org.gtk.Menus", "Start", params, GLib.VariantType("(a(uuaa{sv}))")
def _find_item(
self, item_id: int, items: Optional[List[DBusMenuItem]] = None
) -> Optional[DBusMenuItem]:
if items is None:
items = self._cached_items
for item in items:
if item.id == item_id:
return item
if item.children:
found = self._find_item(item_id, item.children)
if found:
return found
return None
def _fetch_menu(
self, menu_id: int, _depth: int = 0, _visited: Optional[set] = None
) -> List[DBusMenuItem]:
if _depth > 10:
return []
if _visited is None:
_visited = set()
if menu_id in _visited:
return []
_visited.add(menu_id)
res = self._call(
"org.gtk.Menus",
"Start",
GLib.Variant("(au)", ([menu_id],)),
GLib.VariantType("(a(uuaa{sv}))"),
)
if not res:
return []
try:
# Unpack the response
# res is a tuple containing one element: the array
data = res.get_child_value(0).unpack()
if not data or len(data) == 0:
if not data:
return []
# data[0] is (subscription_id, menu_id, groups)
groups = data[0][2]
entry_map = {}
for entry in data:
gid, mid, sections = entry
entry_map[(gid, mid)] = sections
items = []
# each group is an array of dicts (items)
for group_idx, group in enumerate(groups):
# insert separator between groups
if group_idx > 0 and len(items) > 0 and len(group) > 0:
items.append(DBusMenuItem(id=-1, type="separator"))
for idx, props in enumerate(group):
item_id = hash(f"{menu_id}_{group_idx}_{idx}")
item = DBusMenuItem(id=item_id)
items: List[DBusMenuItem] = []
def resolve(sections, visited=None):
if visited is None:
visited = set()
for props in sections:
if not isinstance(props, dict):
continue
if ":section" in props:
ref = props[":section"]
if isinstance(ref, (list, tuple)) and len(ref) == 2:
key = (int(ref[0]), int(ref[1]))
if key not in visited:
visited.add(key)
child = entry_map.get(key)
if child is not None:
resolve(child, visited)
continue
if not props.get("label") and not any(
k in props for k in ("action", ":submenu", "submenu")
):
continue
item = DBusMenuItem(id=hash((menu_id, len(items))))
item.label = props.get("label", "").replace("_", "")
if "submenu" in props:
item.enabled = props.get("enabled", True)
item.visible = props.get("visible", True)
sub_ref = props.get(":submenu") or props.get("submenu")
if (
sub_ref
and isinstance(sub_ref, (list, tuple))
and len(sub_ref) == 2
):
item.has_submenu = True
submenu_id = props["submenu"]
item.children = self._fetch_menu(submenu_id)
item.children = self._fetch_menu(
int(sub_ref[0]), _depth + 1, _visited
)
if "action" in props:
item.action_name = props["action"]
if "target" in props:
# Variant target
item.action_target = GLib.Variant.new_variant(
GLib.Variant("v", props["target"])
)
try:
item.action_target = GLib.Variant.new_variant(
GLib.Variant("v", props["target"])
)
except Exception:
pass
items.append(item)
if data:
resolve(data[0][2])
return items
except Exception as e:
logger.error(f"[GtkMenuClient] Parse error: {e}")
return []
def about_to_show(self, item_id: int) -> bool:
# org.gtk.Menus doesn't have AboutToShow
return False
def click_item(
self, item_id: int, action_name: str = "", action_target: GLib.Variant = None
self, item_id: int, action_name: str = "", action_target=None
) -> None:
if not action_name:
item = self._find_item(item_id)
if item and item.action_name:
action_name = item.action_name
action_target = item.action_target
if not action_name:
return
# Strip the unity. prefix added by appmenu-gtk-module
if action_name.startswith("unity."):
action_name = action_name[6:]
try:
# Action activation
# Signature: Activate (sava{sv})
# s: action_name
# av: parameter (array of variants, empty if no target)
# a{sv}: platform data
params_arr = []
if action_target is not None:
params_arr = [action_target]
params = GLib.Variant("(sava{sv})", (action_name, params_arr, {}))
self._call_method(
"org.gtk.Actions",
"Activate",
params,
None,
path_override=self._action_group_path,
)
params_arr = [action_target] if action_target is not None else []
# Try menu path first (Thunar), fall back to action path (LibreOffice)
for act_path in (self.object_path, self._action_path):
try:
self._call(
"org.gtk.Actions",
"Activate",
GLib.Variant("(sava{sv})", (action_name, params_arr, {})),
path=act_path,
)
return
except Exception:
continue
except Exception as e:
logger.error(f"[GtkMenuClient] Failed to click item: {e}")
logger.error(f"[GtkMenuClient] click failed: {e}")