Thursday, January 21, 2010

Message Bus for Gnome Object based Python projects

Whilst developing my latest software project (Rhythmbox plugin Last.fm sync), I have crafted a "Message Bus" based on Gnome GObject. Since I didn't have any experience at all with Gnome GObject previously, it took me sometimes to work out the details of the architecture.  For an example usage, please refer to my Git Repository.

""" 
    GObject based "Message Bus"
    
    @author: Jean-Lou Dupont
"""

import gobject #@UnresolvedImport

class Signals(gobject.GObject):
    """
    List of the application level signals
    """
    __gsignals__ = {

         ## NOTE: customization required here
         ## =================================
         ## Announces changes in the user's Last.fm properties
        "lastfm_username_changed":  (gobject.SIGNAL_RUN_FIRST, gobject.TYPE_NONE, (gobject.TYPE_STRING,)) 
        ,'lastfm_password_changed': (gobject.SIGNAL_RUN_FIRST, gobject.TYPE_NONE, (gobject.TYPE_STRING,))

        ## Used to signal a change in the currently playing track
        ,"playing_song_changed":    (gobject.SIGNAL_RUN_FIRST, gobject.TYPE_NONE, (gobject.TYPE_OBJECT,))
        
        ## Used to report a failure when accessing Last.fm web service
        ,'lastfm_request_failed':   (gobject.SIGNAL_RUN_FIRST, gobject.TYPE_NONE, ())
        
        ## Used for distributing the results of the query against the Last.fm web service
        ,"user_track_info":         (gobject.SIGNAL_RUN_FIRST, gobject.TYPE_NONE, (gobject.TYPE_OBJECT,))
        
        ## Used to pass around the "shell" global object
        ,"rb_shell":                (gobject.SIGNAL_RUN_FIRST, gobject.TYPE_NONE, (gobject.TYPE_OBJECT,))              
    }

    def __init__(self):
        gobject.GObject.__init__(self)
        
        
class Bus(object):
    """
    Message Bus
    
    Borg Pattern
    """
    _signals=Signals()

    @classmethod
    def emit(cls, name, *pa, **kwa):
        cls._signals.emit(name, *pa, **kwa)

    @classmethod
    def add_emission_hook(cls, name, callback):
        gobject.add_emission_hook(cls._signals, name, callback)
    
mbus=Bus()

## =============================================================== Tests

if __name__=="__main__":

    def callback(signal, data):
        print "callback: signal=%s, data=%s" % (signal, data)
    
    Bus.add_emission_hook("lastfm_username_changed", callback)
    
    Bus.emit("lastfm_username_changed", "jldupont")

Relevant Stackoverflow posts
Here are two Stackoverflow posts in relation with my exercise: question 1, question 2.

No comments:

Post a Comment