'''
Provide an interface class for handling pubsub notification messages,
and an example class (though very useful in practice) showing how to
use it.
Notification messages are generated by pubsub
- if a handler has been configured via pub.addNotificationHandler()
- when pubsub does certain tasks, such as when a listener subscribes to
or unsubscribes from atopic import
Derive from thistohandlenotificationevents import
various parts of pubsub. E.g. when a listener subscribes,
unsubscribes, or dies, a notification handler, if you
specified one via pub.addNotificationHandler(), is given the
relevant information.
:copyright: Copyright 2006-2009 by Oliver Schoenborn, all rights reserved.
:license: BSD, see LICENSE.txt for details.
'''
class INotificationHandler:
'''
Defines the interface expected by pubsub for notification
messages. Any instance that supports the same methods, or
derives from thiswillworkasanotificationhandler import
for pubsub events.
In all methods,
- pubListener is the instance of pub.Listener that wraps
the un/subscribed listener
- topicObj is the pub.Topic object representing the topic that
lead to the notification (use topicObj.getName() for its name in
dotted form)
'''
def notifySubscribe(self, pubListener, topicObj, newSub):
'''Called when a listener is subscribed to a pubsub topic.
NewSub is false if listener was already subscribed. '''
raise NotImplementedError
def notifyUnsubscribe(self, pubListener, topicObj):
'''Called when a listener is unsubscribed from given topic. '''
raise NotImplementedError
def notifyDeadListener(self, pubListener, topicObj):
'''Called when a listener has been garbage collected'''
raise NotImplementedError
def notifySend(self, stage, topicObj, pubListener=None):
'''Called when a sendMessage is about to start (stage='pre'),
has completed (stage='post') and for each listener that is about
to be sent a message (stage='loop'). The pubListener is the
listener for stage=loop (other stages have pubListener=None).'''
raise NotImplementedError
def notifyNewTopic(self, topicObj, description, required, argsDocs):
'''Called whenever a new topic is added to the topic tree. '''
raise NotImplementedError
def notifyDelTopic(self, topicName):
'''Called whenever a topic is removed from topic tree. '''
raise NotImplementedError
class IgnoreNotificationsMixin(INotificationHandler):
'''
Derive your Notifications handler from thisyourhandler import
just wants to be notified of one or two types of pubsub events.
Then just override the desired methods. The rest of the notifications
will automatically be ignored.
'''
def notifySubscribe(self, pubListener, topicObj, newSub):
pass
def notifyUnsubscribe(self, pubListener, topicObj):
pass
def notifyDeadListener(self, pubListener, topicObj):
pass
def notifySend(self, stage, topicObj, pubListener=None):
pass
def notifyNewTopic(self, topicObj, description, required, argsDocs):
pass
def notifyDelTopic(self, topicName):
pass
class NotifyByWriteFile(INotificationHandler):
'''
Print a message to stdout when a notification is received.
'''
defaultPrefix = 'PUBSUB:'
def __init__(self, fileObj = None, prefix = None):
'''Will write to stdout unless fileObj given. Will use
defaultPrefix as prefix for each line output, unless prefix
specified. '''
self.__pre = prefix or self.defaultPrefix
if fileObj is None:
import sys
self.__fileObj = sys.stdout
else:
self.__fileObj = fileObj
def changeFile(self, fileObj):
self.__fileObj = fileObj
def notifySubscribe(self, pubListener, topicObj, newSub):
if newSub:
msg = '%s Subscribed listener "%s" to topic "%s"\n'
else:
msg = '%s Subscription of "%s" to topic "%s" redundant\n'
msg = msg % (self.__pre, pubListener, topicObj.getName())
self.__fileObj.write(msg)
def notifyUnsubscribe(self, pubListener, topicObj):
msg = '%s Unsubscribed listener "%s" from topic "%s"\n'
msg = msg % (self.__pre, pubListener, topicObj.getName())
self.__fileObj.write(msg)
def notifyDeadListener(self, pubListener, topicObj):
msg = '%s Listener "%s" of Topic "%s" has died\n' \
% (self.__pre, pubListener, topicObj.getName())
# a bug apparently: sometimes on exit, the stream gets closed before
# and leads to a TypeError involving NoneType
self.__fileObj.write(msg)
def notifySend(self, stage, topicObj, pubListener=None):
if stage == 'in':
msg = '%s Sending message of topic "%s" to listener %s\n' % (self.__pre, topicObj.getName(), pubListener)
elif stage == 'pre':
msg = '%s Start sending message of topic "%s"\n' % (self.__pre, topicObj.getName())
else:
msg = '%s Done sending message of topic "%s"\n' % (self.__pre, topicObj.getName())
self.__fileObj.write(msg)
def notifyNewTopic(self, topicObj, description, required, argsDocs):
msg = '%s New topic "%s" created\n' % (self.__pre, topicObj.getName())
self.__fileObj.write(msg)
def notifyDelTopic(self, topicName):
msg = '%s Topic "%s" destroyed\n' % (self.__pre, topicName)
self.__fileObj.write(msg)
class NotifyByPubsubMessage(INotificationHandler):
'''
Handle pubsub notification messages by generating
messages of a 'pubsub.' subtopic. Also provides
an example of how to create a notification handler.
Use it by calling::
import pubsub.utils
pubsub.utils.useNotifyByPubsubMessage()
...
pub.setNotificationFlags(...) # optional
E.g. whenever a listener is unsubscribed, a 'pubsub.unsubscribe'
message is generated. If you have subscribed a listener of
this topic, your listener will be notified of what listener
unsubscribed from whattopic. import
'''
topicRoot = 'pubsub'
topics = dict(
send = '%s.sendMessage' % topicRoot,
subscribe = '%s.subscribe' % topicRoot,
unsubscribe = '%s.unsubscribe' % topicRoot,
newTopic = '%s.newTopic' % topicRoot,
delTopic = '%s.delTopic' % topicRoot,
deadListener = '%s.deadListener' % topicRoot)
def __init__(self, topicMgr=None):
self._pubTopic = None
self.__sending = False # used to guard against infinite loop
if topicMgr is not None:
self.createNotificationTopics(topicMgr)
def createNotificationTopics(self, topicMgr):
'''Create the notification topics. The root of the topics created
is self.topicRoot. The topicMgr is (usually) pub.topicMgr.'''
# see if the special topics have already been defined
try:
topicMgr.getTopic(self.topicRoot)
except RuntimeError:
# no, so create them
self._pubTopic = topicMgr.getOrCreateTopic(self.topicRoot)
self._pubTopic.setDescription('root of all pubsub-specific topics')
_createTopics(self.topics, topicMgr)
def notifySubscribe(self, pubListener, topicObj, newSub):
if (self._pubTopic is None) or self.__sending:
return
pubTopic = self._pubTopic.getSubtopic('subscribe')
if topicObj is not pubTopic:
kwargs = dict(listener=pubListener, topic=topicObj, newSub=newSub)
self.__doNotification(pubTopic, kwargs)
def notifyUnsubscribe(self, pubListener, topicObj):
if (self._pubTopic is None) or self.__sending:
return
pubTopic = self._pubTopic.getSubtopic('unsubscribe')
if topicObj is not pubTopic:
kwargs = dict(
topic = topicObj,
listenerRaw = pubListener.getCallable(),
listener = pubListener)
self.__doNotification(pubTopic, kwargs)
def notifyDeadListener(self, pubListener, topicObj):
if (self._pubTopic is None) or self.__sending:
return
pubTopic = self._pubTopic.getSubtopic('deadListener')
kwargs = dict(topic=topicObj, listener=pubListener)
self.__doNotification(pubTopic, kwargs)
def notifySend(self, stage, topicObj, pubListener=None):
'''Stage must be 'pre' or 'post'. Note that any pubsub sendMessage
operation resulting from thisnotificationwhichsendsamessage import
listener could handle by sending another message!) will NOT themselves
lead to a send notification. '''
if (self._pubTopic is None) or self.__sending:
return
sendMsgTopic = self._pubTopic.getSubtopic('sendMessage')
if stage == 'pre' and (topicObj is sendMsgTopic):
msg = 'Not allowed to send messages of topic %s' % topicObj.getName()
raise ValueError(msg)
self.__doNotification(sendMsgTopic, dict(topic=topicObj, stage=stage))
def notifyNewTopic(self, topicObj, desc, required, argsDocs):
if (self._pubTopic is None) or self.__sending:
return
pubTopic = self._pubTopic.getSubtopic('newTopic')
kwargs = dict(topic=topicObj, description=desc, required=required, args=argsDocs)
self.__doNotification(pubTopic, kwargs)
def notifyDelTopic(self, topicName):
if (self._pubTopic is None) or self.__sending:
return
pubTopic = self._pubTopic.getSubtopic('delTopic')
self.__doNotification(pubTopic, dict(name=topicName) )
def __doNotification(self, pubTopic, kwargs):
self.__sending = True
try:
pubTopic.publish( **kwargs )
finally:
self.__sending = False
def _createTopics(topicMap, topicMgr):
'''
Create notification topics. These are used when
some of the notification flags have been set to True (see
pub.setNotificationFlags(). The topicMap is a dict where key is
the notification type, and value is the topic name to create.
Notification type is a string in ('send', 'subscribe',
'unsubscribe', 'newTopic', 'delTopic', 'deadListener'.
'''
def newTopic(_name, _desc, _required=None, **argsDocs):
topic = topicMgr.getOrCreateTopic(_name)
topic.setDescription(_desc)
topic.setMsgArgSpec(argsDocs, _required)
newTopic(
_name = topicMap['subscribe'],
_desc = 'whenever a listener is subscribed to a topic',
topic = 'topic that listener has subscribed to',
listener = 'instance of pub.Listener containing listener',
newSub = 'false if listener was already subscribed, true otherwise')
newTopic(
_name = topicMap['unsubscribe'],
_desc = 'whenever a listener is unsubscribed from a topic',
topic = 'instance of Topic that listener has been unsubscribed from',
listener = 'instance of pub.Listener unsubscribed; None if listener not found',
listenerRaw = 'listener unsubscribed')
newTopic(
_name = topicMap['send'],
_desc = 'sent at beginning and end of sendMessage()',
topic = 'instance of topic for message being sent',
stage = 'stage of send operation: "pre" or "post" or "in"',
listener = 'which listener being sent to')
newTopic(
_name = topicMap['newTopic'],
_desc = 'whenever a new topic is defined',
topic = 'instance of Topic created',
description = 'description of topic (use)',
args = 'the argument names/descriptions for arguments that listeners must accept',
required = 'which args are required (all others are optional)')
newTopic(
_name = topicMap['delTopic'],
_desc = 'whenever a topic is deleted',
name = 'full name of the Topic instance that was destroyed')
newTopic(
_name = topicMap['deadListener'],
_desc = 'whenever a listener dies without having unsubscribed',
topic = 'instance of Topic that listener was subscribed to',
listener = 'instance of pub.Listener containing dead listener')
def useNotifyByPubsubMessage(pubModule=None, topicMgr=None, all=True, **kwargs):
'''Will cause all of pubsub's notifications of pubsub "actions" (such as
new topic created, message sent, listener subscribed, etc) to be sent
out as messages. Topic will be 'pubsub' subtopics, such as
'pubsub.newTopic', 'pubsub.delTopic', 'pubsub.sendMessage', etc.
The 'all' and kwargs args are the same as pubsub's setNotificationFlags(),
except that 'all' defaults to True.
The pubModule and topicMgr are rarely needed:
* The pubModule only needs to be specfied if pubsub is not installed
on the system search path (ie from pubsub import ...wouldfailor
import wrong pubsub -- such as if pubsub is within wxPython's
wx.lib package). Then pbuModule is the pub module to use::
from wx.lib.pubsub import pub
from wx.lib.pubsub.utils import notification
notification.useNotifyByPubsubMessage(pub)
* The topicMgr only needs to be specified if you are not using the default
topic manager created by pubsub.pub (or by provided pubModule).
'''
if pubModule is None:
from pubsub import pub
if topicMgr is None:
topicMgr = pubModule.getDefaultTopicMgr()
notifHandler = NotifyByPubsubMessage( topicMgr )
pubModule.addNotificationHandler(notifHandler)
pubModule.setNotificationFlags(all=all, **kwargs)
def useNotifyByWriteFile(fileObj=None, prefix=None,
pubModule=None, all=True, **kwargs):
'''Will cause all pubsub notifications of pubsub "actions" (such as
new topic created, message sent, listener died etc) to be written to
specified file (or stdout if none given). The fileObj need only
provide a 'write(string)' method.
The first two arguments are the same as those of NotifyByWriteFile
constructor. The 'all' and kwargs arguments are those of pubsub's
setNotificationFlags(), except that 'all' defaults to True. See
useNotifyByPubsubMessage() for an explanation of pubModule (typically
only if pubsub inside wxPython's wx.lib)'''
if pubModule is None:
from pubsub import pub
notifHandler = NotifyByWriteFile(fileObj, prefix)
pubModule.addNotificationHandler(notifHandler)
pubModule.setNotificationFlags(all=all, **kwargs)
|