Merge chrism-publishfile-branch. See...

Merge chrism-publishfile-branch.  See http://dev.zope.org/Wikis/DevSite/Proposals/FasterStaticContentServing for more information.

lib/python/OFS/Image.py

@@ -12,7 +12,7 @@
 ##############################################################################
 """Image object"""
 
-__version__='$Revision: 1.150 $'[11:-2]
+__version__='$Revision: 1.151 $'[11:-2]
 
 import Globals, struct
 from OFS.content_types import guess_content_type
@@ -31,6 +31,7 @@ from Cache import Cacheable
 from mimetools import choose_boundary
 from ZPublisher import HTTPRangeSupport
 from ZPublisher.HTTPRequest import FileUpload
+from ZPublisher.Iterators import filestream_iterator
 from zExceptions import Redirect
 from cgi import escape
 
@@ -127,14 +128,9 @@ class File(Persistent, Implicit, PropertyManager,
     def id(self):
         return self.__name__
 
-    def index_html(self, REQUEST, RESPONSE):
-        """
-        The default view of the contents of a File or Image.
-
-        Returns the contents of the file or image.  Also, sets the
-        Content-Type HTTP header to the objects content type.
-        """
-        # HTTP If-Modified-Since header handling.
+    def _if_modified_since_request_handler(self, REQUEST, RESPONSE):
+        # HTTP If-Modified-Since header handling: return True if
+        # we can handle this request by returning a 304 response
         header=REQUEST.get_header('If-Modified-Since', None)
         if header is not None:
             header=header.split( ';')[0]
@@ -154,27 +150,19 @@ class File(Persistent, Implicit, PropertyManager,
                 else:
                     last_mod = long(0)
                 if last_mod > 0 and last_mod <= mod_since:
-                    # Set header values since apache caching will return Content-Length
-                    # of 0 in response if size is not set here
-                    RESPONSE.setHeader('Last-Modified', rfc1123_date(self._p_mtime))
+                    # Set header values since apache caching will return
+                    # Content-Length of 0 in response if size is not set here
+                    RESPONSE.setHeader('Last-Modified',
+                                       rfc1123_date(self._p_mtime))
                     RESPONSE.setHeader('Content-Type', self.content_type)
                     RESPONSE.setHeader('Content-Length', self.size)
                     RESPONSE.setHeader('Accept-Ranges', 'bytes')
                     RESPONSE.setStatus(304)
-                    self.ZCacheable_set(None)
-                    return ''
+                    return True
 
-        if self.precondition and hasattr(self, str(self.precondition)):
-            # Grab whatever precondition was defined and then
-            # execute it.  The precondition will raise an exception
-            # if something violates its terms.
-            c=getattr(self, str(self.precondition))
-            if hasattr(c,'isDocTemp') and c.isDocTemp:
-                c(REQUEST['PARENTS'][1],REQUEST)
-            else:
-                c()
-
-        # HTTP Range header handling
+    def _range_request_handler(self, REQUEST, RESPONSE):
+        # HTTP Range header handling: return True if we've served a range
+        # chunk out of our data.
         range = REQUEST.get_header('Range', None)
         request_range = REQUEST.get_header('Request-Range', None)
         if request_range is not None:
@@ -228,7 +216,7 @@ class File(Persistent, Implicit, PropertyManager,
                     RESPONSE.setHeader('Content-Type', self.content_type)
                     RESPONSE.setHeader('Content-Length', self.size)
                     RESPONSE.setStatus(416)
-                    return ''
+                    return True
 
                 ranges = HTTPRangeSupport.expandRanges(ranges, self.size)
                                 
@@ -248,7 +236,8 @@ class File(Persistent, Implicit, PropertyManager,
 
                     data = self.data
                     if type(data) is StringType:
-                        return data[start:end]
+                        RESPONSE.write(data[start:end])
+                        return True
 
                     # Linked Pdata objects. Urgh.
                     pos = 0
@@ -274,7 +263,7 @@ class File(Persistent, Implicit, PropertyManager,
 
                         data = data.next
 
-                    return ''
+                    return True
 
                 else:
                     boundary = choose_boundary()
@@ -364,6 +353,32 @@ class File(Persistent, Implicit, PropertyManager,
                     del pdata_map
 
                     RESPONSE.write('\r\n--%s--\r\n' % boundary)
+                    return True
+
+    def index_html(self, REQUEST, RESPONSE):
+        """
+        The default view of the contents of a File or Image.
+
+        Returns the contents of the file or image.  Also, sets the
+        Content-Type HTTP header to the objects content type.
+        """
+
+        if self._if_modified_since_request_handler(REQUEST, RESPONSE):
+            # we were able to handle this by returning a 304
+            return ''
+
+        if self.precondition and hasattr(self, str(self.precondition)):
+            # Grab whatever precondition was defined and then
+            # execute it.  The precondition will raise an exception
+            # if something violates its terms.
+            c=getattr(self, str(self.precondition))
+            if hasattr(c,'isDocTemp') and c.isDocTemp:
+                c(REQUEST['PARENTS'][1],REQUEST)
+            else:
+                c()
+
+        if self._range_request_handler(REQUEST, RESPONSE):
+            # we served a chunk of content in response to a range request.
             return ''
 
         RESPONSE.setHeader('Last-Modified', rfc1123_date(self._p_mtime))
@@ -371,8 +386,15 @@ class File(Persistent, Implicit, PropertyManager,
         RESPONSE.setHeader('Content-Length', self.size)
         RESPONSE.setHeader('Accept-Ranges', 'bytes')
 
-        # Don't cache the data itself, but provide an opportunity
-        # for a cache manager to set response headers.
+        if self.ZCacheable_isCachingEnabled():
+            result = self.ZCacheable_get(default=None)
+            if result is not None:
+                # We will always get None from RAMCacheManager and HTTP
+                # Accelerated Cache Manager but we will get
+                # something implementing the IStreamIterator interface
+                # from a "FileCacheManager"
+                return result
+
         self.ZCacheable_set(None)
             
         data=self.data
@@ -400,6 +422,7 @@ class File(Persistent, Implicit, PropertyManager,
         self.size=size
         self.data=data
         self.ZCacheable_invalidate()
+        self.ZCacheable_set(None)
         self.http__refreshEtag()
 
     def manage_edit(self, title, content_type, precondition='',
@@ -556,6 +579,16 @@ class File(Persistent, Implicit, PropertyManager,
 
     def manage_FTPget(self):
         """Return body for ftp."""
+        if self.ZCacheable_isCachingEnabled():
+            result = self.ZCacheable_get(default=None)
+            if result is not None:
+                # We will always get None from RAMCacheManager but we will get
+                # something implementing the IStreamIterator interface
+                # from FileCacheManager.
+                # the content-length is required here by HTTPResponse, even
+                # though FTP doesn't use it.
+                self.REQUEST.RESPONSE.setHeader('Content-Length', self.size)
+                return result
         return str(self.data)
 
 
@@ -721,7 +754,7 @@ class Image(File):
         if content_type is not None: self.content_type = content_type
 
         self.ZCacheable_invalidate()
-
+        self.ZCacheable_set(None)
 
     def __str__(self):
         return self.tag()

lib/python/ZPublisher/Iterators.py

+from Interface import Interface
+
+class IStreamIterator(Interface):
+    def next(self):
+        """
+        Return a sequence of bytes out of the bytestream, or raise
+        StopIeration if we've reached the end of the bytestream.
+        """
+class filestream_iterator(file):
+    """
+    a file subclass which implements an iterator that returns a
+    fixed-sized sequence of bytes.
+    """
+
+    __implements__ = (IStreamIterator,)
+
+    def __init__(self, name, mode='r', bufsize=-1, streamsize=1<<16):
+        file.__init__(self, name, mode, bufsize)
+        self.streamsize = streamsize
+
+    def next(self):
+        data = self.read(self.streamsize)
+        if not data:
+            raise StopIteration
+        return data
+    
+    

lib/python/ZServer/FTPResponse.py

@@ -26,9 +26,6 @@ class FTPResponse(ZServerHTTPResponse):
     """
 
     def __str__(self):
-#        return ZServerHTTPResponse.__str__(self)
-        # ZServerHTTPResponse.__str__(self) return HTTP headers
-        # Why should be send them to the FTP client ??? (ajung)
         return ''
 
     def outputBody(self):

lib/python/ZServer/FTPServer.py

@@ -310,7 +310,19 @@ class zope_ftp_channel(ftp_channel):
         if status==200:
             self.make_xmit_channel()
             if not response._wrote:
-                self.client_dc.push(response.body)
+                # chrism: we explicitly use a large-buffered producer here to
+                # increase speed.  Using "client_dc.push" with the body causes
+                # a simple producer with a buffer size of 512 to be created
+                # to serve the data, and it's very slow
+                # (about 100 times slower than the large-buffered producer)
+                self.client_dc.push_with_producer(
+                    asynchat.simple_producer(response.body, 1<<16))
+                # chrism: if the response has a bodyproducer, it means that
+                # the actual body was likely an empty string.  This happens
+                # typically when someone returns a StreamIterator from
+                # Zope application code.
+                if response._bodyproducer:
+                    self.client_dc.push_with_producer(response._bodyproducer)
             else:
                 for producer in self._response_producers:
                     self.client_dc.push_with_producer(producer)
@@ -647,3 +659,4 @@ class FTPServer(ftp_server):
         # override asyncore limits for nt's listen queue size
         self.accepting = 1
         return self.socket.listen (num)
+

lib/python/ZServer/HTTPResponse.py

@@ -21,13 +21,14 @@ import time, re,  sys, tempfile
 from cStringIO import StringIO
 import thread
 from ZPublisher.HTTPResponse import HTTPResponse
+from ZPublisher.Iterators import IStreamIterator
 from medusa.http_date import build_http_date
 from PubCore.ZEvent import Wakeup
 from medusa.producers import hooked_producer
 from medusa import http_server
 import asyncore
 from Producers import ShutdownProducer, LoggingProducer, CallbackProducer, \
-    file_part_producer, file_close_producer
+    file_part_producer, file_close_producer, iterator_producer
 from types import LongType
 import DebugLogger
 
@@ -49,6 +50,7 @@ class ZServerHTTPResponse(HTTPResponse):
     _streaming=0
     # using chunking transfer-encoding
     _chunking=0
+    _bodyproducer = None
 
     def __str__(self,
                 html_search=re.compile('<html>',re.I).search,
@@ -230,6 +232,22 @@ class ZServerHTTPResponse(HTTPResponse):
         self._retried_response = response
         return response
 
+    def outputBody(self):
+        """Output the response body"""
+        self.stdout.write(str(self))
+        if self._bodyproducer:
+            self.stdout.write(self._bodyproducer, 0)
+
+    def setBody(self, body, title='', is_error=0, **kw):
+        """ Accept either a stream iterator or a string as the body """
+        if IStreamIterator.isImplementedBy(body):
+            assert(self.headers.has_key('content-length'))
+            # wrap the iterator up in a producer that medusa can understand
+            self._bodyproducer = iterator_producer(body)
+            HTTPResponse.setBody(self, '', title, is_error, **kw)
+            return self
+        else:
+            HTTPResponse.setBody(self, body, title, is_error, **kw)
 
 class ChannelPipe:
     """Experimental pipe from ZPublisher to a ZServer Channel.

lib/python/ZServer/Producers.py

@@ -19,14 +19,12 @@ import sys
 
 class ShutdownProducer:
     "shuts down medusa"
-
     def more(self):
         asyncore.close_all()
 
 
 class LoggingProducer:
     "logs request"
-
     def __init__(self, logger, bytes, method='log'):
         self.logger=logger
         self.bytes=bytes
@@ -40,7 +38,6 @@ class LoggingProducer:
 
 class CallbackProducer:
     "Performs a callback in the channel's thread"
-
     def __init__(self, callback):
         self.callback=callback
 
@@ -52,7 +49,6 @@ class CallbackProducer:
 
 class file_part_producer:
     "producer wrapper for part of a file[-like] objects"
-
     # match http_channel's outgoing buffer size
     out_buffer_size = 1<<16
 
@@ -91,7 +87,6 @@ class file_part_producer:
 
         return data
 
-
 class file_close_producer:
     def __init__(self, file):
         self.file=file
@@ -102,3 +97,13 @@ class file_close_producer:
             file.close()
             self.file=None
         return ''
+
+class iterator_producer:
+    def __init__(self, iterator):
+        self.iterator = iterator
+
+    def more(self):
+        try:
+            return self.iterator.next()
+        except StopIteration:
+            return ''

lib/python/ZServer/tests/test_responses.py

@@ -18,8 +18,9 @@ from ZServer.HTTPResponse import ZServerHTTPResponse
 from ZServer.FTPResponse import FTPResponse
 from ZServer.PCGIServer import PCGIResponse
 from ZServer.FCGIServer import FCGIResponse
+from ZPublisher.Iterators import IStreamIterator
 import unittest
-
+from cStringIO import StringIO
 
 class ZServerResponseTestCase(unittest.TestCase):
     """Test ZServer response objects."""
@@ -40,7 +41,56 @@ class ZServerResponseTestCase(unittest.TestCase):
         response = FCGIResponse()
         self.assertRaises(TypeError, response.write, u'bad')
 
+    def test_setBodyIterator(self):
+        channel = DummyChannel()
+        one = ZServerHTTPResponse(stdout=channel)
+        one.setHeader('content-length', 5)
+        one.setBody(test_streamiterator())
+        one.outputBody()
+        all = channel.all()
+        lines = all.split('\r\n')
+        self.assertEqual(lines[-2], '')    # end of headers
+        self.assertEqual(lines[-1], 'hello') # payload
+
+    def test_setBodyIteratorFailsWithoutContentLength(self):
+        one = ZServerHTTPResponse(stdout=DummyChannel())
+        self.assertRaises(AssertionError,
+                          one.setBody, test_streamiterator())
+
+class DummyChannel:
+    def __init__(self):
+        self.out = StringIO()
+
+    def all(self):
+        self.out.seek(0)
+        return self.out.read()
+
+    def read(self):
+        pass
+
+    def write(self, data, len=None):
+        try:
+            if isinstance(data, str):
+                self.out.write(data)
+                return
+        except TypeError:
+            pass
+        while 1:
+            s = data.more()
+            if not s:
+                break
+            self.out.write(s)
+
+class test_streamiterator:
+    __implements__ = IStreamIterator
+    data = "hello"
+    done = 0
 
+    def next(self):
+        if not self.done:
+            self.done = 1
+            return self.data
+        raise StopIteration
 
 def test_suite():
     return unittest.makeSuite(ZServerResponseTestCase)