[Zope3-checkins] CVS: Zope3/src/zope/app/observable -
observers.py:1.1.2.1 observers.txt:1.1.2.1
observable.py:1.1.2.4 tests.py:1.1.2.2
Jim Fulton
jim at zope.com
Mon Mar 29 16:02:58 EST 2004
Update of /cvs-repository/Zope3/src/zope/app/observable
In directory cvs.zope.org:/tmp/cvs-serv22276/src/zope/app/observable
Modified Files:
Tag: observable-branch
observable.py tests.py
Added Files:
Tag: observable-branch
observers.py observers.txt
Log Message:
Factored out the data structure for keeping track of observers into a
separate module.
Wrote a number of new tests for the data structure.
Added an unsubscription method to the oberervers
=== Added File Zope3/src/zope/app/observable/observers.py ===
##############################################################################
#
# Copyright (c) 2004 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
"""Observer Registry
Observers observe other objects by getting notified of object events
on those objects. Observers subscribe to particular types of events.
>>> registry = Observers()
>>> import zope.interface
>>> class IR1(zope.interface.Interface):
... pass
>>> class IP1(zope.interface.Interface):
... pass
>>> class IP2(IP1):
... pass
>>> class IQ(zope.interface.Interface):
... pass
>>> registry.subscribe([IR1], IP2, 'sub12 1')
>>> registry.subscriptions([IR1], IP2)
['sub12 1']
You can have multiple subscribers for the same specification::
>>> registry.subscribe([IR1], IP2, 'sub12 2')
>>> subs = registry.subscriptions([IR1], IP2)
>>> subs.sort()
>>> subs
['sub12 1', 'sub12 2']
You can register subscribers for all specifications using None::
>>> class IR2(IR1):
... pass
>>> registry.subscribe([None], IP1, 'sub_1')
>>> subs = registry.subscriptions([IR2], IP1)
>>> subs.sort()
>>> subs
['sub12 1', 'sub12 2', 'sub_1']
Subscriptions may be combined over multiple compatible specifications::
>>> subs = registry.subscriptions([IR2], IP1)
>>> subs.sort()
>>> subs
['sub12 1', 'sub12 2', 'sub_1']
>>> registry.subscribe([IR1], IP1, 'sub11')
>>> subs = registry.subscriptions([IR2], IP1)
>>> subs.sort()
>>> subs
['sub11', 'sub12 1', 'sub12 2', 'sub_1']
>>> registry.subscribe([IR2], IP2, 'sub22')
>>> subs = registry.subscriptions([IR2], IP1)
>>> subs.sort()
>>> subs
['sub11', 'sub12 1', 'sub12 2', 'sub22', 'sub_1']
>>> subs = registry.subscriptions([IR2], IP2)
>>> subs.sort()
>>> subs
['sub12 1', 'sub12 2', 'sub22']
Subscriptions can be on multiple specifications::
>>> class IQ(zope.interface.Interface):
... pass
>>> registry.subscribe([IR1, IQ], IP2, 'sub1q2')
>>> registry.subscriptions([IR1, IQ], IP2)
['sub1q2']
As with single subscriptions, you can specify None for the first
required interface, to specify a default::
>>> registry.subscribe([None, IQ], IP2, 'sub_q2')
>>> class IS(zope.interface.Interface):
... pass
>>> registry.subscriptions([IS, IQ], IP2)
['sub_q2']
>>> subs = registry.subscriptions([IR1, IQ], IP2)
>>> subs.sort()
>>> subs
['sub1q2', 'sub_q2']
You can unsubscribe:
>>> registry.unsubscribe([IR1], IP2, 'sub12 2')
>>> subs = registry.subscriptions([IR2], IP1)
>>> subs.sort()
>>> subs
['sub11', 'sub12 1', 'sub22', 'sub_1']
"""
from persistent import Persistent
from zope.interface.adapter import Default
from zope.interface.adapter import Surrogate, AdapterRegistry
class LocalSurrogate(Surrogate):
"""Local surrogates
Local surrogates are transient, rather than persistent.
Their adapter data are stored in their registry objects.
"""
def __init__(self, spec, registry):
Surrogate.__init__(self, spec, registry)
self.registry = registry
def clean(self):
spec = self.spec()
ladapters = self.registry.adapters.get(spec)
if ladapters:
self.adapters = dict(
[((True, required, '', provided), subs)
for ((required, provided), subs) in ladapters.iteritems()
]
)
else:
self.adapters = {}
Surrogate.clean(self)
class Observers(AdapterRegistry, Persistent):
"""Local/persistent surrogate registry
"""
_surrogateClass = LocalSurrogate
def __init__(self):
self.adapters = {}
AdapterRegistry.__init__(self)
def __getstate__(self):
state = Persistent.__getstate__(self).copy()
del state['_surrogates']
del state['_default']
del state['_null']
del state['_remove']
return state
def __setstate__(self, state):
Persistent.__setstate__(self, state)
AdapterRegistry.__init__(self)
def subscribe(self, required, provided, subscriber):
if len(required) == 0:
raise ValueError("required can not be zero length")
akey = required[0]
if akey is None:
akey = Default
adapters = self.adapters.get(akey)
if not adapters:
adapters = self.adapters[akey] = {}
key = tuple(required[1:]), provided
adapters[key] = adapters.get(key, ()) + (subscriber, )
# reinitialize, thus clearing surrogates *and* marking as changed :)
AdapterRegistry.__init__(self)
def unsubscribe(self, required, provided, subscriber):
akey = required[0]
if akey is None:
akey = Default
adapters = self.adapters.get(akey)
if not adapters:
return
key = tuple(required[1:]), provided
subscribers = adapters.get(key, ())
if subscriber in subscribers:
subscribers = list(subscribers)
subscribers.remove(subscriber)
if subscribers:
adapters[key] = tuple(subscribers)
else:
del adapters[key]
# reinitialize, thus clearing surrogates *and* marking as changed
AdapterRegistry.__init__(self)
self._p_changed = True
=== Added File Zope3/src/zope/app/observable/observers.txt ===
=================
Observer Registry
=================
Observer registries provide a way to register observers that depend on
one or more interface specifications and provide (perhaps indirectly)
some interface.
The term "interface specification" refers both to interfaces and to
interface declarations, such as declarations of interfaces implemented
by a class.
>>> from zope.interface.adapter import AdapterRegistry
>>> import zope.interface
>>> class IR1(zope.interface.Interface):
... pass
>>> class IP1(zope.interface.Interface):
... pass
>>> class IP2(IP1):
... pass
>>> registry = AdapterRegistry()
>>> class IR2(IR1):
... pass
>>> registry.lookup([IR2], IP2, '')
12
>>> class C2:
... zope.interface.implements(IR2)
>>> class IP3(IP2):
... pass
Normally, we want to look up an object that most-closely matches a
specification. Sometimes, we want to get all of the objects that
match some specification. We use subscriptions for this. We
subscribe objects against specifications and then later find all of
the subscribed objects::
>>> registry.subscribe([IR1], IP2, 'sub12 1')
>>> registry.subscriptions([IR1], IP2)
['sub12 1']
Note that, unlike regular adapters, subscriptions are unnamed.
The order of returned subscriptions is not specified.
You can have multiple subscribers for the same specification::
>>> registry.subscribe([IR1], IP2, 'sub12 2')
>>> subs = registry.subscriptions([IR1], IP2)
>>> subs.sort()
>>> subs
['sub12 1', 'sub12 2']
You can register subscribers for all specifications using None::
>>> registry.subscribe([None], IP1, 'sub_1')
>>> subs = registry.subscriptions([IR2], IP1)
>>> subs.sort()
>>> subs
['sub12 1', 'sub12 2', 'sub_1']
Subscriptions may be combined over multiple compatible specifications::
>>> subs = registry.subscriptions([IR2], IP1)
>>> subs.sort()
>>> subs
['sub12 1', 'sub12 2', 'sub_1']
>>> registry.subscribe([IR1], IP1, 'sub11')
>>> subs = registry.subscriptions([IR2], IP1)
>>> subs.sort()
>>> subs
['sub11', 'sub12 1', 'sub12 2', 'sub_1']
>>> registry.subscribe([IR2], IP2, 'sub22')
>>> subs = registry.subscriptions([IR2], IP1)
>>> subs.sort()
>>> subs
['sub11', 'sub12 1', 'sub12 2', 'sub22', 'sub_1']
>>> subs = registry.subscriptions([IR2], IP2)
>>> subs.sort()
>>> subs
['sub12 1', 'sub12 2', 'sub22']
Subscriptions can be on multiple specifications::
>>> registry.subscribe([IR1, IQ], IP2, 'sub1q2')
>>> registry.subscriptions([IR1, IQ], IP2)
['sub1q2']
As with single subscriptions and non-subscription adapters, you can
specify None for the first required interface, to specify a default::
>>> registry.subscribe([None, IQ], IP2, 'sub_q2')
>>> registry.subscriptions([IS, IQ], IP2)
['sub_q2']
>>> subs = registry.subscriptions([IR1, IQ], IP2)
>>> subs.sort()
>>> subs
['sub1q2', 'sub_q2']
You can have subscriptions that are indepenent of any specifications::
>>> registry.subscriptions([], IP1)
[]
>>> registry.subscribe([], IP2, 'sub2')
>>> registry.subscriptions([], IP1)
['sub2']
>>> registry.subscribe([], IP1, 'sub1')
>>> subs = registry.subscriptions([], IP1)
>>> subs.sort()
>>> subs
['sub1', 'sub2']
>>> registry.subscriptions([], IP2)
['sub2']
=== Zope3/src/zope/app/observable/observable.py 1.1.2.3 => 1.1.2.4 ===
--- Zope3/src/zope/app/observable/observable.py:1.1.2.3 Tue Mar 23 16:24:19 2004
+++ Zope3/src/zope/app/observable/observable.py Mon Mar 29 16:02:57 2004
@@ -15,39 +15,13 @@
$Id$
"""
-
from zope.interface import implements, providedBy
from zope.app.observable.interfaces import IObservable
from zope.app.annotation.interfaces import IAnnotations
-from zope.app.adapter.adapter import LocalAdapterRegistry
+from zope.app.observable.observers import Observers
key = 'zope.app.observable'
-class DummyBase:
-
- adapters = {}
-
- def subscribe(self, arg):
- pass
-
-class ObjectAdapterRegistry(LocalAdapterRegistry):
-
- dummybase = DummyBase()
-
- def baseFor(self, spec):
- return self.dummybase
-
- def subscribe(self, required, provided, subscriber):
- if len(required) == 0:
- raise ValueError("required can not be zero length")
-
- adapters = self.adapters.get(required[0])
- if not adapters:
- adapters = self.adapters[required[0]] = {}
- key = True, tuple(required[1:]), '', provided
- adapters[key] = adapters.get(key, ()) + (subscriber, )
- self._p_changed = True
-
class ObservableAdapter:
implements(IObservable)
@@ -60,7 +34,7 @@
registry = annotations.get(key)
if registry is None:
- annotations[key] = registry = ObjectAdapterRegistry(DummyBase())
+ annotations[key] = registry = Observers()
registry.subscribe(required, provided, subscriber)
=== Zope3/src/zope/app/observable/tests.py 1.1.2.1 => 1.1.2.2 ===
--- Zope3/src/zope/app/observable/tests.py:1.1.2.1 Tue Mar 23 16:23:52 2004
+++ Zope3/src/zope/app/observable/tests.py Mon Mar 29 16:02:57 2004
@@ -82,7 +82,8 @@
def test_suite():
import sys
return unittest.TestSuite((
- doctest.DocTestSuite(sys.modules[__name__]),
+ doctest.DocTestSuite(),
+ doctest.DocTestSuite('zope.app.observable.observers'),
))
if __name__ == '__main__':
More information about the Zope3-Checkins
mailing list