[Zope3-checkins] CVS: Zope3/src/zope/app/services - hub.py:1.7
Steve Alexander
steve@cat-box.net
Wed, 19 Mar 2003 13:05:34 -0500
Update of /cvs-repository/Zope3/src/zope/app/services
In directory cvs.zope.org:/tmp/cvs-serv29463/src/zope/app/services
Modified Files:
hub.py
Log Message:
Changed the hubIds service / ObjectHub implementation to use unicode
strings for paths, rather than tuples of unicode strings.
This is part of the ongoing work Albert and I are doing to standardize on
unicode strings for paths.
=== Zope3/src/zope/app/services/hub.py 1.6 => 1.7 ===
--- Zope3/src/zope/app/services/hub.py:1.6 Tue Mar 11 11:11:22 2003
+++ Zope3/src/zope/app/services/hub.py Wed Mar 19 13:05:03 2003
@@ -25,13 +25,13 @@
from zodb.btrees.IOBTree import IOBTree
from zodb.btrees.OIBTree import OIBTree
-from zope.app.traversing import getPhysicalPath
-from zope.app.traversing import locationAsTuple, locationAsUnicode
+#from zope.app.traversing import getPath, canonicalPath
+from zope.app.traversing import getPhysicalPathString as getPath
+from zope.app.traversing import locationAsUnicode as canonicalPath
+
from zope.component import getAdapter
from zope.exceptions import NotFoundError
-from zope.proxy.context import ContextWrapper
-from zope.proxy.context import isWrapper
-from zope.proxy.context import ContextMethod
+from zope.proxy.context import ContextWrapper, isWrapper, ContextMethod
from zope.proxy.introspection import removeAllProxies
from zope.app.interfaces.traversing import ITraverser
@@ -76,7 +76,7 @@
def __getLocation(self):
loc = self.__location
if loc is None:
- loc = self.__location = self.hub.getLocation(self.hubid)
+ loc = self.__location = self.hub.getPath(self.hubid)
return loc
location = property(__getLocation)
@@ -92,7 +92,6 @@
"""We are no longer interested in this object.
"""
-
hub = None
hubid = None
# object = None
@@ -163,8 +162,11 @@
else:
return abs
+def canonicalSlash(path):
+ # Return a canonical path, with a slash appended
+ return canonicalPath(path) + u'/'
-class ObjectHub(ServiceSubscriberEventChannel, ):
+class ObjectHub(ServiceSubscriberEventChannel):
# this implementation makes the decision to not interact with any
# object hubs above it: it is a world unto itself, as far as it is
@@ -178,21 +180,11 @@
def __init__(self):
ServiceSubscriberEventChannel.__init__(self)
- # int --> tuple of unicodes
- self.__hubid_to_location = IOBTree()
- # tuple of unicodes --> int
- self.__location_to_hubid = OIBTree()
-
- # XXX this is copied because of some context method problems
- # with moving LocalEventChannel.notify to this _notify via a simple
- # assignment, i.e. _notify = LocalEventChannel.notify
- #def _notify(clean_self, wrapped_self, event):
- # subscriptionsForEvent = clean_self._registry.getAllForObject(event)
- # for subscriptions in subscriptionsForEvent:
- # for subscriber, filter in subscriptions:
- # if filter is not None and not filter(event):
- # continue
- # ContextWrapper(subscriber, wrapped_self).notify(event)
+ # A pathslash is a path with a '/' on the end.
+ # int --> unicode pathslash
+ self.__hubid_to_path = IOBTree()
+ # unicode pathslash --> int
+ self.__path_to_hubid = OIBTree()
def notify(wrapped_self, event):
'''See interface ISubscriber'''
@@ -202,29 +194,24 @@
# generate NotificationHubEvents only if object is known
# ie registered
if IObjectMovedEvent.isImplementedBy(event):
- canonical_location = locationAsTuple(event.fromLocation)
- hubid = clean_self.__location_to_hubid.get(canonical_location)
+ pathslash = canonicalSlash(event.fromLocation)
+ hubid = clean_self.__path_to_hubid.get(pathslash)
if hubid is not None:
- canonical_new_location = locationAsTuple(
- event.location)
- location_to_hubid = clean_self.__location_to_hubid
- if location_to_hubid.has_key(canonical_new_location):
+ new_pathslash = canonicalSlash(event.location)
+ path_to_hubid = clean_self.__path_to_hubid
+ if path_to_hubid.has_key(new_pathslash):
raise ObjectHubError(
'Cannot move to location %s, '
'as there is already something there'
- % locationAsUnicode(canonical_new_location))
- hubid = location_to_hubid[canonical_location]
- del location_to_hubid[canonical_location]
- location_to_hubid[canonical_new_location] = hubid
- clean_self.__hubid_to_location[hubid] = (
- canonical_new_location)
+ % new_pathslash[:-1])
+ hubid = path_to_hubid[pathslash]
+ del path_to_hubid[pathslash]
+ path_to_hubid[new_pathslash] = hubid
+ clean_self.__hubid_to_path[hubid] = new_pathslash
# send out IObjectMovedHubEvent to plugins
event = ObjectMovedHubEvent(
- wrapped_self,
- hubid,
- canonical_location,
- canonical_new_location,
- event.object)
+ wrapped_self, hubid, pathslash[:-1],
+ new_pathslash[:-1], event.object)
clean_self._notify(wrapped_self, event)
elif IObjectCreatedEvent.isImplementedBy(event):
# a newly created object that has not been added to a
@@ -232,171 +219,154 @@
# it.
pass
else:
- canonical_location = locationAsTuple(event.location)
- hubid = clean_self.__location_to_hubid.get(canonical_location)
+ pathslash = canonicalSlash(event.location)
+ hubid = clean_self.__path_to_hubid.get(pathslash)
if hubid is not None:
if IObjectModifiedEvent.isImplementedBy(event):
# send out IObjectModifiedHubEvent to plugins
event = ObjectModifiedHubEvent(
- wrapped_self,
- hubid,
- canonical_location,
- event.object)
+ wrapped_self, hubid, pathslash[:-1], event.object)
clean_self._notify(wrapped_self, event)
elif IObjectRemovedEvent.isImplementedBy(event):
- del clean_self.__hubid_to_location[hubid]
- del clean_self.__location_to_hubid[canonical_location]
+ del clean_self.__hubid_to_path[hubid]
+ del clean_self.__path_to_hubid[pathslash]
# send out IObjectRemovedHubEvent to plugins
event = ObjectRemovedHubEvent(
- event.object,
- hubid,
- canonical_location,
- event.object)
+ event.object, hubid, pathslash[:-1], event.object)
clean_self._notify(wrapped_self, event)
notify = ContextMethod(notify)
- def getHubId(self, location):
+ def getHubId(self, path_or_object):
'''See interface ILocalObjectHub'''
- if isWrapper(location):
- location = getPhysicalPath(location)
+ if isinstance(path_or_object, (unicode, str)):
+ path = path_or_object
else:
- location = locationAsTuple(location)
- hubid = self.__location_to_hubid.get(location)
+ path = getPath(path_or_object)
+
+ pathslash = canonicalSlash(path)
+ hubid = self.__path_to_hubid.get(pathslash)
if hubid is None:
- raise NotFoundError(locationAsUnicode(location))
+ raise NotFoundError(path)
else:
return hubid
- def getLocation(self, hubid):
+ def getPath(self, hubid):
'''See interface IObjectHub'''
try:
- return self.__hubid_to_location[hubid]
+ return self.__hubid_to_path[hubid][:-1]
except KeyError:
raise NotFoundError(hubid)
def getObject(wrapped_self, hubid):
'''See interface IObjectHub'''
- location = wrapped_self.getLocation(hubid)
+ path = wrapped_self.getPath(hubid)
adapter = getAdapter(wrapped_self, ITraverser)
- return adapter.traverse(location)
+ return adapter.traverse(path)
getObject = ContextMethod(getObject)
- def register(wrapped_self, obj_or_loc):
+ def register(wrapped_self, path_or_object):
'''See interface ILocalObjectHub'''
clean_self = removeAllProxies(wrapped_self)
# XXX Need a new unit test for this; previously we tested
# whether it's wrapped, which is wrong because the root
# isn't wrapped (and it certainly makes sense to want to
# register the root).
- if isinstance(obj_or_loc, (str, unicode, tuple)):
+ if isinstance(path_or_object, (str, unicode)):
obj = None
- location = obj_or_loc
+ path = path_or_object
else:
- obj = obj_or_loc
- location = getPhysicalPath(obj_or_loc)
- canonical_location = locationAsTuple(location)
- if not canonical_location[0] == u'':
- raise ValueError("Location must be absolute")
-
- # This is here to make sure the 'registrations' method won't
- # trip up on using unichar ffff as a sentinel.
- for segment in canonical_location:
- if segment.startswith(u'\uffff'):
- raise ValueError(
- "Location contains a segment starting with \\uffff")
-
- location_to_hubid = clean_self.__location_to_hubid
- if location_to_hubid.has_key(canonical_location):
- # XXX It would be more convenient if register() returned
- # a bool indicating whether the object is already
- # registered, rather than raising an exception.
- # Then a more useful distinction between real errors
- # and this (common) condition could be made.
- raise ObjectHubError(
- 'location %s already in object hub' %
- locationAsUnicode(canonical_location))
- hubid = clean_self._generateHubId(canonical_location)
- location_to_hubid[canonical_location] = hubid
+ obj = path_or_object
+ path = getPath(path_or_object)
+
+ pathslash = canonicalSlash(path)
+
+ # XXX This check should be done by canonicalPath, but Albert is still
+ # refactoring that. So, I'll do it here for now.
+ if not pathslash.startswith(u'/'):
+ raise ValueError('Path must be absolute, not relative:', path)
+
+ path_to_hubid = clean_self.__path_to_hubid
+ if path_to_hubid.has_key(pathslash):
+ raise ObjectHubError('path %s already in object hub' % path)
+ hubid = clean_self._generateHubId(pathslash)
+ path_to_hubid[pathslash] = hubid
# send out IObjectRegisteredHubEvent to plugins
event = ObjectRegisteredHubEvent(
- wrapped_self,
- hubid,
- canonical_location,
- obj)
+ wrapped_self, hubid, pathslash[:-1], obj)
clean_self._notify(wrapped_self, event)
return hubid
register = ContextMethod(register)
- def unregister(wrapped_self, location):
+ def unregister(wrapped_self, path_or_object_or_hubid):
'''See interface ILocalObjectHub'''
clean_self = removeAllProxies(wrapped_self)
- if isWrapper(location):
- location = getPhysicalPath(location) # XXX this branch is
- # not exercised: needs unit test
- canonical_location = locationAsTuple(location)
- elif isinstance(location, int):
- canonical_location = clean_self.getLocation(location)
+ if isinstance(path_or_object_or_hubid, (unicode, str)):
+ path = canonicalPath(path_or_object_or_hubid)
+ elif isinstance(path_or_object_or_hubid, int):
+ path = clean_self.getPath(path_or_object_or_hubid)
else:
- canonical_location = locationAsTuple(location)
- location_to_hubid = clean_self.__location_to_hubid
- hubid_to_location = clean_self.__hubid_to_location
+ path = getPath(path_or_object_or_hubid)
+
+ pathslash = canonicalSlash(path)
+
+ path_to_hubid = clean_self.__path_to_hubid
+ hubid_to_path = clean_self.__hubid_to_path
try:
- hubid = location_to_hubid[canonical_location]
+ hubid = path_to_hubid[pathslash]
except KeyError:
- raise NotFoundError('location %s is not in object hub' %
- locationAsUnicode(canonical_location))
+ raise NotFoundError('path %s is not in object hub' % path)
else:
- del hubid_to_location[hubid]
- del location_to_hubid[canonical_location]
+ del hubid_to_path[hubid]
+ del path_to_hubid[pathslash]
# send out IObjectUnregisteredHubEvent to plugins
event = ObjectUnregisteredHubEvent(
- wrapped_self,
- hubid,
- canonical_location)
+ wrapped_self, hubid, pathslash[:-1])
clean_self._notify(wrapped_self, event)
unregister = ContextMethod(unregister)
def numRegistrations(self):
"""See interface IObjectHub"""
- # The hubid<-->location mappings should be the same size.
- # The IOBTree of hubid-->location might be faster to find the
+ # The hubid<-->path mapping should be the same size.
+ # The IOBTree of hubid-->path might be faster to find the
# size of, as the keys are ints. But, I haven't tested that.
- # assert len(self.__hubid_to_location)==len(self.__location_to_hubid)
- return len(self.__hubid_to_location)
+ # assert len(self.__hubid_to_path)==len(self.__path_to_hubid)
+ return len(self.__hubid_to_path)
- def getRegistrations(self, location=(u'',)):
+ def iterRegistrations(self, path=u'/'):
"""See interface IObjectHub"""
- # Location can be an ascii string a unicode or a tuple of strings
- # or unicodes. So, get a canonical location first of all.
- location = locationAsTuple(location)
- if location == (u'',):
+ # or unicodes. So, get a canonical path first of all.
+ pathslash = canonicalSlash(path)
+ if pathslash == u'//':
# Optimisation when we're asked for all the registered objects.
# Returns an IOBTreeItems object.
- return self.__location_to_hubid.items()
+ for pathslash, hubId in self.__path_to_hubid.iteritems():
+ yield pathslash[:-1], hubId
- # BTrees only support searches including the min and max.
- # So, I need to add to the end of the location a string that will
- # be larger than any other. I could also use a type that
- # sorts after unicodes.
- return self.__location_to_hubid.items(location, location+(u'\uffff',))
+ else:
+ # For a search /foo/bar, constrain the paths returned to those
+ # between /foo/bar/ and /foo/bar0, excluding /foo/bar0.
+ # chr(ord('/')+1) == '0'
+ for pathslash, hubId in self.__path_to_hubid.iteritems(
+ min=pathslash, max=pathslash[:-1]+u'0', excludemax=True):
+ yield pathslash[:-1], hubId
def iterObjectRegistrations(wrapped_self):
"""See interface IHubEventChannel"""
traverser = getAdapter(wrapped_self, ITraverser)
- for location, hubId in wrapped_self.getRegistrations():
- yield (location, hubId, traverser.traverse(location))
+ for path, hubId in wrapped_self.iterRegistrations():
+ yield (path, hubId, traverser.traverse(path))
iterObjectRegistrations = ContextMethod(iterObjectRegistrations)
############################################################
- def _generateHubId(self, location):
+ def _generateHubId(self, pathslash):
index = getattr(self, '_v_nextid', 0)
if index%4000 == 0:
index = randid()
- hubid_to_location = self.__hubid_to_location
- while not hubid_to_location.insert(index, location):
+ hubid_to_path = self.__hubid_to_path
+ while not hubid_to_path.insert(index, pathslash):
index = randid()
self._v_nextid = index + 1
return index