Tuesday, May 3, 2011

Using Twisted's twisted.web classes, how do I flush my outgoing buffers?

I've made a simple http server using Twisted, which sends the Content-Type: multipart/x-mixed-replace header. I'm using this to test an http client which I want to set up to accept a long-term stream.

The problem that has arisen is that my client request hangs until the http.Request calls self.finish(), then it receives all multipart documents at once.

Is there a way to manually flush the output buffers down to the client? I'm assuming this is why I'm not receiving the individual multipart documents.

#!/usr/bin/env python

import time

from twisted.web import http
from twisted.internet import protocol

class StreamHandler(http.Request):
    BOUNDARY = 'BOUNDARY'

    def writeBoundary(self):
        self.write("--%s\n" % (self.BOUNDARY))

    def writeStop(self):
        self.write("--%s--\n" % (self.BOUNDARY))

    def process(self):
        self.setHeader('Connection', 'Keep-Alive')
        self.setHeader('Content-Type', "multipart/x-mixed-replace;boundary=%s" % (self.BOUNDARY))

        self.writeBoundary()

        self.write("Content-Type: text/html\n")
        s = "<html>foo</html>\n"
        self.write("Content-Length: %s\n\n" % (len(s)))
        self.write(s)
        self.writeBoundary()
        time.sleep(2)

        self.write("Content-Type: text/html\n")
        s = "<html>bar</html>\n"
        self.write("Content-Length: %s\n\n" % (len(s)))
        self.write(s)
        self.writeBoundary()
        time.sleep(2)

        self.write("Content-Type: text/html\n")
        s = "<html>baz</html>\n"
        self.write("Content-Length: %s\n\n" % (len(s)))
        self.write(s)

        self.writeStop()

        self.finish()

class StreamProtocol(http.HTTPChannel):
    requestFactory = StreamHandler

class StreamFactory(http.HTTPFactory):
    protocol = StreamProtocol


if __name__ == '__main__':
    from twisted.internet import reactor
    reactor.listenTCP(8800, StreamFactory())
    reactor.run()
From stackoverflow
  • The reason seems to be explained in the FAQ for twisted. The twisted server does not actually write anything to the underlining connection until the reactor thread is free to run, in this case at the end of your method. However you can use reactor.doSelect(timeout) before each of your sleeps to make the reactor write what it has to the connection.

    Jean-Paul Calderone : You should never call reactor.doSelect. This isn't portable across reactors, and it could easily break the reactor by re-entering it where it isn't expecting to be re-entered.
    Mick : Notwithstanding the comment/correction around doSelect above, for anyone trying to figure out what is happening with their transport code the pointer to the FAQ is spot on - in particular "Twisted can only send data after you give up control of execution to the reactor. For example, if you have an infinite loop writing data to a transport, the data will never actually be sent since control will never leave your code and return to the reactor."
  • Using time.sleep() prevents twisted from doing its job. To make it work you can't use time.sleep(), you must return control to twisted instead. The easiest way to modify your existing code to do that is by using twisted.internet.defer.inlineCallbacks, which is the next best thing since sliced bread:

    #!/usr/bin/env python
    
    import time
    
    from twisted.web import http
    from twisted.internet import protocol
    from twisted.internet import reactor
    from twisted.internet import defer
    
    def wait(seconds, result=None):
        """Returns a deferred that will be fired later"""
        d = defer.Deferred()
        reactor.callLater(seconds, d.callback, result)
        return d
    
    class StreamHandler(http.Request):
        BOUNDARY = 'BOUNDARY'
    
        def writeBoundary(self):
            self.write("--%s\n" % (self.BOUNDARY))
    
        def writeStop(self):
            self.write("--%s--\n" % (self.BOUNDARY))
    
        @defer.inlineCallbacks
        def process(self):
            self.setHeader('Connection', 'Keep-Alive')
            self.setHeader('Content-Type', "multipart/x-mixed-replace;boundary=%s" % (self.BOUNDARY))
    
            self.writeBoundary()
    
            self.write("Content-Type: text/html\n")
            s = "<html>foo</html>\n"
            self.write("Content-Length: %s\n\n" % (len(s)))
            self.write(s)
            self.writeBoundary()
    
    
            yield wait(2)
    
            self.write("Content-Type: text/html\n")
            s = "<html>bar</html>\n"
            self.write("Content-Length: %s\n\n" % (len(s)))
            self.write(s)
            self.writeBoundary()
    
            yield wait(2)
    
            self.write("Content-Type: text/html\n")
            s = "<html>baz</html>\n"
            self.write("Content-Length: %s\n\n" % (len(s)))
            self.write(s)
    
            self.writeStop()
    
            self.finish()
    
    
    class StreamProtocol(http.HTTPChannel):
        requestFactory = StreamHandler
    
    class StreamFactory(http.HTTPFactory):
        protocol = StreamProtocol
    
    
    if __name__ == '__main__':   
        reactor.listenTCP(8800, StreamFactory())
        reactor.run()
    

    That works in firefox, I guess it answers your question correctly.

    Mick : Does this effectively allow any other (as selected by reactor) code run when you are yielded? If so you would need to be very careful that data you are using is not overwritten or altered by the other code. See the comment at the URL below which I have not seen refuted anywhere: http://twistedmatrix.com/pipermail/twisted-python/2007-February/014869.html

0 comments:

Post a Comment