Author: dmeyer
Date: Thu Jan 26 21:31:53 2006
New Revision: 7884
Modified:
trunk/core/src/ipc/mbus_wrapper.py
Log:
some data hiding and doc update
Modified: trunk/core/src/ipc/mbus_wrapper.py
==============================================================================
--- trunk/core/src/ipc/mbus_wrapper.py (original)
+++ trunk/core/src/ipc/mbus_wrapper.py Thu Jan 26 21:31:53 2006
@@ -96,51 +96,21 @@
return True
-class RPC(kaa.notifier.Callback):
+def expose(command, add_command=False, add_source=False):
"""
- Wrapper for a RPC function in the server.
+ Decorator to expose a function to the mbus.
"""
- def __init__(self, entity, add_source, add_command, *args, **kwargs):
- super(RPC, self).__init__(*args, **kwargs)
- self.entity = entity
- self.add_source = add_source
- self.add_command = add_command
-
+ def decorator(func):
+ func._mbus_rpc = (command, add_command, add_source)
+ return func
+ return decorator
- def __call__(self, command):
- """
- Call the RPC function. This function is called from the mbus module.
- """
- try:
- args = []
- if self.add_command:
- args.append(command)
- if self.add_source:
- args.append(self.entity.get_entity(command.source))
- args += _fromMType(command.arguments)
- result = super(RPC, self).__call__(*args)
- if result != None:
- if not isinstance(result, (list, tuple, dict)):
- log.error('return value is no list')
- result = [ result ]
- # send result as rpc return
- ret = mbus.RPCReturn(command)
- ret.setAppStatus( True, 'OK', 'success' )
- ret.arguments = _toMType(result)
- return ret
- except (KeyboardInterrupt, SystemExit), e:
- OneShotTimer(sys.exit, 0).start(0)
- except Exception, e:
- # send all exceptions as rpc error return
- log.exception('Error in RPC handling')
- ret = mbus.RPCReturn(command)
- ret.setAppStatus( True, 'FAILED', str(e) )
- ret.arguments = []
- return ret
class RPCReturn(list):
"""
- RPC Return object.
+ RPC Return class. An object of this class will be passed to
+ the callback for a rpc. This class inherits from list, the arguments
+ of the return can be parsed with normal list memeber functions.
"""
def __init__(self, result):
self.result = result
@@ -150,12 +120,18 @@
else:
list.__init__(self)
self.command = ''
-
+
def __nonzero__(self):
+ """
+ Check return if it is valid or not.
+ """
return hasattr(self.result, 'success') and \
self.result.success and self.result.result == 'OK'
def __str__(self):
+ """
+ Convert to string for debugging.
+ """
if self:
return '<RPCReturn %s(%s)>' % (self.command, list.__str__(self))
if hasattr(self.result, 'description'):
@@ -163,43 +139,33 @@
return str(self.result)
-class RPCClient(kaa.notifier.Callback):
+class RPC(kaa.notifier.Callback):
"""
- Client part of RPC.
+ Client part of RPC. An object of this class has a call function to do
+ the real calling of the RPC.
"""
def __init__(self, entity, command, *args, **kwargs):
- super(RPCClient, self).__init__(*args, **kwargs)
+ super(RPC, self).__init__(*args, **kwargs)
self.entity = entity
self.command = command
-
def call(self, *args):
args = _toMType(args)
log.debug('send %s' % args)
self.entity.entity.sendRPC(self.entity.addr, self.command, args, self)
-
+
def __call__(self, result):
try:
result = RPCReturn(result)
log.debug('recv %s' % result)
- super(RPCClient, self).__call__(result)
+ super(RPC, self).__call__(result)
except (KeyboardInterrupt, SystemExit), e:
OneShotTimer(sys.exit, 0).start(0)
except Exception, e:
log.exception('RPC Return handling %s' % result)
return True
-
-def expose(command, add_command=False, add_source=False):
- """
- Decorator to expose a function to the mbus.
- """
- def decorator(func):
- func._mbus_rpc = (command, add_command, add_source)
- return func
- return decorator
-
class RemoteEntity(object):
"""
A remote mbus entity.
@@ -210,28 +176,28 @@
self.valid = True
self.signals = { 'lost-entity': Signal() }
self.registrations = []
-
+
def send(self, event, *args):
self.entity.sendReliable(self.addr, event, _toMType(args))
def rpc(self, command, callback, *args, **kwargs):
if callback == None:
callback = _dummy_callback
- return RPCClient(self, command, callback, *args, **kwargs)
+ return RPC(self, command, callback, *args, **kwargs)
def register(self, prefix):
self.rpc('mbus.register', None).call(prefix)
-
+
def __str__(self):
return '<RemoteEntity %s>' % self.addr
-
+
def __getattr__(self, attr):
log.error('no attr %s for %s' % (attr, self))
raise AttributeError('no %s for %s' % (attr, self))
def __nonzero__(self):
return self.valid
-
+
def matches(self, addr):
return mbus.isAddressedTo(addr, self.addr)
@@ -245,44 +211,6 @@
def __str__(self):
return '<Event %s>' % self.name
-
-class EventDict(object):
- def __init__(self, entity):
- self.entity = entity
- self.signals = {}
-
- def __getitem__(self, event):
- if not self.signals.has_key(event):
- self.signals[event] = Signal()
- self.entity.addCallback(event, self)
- return self.signals[event]
-
- def __call__(self, event):
- source = self.entity.get_entity(event.header.source)
- event = Event(source, event.payload[0])
- try:
- self.signals[event.name].emit(event)
- except (KeyboardInterrupt, SystemExit), e:
- OneShotTimer(sys.exit, 0).start(0)
- except Exception, e:
- log.exception('event handling')
- return True
-
-
-class EntitySignal(Signal):
- def __init__(self, entities):
- Signal.__init__(self)
- self.entities = entities
-
- def connect(self, callback, *args, **kwargs):
- Signal.connect(self, callback, *args, **kwargs)
- for e in self.entities.values():
- callback(e, *args, **kwargs)
-
- def connect_weak(self, callback, *args, **kwargs):
- Signal.connect(self, callback, *args, **kwargs)
- for e in self.entities.values():
- callback(e, *args, **kwargs)
class Instance(mbus.Guides):
"""
@@ -290,7 +218,99 @@
is provided a second time, the already existing object will be returned.
"""
_instances = {}
-
+
+ class RPC(kaa.notifier.Callback):
+ """
+ Wrapper for a RPC function in the server.
+ """
+ def __init__(self, entity, add_source, add_command, *args, **kwargs):
+ super(Instance.RPC, self).__init__(*args, **kwargs)
+ self.entity = entity
+ self.add_source = add_source
+ self.add_command = add_command
+
+
+ def __call__(self, command):
+ """
+ Call the RPC function. This function is called from the mbus
module.
+ """
+ try:
+ args = []
+ if self.add_command:
+ args.append(command)
+ if self.add_source:
+ args.append(self.entity._get_entity(command.source))
+ args += _fromMType(command.arguments)
+ result = super(Instance.RPC, self).__call__(*args)
+ if result != None:
+ if not isinstance(result, (list, tuple, dict)):
+ log.error('return value is no list')
+ result = [ result ]
+ # send result as rpc return
+ ret = mbus.RPCReturn(command)
+ ret.setAppStatus( True, 'OK', 'success' )
+ ret.arguments = _toMType(result)
+ return ret
+ except (KeyboardInterrupt, SystemExit), e:
+ OneShotTimer(sys.exit, 0).start(0)
+ except Exception, e:
+ # send all exceptions as rpc error return
+ log.exception('Error in RPC handling')
+ ret = mbus.RPCReturn(command)
+ ret.setAppStatus( True, 'FAILED', str(e) )
+ ret.arguments = []
+ return ret
+
+
+ class EventDict(object):
+ """
+ Special case of a dict for events. When the key is not known,
+ create it and register it to the mbus. The value of a key is
+ a Signal.
+ """
+ def __init__(self, entity):
+ self.entity = entity
+ self.signals = {}
+
+ def __getitem__(self, event):
+ if not self.signals.has_key(event):
+ self.signals[event] = Signal()
+ self.entity.addCallback(event, self)
+ return self.signals[event]
+
+ def __call__(self, event):
+ source = self.entity._get_entity(event.header.source)
+ event = Event(source, event.payload[0])
+ try:
+ self.signals[event.name].emit(event)
+ except (KeyboardInterrupt, SystemExit), e:
+ OneShotTimer(sys.exit, 0).start(0)
+ except Exception, e:
+ log.exception('event handling')
+ return True
+
+
+ class EntitySignal(Signal):
+ """
+ Special case of a signal for entities. It will pass all already
+ known entities to the callback. Because of that, one time variants
+ of a signal make no sense and have no special handling.
+ """
+ def __init__(self, entities):
+ Signal.__init__(self)
+ self.entities = entities
+
+ def connect(self, callback, *args, **kwargs):
+ Signal.connect(self, callback, *args, **kwargs)
+ for e in self.entities.values():
+ callback(e, *args, **kwargs)
+
+ def connect_weak(self, callback, *args, **kwargs):
+ Signal.connect(self, callback, *args, **kwargs)
+ for e in self.entities.values():
+ callback(e, *args, **kwargs)
+
+
def __new__(cls, name='default'):
"""
Create new object or return old one.
@@ -317,70 +337,79 @@
if name in ('main', 'freevo'):
name = 'core'
self.addr = mbus.MAddress({'type': 'home-theatre', 'module': name })
- self.__remote_entities = {}
- self.__modules = []
-
- self.signals = { 'new-entity': EntitySignal(self.__remote_entities),
+
+ # some internal variables
+ self._entity_dict = {}
+ self._connected_modules = []
+
+ # known signals
+ self.signals = { 'new-entity':
Instance.EntitySignal(self._entity_dict),
'lost-entity': Signal(),
'register': Signal()
}
- self.events = EventDict(self)
-
+ # event list
+ self.events = Instance.EventDict(self)
+
+ # call mbus.Guides init function
mbus.Guides.__init__(self, self.addr)
# set callbacks
- self.signal_connect( 'new-entity', self.new_entity )
- self.signal_connect( 'lost-entity', self.lost_entity )
- self.signal_connect( 'unknown-message', self.mbus_error )
- self.signal_connect( 'error', self.mbus_error )
+ self.signal_connect( 'new-entity', self._new_entity )
+ self.signal_connect( 'lost-entity', self._lost_entity )
+ self.signal_connect( 'unknown-message', self._mbus_error )
+ self.signal_connect( 'error', self._mbus_error )
self.onErrorInvokeRPCCallback = True
- self.connect_rpc(self.register_request, 'mbus.register',
add_source=True)
+
+ # connect mbus.register rpc
+ self.connect_rpc(self._register_request, 'mbus.register',
add_source=True)
def get_entities(self):
"""
+ Return a list of all known entities.
"""
- return self.__remote_entities.values()
+ return self._entity_dict.values()
- def get_entity(self, maddr):
+ def _get_entity(self, maddr):
"""
+ Get the entity with the given addr (internal use)
"""
- if self.__remote_entities.has_key(str(maddr)):
- return self.__remote_entities[str(maddr)]
+ if self._entity_dict.has_key(str(maddr)):
+ return self._entity_dict[str(maddr)]
e = RemoteEntity(maddr, self)
- self.__remote_entities[str(maddr)] = e
+ self._entity_dict[str(maddr)] = e
return e
-
- def new_entity(self, maddr):
+
+ def _new_entity(self, maddr):
"""
pyMbus callback for new entity
"""
- self.signals['new-entity'].emit(self.get_entity(maddr))
+ self.signals['new-entity'].emit(self._get_entity(maddr))
- def lost_entity(self, maddr):
+ def _lost_entity(self, maddr):
"""
pyMbus callback for lost entity
"""
log.debug('lost entity %s' % maddr)
- e = self.__remote_entities[str(maddr)]
- del self.__remote_entities[str(maddr)]
+ e = self._entity_dict[str(maddr)]
+ del self._entity_dict[str(maddr)]
e.valid = False
self.signals['lost-entity'].emit(e)
e.signals['lost-entity'].emit()
- def mbus_error(self, error):
+ def _mbus_error(self, error):
"""
Small errorhandler for debugging mbus
"""
log.info('mbus error %s' % error)
- def register_request(self, entity, prefix):
+ def _register_request(self, entity, prefix):
"""
Register request from the other side.
"""
@@ -388,7 +417,7 @@
self.signals['register'].emit(entity, prefix)
return []
-
+
def connect_rpc(self, function, command, *args, **kwargs):
"""
connect rpc to mbus.
@@ -402,7 +431,7 @@
if 'add_source' in kwargs:
add_source = kwargs['add_source']
del kwargs['add_source']
- r = RPC(self, add_source, add_command, function, *args, **kwargs)
+ r = Instance.RPC(self, add_source, add_command, function, *args,
**kwargs)
self.addRPCCallback(command, r)
@@ -413,10 +442,10 @@
if isinstance(obj, str):
exec('import %s as obj' % obj)
if isinstance(obj, types.ModuleType):
- if obj in self.__modules:
+ if obj in self._connected_modules:
log.error('duplicate ipc connect: %s', obj)
- self.__modules.append(obj)
-
+ self._connected_modules.append(obj)
+
class struct(object):
pass
@@ -438,16 +467,18 @@
def send_event(self, event, *args):
- send_to = []
+ """
+ Send an event to all registered entities.
+ """
args = _toMType(args)
- for entity in self.__remote_entities.values():
+ for entity in self._entity_dict.values():
if not entity.valid:
continue
for ev in entity.registrations:
if event.startswith(ev):
self.sendReliable(entity.addr, event, args)
break
-
+
# Check for mbus configuration file. It is either ~/.mbus or defined
# in the environment variable MBUS.
-------------------------------------------------------
This SF.net email is sponsored by: Splunk Inc. Do you grep through log files
for problems? Stop! Download the new AJAX search engine that makes
searching your log files as easy as surfing the web. DOWNLOAD SPLUNK!
http://sel.as-us.falkag.net/sel?cmd=lnk&kid=103432&bid=230486&dat=121642
_______________________________________________
Freevo-cvslog mailing list
[email protected]
https://lists.sourceforge.net/lists/listinfo/freevo-cvslog
