Hoping that silence (or apathy?) is consent here. :) Adding an event to...

Hoping that silence (or apathy?) is consent here. :) Adding an event to indicate when streaming is starting in case of chunked/streamed responses using response.write()

doc/CHANGES.rst

@@ -17,6 +17,11 @@ Features Added
   - ExtensionClass = 2.13.0
   - Persistence = 2.13.0
 
+- There is now an event ZPublisher.interfaces.IPubBeforeStreaming which will
+  be fired just before the first chunk of data is written to the response
+  stream when using the write() method on the response. This is the last
+  possible point at which response headers may be set in this case.
+
 Bugs Fixed
 ++++++++++
 

src/ZPublisher/HTTPResponse.py

@@ -18,10 +18,12 @@ __version__ = '$Revision: 1.81 $'[11:-2]
 import types, os, sys, re
 import zlib, struct
 from string import translate, maketrans
+from zope.event import notify
 from BaseResponse import BaseResponse
 from zExceptions import Unauthorized, Redirect
 from zExceptions.ExceptionFormatter import format_exception
 from ZPublisher import BadRequest, InternalError, NotFound
+from ZPublisher.pubevents import PubBeforeStreaming
 from cgi import escape
 from urllib import quote
 
@@ -921,6 +923,9 @@ class HTTPResponse(BaseResponse):
 
         """
         if not self._wrote:
+            
+            notify(PubBeforeStreaming(self))
+            
             self.outputBody()
             self._wrote = 1
             self.stdout.flush()

src/ZPublisher/interfaces.py

@@ -50,3 +50,11 @@ class IPubBeforeAbort(IPubEvent):
     """
     exc_info = Attribute('''The exception info as returned by 'sys.exc_info()'.''')
     retry = Attribute('Whether the request will be retried')
+
+class IPubBeforeStreaming(Interface):
+    """Event fired just before a streaming response is initiated, i.e. when
+    something calls response.write() for the first time. Note that this is
+    carries a reference to the *response*, not the request.
+    """
+    
+    response = Attribute(u"The current HTTP response")

src/ZPublisher/pubevents.py

@@ -10,7 +10,8 @@ for detailed time related analysis, inline request monitoring.
 from zope.interface import implements
 
 from interfaces import IPubStart, IPubSuccess, IPubFailure, \
-     IPubAfterTraversal, IPubBeforeCommit, IPubBeforeAbort
+     IPubAfterTraversal, IPubBeforeCommit, IPubBeforeAbort, \
+     IPubBeforeStreaming
 
 class _Base(object):
     """PubEvent base class."""
@@ -49,3 +50,11 @@ class PubBeforeAbort(_Base):
     
     def __init__(self, request, exc_info, retry):
         self.request, self.exc_info, self.retry = request, exc_info, retry
+
+class PubBeforeStreaming(object):
+    """Notified immediately before streaming via response.write() commences
+    """
+    implements(IPubBeforeStreaming)
+    
+    def __init__(self, response):
+        self.response = response

src/ZPublisher/tests/testpubevents.py

+from StringIO import StringIO
 from sys import modules, exc_info
 from unittest import TestCase, TestSuite, makeSuite, main
 
@@ -7,11 +8,14 @@ from zope.event import subscribers
 
 from ZPublisher.Publish import publish, Retry
 from ZPublisher.BaseRequest import BaseRequest
+from ZPublisher.HTTPResponse import HTTPResponse
 from ZPublisher.pubevents import PubStart, PubSuccess, PubFailure, \
-     PubAfterTraversal, PubBeforeCommit, PubBeforeAbort
+     PubAfterTraversal, PubBeforeCommit, PubBeforeAbort, \
+     PubBeforeStreaming
 from ZPublisher.interfaces import \
      IPubStart, IPubEnd, IPubSuccess, IPubFailure, \
-     IPubAfterTraversal, IPubBeforeCommit
+     IPubAfterTraversal, IPubBeforeCommit, \
+     IPubBeforeStreaming
 
 PUBMODULE = 'TEST_testpubevents'
 
@@ -41,7 +45,10 @@ class TestInterface(TestCase):
     def testBeforeCommit(self):
         e = PubBeforeCommit(_Request())
         verifyObject(IPubBeforeCommit, e)
-
+    
+    def testBeforeStreaming(self):
+        e = PubBeforeStreaming(_Response())
+        verifyObject(IPubBeforeStreaming, e)
         
 class TestPubEvents(TestCase):
     def setUp(self):
@@ -127,6 +134,21 @@ class TestPubEvents(TestCase):
         self.assert_(isinstance(events[5], PubBeforeCommit))
         self.assert_(isinstance(events[6], PubSuccess))
 
+    def testStreaming(self):
+        
+        out = StringIO()
+        response = HTTPResponse(stdout=out)
+        response.write('datachunk1')
+        response.write('datachunk2')
+        
+        events = self.reporter.events
+        self.assertEqual(len(events), 1)
+        self.assert_(isinstance(events[0], PubBeforeStreaming))
+        self.assertEqual(events[0].response, response)
+        
+        self.failUnless('datachunk1datachunk2' in out.getvalue())
+
+
 # Auxiliaries
 def _succeed():
     ''' '''

src/ZServer/HTTPResponse.py

@@ -20,8 +20,10 @@ and logging duties.
 import time, re,  sys, tempfile
 from cStringIO import StringIO
 import thread
+from zope.event import notify
 from ZPublisher.HTTPResponse import HTTPResponse
 from ZPublisher.Iterators import IStreamIterator
+from ZPublisher.pubevents import PubBeforeStreaming
 from medusa.http_date import build_http_date
 from PubCore.ZEvent import Wakeup
 from medusa.producers import hooked_producer
@@ -165,6 +167,9 @@ class ZServerHTTPResponse(HTTPResponse):
         stdout=self.stdout
 
         if not self._wrote:
+            
+            notify(PubBeforeStreaming(self))
+            
             l=self.headers.get('content-length', None)
             if l is not None:
                 try:

src/ZServer/tests/test_responses.py

@@ -19,17 +19,21 @@ from ZServer.FTPResponse import FTPResponse
 from ZServer.PCGIServer import PCGIResponse
 from ZServer.FCGIServer import FCGIResponse
 from ZPublisher.Iterators import IStreamIterator
+from ZPublisher.pubevents import PubBeforeStreaming
 from zope.interface import implements
 import unittest
 from cStringIO import StringIO
 
+from zope.event import subscribers
+
+
 class ZServerResponseTestCase(unittest.TestCase):
     """Test ZServer response objects."""
 
     def test_http_response_write_unicode(self):
         response = ZServerHTTPResponse()
         self.assertRaises(TypeError, response.write, u'bad')
-
+    
     def test_ftp_response_write_unicode(self):
         response = FTPResponse()
         self.assertRaises(TypeError, response.write, u'bad')
@@ -57,7 +61,7 @@ class ZServerResponseTestCase(unittest.TestCase):
         one = ZServerHTTPResponse(stdout=DummyChannel())
         self.assertRaises(AssertionError,
                           one.setBody, test_streamiterator())
-        
+    
 class DummyChannel:
     def __init__(self):
         self.out = StringIO()
@@ -267,12 +271,39 @@ class ZServerHTTPResponseTestCase(unittest.TestCase):
                                        '',
                                        ''))
 
+class _Reporter(object):
+    def __init__(self): self.events = []
+    def __call__(self, event): self.events.append(event)
+
+class ZServerHTTPResponseEventsTestCase(unittest.TestCase):
+
+    def setUp(self):
+        self._saved_subscribers = subscribers[:]
+        self.reporter = r = _Reporter()
+        subscribers[:] = [r]
+
+    def tearDown(self):
+        subscribers[:] = self._saved_subscribers
+    
+    def testStreaming(self):
+        out = StringIO()
+        response = ZServerHTTPResponse(stdout=out)
+        response.write('datachunk1')
+        response.write('datachunk2')
+        
+        events = self.reporter.events
+        self.assertEqual(len(events), 1)
+        self.assert_(isinstance(events[0], PubBeforeStreaming))
+        self.assertEqual(events[0].response, response)
+        
+        self.failUnless('datachunk1datachunk2' in out.getvalue())
 
 def test_suite():
     suite = unittest.TestSuite()
     suite.addTests((
         unittest.makeSuite(ZServerResponseTestCase),
-        unittest.makeSuite(ZServerHTTPResponseTestCase)
+        unittest.makeSuite(ZServerHTTPResponseTestCase),
+        unittest.makeSuite(ZServerHTTPResponseEventsTestCase)
     ))
     return suite