On Nov 14, 2007 8:55 AM, Marty Alchin <[EMAIL PROTECTED]> wrote:
> I dunno, that's all just off the top of my head, but maybe it could work.
Well, I did a little digging and put together something that seems to
work. And when I say "seems to work", I mean that there are almost 100
lines of doctests for about 30 lines of code, and it all works as
expected. I expect there are some corner cases the tests don't cover
yet, but that's why I'm sending it around here instead of just
throwing it up on the wiki.
And no, I'm not making a ticket for it yet, because it occurred to me
while working on it, that it doesn't require any changes to
PyDispatcher itself. That means, Jeremy, that you can just drop this
somewhere and use it as I described earlier. Of course, if the rest of
you think it might be useful for other people, I'll see if I can track
down any corner cases that might break, and put it up on
djangosnippets or something.
Of course, it'd feel a bit cleaner if it were part of our PyDispatcher
distribution, since it reaches in and modifies a module-level variable
to make things work right. But, for the few who need it, I think it'd
be fine to have it as an external file.
Also, I apologize if Google Groups doesn't play nice with attachments.
I'll put it somewhere if that's a problem.
-Gul
--~--~---------~--~----~------------~-------~--~----~
You received this message because you are subscribed to the Google Groups
"Django developers" group.
To post to this group, send email to django-developers@googlegroups.com
To unsubscribe from this group, send email to [EMAIL PROTECTED]
For more options, visit this group at
http://groups.google.com/group/django-developers?hl=en
-~----------~----~----~----~------~----~------~--~---
"""
DispatchQueue: Stores dispatches in a queue to be sent later
# Set up some signals, a receiver and some senders
>>> sig_number, sig_letter = object(), object()
>>> def receiver(sender, arg): print sender.name, arg
>>> class SenderOne: name = 'One'
>>> class SenderTwo: name = 'Two'
# Catch all dispatches
>>> dispatcher.connect(receiver, sig_number)
>>> dispatcher.connect(receiver, sig_letter)
# Define a function which will send signals
>>> def test():
... dispatcher.send(sig_number, arg=1, sender=SenderOne)
... dispatcher.send(sig_letter, arg='a', sender=SenderOne)
... dispatcher.send(sig_number, arg=2, sender=SenderTwo)
... dispatcher.send(sig_letter, arg='b', sender=SenderTwo)
# Standard dispatch mechanism
>>> test()
One 1
One a
Two 2
Two b
# Queued dispatch should only catch those matching this signal
>>> queue = DispatchQueue(sig_number)
>>> test()
One a
Two b
# Dispatching the queue should send any stored dispatches
>>> queue.dispatch()
One 1
Two 2
# Make sure the standard mechanism is wholly restored
>>> test()
One 1
One a
Two 2
Two b
# Clean up a bit before further testing
>>> dispatcher.disconnect(receiver, sig_number)
>>> dispatcher.disconnect(receiver, sig_letter)
# Limiting by sender complicates things, but it shouldn't break queues
>>> dispatcher.connect(receiver, sig_number, sender=SenderOne)
>>> dispatcher.connect(receiver, sig_number, sender=SenderTwo)
>>> dispatcher.connect(receiver, sig_letter, sender=SenderOne)
>>> dispatcher.connect(receiver, sig_letter, sender=SenderTwo)
# Try it out without limiting by sender
>>> queue = DispatchQueue(sig_number)
>>> test()
One a
Two b
>>> queue.dispatch()
One 1
Two 2
>>> queue = DispatchQueue(sig_letter, sender=SenderOne)
>>> test()
One 1
Two 2
Two b
>>> queue.dispatch()
One a
# Try another one, just to be sure
>>> queue = DispatchQueue(sig_number, sender=SenderTwo)
>>> test()
One 1
One a
Two b
>>> queue.dispatch()
Two 2
# Make sure all this mucking didn't break the standard mechanism
>>> test()
One 1
One a
Two 2
Two b
# Stacked queues should work properly as well
>>> number_queue = DispatchQueue(sig_number)
>>> letter_queue = DispatchQueue(sig_letter, sender=SenderTwo)
# One letter should squeak through
>>> test()
One a
>>> letter_queue.dispatch()
Two b
# Number queue is still in effect
>>> test()
One a
Two b
>>> number_queue.dispatch()
One 1
Two 2
One 1
Two 2
# Once again, make sure the standard mechanism is intact
>>> test()
One 1
One a
Two 2
Two b
"""
from django.dispatch import dispatcher
class DispatchQueue:
"""A store of dispatched signals, waiting to be sent"""
def __init__(self, signal, sender=dispatcher.Any):
self.connections = {}
self.store = []
# Pull relevant receivers out of PyDispatcher for now
for snd, sigs in dispatcher.connections.items():
if sender is dispatcher.Any or id(sender) == snd:
self.connections[snd] = {}
for sig, receivers in sigs.items():
if sig is signal:
self.connections[snd][sig] = receivers
dispatcher.connections[snd][sig] = [self]
def __call__(self, *args, **kwargs):
"""Catch a dispatched signal and add it to the queue"""
self.store.append((args, kwargs))
def dispatch(self):
"""Send all queued signals, in order, as originally intended"""
# Put the captured listeners back where they belong
for snd, sigs in self.connections.items():
for sig, receivers in sigs.items():
dispatcher.connections[snd][sig] = receivers
# Dispatch the signals as intended
for args, kwargs in self.store:
dispatcher.send(*args, **kwargs)
if __name__ == '__main__':
import doctest
doctest.testmod()
