shaba/virt-manager
1
0
Fork 0
mirror of https://github.com/virt-manager/virt-manager.git synced 2025-01-05 09:17:57 +03:00
Code Issues Projects Releases Wiki Activity

baseclass: Emulate gobject signals for TUI

Browse Source 
Pretty easy to do with a high level language. Go python!
This commit is contained in:
Cole Robinson 2011-07-22 18:07:57 -04:00
parent 099753d146
commit ec38df2ee4
1 changed files with 54 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

74
src/virtManager/baseclass.py
View File

@ -44,15 +44,16 @@ class vmmGObject(GObject):
@staticmethod @staticmethod
def signal_new(klass, signal, args): def signal_new(klass, signal, args):
if not hasattr(gobject, "signal_new"): if hasattr(gobject, "signal_new"):
return gobject.signal_new(signal, klass,
gobject.SIGNAL_RUN_FIRST,
gobject.signal_new(signal, klass, gobject.TYPE_NONE,
gobject.SIGNAL_RUN_FIRST, args)
gobject.TYPE_NONE, else:
args) klass.fake_signal_listeners[signal] = []
_leak_check = True _leak_check = True
fake_signal_listeners = {}
def __init__(self): def __init__(self):
GObject.__init__(self) GObject.__init__(self)
@ -62,6 +63,9 @@ class vmmGObject(GObject):
self._gobject_timeouts = [] self._gobject_timeouts = []
self._gconf_handles = [] self._gconf_handles = []
self._signal_id_map = {}
self._next_signal_id = 1
self.object_key = str(self) self.object_key = str(self)
# Config might not be available if we error early in startup # Config might not be available if we error early in startup
@ -81,18 +85,42 @@ class vmmGObject(GObject):
except: except:
logging.exception("Error cleaning up %s" % self) logging.exception("Error cleaning up %s" % self)
# Custom signal implementations
# These functions duplicate gobject signal behavior since it's
# needed for some TUI functionality
def _get_signal_listeners(self, signal_name):
return self.fake_signal_listeners[signal_name]
def _add_signal_listener(self, signal_name, cb, *args):
sigid = self._next_signal_id
self._next_signal_id += 1
self._signal_id_map[sigid] = (signal_name, cb, args)
self.fake_signal_listeners[signal_name].append((cb, args))
return sigid
def _remove_signal_listener(self, sigid):
signame, cb, args = self._signal_id_map[sigid]
listener_data = (cb, args)
if listener_data not in self.fake_signal_listeners[signame]:
raise RuntimeError("Didn't find signal listener to remove: %d" %
sigid)
self.fake_signal_listeners[signame].remove(listener_data)
del(self._signal_id_map[sigid])
def connect(self, name, callback, *args): def connect(self, name, callback, *args):
if not hasattr(GObject, "connect"): if hasattr(GObject, "connect"):
return ret = GObject.connect(self, name, callback, *args)
ret = GObject.connect(self, name, callback, *args) self._gobject_handles.append(ret)
self._gobject_handles.append(ret) return ret
return ret else:
return self._add_signal_listener(name, callback, *args)
def disconnect(self, handle): def disconnect(self, handle):
if not hasattr(GObject, "disconnect"): if hasattr(GObject, "disconnect"):
return ret = GObject.disconnect(self, handle)
ret = GObject.disconnect(self, handle) self._gobject_handles.remove(handle)
self._gobject_handles.remove(handle) return ret
return ret else:
return self._remove_signal_listener(handle)
def add_gconf_handle(self, handle): def add_gconf_handle(self, handle):
self._gconf_handles.append(handle) self._gconf_handles.append(handle)
@ -171,12 +199,18 @@ class vmmGObject(GObject):
""" """
Make sure timeout functions are run thread safe Make sure timeout functions are run thread safe
""" """
if not hasattr(gobject, "timout_add"):
return
return gobject.timeout_add(timeout, _safe_wrapper, func, *args) return gobject.timeout_add(timeout, _safe_wrapper, func, *args)
def emit(self, *args, **kwargs): def emit(self, signal_name, *args):
if not hasattr(GObject, "emit"): if hasattr(GObject, "emit"):
return GObject.emit(self, signal_name, *args)
else:
for cb, customargs in self._get_signal_listeners(signal_name):
cbargs = (self,) + args + customargs
cb(*cbargs)
return return
return GObject.emit(self, *args, **kwargs)
def __del__(self): def __del__(self):
if hasattr(GObject, "__del__"): if hasattr(GObject, "__del__"):