root/trac/hacks/marketplugin/0.9/tracmarket/mq.py

from cPickle import loads, dumps
import random
import os
import re
import sys
import types
import time

from trac.core import *

# from trac.db import SQLiteConnection
# from tracmarket.db import *

'''
See the "Enterprise Integration Patterns" book by Hohpe and Woolf for
much of the terminology and concepts used in this module.  

Additional terms we use here:

- group: a named publish/subscribe channel; clients subscribe to
  groups; clients send messages to groups; groups can be named using
  any string, but a good convention is "foo.bar.baz" or "foo/bar/baz"

- reader: the callback function which the bus calls to transfer a
  message to a client

'''

class IMessageBusSubscriber(Interface): 
    '''
    A component which subscribes to message bus channels.  Note that
    anything can call bus.subscribe() etc. -- implementing this
    interface is not required for that.  This interface exists solely
    for components to implement solicit() if they need to, so the bus
    can tickle them for subscriptions at startup.

    '''

    def solicit(bus):
        '''Ask subscriber for subscription requests.  Subscriber will call
        bus.subscribe() to respond.'''

class MemoryBus(object): 
    '''
    A message bus which routes and delivers messages using a
    publish/subscribe model.  This is intended to be used as a
    per-query, in-memory message bus, with no persisent message store.
    The goals are simplicity, speed, and no need to isolate queries
    from each other by using any correlation id or special reply
    channel names.  Keeping everything in memory should also avoid
    most of the need for a transactional interface to the bus,
    assuming each http query is handled within a single db
    transaction.  
    
    A hypothetical DbBus might subclass MemoryBus to create a bus with
    transactions and persistent message storage; that might be worth
    doing if the DbBus was running in its own daemon, with MemoryBus
    clients attaching to it.
    
    '''

    def __init__(self, subscribers=None, deadletter_id=None, ctx=None,
            delimiter='.'):
        self.pattern_type = type(re.compile(''))
        self.grep = os.environ.get('MQGREP', None)
        self.debug = os.environ.get('MQDEBUG', None)
        # self.queues[subscription_id] = Queue()
        # self.subscriptions[subscription_id] = dict(**subscribe_args)
        # self.patterns[pattern][subscription_id] = True
        # self.query_cache[group] = sendto_queues
        self.queues = {}
        self.subscriptions = {}
        self.patterns = {}
        self.query_cache = {}
        # delimiter between parts of heirarchical group names
        self.delimiter = delimiter
        # bulletin board space for storage of miscellaneous local
        # context bits (db handles, etc.)
        self.ctx = ctx
        # caller is supposed to create a dead letter group and reader
        # and then give us the subscription id; if they didn't do
        # this, then create a group named 'deadletter', and subscribe
        # ourselves to it
        if not deadletter_id: 
            self.deadletter_id = self.subscribe('deadletter', reader=self.error)
        else:
            self.deadletter_id = deadletter_id
        # wake up subscribers (this needs to be *last*) 
        if subscribers:
            for s in subscribers:
                s.solicit(self)

    def error(self, bus, group, msg):
        print >>sys.stderr, "dead letter: no readers for", msg.group

    def subscribe(self, group, reader=None, id=None):
        '''Called by client to attach to bus; returns subscription id.
        Can be called multiple times by same client using different
        group.  Group is a string or compiled regex.  Bus will enqueue
        for client any message addressed to any group matching the
        group string or regex.  Reader is a callback function.
        Messages will be delivered to client using reader(bus, group,
        msg).  If reader is None, then client will need to call recv()
        to pick up messages.  Client can assign any globally unique
        string as subscription id; if id is None, then a random id
        will be assigned.  Client can re-subscribe (to change callme
        or reader, for instance) by re-using the same id.'''
        self.query_cache = {}
        # convert to regex 
        if type(group) is self.pattern_type:
            pattern = group
        elif group.endswith(self.delimiter):
            pattern = re.compile("^%s.*$" % re.escape(group))
        else:
            pattern = re.compile("^%s$" % re.escape(group))
        # assign id
        if not id:
            id = pattern.pattern + "." + str(time.time()) + str(random.random())
        id = str(id)
        # add subscription to db and indexes
        if reader:
            dst = str(reader)
        else:
            dst = sys._getframe(1).f_code.co_name
        self.queues.setdefault(id, Queue(dst=dst))
        self.subscriptions[id] = dict(reader=reader, group=group,
                pattern=pattern, id=id)
        self.patterns.setdefault(pattern, {})
        self.patterns[pattern][id] = True
        # catch up on messages in case we just now added reader 
        self._notify(id)
        return id

    def _notify(self, id):
        # send all queued messages to reader
        reader = self.subscriptions[id]['reader']
        if not reader:
            return
        queue = self.queues[id]
        while len(queue):
            msg = self.recv(id)
            reader(bus=self, group=msg.group, msg=msg)

    def send(self, group, msg):
        '''Called by anyone to deliver a message to all subscribers of
        group.  Senders need not be subscribed.  Group is a string,
        msg is a Msg object.'''
        grep_hit =  self.grep and re.match(self.grep, group)
        # check cache
        sendto = self.query_cache.get(group, None)
        if sendto is None:
            sendto = {}
            # cache miss -- slog through all subscriptions
            for pattern,id_dict in self.patterns.items():
                if re.match(pattern, group):
                    ids = id_dict.keys()
                    for id in ids:
                        queue = self.queues[id]
                        sendto[id] = queue
            # populate cache
            self.query_cache[group] = sendto
        if not sendto:
            # no readers found -- route to dead letter queue
            sendto[self.deadletter_id] = self.queues[self.deadletter_id]
        # XXX deprecate
        assert not msg.group
        msg.group = group
        if not msg.sender:
            # msg.sender = sys._getframe(1).f_code.co_name
            msg.sender = str(sys._getframe(1).f_code)
        sender = msg.sender
        if msg.cp:
            # use pickle instead of copy, because a dump into queue on
            # send and a load into reader on recv is usually faster than
            # multiple deep copies
            try:
                msg = dumps(msg)
            except Exception, e:
                raise ValueError, "object not picklable: %s: %s: %s" % (
                    type(msg.data), str(e), str(msg.data))
        # enqueue message
        for queue in sendto.values():
            if grep_hit:
                print "%s: %s -> %s" % (group, sender, queue.dst)
                if self.debug:
                    import pdb
                    pdb.set_trace()
            queue.send(msg)
        # notify readers
        for id in sendto.keys():
            self._notify(id)
        
    def recv(self, id, i=0, peek=False):
        '''Called by client to receive the next message in queue for
        subscription id.  Id must be the same as that
        returned by open().  Returns exactly one Msg object, or None
        if no messages are in queue.'''
        queue = self.queues.get(id, None)
        if not queue:
            return None
        msg = queue.recv(i, peek=peek)
        if not msg:
            return None
        if type(msg) is str:
            msg = loads(msg)
        return msg

    def peek(self, id, i=0):
        return self.recv(id, i, peek=True)

    # XXX replace with include/exclude or allow/deny
    def XXXshutdown(reader, group, func):
        '''Called by client to stop queueing messages into one queue.
        Bus will continue to serve messages from queue in response to
        recv(), but will not enqueue any new ones.  Queue will be
        destroyed after last message is dequeued.''' 

    def XXXdestroy(reader, group, func):
        '''Called by client to stop queueing messages into queue and
        remove queue immediately.  Any messages currently in queue
        will be destroyed.'''

    # XXX implement request/response pattern
    # def reader(self, bus, group, msg):
    #     foo = bus.rpc('get', Msg('foo')).data
    #     # or
    #     foo = bus.rpc.get('foo')

    # XXX enable bus.foo.bar.baz(msg)
    # XXX -- this is a bad way to do it; use a msg factory instead:
    # XXX factory.foo.bar.baz(data) generates message and calls bus.send()
    class Path(object):
        pass
    def XXX__call__(self, *args, **kwargs):
        pass
    def XXX__getattr__(self, name):
        val = self[name]
        return val
    def XXX__getitem__(self, name):
        """enable self['foo.bar']"""
        name = str(name)
        item, child = self._parts(name)
        val = super(Bus, self).get(item, None)
        if val is None:
            print item
            val = self[item] = Bus(name=item)
        if child:
            return val[child]
        return val 

class Queue(list):
    '''A publish/subscribe message channel for one reader (or a group
    of load-balancing readers).
    
    >>> g = Queue()
    >>> g.send('a')
    >>> g.send('b')
    >>> g.peek()
    'a'
    >>> g.peek(1)
    'b'
    >>> g.recv()
    'a'
    >>> g.recv()
    'b'
    >>> g.peek()
    
    '''

    def __init__(self, dst):
        self.dst=dst

    def send(self, msg):
        self.append(msg)

    def recv(self, i=0, peek=False):
        if len(self) -1 < i: 
            return None
        if peek:
            return self[i]
        else:
            return self.pop(i)

    def peek(self, i=0):
        return self.recv(i, peek=True)


class Msg(object):
    '''A basic message.'''

    def __init__(self, data, sender=None, ttl=None, cp=True, **kwargs):
        '''Create message.  Message payload is in data, envelope info
        is in other attributes.  Data must be a pickleable python
        object.  Sender is a unique string.  The cp flag is
        a signal to bus to make copies of this message on send and
        receive; if False, senders and receivers need to take extra
        care to not modify the message contents, lest they disturb
        peers who are also holding references to the same message.'''
        self.data = data
        self.sender = sender
        self.cp = cp
        self.time = time.time()
        if ttl:
            self.expires = self.time + ttl
        # destination group (channel) name 
        self.group = None  
        # group to post replies to
        self.reply_to = None  
        self.__dict__.update(kwargs)

    def __getitem__(self, name):
        return getattr(self, name)

"""
class ListMsg(object):
    '''A message with attributes useful for updating a list.'''

    def __init__(self, data, **kwargs):
        Msg.__init__(self, data, index, **kwargs)
        
        self.index = index






        # series type:
        #   list    payload is an element of a sequence 
        #   dict    payload is an element of a dict
        #   None    payload is a standalone scalar value
        self.type = None
        # series operation:
        #   reset   start a new data series; message payload supercedes 
        #           all prior messages posted to this group
        self.op = None
        # series key:
        #   x       add/replace element; use x as list index or dict key 
        #   None    append data to existing sequence, incrementing
        #           previous index value; message payload is a 
        #           continuation of prior message(s)
        self.key = None
        # string describing message sender
        self.__dict__.update(kwargs)

    def __getitem__(self, name):
        return getattr(self, name)

"""