Streaming responses with Silex.

00_LICENSE

            DO WHAT THE FUCK YOU WANT TO PUBLIC LICENSE
                    Version 2, December 2004

 Copyright (C) 2004 Sam Hocevar <sam@hocevar.net>

 Everyone is permitted to copy and distribute verbatim or modified
 copies of this license document, and changing it is allowed as long
 as the name is changed.

            DO WHAT THE FUCK YOU WANT TO PUBLIC LICENSE
   TERMS AND CONDITIONS FOR COPYING, DISTRIBUTION AND MODIFICATION

  0. You just DO WHAT THE FUCK YOU WANT TO.

index.php

<?php

require 'phar://'.__DIR__.'/silex.phar/autoload.php';
require 'IntegerSequence.php';
require 'StreamingResponse.php';

$app = new Silex\Application();

$app->get('/', function() use($app) {
    // Sample iterator that generates numbers up to 50, it could also be
    // an iterator that reads chunks of a file (just the first thing that
    // comes to mind and that could have a tiny bit of sense).
    $iterator = new Nrk\IntegerSequence(50);

    // Our class extends Symfony's standard response class
    // and accepts only instances of classes that implement
    // the Iterator interface.
    $response = new Nrk\StreamingResponse($iterator);

    // This callback gets invoked on each element returned
    // by the underlying iterator.
    $response->onData(function($data) {
        return "* $data<br/>";
    });

    // This callback gets invoked after each call
    // to 'echo' performed by the response class.
    $response->onOutput(function() {
        flush();
        ob_flush();
    });

    return $response;
});

$app->run();

IntegerSequence.php

<?php

namespace Nrk;

class IntegerSequence implements \Iterator
{
    private $_key = 0;    
    private $_limit = -1;
        
    public function __construct($limit = -1)
    {
        $this->_limit = (integer) $limit;
    }
    
    public function current()
    {
        return $this->_key;
    }

    public function key()
    {
        return $this->_key;
    }

    public function next()
    {
        $this->_key++;
    }

    public function rewind()
    {
        $this->_key = 0;
    }

    public function valid()
    {
        if ($this->_limit < 0) {
            return true;
        }
        return $this->_key <= $this->_limit;
    } 
}

StreamingResponse.php

<?php

// Disclaimer: this class is just a fun experiment, do not consider it as being
// anything more than that. I am releasing this code under the glorious WTFPL
// (see the LICENSE file in this gist).

namespace Nrk;

class StreamingResponse extends \Symfony\Component\HttpFoundation\Response
{
    private $_onData;
    private $_onOutput;
    private $_onAbort;

    public function __construct(\Iterator $content, $status = 200, $headers = array())
    {
        $this->onData(array($this, 'doData'));
        parent::__construct($content, $status, $headers);
    }

    public function __toString()
    {
        throw new \RuntimeException('A streaming response cannot be represented as a string');
    }

    public function sendContent()
    {
        $onData = $this->getDataCallback();
        $onOutput = $this->getOutputCallback();

        foreach ($this->getContent() as $data) {
            $output = call_user_func($onData, $data);
            if (isset($output)) {
                echo $output;
                if (isset($onOutput)) {
                    call_user_func($onOutput);
                }
            }
        }
    }

    public function sendHeaders()
    {
        parent::sendHeaders();
        $this->doFlush();
    }

    protected function checkCallableArgument($callback)
    {
        if (!is_callable($callback)) {
            throw new \InvalidArgumentException('Argument must be a valid callable object');
        }
    }

    private function setCallback($field, $callback, $default = null)
    {
        if (!isset($callback)) {
            if (isset($default)) {
                $this->checkCallableArgument($default);
            }
            $this->$field = $default;
        }
        else {
            $this->checkCallableArgument($callback);
            $this->$field = $callback;
        }
    }

    protected function doData($chunk)
    {
        return $chunk;
    }

    protected function doFlush()
    {
        flush();
        ob_flush();
    }

    public function onData($callback = null)
    {
        $this->setCallback('_onData', $callback, array($this, 'doData'));
    }

    public function getDataCallback() 
    {
        return $this->_onData;
    }

    public function onOutput($callback = null)
    {
        $callback = $callback === 'flush' ? array($this, 'doFlush') : $callback;
        $this->setCallback('_onOutput', $callback);
    }

    public function getOutputCallback() 
    {
        return $this->_onOutput;
    }

    public function onAbort($callback = null)
    {
        $this->setCallback('_onAbort', $callback);
        $response = $this;
        register_shutdown_function(function() use($response) {
            if (connection_aborted()) {
                $callback = $response->getAbortCallback();
                if (isset($callback)) {
                    call_user_func($callback);
                }
            }
        });
    }

    public function getAbortCallback() 
    {
        return $this->_onAbort;
    }
}

wireshark_capture.txt

HTTP/1.1 200 OK
X-Powered-By: PHP/5.3.3-1ubuntu9.5
cache-control: no-cache
content-type: text/html; charset=UTF-8
Transfer-Encoding: chunked
Date: Mon, 09 May 2011 19:05:21 GMT
Server: lighttpd/1.4.26

8
* 0<br/>
8
* 1<br/>
8
* 2<br/>
8
* 3<br/>
8
* 4<br/>
8
* 5<br/>
8
* 6<br/>
8
* 7<br/>
8
* 8<br/>
8
* 9<br/>
9
* 10<br/>
9
* 11<br/>
9
* 12<br/>
9
* 13<br/>
9
* 14<br/>
9
* 15<br/>
9
* 16<br/>
9
* 17<br/>
9
* 18<br/>
9
* 19<br/>
9
* 20<br/>
9
* 21<br/>
9
* 22<br/>
9
* 23<br/>
9
* 24<br/>
9
* 25<br/>
9
* 26<br/>
9
* 27<br/>
9
* 28<br/>
9
* 29<br/>
9
* 30<br/>
9
* 31<br/>
9
* 32<br/>
9
* 33<br/>
9
* 34<br/>
9
* 35<br/>
9
* 36<br/>
9
* 37<br/>
9
* 38<br/>
9
* 39<br/>
9
* 40<br/>
9
* 41<br/>
9
* 42<br/>
9
* 43<br/>
9
* 44<br/>
9
* 45<br/>
9
* 46<br/>
9
* 47<br/>
9
* 48<br/>
9
* 49<br/>
9
* 50<br/>
0

Just a couple of additional notes:

• please note that your frontend server might still be buffering your responses depending on its configuration.
• streaming long replies (e.g. files) will block the PHP process in charge of processing your request for a long time.
• do not flush the output buffer on every tiny chunk unless you have valid reasons to do that! Mine was just an example.
• I am not really sure if this approach could lead to weird issues with caching and such, completely untested.

Now do not tell me that you have not been warned :-)