[CalendarServer-changes] [2473] CalendarServer/trunk
source_changes at macosforge.org
source_changes at macosforge.org
Fri May 23 17:57:55 PDT 2008
Revision: 2473
http://trac.macosforge.org/projects/calendarserver/changeset/2473
Author: dreid at apple.com
Date: 2008-05-23 17:57:54 -0700 (Fri, 23 May 2008)
Log Message:
-----------
Merge unified-cache, with memcached.
Modified Paths:
--------------
CalendarServer/trunk/run
CalendarServer/trunk/twistedcaldav/cache.py
CalendarServer/trunk/twistedcaldav/cluster.py
CalendarServer/trunk/twistedcaldav/config.py
CalendarServer/trunk/twistedcaldav/directory/calendaruserproxy.py
CalendarServer/trunk/twistedcaldav/directory/principal.py
CalendarServer/trunk/twistedcaldav/directory/test/test_principal.py
CalendarServer/trunk/twistedcaldav/directory/test/test_sudo.py
CalendarServer/trunk/twistedcaldav/extensions.py
CalendarServer/trunk/twistedcaldav/itip.py
CalendarServer/trunk/twistedcaldav/method/delete.py
CalendarServer/trunk/twistedcaldav/method/put_common.py
CalendarServer/trunk/twistedcaldav/resource.py
CalendarServer/trunk/twistedcaldav/root.py
CalendarServer/trunk/twistedcaldav/static.py
CalendarServer/trunk/twistedcaldav/test/test_cache.py
CalendarServer/trunk/twistedcaldav/test/test_resource.py
CalendarServer/trunk/twistedcaldav/test/test_static.py
CalendarServer/trunk/twistedcaldav/test/util.py
Added Paths:
-----------
CalendarServer/trunk/twistedcaldav/memcache.py
CalendarServer/trunk/twistedcaldav/test/test_memcache.py
Modified: CalendarServer/trunk/run
===================================================================
--- CalendarServer/trunk/run 2008-05-24 00:47:17 UTC (rev 2472)
+++ CalendarServer/trunk/run 2008-05-24 00:57:54 UTC (rev 2473)
@@ -518,6 +518,42 @@
fi;
#
+# libevent
+#
+libevent="${top}/libevent-1.4.4-stable"
+
+www_get "libevent" "${libevent}" http://www.monkey.org/~provos/libevent-1.4.4-stable.tar.gz
+
+if ! "${disable_setup}" && (
+ "${force_setup}" || [ ! -d "${libevent}/_root" ]
+); then
+ echo "";
+ echo "Building libevent...";
+ cd "${libevent}";
+ ./configure --prefix="${libevent}/_root";
+ make;
+ make install;
+fi;
+
+#
+# memcached
+#
+memcached="${top}/memcached-1.2.5"
+
+www_get "memcached" "${memcached}" http://www.danga.com/memcached/dist/memcached-1.2.5.tar.gz;
+
+if ! "${disable_setup}" && (
+ "${force_setup}" || [ ! -d "${memcached}/_root" ]
+); then
+ echo "";
+ echo "Building memcached...";
+ cd "${memcached}";
+ ./configure --prefix="${memcached}/_root" --with-libevent="${libevent}/_root";
+ make;
+ make install;
+fi;
+
+#
# Twisted
#
Modified: CalendarServer/trunk/twistedcaldav/cache.py
===================================================================
--- CalendarServer/trunk/twistedcaldav/cache.py 2008-05-24 00:47:17 UTC (rev 2472)
+++ CalendarServer/trunk/twistedcaldav/cache.py 2008-05-24 00:57:54 UTC (rev 2473)
@@ -17,16 +17,21 @@
import uuid
import time
import os
+import hashlib
+import cPickle
from zope.interface import implements
from twisted.python.filepath import FilePath
-from twisted.python import log
+from twisted.internet.defer import succeed, fail
+from twisted.internet.protocol import ClientCreator
+
from twisted.web2.iweb import IResource
from twisted.web2.dav import davxml
from twisted.web2.dav.util import allDataFromStream
from twisted.web2.http import HTTPError, Response
+from twisted.web2.http_headers import Headers
from twisted.web2.stream import MemoryStream
from twisted.web2.dav.xattrprops import xattrPropertyStore
@@ -34,57 +39,126 @@
from twisted.internet.threads import deferToThread
from twistedcaldav.log import LoggingMixIn
+from twistedcaldav.memcache import MemCacheProtocol
+from twistedcaldav.config import config
-
class CacheTokensProperty(davxml.WebDAVTextElement):
namespace = davxml.twisted_private_namespace
name = "cacheTokens"
-class CacheChangeNotifier(LoggingMixIn):
+class XattrCacheChangeNotifier(LoggingMixIn):
def __init__(self, propertyStore):
self._propertyStore = propertyStore
self._token = None
def _newCacheToken(self):
- return uuid.uuid4()
+ return str(uuid.uuid4())
def changed(self):
+ """
+ Change the cache token for a resource.
+
+ return: A L{Deferred} that fires when the token has been changed.
+ """
self.log_debug("Changing Cache Token for %r" % (
- self._propertyStore))
+ self._propertyStore.resource.fp))
property = CacheTokensProperty.fromString(self._newCacheToken())
self._propertyStore.set(property)
+ return succeed(True)
+class MemcacheChangeNotifier(LoggingMixIn):
+ _memcacheProtocol = None
-class ResponseCache(LoggingMixIn):
- """
- An object that caches responses to given requests.
+ def __init__(self, propertyStore):
+ self._path = propertyStore.resource.fp.path
+ self._host = config.Memcached['BindAddress']
+ self._port = config.Memcached['Port']
- @ivar CACHE_TIMEOUT: The number of seconds that a cache entry is valid,
- (default 3600 seconds or 1 hour).
+ from twisted.internet import reactor
+ self._reactor = reactor
- @ivar _docroot: An L{FilePath} that points to the document root.
- @ivar _responses: A C{dict} with (request-method, request-uri,
- principal-uri) keys and (principal-token, uri-token, cache-time,
- response) values.
+
+ def _newCacheToken(self):
+ return str(uuid.uuid4())
+
+
+ def _getMemcacheProtocol(self):
+ if MemcacheChangeNotifier._memcacheProtocol is not None:
+ return succeed(self._memcacheProtocol)
+
+ d = ClientCreator(self._reactor, MemCacheProtocol).connectTCP(
+ self._host,
+ self._port)
+
+ def _cacheProtocol(proto):
+ MemcacheChangeNotifier._memcacheProtocol = proto
+ return proto
+
+ return d.addCallback(_cacheProtocol)
+
+
+ def changed(self):
+ """
+ Change the cache token for a resource
+
+ return: A L{Deferred} that fires when the token has been changed.
+ """
+ def _updateCacheToken(proto):
+ return proto.set('cacheToken:%s' % (self._path,),
+ self._newCacheToken())
+
+ self.log_debug("Changing Cache Token for %r" % (self._path,))
+ d = self._getMemcacheProtocol()
+ d.addCallback(_updateCacheToken)
+ return d
+
+
+
+class BaseResponseCache(LoggingMixIn):
"""
-
- CACHE_SIZE = 1000
+ A base class which provides some common operations
+ """
propertyStoreFactory = xattrPropertyStore
- def __init__(self, docroot, cacheSize=None):
- self._docroot = docroot
- self._responses = {}
+ def _principalURI(self, principal):
+ return str(principal.children[0])
- if cacheSize is not None:
- self.CACHE_SIZE = cacheSize
+ def _requestKey(self, request):
+ def _getKey(requestBody):
+ if requestBody is not None:
+ request.stream = MemoryStream(requestBody)
+ request.stream.doStartReading = None
+ request.cacheKey = (request.method,
+ self._principalURI(request.authnUser),
+ request.uri,
+ request.headers.getHeader('depth'),
+ hash(requestBody))
+
+ return request.cacheKey
+
+ d = allDataFromStream(request.stream)
+ d.addCallback(_getKey)
+ return d
+
+
+ def _getTokensInThread(self, principalURI, requestURI):
+ def _getTokens():
+ pToken = self._tokenForURI(principalURI)
+ uToken = self._tokenForURI(requestURI)
+
+ return (pToken, uToken)
+
+ return deferToThread(_getTokens)
+
+
def _tokenForURI(self, uri):
"""
Get a property store for the given C{uri}.
@@ -117,10 +191,36 @@
pass
- def _principalURI(self, principal):
- return str(principal.children[0])
+ def _getResponseBody(self, key, response):
+ d1 = allDataFromStream(response.stream)
+ d1.addCallback(lambda responseBody: (key, responseBody))
+ return d1
+
+class ResponseCache(BaseResponseCache):
+ """
+ An object that caches responses to given requests.
+
+ @ivar CACHE_TIMEOUT: The number of seconds that a cache entry is valid,
+ (default 3600 seconds or 1 hour).
+
+ @ivar _docroot: An L{FilePath} that points to the document root.
+ @ivar _responses: A C{dict} with (request-method, request-uri,
+ principal-uri) keys and (principal-token, uri-token, cache-time,
+ response) values.
+ """
+
+ CACHE_SIZE = 1000
+
+ def __init__(self, docroot, cacheSize=None):
+ self._docroot = docroot
+ self._responses = {}
+
+ if cacheSize is not None:
+ self.CACHE_SIZE = cacheSize
+
+
def _time(self):
"""
Return the current time in seconds since the epoch
@@ -138,13 +238,8 @@
@return: An L{IResponse} or C{None} if the response has not been cached.
"""
- def _getTokens(pURI, rURI):
- pToken = self._tokenForURI(pURI)
- uToken = self._tokenForURI(rURI)
+ principalURI = self._principalURI(request.authnUser)
- return (pToken, uToken)
-
-
def _checkTokens((newPrincipalToken, newURIToken), key):
(principalToken,
uriToken,
@@ -184,37 +279,20 @@
return responseObj
- def _returnRequest(requestBody):
-
- if requestBody is not None:
- request.stream = MemoryStream(requestBody)
- request.stream.doStartReading = None
-
- principalURI = self._principalURI(request.authnUser)
-
- key = (request.method,
- request.uri,
- principalURI,
- request.headers.getHeader('depth'),
- hash(requestBody))
-
+ def _checkKeyInCache(key):
self.log_debug("Checking cache for: %r" % (key,))
- request.cacheKey = key
-
if key not in self._responses:
self.log_debug("Not in cache: %r" % (key,))
return None
- d1 = deferToThread(_getTokens,
- principalURI,
- request.uri)
+ d1 = self._getTokensInThread(principalURI, request.uri)
d1.addCallback(_checkTokens, key)
return d1
- d = allDataFromStream(request.stream)
- d.addCallback(_returnRequest)
+ d = self._requestKey(request)
+ d.addCallback(_checkKeyInCache)
return d
@@ -231,27 +309,9 @@
@return: A deferred that fires when the response has been added
to the cache.
"""
- def _getRequestBody(responseBody):
- d1 = allDataFromStream(request.stream)
- d1.addCallback(lambda requestBody: (requestBody, responseBody))
- return d1
-
- def _cacheResponse((requestBody, responseBody)):
- if requestBody is not None:
- request.stream = MemoryStream(requestBody)
- request.stream.doStartReading = None
-
+ def _cacheResponse((key, responseBody)):
principalURI = self._principalURI(request.authnUser)
- if hasattr(request, 'cacheKey'):
- key = request.cacheKey
- else:
- key = (request.method,
- request.uri,
- principalURI,
- request.headers.getHeader('depth'),
- hash(requestBody))
-
self.log_debug("Adding to cache: %r = %r" % (key,
response))
@@ -287,14 +347,186 @@
response.stream = MemoryStream(responseBody)
return response
+ if hasattr(request, 'cacheKey'):
+ request.cacheKey
+ d = succeed(request.cacheKey)
+ else:
+ d = self._requestKey(request)
- d = allDataFromStream(response.stream)
- d.addCallback(_getRequestBody)
+ d.addCallback(self._getResponseBody, response)
d.addCallback(_cacheResponse)
return d
+class MemcacheResponseCache(BaseResponseCache):
+ def __init__(self, docroot, host, port, reactor=None):
+ self._docroot = docroot
+ self._host = host
+ self._port = port
+ if reactor is None:
+ from twisted.internet import reactor
+
+ self._reactor = reactor
+
+ self._memcacheProtocol = None
+
+
+ def _tokenForURI(self, uri):
+ """
+ Get a property store for the given C{uri}.
+
+ @param uri: The URI we'd like the token for.
+ @return: A C{str} representing the token for the URI.
+ """
+
+ class __FauxStaticResource(object):
+ def __init__(self, fp):
+ self.fp = fp
+
+
+ fp = self._docroot
+ for childPath in uri.split('/')[:4]:
+ fp = fp.child(childPath)
+
+ return self._getMemcacheProtocol().addCallback(
+ lambda p: p.get('cacheToken:%s' % (fp.path,)))
+
+
+ def _getTokens(self, principalURI, requestURI):
+ def _getSecondToken(pToken):
+ d1 = self._tokenForURI(requestURI)
+ d1.addCallback(lambda uToken: (pToken, uToken))
+ return d1
+
+ d = self._tokenForURI(principalURI)
+ d.addCallback(_getSecondToken)
+ return d
+
+
+ def _getMemcacheProtocol(self):
+ if self._memcacheProtocol is not None:
+ return succeed(self._memcacheProtocol)
+
+ d = ClientCreator(self._reactor, MemCacheProtocol).connectTCP(
+ self._host,
+ self._port)
+
+ def _cacheProtocol(proto):
+ self._memcacheProtocol = proto
+ return proto
+
+ return d.addCallback(_cacheProtocol)
+
+
+ def _hashedRequestKey(self, request):
+ def _hashKey(key):
+ oldkey = key
+ request.cacheKey = key = hashlib.md5(
+ ':'.join([str(t) for t in key])).hexdigest()
+ self.log_debug("hashing key for get: %r to %r" % (oldkey, key))
+ return request.cacheKey
+
+ d = self._requestKey(request)
+ d.addCallback(_hashKey)
+ return d
+
+
+ def getResponseForRequest(self, request):
+ def _checkTokens(curTokens, expectedTokens, (code, headers, body)):
+ if curTokens[0] != expectedTokens[0]:
+ self.log_debug(
+ "Principal token doesn't match for %r: %r != %r" % (
+ request.cacheKey,
+ curTokens[0],
+ expectedTokens[0]))
+ return None
+
+ if curTokens[1] != expectedTokens[1]:
+ self.log_debug(
+ "URI token doesn't match for %r: %r != %r" % (
+ request.cacheKey,
+ curTokens[1],
+ expectedTokens[1]))
+ return None
+
+ r = Response(code,
+ stream=MemoryStream(body))
+
+ for key, value in headers.iteritems():
+ r.headers.setRawHeaders(key, value)
+
+ return r
+
+ def _unpickleResponse((flags, value), key):
+ if value is None:
+ self.log_debug("Not in cache: %r" % (key,))
+ return None
+
+ self.log_debug("Found in cache: %r = %r" % (key, value))
+
+ (principalToken, uriToken,
+ resp) = cPickle.loads(value)
+
+ d2 = self._getTokens(self._principalURI(request.authnUser),
+ request.uri)
+
+ d2.addCallback(_checkTokens, (principalToken, uriToken), resp)
+
+ return d2
+
+ def _getCache(proto, key):
+ self.log_debug("Checking cache for: %r" % (key,))
+ d1 = proto.get(key)
+ return d1.addCallback(_unpickleResponse, key)
+
+ def _getProtocol(key):
+ return self._getMemcacheProtocol().addCallback(_getCache, key)
+
+ d = self._hashedRequestKey(request)
+ d.addCallback(_getProtocol)
+ return d
+
+
+ def cacheResponseForRequest(self, request, response):
+ def _setCacheEntry(proto, key, cacheEntry):
+ self.log_debug("Adding to cache: %r = %r" % (key, cacheEntry))
+ return proto.set(key, cacheEntry).addCallback(
+ lambda _: response)
+
+ def _makeCacheEntry((pToken, uToken), (key, responseBody)):
+ cacheEntry = cPickle.dumps(
+ (pToken,
+ uToken,
+ (response.code,
+ dict(list(response.headers.getAllRawHeaders())),
+ responseBody)))
+
+ d2 = self._getMemcacheProtocol()
+ d2.addCallback(_setCacheEntry, key, cacheEntry)
+ return d2
+
+ def _cacheResponse((key, responseBody)):
+ principalURI = self._principalURI(request.authnUser)
+
+ response.headers.removeHeader('date')
+ response.stream = MemoryStream(responseBody)
+
+ d1 = self._getTokens(principalURI, request.uri)
+ d1.addCallback(_makeCacheEntry, (key, responseBody))
+ return d1
+
+ if hasattr(request, 'cacheKey'):
+ d = succeed(request.cacheKey)
+ else:
+ d = self._hashedRequestKey(request)
+
+ d.addCallback(self._getResponseBody, response)
+ d.addCallback(_cacheResponse)
+ return d
+
+
+
class _CachedResponseResource(object):
implements(IResource)
@@ -308,6 +540,7 @@
return self, []
+
class PropfindCacheMixin(object):
def http_PROPFIND(self, request):
def _cacheResponse(responseCache, response):
Modified: CalendarServer/trunk/twistedcaldav/cluster.py
===================================================================
--- CalendarServer/trunk/twistedcaldav/cluster.py 2008-05-24 00:47:17 UTC (rev 2472)
+++ CalendarServer/trunk/twistedcaldav/cluster.py 2008-05-24 00:57:54 UTC (rev 2473)
@@ -102,6 +102,10 @@
'-o', 'MultiProcess/ProcessCount=%d' % (
config.MultiProcess['ProcessCount'],)])
+ if config.Memcached["ServerEnabled"]:
+ args.extend(
+ ['-o', 'Memcached/ClientEnabled=True'])
+
if self.ports:
args.extend([
'-o',
@@ -287,6 +291,24 @@
fname],
env=parentEnv)
+
+ if config.Memcached["ServerEnabled"]:
+ log.msg("Adding memcached service")
+
+ memcachedArgv = [
+ config.Memcached["memcached"],
+ '-p', str(config.Memcached["Port"]),
+ '-l', config.Memcached["BindAddress"]]
+
+ if config.Memcached["MaxMemory"] is not 0:
+ memcachedArgv.extend([
+ '-m', str(config.Memcached["MaxMemory"])])
+
+ memcachedArgv.extend(config.Memcached["Options"])
+
+ monitor.addProcess('memcached', memcachedArgv, env=parentEnv)
+
+
logger = AMPLoggingFactory(
RotatingFileAccessLoggingObserver(config.AccessLogFile))
Modified: CalendarServer/trunk/twistedcaldav/config.py
===================================================================
--- CalendarServer/trunk/twistedcaldav/config.py 2008-05-24 00:47:17 UTC (rev 2472)
+++ CalendarServer/trunk/twistedcaldav/config.py 2008-05-24 00:57:54 UTC (rev 2473)
@@ -202,8 +202,18 @@
},
"ThreadPoolSize": 10,
-
+
"ListenBacklog": 50,
+
+ "Memcached": {
+ "ClientEnabled": False,
+ "ServerEnabled": False,
+ "BindAddress": "127.0.0.1",
+ "Port": 11211,
+ "memcached": "/usr/share/caldavd/bin/memcached",
+ "MaxMemory": 0, # Megabytes
+ "Options": [],
+ },
}
class Config (object):
Modified: CalendarServer/trunk/twistedcaldav/directory/calendaruserproxy.py
===================================================================
--- CalendarServer/trunk/twistedcaldav/directory/calendaruserproxy.py 2008-05-24 00:47:17 UTC (rev 2472)
+++ CalendarServer/trunk/twistedcaldav/directory/calendaruserproxy.py 2008-05-24 00:57:54 UTC (rev 2473)
@@ -22,7 +22,7 @@
"CalendarUserProxyPrincipalResource",
]
-from twisted.internet.defer import succeed
+from twisted.internet.defer import succeed, inlineCallbacks
from twisted.web2 import responsecode
from twisted.web2.dav import davxml
from twisted.web2.dav.element.base import dav_namespace
@@ -154,6 +154,7 @@
return super(CalendarUserProxyPrincipalResource, self).writeProperty(property, request)
+ @inlineCallbacks
def setGroupMemberSet(self, new_members, request):
# FIXME: as defined right now it is not possible to specify a calendar-user-proxy group as
# a member of any other group since the directory service does not know how to lookup
@@ -178,14 +179,15 @@
"Attempt to use a non-existent principal %s as a group member of %s." % (uri, self.principalURL(),)
))
principals.append(principal)
- principal.cacheNotifier.changed()
+ changed = yield principal.cacheNotifier.changed()
# Map the principals to UIDs.
uids = [p.principalUID() for p in principals]
self._index().setGroupMembers(self.uid, uids)
- self.parent.cacheNotifier.changed()
- return succeed(True)
+ changed = yield self.parent.cacheNotifier.changed()
+ yield True
+ return
##
# HTTP
Modified: CalendarServer/trunk/twistedcaldav/directory/principal.py
===================================================================
--- CalendarServer/trunk/twistedcaldav/directory/principal.py 2008-05-24 00:47:17 UTC (rev 2472)
+++ CalendarServer/trunk/twistedcaldav/directory/principal.py 2008-05-24 00:57:54 UTC (rev 2473)
@@ -40,7 +40,7 @@
from twisted.web2.dav.util import joinURL
from twistedcaldav.config import config
-from twistedcaldav.cache import CacheChangeNotifier, PropfindCacheMixin
+from twistedcaldav.cache import XattrCacheChangeNotifier, PropfindCacheMixin
from twistedcaldav.directory.calendaruserproxy import CalendarUserProxyDatabase
from twistedcaldav.directory.calendaruserproxy import CalendarUserProxyPrincipalResource
@@ -361,7 +361,7 @@
"""
Directory principal resource.
"""
- cacheNotifierFactory = CacheChangeNotifier
+ cacheNotifierFactory = XattrCacheChangeNotifier
def __init__(self, path, parent, record):
"""
Modified: CalendarServer/trunk/twistedcaldav/directory/test/test_principal.py
===================================================================
--- CalendarServer/trunk/twistedcaldav/directory/test/test_principal.py 2008-05-24 00:47:17 UTC (rev 2472)
+++ CalendarServer/trunk/twistedcaldav/directory/test/test_principal.py 2008-05-24 00:57:54 UTC (rev 2473)
@@ -40,7 +40,7 @@
from twistedcaldav.directory.principal import DirectoryPrincipalResource
from twistedcaldav.directory.principal import DirectoryCalendarPrincipalResource
-from twistedcaldav.cache import CacheChangeNotifier
+from twistedcaldav.cache import XattrCacheChangeNotifier
import twistedcaldav.test.util
@@ -231,11 +231,11 @@
def test_cacheNotifier(self):
"""
Each DirectoryPrincipalResource should have a cacheNotifier attribute
- that is an instance of CacheChangeNotifier
+ that is an instance of XattrCacheChangeNotifier
"""
for provisioningResource, recordType, recordResource, record in self._allRecords():
self.failUnless(isinstance(recordResource.cacheNotifier,
- CacheChangeNotifier))
+ XattrCacheChangeNotifier))
def test_displayName(self):
"""
Modified: CalendarServer/trunk/twistedcaldav/directory/test/test_sudo.py
===================================================================
--- CalendarServer/trunk/twistedcaldav/directory/test/test_sudo.py 2008-05-24 00:47:17 UTC (rev 2472)
+++ CalendarServer/trunk/twistedcaldav/directory/test/test_sudo.py 2008-05-24 00:57:54 UTC (rev 2473)
@@ -66,23 +66,6 @@
record = service.recordWithShortName(self.recordType, 'bob')
self.failIf(record)
- def test_recordChanges(self):
- service = self.service()
-
- record = service.recordWithShortName(self.recordType, 'alice')
- self.assertEquals(record.password, 'alice')
-
- record = service.recordWithShortName(self.recordType, 'bob')
- self.failIf(record)
-
- plistFile2.copyTo(self._plistFile)
-
- record = service.recordWithShortName(self.recordType, 'alice')
- self.assertEquals(record.password, 'alice')
-
- record = service.recordWithShortName(self.recordType, 'bob')
- self.assertEquals(record.password, 'bob')
-
def test_calendaringDisabled(self):
service = self.service()
Modified: CalendarServer/trunk/twistedcaldav/extensions.py
===================================================================
--- CalendarServer/trunk/twistedcaldav/extensions.py 2008-05-24 00:47:17 UTC (rev 2472)
+++ CalendarServer/trunk/twistedcaldav/extensions.py 2008-05-24 00:57:54 UTC (rev 2473)
@@ -34,6 +34,7 @@
import time
from twisted.internet.defer import succeed, deferredGenerator, waitForDeferred
+from twisted.internet.defer import maybeDeferred
from twisted.web2 import responsecode
from twisted.web2.http import HTTPError, Response, RedirectResponse
from twisted.web2.http_headers import MimeType
@@ -185,10 +186,42 @@
authorizationPrincipal = deferredGenerator(authorizationPrincipal)
+def updateCacheTokenOnCallback(f):
+ def fun(self, *args, **kwargs):
+ def _updateToken(response):
+ return self.cacheNotifier.changed().addCallback(
+ lambda _: response)
+
+ d = maybeDeferred(f, self, *args, **kwargs)
+
+ if hasattr(self, 'cacheNotifier'):
+ d.addCallback(_updateToken)
+
+ return d
+
+ return fun
+
+
class DAVResource (SudoSACLMixin, SuperDAVResource, LoggingMixIn):
"""
Extended L{twisted.web2.dav.resource.DAVResource} implementation.
"""
+
+ @updateCacheTokenOnCallback
+ def http_PROPPATCH(self, request):
+ return super(DAVResource, self).http_PROPPATCH(request)
+
+
+ @updateCacheTokenOnCallback
+ def http_DELETE(self, request):
+ return super(DAVResource, self).http_DELETE(request)
+
+
+ @updateCacheTokenOnCallback
+ def http_ACL(self, request):
+ return super(DAVResource, self).http_ACL(request)
+
+
def findChildrenFaster(self, depth, request, okcallback, badcallback, names, privileges, inherited_aces):
"""
See L{IDAVResource.findChildren}.
Modified: CalendarServer/trunk/twistedcaldav/itip.py
===================================================================
--- CalendarServer/trunk/twistedcaldav/itip.py 2008-05-24 00:47:17 UTC (rev 2472)
+++ CalendarServer/trunk/twistedcaldav/itip.py 2008-05-24 00:57:54 UTC (rev 2473)
@@ -719,9 +719,7 @@
def _deletedResourced(result):
# Change CTag on the parent calendar collection
- collection.updateCTag()
-
- return result
+ return collection.updateCTag().addCallback(lambda _: result)
d = maybeDeferred(delete, "", delchild.fp, "0")
d.addCallback(_deletedResourced)
Copied: CalendarServer/trunk/twistedcaldav/memcache.py (from rev 2472, CalendarServer/branches/unified-cache/twistedcaldav/memcache.py)
===================================================================
--- CalendarServer/trunk/twistedcaldav/memcache.py (rev 0)
+++ CalendarServer/trunk/twistedcaldav/memcache.py 2008-05-24 00:57:54 UTC (rev 2473)
@@ -0,0 +1,657 @@
+# -*- test-case-name: twisted.test.test_memcache -*-
+# Copyright (c) 2007 Twisted Matrix Laboratories.
+# See LICENSE for details.
+
+"""
+Memcache client protocol. Memcached is a caching server, storing data in the
+form of pairs key/value, and memcache is the protocol to talk with it.
+
+To connect to a server, create a factory for L{MemCacheProtocol}::
+
+ from twisted.internet import reactor, protocol
+ from twisted.protocols.memcache import MemCacheProtocol, DEFAULT_PORT
+ d = protocol.ClientCreator(reactor, MemCacheProtocol
+ ).connectTCP("localhost", DEFAULT_PORT)
+ def doSomething(proto):
+ # Here you call the memcache operations
+ return proto.set("mykey", "a lot of data")
+ d.addCallback(doSomething)
+ reactor.run()
+
+All the operations of the memcache protocol are present, but
+L{MemCacheProtocol.set} and L{MemCacheProtocol.get} are the more important.
+
+See U{http://code.sixapart.com/svn/memcached/trunk/server/doc/protocol.txt} for
+more information about the protocol.
+"""
+
+try:
+ from collections import deque
+except ImportError:
+ class deque(list):
+ def popleft(self):
+ return self.pop(0)
+
+
+from twisted.protocols.basic import LineReceiver
+from twisted.protocols.policies import TimeoutMixin
+from twisted.internet.defer import Deferred, fail, TimeoutError
+from twisted.python import log
+
+
+
+DEFAULT_PORT = 11211
+
+
+
+class NoSuchCommand(Exception):
+ """
+ Exception raised when a non existent command is called.
+ """
+
+
+
+class ClientError(Exception):
+ """
+ Error caused by an invalid client call.
+ """
+
+
+
+class ServerError(Exception):
+ """
+ Problem happening on the server.
+ """
+
+
+
+class Command(object):
+ """
+ Wrap a client action into an object, that holds the values used in the
+ protocol.
+
+ @ivar _deferred: the L{Deferred} object that will be fired when the result
+ arrives.
+ @type _deferred: L{Deferred}
+
+ @ivar command: name of the command sent to the server.
+ @type command: C{str}
+ """
+
+ def __init__(self, command, **kwargs):
+ """
+ Create a command.
+
+ @param command: the name of the command.
+ @type command: C{str}
+
+ @param kwargs: this values will be stored as attributes of the object
+ for future use
+ """
+ self.command = command
+ self._deferred = Deferred()
+ for k, v in kwargs.items():
+ setattr(self, k, v)
+
+
+ def success(self, value):
+ """
+ Shortcut method to fire the underlying deferred.
+ """
+ self._deferred.callback(value)
+
+
+ def fail(self, error):
+ """
+ Make the underlying deferred fails.
+ """
+ self._deferred.errback(error)
+
+
+
+class MemCacheProtocol(LineReceiver, TimeoutMixin):
+ """
+ MemCache protocol: connect to a memcached server to store/retrieve values.
+
+ @ivar persistentTimeOut: the timeout period used to wait for a response.
+ @type persistentTimeOut: C{int}
+
+ @ivar _current: current list of requests waiting for an answer from the
+ server.
+ @type _current: C{deque} of L{Command}
+
+ @ivar _lenExpected: amount of data expected in raw mode, when reading for
+ a value.
+ @type _lenExpected: C{int}
+
+ @ivar _getBuffer: current buffer of data, used to store temporary data
+ when reading in raw mode.
+ @type _getBuffer: C{list}
+
+ @ivar _bufferLength: the total amount of bytes in C{_getBuffer}.
+ @type _bufferLength: C{int}
+ """
+ MAX_KEY_LENGTH = 250
+
+ def __init__(self, timeOut=60):
+ """
+ Create the protocol.
+
+ @param timeOut: the timeout to wait before detecting that the
+ connection is dead and close it. It's expressed in seconds.
+ @type timeOut: C{int}
+ """
+ self._current = deque()
+ self._lenExpected = None
+ self._getBuffer = None
+ self._bufferLength = None
+ self.persistentTimeOut = self.timeOut = timeOut
+
+
+ def timeoutConnection(self):
+ """
+ Close the connection in case of timeout.
+ """
+ for cmd in self._current:
+ cmd.fail(TimeoutError("Connection timeout"))
+ self.transport.loseConnection()
+
+
+ def sendLine(self, line):
+ """
+ Override sendLine to add a timeout to response.
+ """
+ if not self._current:
+ self.setTimeout(self.persistentTimeOut)
+ LineReceiver.sendLine(self, line)
+
+
+ def rawDataReceived(self, data):
+ """
+ Collect data for a get.
+ """
+ self.resetTimeout()
+ self._getBuffer.append(data)
+ self._bufferLength += len(data)
+ if self._bufferLength >= self._lenExpected + 2:
+ data = "".join(self._getBuffer)
+ buf = data[:self._lenExpected]
+ rem = data[self._lenExpected + 2:]
+ val = buf
+ self._lenExpected = None
+ self._getBuffer = None
+ self._bufferLength = None
+ cmd = self._current[0]
+ cmd.value = val
+ self.setLineMode(rem)
+
+
+ def cmd_STORED(self):
+ """
+ Manage a success response to a set operation.
+ """
+ self._current.popleft().success(True)
+
+
+ def cmd_NOT_STORED(self):
+ """
+ Manage a specific 'not stored' response to a set operation: this is not
+ an error, but some condition wasn't met.
+ """
+ self._current.popleft().success(False)
+
+
+ def cmd_END(self):
+ """
+ This the end token to a get or a stat operation.
+ """
+ cmd = self._current.popleft()
+ if cmd.command == "get":
+ cmd.success((cmd.flags, cmd.value))
+ elif cmd.command == "gets":
+ cmd.success((cmd.flags, cmd.cas, cmd.value))
+ elif cmd.command == "stats":
+ cmd.success(cmd.values)
+
+
+ def cmd_NOT_FOUND(self):
+ """
+ Manage error response for incr/decr/delete.
+ """
+ self._current.popleft().success(False)
+
+
+ def cmd_VALUE(self, line):
+ """
+ Prepare the reading a value after a get.
+ """
+ cmd = self._current[0]
+ if cmd.command == "get":
+ key, flags, length = line.split()
+ cas = ""
+ else:
+ key, flags, length, cas = line.split()
+ self._lenExpected = int(length)
+ self._getBuffer = []
+ self._bufferLength = 0
+ if cmd.key != key:
+ raise RuntimeError("Unexpected commands answer.")
+ cmd.flags = int(flags)
+ cmd.length = self._lenExpected
+ cmd.cas = cas
+ self.setRawMode()
+
+
+ def cmd_STAT(self, line):
+ """
+ Reception of one stat line.
+ """
+ cmd = self._current[0]
+ key, val = line.split(" ", 1)
+ cmd.values[key] = val
+
+
+ def cmd_VERSION(self, versionData):
+ """
+ Read version token.
+ """
+ self._current.popleft().success(versionData)
+
+
+ def cmd_ERROR(self):
+ """
+ An non-existent command has been sent.
+ """
+ log.err("Non-existent command sent.")
+ cmd = self._current.popleft()
+ cmd.fail(NoSuchCommand())
+
+
+ def cmd_CLIENT_ERROR(self, errText):
+ """
+ An invalid input as been sent.
+ """
+ log.err("Invalid input: %s" % (errText,))
+ cmd = self._current.popleft()
+ cmd.fail(ClientError(errText))
+
+
+ def cmd_SERVER_ERROR(self, errText):
+ """
+ An error has happened server-side.
+ """
+ log.err("Server error: %s" % (errText,))
+ cmd = self._current.popleft()
+ cmd.fail(ServerError(errText))
+
+
+ def cmd_DELETED(self):
+ """
+ A delete command has completed successfully.
+ """
+ self._current.popleft().success(True)
+
+
+ def cmd_OK(self):
+ """
+ The last command has been completed.
+ """
+ self._current.popleft().success(True)
+
+
+ def cmd_EXISTS(self):
+ """
+ A C{checkAndSet} update has failed.
+ """
+ self._current.popleft().success(False)
+
+
+ def lineReceived(self, line):
+ """
+ Receive line commands from the server.
+ """
+ self.resetTimeout()
+ token = line.split(" ", 1)[0]
+ # First manage standard commands without space
+ cmd = getattr(self, "cmd_%s" % (token,), None)
+ if cmd is not None:
+ args = line.split(" ", 1)[1:]
+ if args:
+ cmd(args[0])
+ else:
+ cmd()
+ else:
+ # Then manage commands with space in it
+ line = line.replace(" ", "_")
+ cmd = getattr(self, "cmd_%s" % (line,), None)
+ if cmd is not None:
+ cmd()
+ else:
+ # Increment/Decrement response
+ cmd = self._current.popleft()
+ val = int(line)
+ cmd.success(val)
+ if not self._current:
+ # No pending request, remove timeout
+ self.setTimeout(None)
+
+
+ def increment(self, key, val=1):
+ """
+ Increment the value of C{key} by given value (default to 1).
+ C{key} must be consistent with an int. Return the new value.
+
+ @param key: the key to modify.
+ @type key: C{str}
+
+ @param val: the value to increment.
+ @type val: C{int}
+
+ @return: a deferred with will be called back with the new value
+ associated with the key (after the increment).
+ @rtype: L{Deferred}
+ """
+ return self._incrdecr("incr", key, val)
+
+
+ def decrement(self, key, val=1):
+ """
+ Decrement the value of C{key} by given value (default to 1).
+ C{key} must be consistent with an int. Return the new value, coerced to
+ 0 if negative.
+
+ @param key: the key to modify.
+ @type key: C{str}
+
+ @param val: the value to decrement.
+ @type val: C{int}
+
+ @return: a deferred with will be called back with the new value
+ associated with the key (after the decrement).
+ @rtype: L{Deferred}
+ """
+ return self._incrdecr("decr", key, val)
+
+
+ def _incrdecr(self, cmd, key, val):
+ """
+ Internal wrapper for incr/decr.
+ """
+ if not isinstance(key, str):
+ return fail(ClientError(
+ "Invalid type for key: %s, expecting a string" % (type(key),)))
+ if len(key) > self.MAX_KEY_LENGTH:
+ return fail(ClientError("Key too long"))
+ fullcmd = "%s %s %d" % (cmd, key, int(val))
+ self.sendLine(fullcmd)
+ cmdObj = Command(cmd, key=key)
+ self._current.append(cmdObj)
+ return cmdObj._deferred
+
+
+ def replace(self, key, val, flags=0, expireTime=0):
+ """
+ Replace the given C{key}. It must already exist in the server.
+
+ @param key: the key to replace.
+ @type key: C{str}
+
+ @param val: the new value associated with the key.
+ @type val: C{str}
+
+ @param flags: the flags to store with the key.
+ @type flags: C{int}
+
+ @param expireTime: if different from 0, the relative time in seconds
+ when the key will be deleted from the store.
+ @type expireTime: C{int}
+
+ @return: a deferred that will fire with C{True} if the operation has
+ succeeded, and C{False} with the key didn't previously exist.
+ @rtype: L{Deferred}
+ """
+ return self._set("replace", key, val, flags, expireTime, "")
+
+
+ def add(self, key, val, flags=0, expireTime=0):
+ """
+ Add the given C{key}. It must not exist in the server.
+
+ @param key: the key to add.
+ @type key: C{str}
+
+ @param val: the value associated with the key.
+ @type val: C{str}
+
+ @param flags: the flags to store with the key.
+ @type flags: C{int}
+
+ @param expireTime: if different from 0, the relative time in seconds
+ when the key will be deleted from the store.
+ @type expireTime: C{int}
+
+ @return: a deferred that will fire with C{True} if the operation has
+ succeeded, and C{False} with the key already exists.
+ @rtype: L{Deferred}
+ """
+ return self._set("add", key, val, flags, expireTime, "")
+
+
+ def set(self, key, val, flags=0, expireTime=0):
+ """
+ Set the given C{key}.
+
+ @param key: the key to set.
+ @type key: C{str}
+
+ @param val: the value associated with the key.
+ @type val: C{str}
+
+ @param flags: the flags to store with the key.
+ @type flags: C{int}
+
+ @param expireTime: if different from 0, the relative time in seconds
+ when the key will be deleted from the store.
+ @type expireTime: C{int}
+
+ @return: a deferred that will fire with C{True} if the operation has
+ succeeded.
+ @rtype: L{Deferred}
+ """
+ return self._set("set", key, val, flags, expireTime, "")
+
+
+ def checkAndSet(self, key, val, cas, flags=0, expireTime=0):
+ """
+ Change the content of C{key} only if the C{cas} value matches the
+ current one associated with the key. Use this to store a value which
+ hasn't been modified since last time you fetched it.
+
+ @param key: The key to set.
+ @type key: C{str}
+
+ @param val: The value associated with the key.
+ @type val: C{str}
+
+ @param cas: Unique 64-bit value returned by previous call of C{get}.
+ @type cas: C{str}
+
+ @param flags: The flags to store with the key.
+ @type flags: C{int}
+
+ @param expireTime: If different from 0, the relative time in seconds
+ when the key will be deleted from the store.
+ @type expireTime: C{int}
+
+ @return: A deferred that will fire with C{True} if the operation has
+ succeeded, C{False} otherwise.
+ @rtype: L{Deferred}
+ """
+ return self._set("cas", key, val, flags, expireTime, cas)
+
+
+ def _set(self, cmd, key, val, flags, expireTime, cas):
+ """
+ Internal wrapper for setting values.
+ """
+ if not isinstance(key, str):
+ return fail(ClientError(
+ "Invalid type for key: %s, expecting a string" % (type(key),)))
+ if len(key) > self.MAX_KEY_LENGTH:
+ return fail(ClientError("Key too long"))
+ if not isinstance(val, str):
+ return fail(ClientError(
+ "Invalid type for value: %s, expecting a string" %
+ (type(val),)))
+ if cas:
+ cas = " " + cas
+ length = len(val)
+ fullcmd = "%s %s %d %d %d%s" % (
+ cmd, key, flags, expireTime, length, cas)
+ self.sendLine(fullcmd)
+ self.sendLine(val)
+ cmdObj = Command(cmd, key=key, flags=flags, length=length)
+ self._current.append(cmdObj)
+ return cmdObj._deferred
+
+
+ def append(self, key, val):
+ """
+ Append given data to the value of an existing key.
+
+ @param key: The key to modify.
+ @type key: C{str}
+
+ @param val: The value to append to the current value associated with
+ the key.
+ @type val: C{str}
+
+ @return: A deferred that will fire with C{True} if the operation has
+ succeeded, C{False} otherwise.
+ @rtype: L{Deferred}
+ """
+ # Even if flags and expTime values are ignored, we have to pass them
+ return self._set("append", key, val, 0, 0, "")
+
+
+ def prepend(self, key, val):
+ """
+ Prepend given data to the value of an existing key.
+
+ @param key: The key to modify.
+ @type key: C{str}
+
+ @param val: The value to prepend to the current value associated with
+ the key.
+ @type val: C{str}
+
+ @return: A deferred that will fire with C{True} if the operation has
+ succeeded, C{False} otherwise.
+ @rtype: L{Deferred}
+ """
+ # Even if flags and expTime values are ignored, we have to pass them
+ return self._set("prepend", key, val, 0, 0, "")
+
+
+ def get(self, key, withIdentifier=False):
+ """
+ Get the given C{key}. It doesn't support multiple keys. If
+ C{withIdentifier} is set to C{True}, the command issued is a C{gets},
+ that will return the current identifier associated with the value. This
+ identifier has to be used when issuing C{checkAndSet} update later,
+ using the corresponding method.
+
+ @param key: The key to retrieve.
+ @type key: C{str}
+
+ @param withIdentifier: If set to C{True}, retrieve the current
+ identifier along with the value and the flags.
+ @type withIdentifier: C{bool}
+
+ @return: A deferred that will fire with the tuple (flags, value) if
+ C{withIdentifier} is C{False}, or (flags, cas identifier, value)
+ if C{True}.
+ @rtype: L{Deferred}
+ """
+ if not isinstance(key, str):
+ return fail(ClientError(
+ "Invalid type for key: %s, expecting a string" % (type(key),)))
+ if len(key) > self.MAX_KEY_LENGTH:
+ return fail(ClientError("Key too long"))
+ if withIdentifier:
+ cmd = "gets"
+ else:
+ cmd = "get"
+ fullcmd = "%s %s" % (cmd, key)
+ self.sendLine(fullcmd)
+ cmdObj = Command(cmd, key=key, value=None, flags=0, cas="")
+ self._current.append(cmdObj)
+ return cmdObj._deferred
+
+
+ def stats(self):
+ """
+ Get some stats from the server. It will be available as a dict.
+
+ @return: a deferred that will fire with a C{dict} of the available
+ statistics.
+ @rtype: L{Deferred}
+ """
+ self.sendLine("stats")
+ cmdObj = Command("stats", values={})
+ self._current.append(cmdObj)
+ return cmdObj._deferred
+
+
+ def version(self):
+ """
+ Get the version of the server.
+
+ @return: a deferred that will fire with the string value of the
+ version.
+ @rtype: L{Deferred}
+ """
+ self.sendLine("version")
+ cmdObj = Command("version")
+ self._current.append(cmdObj)
+ return cmdObj._deferred
+
+
+ def delete(self, key):
+ """
+ Delete an existing C{key}.
+
+ @param key: the key to delete.
+ @type key: C{str}
+
+ @return: a deferred that will be called back with C{True} if the key
+ was successfully deleted, or C{False} if not.
+ @rtype: L{Deferred}
+ """
+ if not isinstance(key, str):
+ return fail(ClientError(
+ "Invalid type for key: %s, expecting a string" % (type(key),)))
+ self.sendLine("delete %s" % key)
+ cmdObj = Command("delete", key=key)
+ self._current.append(cmdObj)
+ return cmdObj._deferred
+
+
+ def flushAll(self):
+ """
+ Flush all cached values.
+
+ @return: a deferred that will be called back with C{True} when the
+ operation has succeeded.
+ @rtype: L{Deferred}
+ """
+ self.sendLine("flush_all")
+ cmdObj = Command("flush_all")
+ self._current.append(cmdObj)
+ return cmdObj._deferred
+
+
+
+__all__ = ["MemCacheProtocol", "DEFAULT_PORT", "NoSuchCommand", "ClientError",
+ "ServerError"]
+
Modified: CalendarServer/trunk/twistedcaldav/method/delete.py
===================================================================
--- CalendarServer/trunk/twistedcaldav/method/delete.py 2008-05-24 00:47:17 UTC (rev 2472)
+++ CalendarServer/trunk/twistedcaldav/method/delete.py 2008-05-24 00:57:54 UTC (rev 2473)
@@ -36,9 +36,11 @@
if isPseudoCalendarCollectionResource(parent):
index = parent.index()
index.deleteResource(self.fp.basename())
-
+
# Change CTag on the parent calendar collection
- parent.updateCTag()
+ d = parent.updateCTag()
+ d.addCallback(lambda _: response)
+ return d
return response
Modified: CalendarServer/trunk/twistedcaldav/method/put_common.py
===================================================================
--- CalendarServer/trunk/twistedcaldav/method/put_common.py 2008-05-24 00:47:17 UTC (rev 2472)
+++ CalendarServer/trunk/twistedcaldav/method/put_common.py 2008-05-24 00:57:54 UTC (rev 2473)
@@ -627,7 +627,9 @@
if sourcecal:
# Change CTag on the parent calendar collection
- sourceparent.updateCTag()
+ d = waitForDeferred(sourceparent.updateCTag())
+ yield d
+ d.getResult()
if destinationcal:
result = doDestinationIndex(calendar)
@@ -653,7 +655,9 @@
if destinationcal:
# Change CTag on the parent calendar collection
- destinationparent.updateCTag()
+ d = waitForDeferred(destinationparent.updateCTag())
+ yield d
+ d.getResult()
# Can now commit changes and forget the rollback details
rollback.Commit()
Modified: CalendarServer/trunk/twistedcaldav/resource.py
===================================================================
--- CalendarServer/trunk/twistedcaldav/resource.py 2008-05-24 00:47:17 UTC (rev 2472)
+++ CalendarServer/trunk/twistedcaldav/resource.py 2008-05-24 00:57:54 UTC (rev 2473)
@@ -75,6 +75,22 @@
return tuple(super(CalDAVComplianceMixIn, self).davComplianceClasses()) + extra_compliance
+def updateCacheTokenOnCallback(f):
+ def fun(self, *args, **kwargs):
+ def _updateToken(response):
+ return self.cacheNotifier.changed().addCallback(
+ lambda _: response)
+
+ d = maybeDeferred(f, self, *args, **kwargs)
+
+ if hasattr(self, 'cacheNotifier'):
+ d.addCallback(_updateToken)
+
+ return d
+
+ return fun
+
+
class CalDAVResource (CalDAVComplianceMixIn, DAVResource):
"""
CalDAV resource.
@@ -129,6 +145,19 @@
return response
+ @updateCacheTokenOnCallback
+ def http_PROPPATCH(self, request):
+ return super(CalDAVResource, self).http_PROPPATCH(request)
+
+ @updateCacheTokenOnCallback
+ def http_DELETE(self, request):
+ return super(CalDAVResource, self).http_DELETE(request)
+
+ @updateCacheTokenOnCallback
+ def http_ACL(self, request):
+ return super(CalDAVResource, self).http_ACL(request)
+
+
##
# WebDAV
##
@@ -217,9 +246,6 @@
def writeDeadProperty(self, property):
val = super(CalDAVResource, self).writeDeadProperty(property)
- if hasattr(self, 'cacheNotifier'):
- self.cacheNotifier.changed()
-
return val
Modified: CalendarServer/trunk/twistedcaldav/root.py
===================================================================
--- CalendarServer/trunk/twistedcaldav/root.py 2008-05-24 00:47:17 UTC (rev 2472)
+++ CalendarServer/trunk/twistedcaldav/root.py 2008-05-24 00:57:54 UTC (rev 2473)
@@ -26,7 +26,10 @@
from twistedcaldav.extensions import DAVFile
from twistedcaldav.config import config
from twistedcaldav.cache import ResponseCache, _CachedResponseResource
+from twistedcaldav.cache import MemcacheResponseCache, MemcacheChangeNotifier
from twistedcaldav.log import Logger
+from twistedcaldav.static import CalendarHomeFile
+from twistedcaldav.directory.principal import DirectoryPrincipalResource
log = Logger()
@@ -52,8 +55,19 @@
self.contentFilters = []
- self.responseCache = ResponseCache(self.fp, config.ResponseCacheSize)
+ if config.Memcached['ClientEnabled']:
+ self.responseCache = MemcacheResponseCache(
+ self.fp,
+ config.Memcached['BindAddress'],
+ config.Memcached['Port'])
+ CalendarHomeFile.cacheNotifierFactory = MemcacheChangeNotifier
+ DirectoryPrincipalResource.cacheNotifierFactory = MemcacheChangeNotifier
+
+ else:
+ self.responseCache = ResponseCache(self.fp,
+ config.ResponseCacheSize)
+
if config.ResponseCompression:
from twisted.web2.filter import gzip
self.contentFilters.append((gzip.gzipfilter, True))
@@ -148,6 +162,7 @@
return _CachedResponseResource(response), []
def _resourceNotInCacheEb(failure):
+ failure.trap(KeyError)
return super(RootResource, self).locateChild(request,segments)
if request.method == 'PROPFIND' and not getattr(
Modified: CalendarServer/trunk/twistedcaldav/static.py
===================================================================
--- CalendarServer/trunk/twistedcaldav/static.py 2008-05-24 00:47:17 UTC (rev 2472)
+++ CalendarServer/trunk/twistedcaldav/static.py 2008-05-24 00:57:54 UTC (rev 2473)
@@ -39,6 +39,7 @@
from urlparse import urlsplit
from twisted.internet.defer import deferredGenerator, fail, succeed, waitForDeferred
+from twisted.python.failure import Failure
from twisted.web2 import responsecode
from twisted.web2.http import HTTPError, StatusResponse
from twisted.web2.dav import davxml
@@ -69,7 +70,7 @@
from twistedcaldav.log import Logger
from twistedcaldav.timezoneservice import TimezoneServiceResource
-from twistedcaldav.cache import CacheChangeNotifier, PropfindCacheMixin
+from twistedcaldav.cache import XattrCacheChangeNotifier, PropfindCacheMixin
log = Logger()
@@ -135,13 +136,13 @@
raise HTTPError(status)
# Initialize CTag on the calendar collection
- self.updateCTag()
+ d1 = self.updateCTag()
# Create the index so its ready when the first PUTs come in
- self.index().create()
+ d1.addCallback(lambda _: self.index().create())
+ d1.addCallback(lambda _: status)
+ return d1
- return status
-
d = self.createSpecialCollection(davxml.ResourceType.calendar)
d.addCallback(onCalendarCollection)
return d
@@ -286,8 +287,17 @@
def updateCTag(self):
assert self.isCollection()
- self.writeDeadProperty(customxml.GETCTag(str(datetime.datetime.now())))
+ try:
+ self.writeDeadProperty(customxml.GETCTag(
+ str(datetime.datetime.now())))
+ except:
+ return fail(Failure())
+ if hasattr(self, 'cacheNotifier'):
+ return self.cacheNotifier.changed()
+
+ return succeed(True)
+
##
# Quota
##
@@ -549,20 +559,22 @@
"""
Calendar home collection resource.
"""
+ cacheNotifierFactory = XattrCacheChangeNotifier
+
def __init__(self, path, parent, record):
"""
@param path: the path to the file which will back the resource.
"""
CalDAVFile.__init__(self, path)
DirectoryCalendarHomeResource.__init__(self, parent, record)
- self.cacheNotifier = CacheChangeNotifier(self.deadProperties())
+ self.cacheNotifier = self.cacheNotifierFactory(self.deadProperties())
def provisionChild(self, name):
if config.EnableDropBox:
DropBoxHomeFileClass = DropBoxHomeFile
else:
DropBoxHomeFileClass = None
-
+
cls = {
"inbox" : ScheduleInboxFile,
"outbox" : ScheduleOutboxFile,
Modified: CalendarServer/trunk/twistedcaldav/test/test_cache.py
===================================================================
--- CalendarServer/trunk/twistedcaldav/test/test_cache.py 2008-05-24 00:47:17 UTC (rev 2472)
+++ CalendarServer/trunk/twistedcaldav/test/test_cache.py 2008-05-24 00:57:54 UTC (rev 2473)
@@ -15,8 +15,12 @@
##
from new import instancemethod
+import hashlib
+import cPickle
from twisted.trial.unittest import TestCase
+from twisted.internet.defer import succeed, fail
+from twisted.python.failure import Failure
from twisted.python.filepath import FilePath
@@ -25,9 +29,11 @@
from twisted.web2.stream import MemoryStream
from twisted.web2.http_headers import Headers
-from twistedcaldav.cache import CacheChangeNotifier
+from twistedcaldav.cache import XattrCacheChangeNotifier
from twistedcaldav.cache import CacheTokensProperty
from twistedcaldav.cache import ResponseCache
+from twistedcaldav.cache import MemcacheResponseCache
+from twistedcaldav.cache import MemcacheChangeNotifier
from twistedcaldav.test.util import InMemoryPropertyStore
@@ -42,7 +48,7 @@
class StubRequest(object):
- def __init__(self, method, uri, authnUser, depth=1, body=None):
+ def __init__(self, method, uri, authnUser, depth='1', body=None):
self.method = method
self.uri = uri
self.authnUser = davxml.Principal(davxml.HRef.fromString(authnUser))
@@ -65,62 +71,90 @@
-class CacheChangeNotifierTests(TestCase):
+class InMemoryMemcacheProtocol(object):
+ def __init__(self):
+ self._cache = {}
+
+
+ def get(self, key):
+ if key not in self._cache:
+ return succeed((0, None))
+
+ return succeed(self._cache[key])
+
+
+ def set(self, key, value, flags=0, expireTime=0):
+ try:
+ self._cache[key] = (flags, value)
+ return succeed(True)
+
+ except Exception, err:
+ return fail(Failure())
+
+
+
+class XattrCacheChangeNotifierTests(TestCase):
def setUp(self):
self.props = InMemoryPropertyStore()
- self.ccn = CacheChangeNotifier(self.props)
+ self.ccn = XattrCacheChangeNotifier(self.props)
self.ccn._newCacheToken = instancemethod(_newCacheToken,
self.ccn,
- CacheChangeNotifier)
+ XattrCacheChangeNotifier)
- def test_cacheTokenPropertyIsProvisioned(self):
- self.ccn.changed()
+ def assertToken(self, expectedToken):
token = self.props._properties[CacheTokensProperty.qname()
].children[0].data
- self.assertEquals(token, 'token0')
+ self.assertEquals(token, expectedToken)
+ def test_cacheTokenPropertyIsProvisioned(self):
+ d = self.ccn.changed()
+ d.addCallback(lambda _: self.assertToken('token0'))
+ return d
+
+
def test_changedChangesToken(self):
- self.ccn.changed()
- self.ccn.changed()
- token = self.props._properties[CacheTokensProperty.qname()
- ].children[0].data
- self.assertEquals(token, 'token1')
+ d = self.ccn.changed()
+ d.addCallback(lambda _: self.ccn.changed())
+ d.addCallback(lambda _: self.assertToken('token1'))
+ return d
-
-class ResponseCacheTests(TestCase):
+class MemCacheChangeNotifierTests(TestCase):
def setUp(self):
- self.tokens = {
- '/calendars/users/cdaboo/': 'uriToken0',
- '/principals/users/cdaboo/': 'principalToken0',
- '/principals/users/dreid/': 'principalTokenX'}
+ self.memcache = InMemoryMemcacheProtocol()
+ self.ccn = MemcacheChangeNotifier(InMemoryPropertyStore())
+ self.ccn._memcacheProtocol = self.memcache
+ self.ccn._newCacheToken = instancemethod(_newCacheToken,
+ self.ccn,
+ MemcacheChangeNotifier)
- self.rc = ResponseCache(None)
- self.rc._tokenForURI = self.tokens.get
- self.rc._time = (lambda:0)
+ def assertToken(self, expectedToken):
+ token = self.memcache._cache['cacheToken::memory:'][1]
+ self.assertEquals(token, expectedToken)
- self.expected_response = (200, Headers({}), "Foo")
- expected_key = (
- 'PROPFIND',
- '/calendars/users/cdaboo/',
- '/principals/users/cdaboo/',
- 1,
- hash('foobar'),
- )
+ def test_cacheTokenPropertyIsProvisioned(self):
+ d = self.ccn.changed()
+ d.addCallback(lambda _: self.assertToken('token0'))
+ return d
- self.rc._responses[expected_key] = (
- 'principalToken0', 'uriToken0', 0, self.expected_response)
- self.rc._accessList = [expected_key]
+ def test_changedChangesToken(self):
+ d = self.ccn.changed()
+ d.addCallback(lambda _: self.ccn.changed())
+ d.addCallback(lambda _: self.assertToken('token1'))
+ return d
+
+class BaseCacheTestMixin(object):
def assertResponse(self, response, expected):
self.assertEquals(response.code, expected[0])
- self.assertEquals(response.headers, expected[1])
+ self.assertEquals(set(response.headers.getAllRawHeaders()),
+ set(expected[1].getAllRawHeaders()))
d = allDataFromStream(response.stream)
d.addCallback(self.assertEquals, expected[2])
@@ -176,7 +210,7 @@
'PROPFIND',
'/calendars/users/cdaboo/',
'/principals/users/cdaboo/',
- depth=0))
+ depth='0'))
d.addCallback(self.assertEquals, None)
return d
@@ -219,6 +253,37 @@
return d
+
+class ResponseCacheTests(BaseCacheTestMixin, TestCase):
+ def setUp(self):
+ self.tokens = {
+ '/calendars/users/cdaboo/': 'uriToken0',
+ '/principals/users/cdaboo/': 'principalToken0',
+ '/principals/users/dreid/': 'principalTokenX'}
+
+ self.rc = ResponseCache(None)
+ self.rc._tokenForURI = self.tokens.get
+
+ self.rc._time = (lambda:0)
+
+ self.expected_response = (200, Headers({}), "Foo")
+
+ expected_key = (
+ 'PROPFIND',
+ '/principals/users/cdaboo/',
+ '/calendars/users/cdaboo/',
+ '1',
+ hash('foobar'),
+ )
+
+ self.rc._responses[expected_key] = (
+ 'principalToken0', 'uriToken0', '0', self.expected_response)
+
+ self.rc._accessList = [expected_key]
+
+
+
+
def test__tokenForURI(self):
docroot = FilePath(self.mktemp())
principal = docroot.child('principals'
@@ -264,37 +329,75 @@
return d
- def test_cacheExpirationBenchmark(self):
- self.rc.CACHE_SIZE = 70000
- import time
+# def test_cacheExpirationBenchmark(self):
+# self.rc.CACHE_SIZE = 70000
+# import time
- self.rc._responses = {}
- self.rc._accessList = []
+# self.rc._responses = {}
+# self.rc._accessList = []
- for x in xrange(0, self.rc.CACHE_SIZE):
- req = StubRequest('PROPFIND',
- '/principals/users/user%d' % (x,),
- '/principals/users/user%d' % (x,))
- self.rc._responses[req] = (
- 'pTokenUser%d' % (x,), 'rTokenUser%d' % (x,), 0,
- (200, {}, 'foobar'))
+# for x in xrange(0, self.rc.CACHE_SIZE):
+# req = StubRequest('PROPFIND',
+# '/principals/users/user%d' % (x,),
+# '/principals/users/user%d' % (x,))
+# self.rc._responses[req] = (
+# 'pTokenUser%d' % (x,), 'rTokenUser%d' % (x,), 0,
+# (200, {}, 'foobar'))
- self.rc._accessList.append(req)
+# self.rc._accessList.append(req)
- def assertTime(result, startTime):
- duration = time.time() - startTime
+# def assertTime(result, startTime):
+# duration = time.time() - startTime
- self.failUnless(
- duration < 0.01,
- "Took to long to add to the cache: %r" % (duration,))
+# self.failUnless(
+# duration < 0.01,
+# "Took to long to add to the cache: %r" % (duration,))
- startTime = time.time()
+# startTime = time.time()
- d = self.rc.cacheResponseForRequest(
- StubRequest('PROPFIND',
- '/principals/users/dreid/',
- '/principals/users/dreid/'),
- StubResponse(200, {}, 'Foobar'))
+# d = self.rc.cacheResponseForRequest(
+# StubRequest('PROPFIND',
+# '/principals/users/dreid/',
+# '/principals/users/dreid/'),
+# StubResponse(200, {}, 'Foobar'))
- d.addCallback(assertTime, startTime)
- return d
+# d.addCallback(assertTime, startTime)
+# return d
+
+
+class MemcacheResponseCacheTests(BaseCacheTestMixin, TestCase):
+ def setUp(self):
+ memcacheStub = InMemoryMemcacheProtocol()
+ self.rc = MemcacheResponseCache(None, None, None, None)
+ self.rc.logger.setLevel('debug')
+ self.tokens = {}
+
+ self.tokens['/calendars/users/cdaboo/'] = 'uriToken0'
+ self.tokens['/principals/users/cdaboo/'] = 'principalToken0'
+ self.tokens['/principals/users/dreid/'] = 'principalTokenX'
+
+ def _getToken(uri):
+ return succeed(self.tokens.get(uri))
+
+ self.rc._tokenForURI = _getToken
+
+ self.expected_response = (200, Headers({}), "Foo")
+
+ expected_key = hashlib.md5(':'.join([str(t) for t in (
+ 'PROPFIND',
+ '/principals/users/cdaboo/',
+ '/calendars/users/cdaboo/',
+ '1',
+ hash('foobar'),
+ )])).hexdigest()
+
+ memcacheStub._cache[expected_key] = (
+ 0, #flags
+ cPickle.dumps((
+ 'principalToken0',
+ 'uriToken0',
+ (self.expected_response[0],
+ dict(list(self.expected_response[1].getAllRawHeaders())),
+ self.expected_response[2]))))
+
+ self.rc._memcacheProtocol = memcacheStub
Copied: CalendarServer/trunk/twistedcaldav/test/test_memcache.py (from rev 2472, CalendarServer/branches/unified-cache/twistedcaldav/test/test_memcache.py)
===================================================================
--- CalendarServer/trunk/twistedcaldav/test/test_memcache.py (rev 0)
+++ CalendarServer/trunk/twistedcaldav/test/test_memcache.py 2008-05-24 00:57:54 UTC (rev 2473)
@@ -0,0 +1,510 @@
+# Copyright (c) 2007 Twisted Matrix Laboratories.
+# See LICENSE for details.
+
+"""
+Test the memcache client protocol.
+"""
+
+from twistedcaldav.memcache import MemCacheProtocol, NoSuchCommand
+from twistedcaldav.memcache import ClientError, ServerError
+
+from twisted.trial.unittest import TestCase
+from twisted.test.proto_helpers import StringTransportWithDisconnection
+from twisted.internet.task import Clock
+from twisted.internet.defer import Deferred, gatherResults, TimeoutError
+
+
+
+class MemCacheTestCase(TestCase):
+ """
+ Test client protocol class L{MemCacheProtocol}.
+ """
+
+ def setUp(self):
+ """
+ Create a memcache client, connect it to a string protocol, and make it
+ use a deterministic clock.
+ """
+ self.proto = MemCacheProtocol()
+ self.clock = Clock()
+ self.proto.callLater = self.clock.callLater
+ self.transport = StringTransportWithDisconnection()
+ self.transport.protocol = self.proto
+ self.proto.makeConnection(self.transport)
+
+
+ def _test(self, d, send, recv, result):
+ """
+ Shortcut method for classic tests.
+
+ @param d: the resulting deferred from the memcache command.
+ @type d: C{Deferred}
+
+ @param send: the expected data to be sent.
+ @type send: C{str}
+
+ @param recv: the data to simulate as reception.
+ @type recv: C{str}
+
+ @param result: the expected result.
+ @type result: C{any}
+ """
+ def cb(res):
+ self.assertEquals(res, result)
+ self.assertEquals(self.transport.value(), send)
+ d.addCallback(cb)
+ self.proto.dataReceived(recv)
+ return d
+
+
+ def test_get(self):
+ """
+ L{MemCacheProtocol.get} should return a L{Deferred} which is
+ called back with the value and the flag associated with the given key
+ if the server returns a successful result.
+ """
+ return self._test(self.proto.get("foo"), "get foo\r\n",
+ "VALUE foo 0 3\r\nbar\r\nEND\r\n", (0, "bar"))
+
+
+ def test_emptyGet(self):
+ """
+ Test getting a non-available key: it should succeed but return C{None}
+ as value and C{0} as flag.
+ """
+ return self._test(self.proto.get("foo"), "get foo\r\n",
+ "END\r\n", (0, None))
+
+
+ def test_set(self):
+ """
+ L{MemCacheProtocol.set} should return a L{Deferred} which is
+ called back with C{True} when the operation succeeds.
+ """
+ return self._test(self.proto.set("foo", "bar"),
+ "set foo 0 0 3\r\nbar\r\n", "STORED\r\n", True)
+
+
+ def test_add(self):
+ """
+ L{MemCacheProtocol.add} should return a L{Deferred} which is
+ called back with C{True} when the operation succeeds.
+ """
+ return self._test(self.proto.add("foo", "bar"),
+ "add foo 0 0 3\r\nbar\r\n", "STORED\r\n", True)
+
+
+ def test_replace(self):
+ """
+ L{MemCacheProtocol.replace} should return a L{Deferred} which
+ is called back with C{True} when the operation succeeds.
+ """
+ return self._test(self.proto.replace("foo", "bar"),
+ "replace foo 0 0 3\r\nbar\r\n", "STORED\r\n", True)
+
+
+ def test_errorAdd(self):
+ """
+ Test an erroneous add: if a L{MemCacheProtocol.add} is called but the
+ key already exists on the server, it returns a B{NOT STORED} answer,
+ which should callback the resulting L{Deferred} with C{False}.
+ """
+ return self._test(self.proto.add("foo", "bar"),
+ "add foo 0 0 3\r\nbar\r\n", "NOT STORED\r\n", False)
+
+
+ def test_errorReplace(self):
+ """
+ Test an erroneous replace: if a L{MemCacheProtocol.replace} is called
+ but the key doesn't exist on the server, it returns a B{NOT STORED}
+ answer, which should callback the resulting L{Deferred} with C{False}.
+ """
+ return self._test(self.proto.replace("foo", "bar"),
+ "replace foo 0 0 3\r\nbar\r\n", "NOT STORED\r\n", False)
+
+
+ def test_delete(self):
+ """
+ L{MemCacheProtocol.delete} should return a L{Deferred} which is
+ called back with C{True} when the server notifies a success.
+ """
+ return self._test(self.proto.delete("bar"), "delete bar\r\n",
+ "DELETED\r\n", True)
+
+
+ def test_errorDelete(self):
+ """
+ Test a error during a delete: if key doesn't exist on the server, it
+ returns a B{NOT FOUND} answer which should callback the resulting
+ L{Deferred} with C{False}.
+ """
+ return self._test(self.proto.delete("bar"), "delete bar\r\n",
+ "NOT FOUND\r\n", False)
+
+
+ def test_increment(self):
+ """
+ Test incrementing a variable: L{MemCacheProtocol.increment} should
+ return a L{Deferred} which is called back with the incremented value of
+ the given key.
+ """
+ return self._test(self.proto.increment("foo"), "incr foo 1\r\n",
+ "4\r\n", 4)
+
+
+ def test_decrement(self):
+ """
+ Test decrementing a variable: L{MemCacheProtocol.decrement} should
+ return a L{Deferred} which is called back with the decremented value of
+ the given key.
+ """
+ return self._test(
+ self.proto.decrement("foo"), "decr foo 1\r\n", "5\r\n", 5)
+
+
+ def test_incrementVal(self):
+ """
+ L{MemCacheProtocol.increment} takes an optional argument C{value} which
+ should replace the default value of 1 when specified.
+ """
+ return self._test(self.proto.increment("foo", 8), "incr foo 8\r\n",
+ "4\r\n", 4)
+
+
+ def test_decrementVal(self):
+ """
+ L{MemCacheProtocol.decrement} takes an optional argument C{value} which
+ should replace the default value of 1 when specified.
+ """
+ return self._test(self.proto.decrement("foo", 3), "decr foo 3\r\n",
+ "5\r\n", 5)
+
+
+ def test_stats(self):
+ """
+ Test retrieving server statistics via the L{MemCacheProtocol.stats}
+ command: it should parse the data sent by the server and call back the
+ resulting L{Deferred} with a dictionary of the received statistics.
+ """
+ return self._test(self.proto.stats(), "stats\r\n",
+ "STAT foo bar\r\nSTAT egg spam\r\nEND\r\n",
+ {"foo": "bar", "egg": "spam"})
+
+
+ def test_version(self):
+ """
+ Test version retrieval via the L{MemCacheProtocol.version} command: it
+ should return a L{Deferred} which is called back with the version sent
+ by the server.
+ """
+ return self._test(self.proto.version(), "version\r\n",
+ "VERSION 1.1\r\n", "1.1")
+
+
+ def test_flushAll(self):
+ """
+ L{MemCacheProtocol.flushAll} should return a L{Deferred} which is
+ called back with C{True} if the server acknowledges success.
+ """
+ return self._test(self.proto.flushAll(), "flush_all\r\n",
+ "OK\r\n", True)
+
+
+ def test_invalidGetResponse(self):
+ """
+ If the value returned doesn't match the expected key of the current, we
+ should get an error in L{MemCacheProtocol.dataReceived}.
+ """
+ self.proto.get("foo")
+ s = "spamegg"
+ self.assertRaises(RuntimeError,
+ self.proto.dataReceived,
+ "VALUE bar 0 %s\r\n%s\r\nEND\r\n" % (len(s), s))
+
+
+ def test_timeOut(self):
+ """
+ Test the timeout on outgoing requests: when timeout is detected, all
+ current commands should fail with a L{TimeoutError}, and the
+ connection should be closed.
+ """
+ d1 = self.proto.get("foo")
+ d2 = self.proto.get("bar")
+ d3 = Deferred()
+ self.proto.connectionLost = d3.callback
+
+ self.clock.advance(self.proto.persistentTimeOut)
+ self.assertFailure(d1, TimeoutError)
+ self.assertFailure(d2, TimeoutError)
+ def checkMessage(error):
+ self.assertEquals(str(error), "Connection timeout")
+ d1.addCallback(checkMessage)
+ return gatherResults([d1, d2, d3])
+
+
+ def test_timeoutRemoved(self):
+ """
+ When a request gets a response, no pending timeout call should remain
+ around.
+ """
+ d = self.proto.get("foo")
+
+ self.clock.advance(self.proto.persistentTimeOut - 1)
+ self.proto.dataReceived("VALUE foo 0 3\r\nbar\r\nEND\r\n")
+
+ def check(result):
+ self.assertEquals(result, (0, "bar"))
+ self.assertEquals(len(self.clock.calls), 0)
+ d.addCallback(check)
+ return d
+
+
+ def test_timeOutRaw(self):
+ """
+ Test the timeout when raw mode was started: the timeout should not be
+ reset until all the data has been received, so we can have a
+ L{TimeoutError} when waiting for raw data.
+ """
+ d1 = self.proto.get("foo")
+ d2 = Deferred()
+ self.proto.connectionLost = d2.callback
+
+ self.proto.dataReceived("VALUE foo 0 10\r\n12345")
+ self.clock.advance(self.proto.persistentTimeOut)
+ self.assertFailure(d1, TimeoutError)
+ return gatherResults([d1, d2])
+
+
+ def test_timeOutStat(self):
+ """
+ Test the timeout when stat command has started: the timeout should not
+ be reset until the final B{END} is received.
+ """
+ d1 = self.proto.stats()
+ d2 = Deferred()
+ self.proto.connectionLost = d2.callback
+
+ self.proto.dataReceived("STAT foo bar\r\n")
+ self.clock.advance(self.proto.persistentTimeOut)
+ self.assertFailure(d1, TimeoutError)
+ return gatherResults([d1, d2])
+
+
+ def test_timeoutPipelining(self):
+ """
+ When two requests are sent, a timeout call should remain around for the
+ second request, and its timeout time should be correct.
+ """
+ d1 = self.proto.get("foo")
+ d2 = self.proto.get("bar")
+ d3 = Deferred()
+ self.proto.connectionLost = d3.callback
+
+ self.clock.advance(self.proto.persistentTimeOut - 1)
+ self.proto.dataReceived("VALUE foo 0 3\r\nbar\r\nEND\r\n")
+
+ def check(result):
+ self.assertEquals(result, (0, "bar"))
+ self.assertEquals(len(self.clock.calls), 1)
+ for i in range(self.proto.persistentTimeOut):
+ self.clock.advance(1)
+ return self.assertFailure(d2, TimeoutError).addCallback(checkTime)
+ def checkTime(ignored):
+ # Check that the timeout happened C{self.proto.persistentTimeOut}
+ # after the last response
+ self.assertEquals(self.clock.seconds(),
+ 2 * self.proto.persistentTimeOut - 1)
+ d1.addCallback(check)
+ return d1
+
+
+ def test_timeoutNotReset(self):
+ """
+ Check that timeout is not resetted for every command, but keep the
+ timeout from the first command without response.
+ """
+ d1 = self.proto.get("foo")
+ d3 = Deferred()
+ self.proto.connectionLost = d3.callback
+
+ self.clock.advance(self.proto.persistentTimeOut - 1)
+ d2 = self.proto.get("bar")
+ self.clock.advance(1)
+ self.assertFailure(d1, TimeoutError)
+ self.assertFailure(d2, TimeoutError)
+ return gatherResults([d1, d2, d3])
+
+
+ def test_tooLongKey(self):
+ """
+ Test that an error is raised when trying to use a too long key: the
+ called command should return a L{Deferred} which fail with a
+ L{ClientError}.
+ """
+ d1 = self.assertFailure(self.proto.set("a" * 500, "bar"), ClientError)
+ d2 = self.assertFailure(self.proto.increment("a" * 500), ClientError)
+ d3 = self.assertFailure(self.proto.get("a" * 500), ClientError)
+ d4 = self.assertFailure(self.proto.append("a" * 500, "bar"), ClientError)
+ d5 = self.assertFailure(self.proto.prepend("a" * 500, "bar"), ClientError)
+ return gatherResults([d1, d2, d3, d4, d5])
+
+
+ def test_invalidCommand(self):
+ """
+ When an unknown command is sent directly (not through public API), the
+ server answers with an B{ERROR} token, and the command should fail with
+ L{NoSuchCommand}.
+ """
+ d = self.proto._set("egg", "foo", "bar", 0, 0, "")
+ self.assertEquals(self.transport.value(), "egg foo 0 0 3\r\nbar\r\n")
+ self.assertFailure(d, NoSuchCommand)
+ self.proto.dataReceived("ERROR\r\n")
+ return d
+
+
+ def test_clientError(self):
+ """
+ Test the L{ClientError} error: when the server send a B{CLIENT_ERROR}
+ token, the originating command should fail with L{ClientError}, and the
+ error should contain the text sent by the server.
+ """
+ a = "eggspamm"
+ d = self.proto.set("foo", a)
+ self.assertEquals(self.transport.value(),
+ "set foo 0 0 8\r\neggspamm\r\n")
+ self.assertFailure(d, ClientError)
+ def check(err):
+ self.assertEquals(str(err), "We don't like egg and spam")
+ d.addCallback(check)
+ self.proto.dataReceived("CLIENT_ERROR We don't like egg and spam\r\n")
+ return d
+
+
+ def test_serverError(self):
+ """
+ Test the L{ServerError} error: when the server send a B{SERVER_ERROR}
+ token, the originating command should fail with L{ServerError}, and the
+ error should contain the text sent by the server.
+ """
+ a = "eggspamm"
+ d = self.proto.set("foo", a)
+ self.assertEquals(self.transport.value(),
+ "set foo 0 0 8\r\neggspamm\r\n")
+ self.assertFailure(d, ServerError)
+ def check(err):
+ self.assertEquals(str(err), "zomg")
+ d.addCallback(check)
+ self.proto.dataReceived("SERVER_ERROR zomg\r\n")
+ return d
+
+
+ def test_unicodeKey(self):
+ """
+ Using a non-string key as argument to commands should raise an error.
+ """
+ d1 = self.assertFailure(self.proto.set(u"foo", "bar"), ClientError)
+ d2 = self.assertFailure(self.proto.increment(u"egg"), ClientError)
+ d3 = self.assertFailure(self.proto.get(1), ClientError)
+ d4 = self.assertFailure(self.proto.delete(u"bar"), ClientError)
+ d5 = self.assertFailure(self.proto.append(u"foo", "bar"), ClientError)
+ d6 = self.assertFailure(self.proto.prepend(u"foo", "bar"), ClientError)
+ return gatherResults([d1, d2, d3, d4, d5, d6])
+
+
+ def test_unicodeValue(self):
+ """
+ Using a non-string value should raise an error.
+ """
+ return self.assertFailure(self.proto.set("foo", u"bar"), ClientError)
+
+
+ def test_pipelining(self):
+ """
+ Test that multiple requests can be sent subsequently to the server, and
+ that the protocol order the responses correctly and dispatch to the
+ corresponding client command.
+ """
+ d1 = self.proto.get("foo")
+ d1.addCallback(self.assertEquals, (0, "bar"))
+ d2 = self.proto.set("bar", "spamspamspam")
+ d2.addCallback(self.assertEquals, True)
+ d3 = self.proto.get("egg")
+ d3.addCallback(self.assertEquals, (0, "spam"))
+ self.assertEquals(self.transport.value(),
+ "get foo\r\nset bar 0 0 12\r\nspamspamspam\r\nget egg\r\n")
+ self.proto.dataReceived("VALUE foo 0 3\r\nbar\r\nEND\r\n"
+ "STORED\r\n"
+ "VALUE egg 0 4\r\nspam\r\nEND\r\n")
+ return gatherResults([d1, d2, d3])
+
+
+ def test_getInChunks(self):
+ """
+ If the value retrieved by a C{get} arrive in chunks, the protocol
+ should be able to reconstruct it and to produce the good value.
+ """
+ d = self.proto.get("foo")
+ d.addCallback(self.assertEquals, (0, "0123456789"))
+ self.assertEquals(self.transport.value(), "get foo\r\n")
+ self.proto.dataReceived("VALUE foo 0 10\r\n0123456")
+ self.proto.dataReceived("789")
+ self.proto.dataReceived("\r\nEND")
+ self.proto.dataReceived("\r\n")
+ return d
+
+
+ def test_append(self):
+ """
+ L{MemCacheProtocol.append} behaves like a L{MemCacheProtocol.set}
+ method: it should return a L{Deferred} which is called back with
+ C{True} when the operation succeeds.
+ """
+ return self._test(self.proto.append("foo", "bar"),
+ "append foo 0 0 3\r\nbar\r\n", "STORED\r\n", True)
+
+
+ def test_prepend(self):
+ """
+ L{MemCacheProtocol.prepend} behaves like a L{MemCacheProtocol.set}
+ method: it should return a L{Deferred} which is called back with
+ C{True} when the operation succeeds.
+ """
+ return self._test(self.proto.prepend("foo", "bar"),
+ "prepend foo 0 0 3\r\nbar\r\n", "STORED\r\n", True)
+
+
+ def test_gets(self):
+ """
+ L{MemCacheProtocol.get} should handle an additional cas result when
+ C{withIdentifier} is C{True} and forward it in the resulting
+ L{Deferred}.
+ """
+ return self._test(self.proto.get("foo", True), "gets foo\r\n",
+ "VALUE foo 0 3 1234\r\nbar\r\nEND\r\n", (0, "1234", "bar"))
+
+
+ def test_emptyGets(self):
+ """
+ Test getting a non-available key with gets: it should succeed but
+ return C{None} as value, C{0} as flag and an empty cas value.
+ """
+ return self._test(self.proto.get("foo", True), "gets foo\r\n",
+ "END\r\n", (0, "", None))
+
+
+ def test_checkAndSet(self):
+ """
+ L{MemCacheProtocol.checkAndSet} passes an additional cas identifier that the
+ server should handle to check if the data has to be updated.
+ """
+ return self._test(self.proto.checkAndSet("foo", "bar", cas="1234"),
+ "cas foo 0 0 3 1234\r\nbar\r\n", "STORED\r\n", True)
+
+
+ def test_casUnknowKey(self):
+ """
+ When L{MemCacheProtocol.checkAndSet} response is C{EXISTS}, the resulting
+ L{Deferred} should fire with C{False}.
+ """
+ return self._test(self.proto.checkAndSet("foo", "bar", cas="1234"),
+ "cas foo 0 0 3 1234\r\nbar\r\n", "EXISTS\r\n", False)
Modified: CalendarServer/trunk/twistedcaldav/test/test_resource.py
===================================================================
--- CalendarServer/trunk/twistedcaldav/test/test_resource.py 2008-05-24 00:47:17 UTC (rev 2472)
+++ CalendarServer/trunk/twistedcaldav/test/test_resource.py 2008-05-24 00:57:54 UTC (rev 2473)
@@ -31,28 +31,9 @@
def setUp(self):
self.resource = CalDAVResource()
self.resource._dead_properties = InMemoryPropertyStore()
- self.resource.cacheNotifier = StubCacheChangeNotifier()
-
- def test_writeDeadPropertyNotifiesCache(self):
- self.resource.writeDeadProperty(StubProperty())
- self.assertEquals(self.resource.cacheNotifier.changedCount, 1)
-
-
def test_writeDeadPropertyWritesProperty(self):
prop = StubProperty()
self.resource.writeDeadProperty(prop)
self.assertEquals(self.resource._dead_properties.get("StubQname"),
prop)
-
-
- def test_writeDeadPropertyFailureDoesntNotifyCache(self):
- self.resource._dead_properties = None
- self.assertRaises(Exception,
- self.resource.writeDeadProperty, StubProperty())
- self.assertEquals(self.resource.cacheNotifier.changedCount, 0)
-
-
- def test_writeDeadPropertyDoesntFailWithoutACacheNotifier(self):
- del self.resource.cacheNotifier
- self.resource.writeDeadProperty(StubProperty())
Modified: CalendarServer/trunk/twistedcaldav/test/test_static.py
===================================================================
--- CalendarServer/trunk/twistedcaldav/test/test_static.py 2008-05-24 00:47:17 UTC (rev 2472)
+++ CalendarServer/trunk/twistedcaldav/test/test_static.py 2008-05-24 00:57:54 UTC (rev 2473)
@@ -17,7 +17,7 @@
from twisted.trial.unittest import TestCase
from twistedcaldav.static import CalendarHomeFile, CalDAVFile
-from twistedcaldav.cache import CacheChangeNotifier
+from twistedcaldav.cache import XattrCacheChangeNotifier
from twistedcaldav.test.util import StubCacheChangeNotifier
class StubParentResource(object):
@@ -34,7 +34,7 @@
def test_hasCacheNotifier(self):
self.failUnless(isinstance(self.calendarHome.cacheNotifier,
- CacheChangeNotifier))
+ XattrCacheChangeNotifier))
def test_childrenHaveCacheNotifier(self):
@@ -52,10 +52,15 @@
def test_updateCTagNotifiesCache(self):
- self.caldavFile.updateCTag()
- self.assertEquals(self.caldavFile.cacheNotifier.changedCount, 1)
+ d = self.caldavFile.updateCTag()
+ d.addCallback(
+ lambda _:
+ self.assertEquals(self.caldavFile.cacheNotifier.changedCount, 1)
+ )
+ return d
def test_updateCTagDoesntFailWithoutACacheNotifier(self):
del self.caldavFile.cacheNotifier
- self.caldavFile.updateCTag()
+ d = self.caldavFile.updateCTag()
+ return d
Modified: CalendarServer/trunk/twistedcaldav/test/util.py
===================================================================
--- CalendarServer/trunk/twistedcaldav/test/util.py 2008-05-24 00:47:17 UTC (rev 2472)
+++ CalendarServer/trunk/twistedcaldav/test/util.py 2008-05-24 00:57:54 UTC (rev 2473)
@@ -17,6 +17,9 @@
import twisted.web2.dav.test.util
from twisted.web2.http import HTTPError, StatusResponse
+from twisted.internet.defer import succeed
+from twisted.python.filepath import FilePath
+
from twistedcaldav.static import CalDAVFile
@@ -26,7 +29,14 @@
class InMemoryPropertyStore(object):
def __init__(self):
+ class _FauxPath(object):
+ path = ':memory:'
+
+ class _FauxResource(object):
+ fp = _FauxPath()
+
self._properties = {}
+ self.resource = _FauxResource()
def get(self, qname):
data = self._properties.get(qname)
@@ -43,3 +53,4 @@
def changed(self):
self.changedCount += 1
+ return succeed(True)
-------------- next part --------------
An HTML attachment was scrubbed...
URL: http://lists.macosforge.org/pipermail/calendarserver-changes/attachments/20080523/7c1a5d2f/attachment-0001.htm
More information about the calendarserver-changes
mailing list