CallbackFilter_06.php

<?php



class CallbackFilter extends php_user_filter
{    
    public function filter($in, $out, &$consumed, $closing) {

        $gen = $this->params['generator'];        
        
        while ($bucket = stream_bucket_make_writeable($in)) {
            $bucket->data = $gen->current();
            stream_bucket_append($out, $bucket);
        }
        return PSFS_PASS_ON;        
    }
}

function printer($out) {
    while (true) {
        $message = yield;
        fwrite($out, $message);
    }
}


$fp = fopen("php://memory", 'w');
$fp = fopen("php://output", 'w');
$gen = printer($fp);

stream_filter_register(CallbackFilter::class, CallbackFilter::class);
stream_filter_append($fp, CallbackFilter::class, STREAM_FILTER_ALL, ['generator' => $gen]);
stream_filter_append($fp, "string.toupper");

$len = 0;
foreach (range('a', 'z') as $i) {
    $message = $i."\n";
    sleep(1);
    
    $gen->send($message);
//     fseek($fp, $len);
//     echo stream_get_contents($fp);
//     $len += strlen($message);
}

// $c = 0;
// $len = 0;
// do {
//     sleep(1);
//     $message = md5(rand(0, 100))."\n";
    
//     $gen->send($message);
//     fseek($fp, $len);
    
//     echo stream_get_contents($fp);
//     $len += strlen($message);
// } while (++$c < 5);