Displaying stream progress in PHP

The Symfony console component is a very useful tool to define and invoke these kind of CLI tasks. You can view the console application as the command line alternative of your index.php front controller and request/response objects for HTTP requests. It handles calling the right commands, parsing and validating arguments and options, displaying usage help, … All you need to do is to create a Command class per task and define its name, description, arguments and options.

You command’s execute method will be called using an InputInterface object and an OutputInterface object. The $input object can be used to retrieve the arguments and options required to run the specific task. The $output object is used for displaying what the task is doing or has done. What is actually printed on the console is very important for the issuer of the task. Think of it as the command’s usability. Too little runtime information, the less usable the task.

Opposite to HTTP requests, tasks run from the command line aren’t supposed to return instantly. They can take a very long time. Imagine a task that loops an entire database table or a task that references an external source repetitively, or maybe a taks that performs a large file transfer. It’s very important to show the issuer what is actually going on, or he/she will be left in the dark for minutes/hours. “Is this task still running?”, “How long has this thing been running yet?”, “Is it almost done?”, “Is it running out of memory?”

Enter the ProgressBar. The ProgressBar is an output helper that wraps the OutputInterface object. Initiate a ProgressBar object with the OutputInterface object, and a number that represents the maximum iterations to be executed, for example number of table rows, number of bytes to be transferred, … For example, do a quick count query first on your database to check the amount of rows that will need to be iterated over. Then while iterating call the ProgressBar::advance or ProgressBar::setProgress method to indicate progress.

15214/455642 [=========>------------------]  33%  21 sec/58 sec  2.1 MB

When you call any Symfony console application you can indicate the verbosity you want from the output (use -vvv option). When invoking a command with the -vvv option, the ProgressBar will also display the elapsed time, the estimated time left and the memory the process is currently consuming.

When transferring a file it’s a bit harder to show the progress of the transfer, but possible. When opening resources (also called “streams”) in PHP one can add a “stream context”. The context defines options and parameters. One parameter is the optional stream_notification_callback.

$context = stream_context_create([], ['notification' => $callback]);
$resource = fopen($source, 'r', null, $context);

A resource can be a FTP, SFTP, HTTP, HTTPS source, or anything else that can be streamed over TCP sockets.

The notification callback will be called to notify you of certain events. See of it as an event dispatching mechanism with a single callable event listener. For example: one can get notified how many bytes are available on the readable stream. After every chunk of data transfer the callback is called with the number of bytes that have been transferred up till then. Also a “completed” event, and several other events are responsible of the callable being called during the lifetime of a resource.

An example of how to use the stream_notification_callback:

namespace Demo;

use League\Flysystem\Filesystem;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Helper\ProgressBar;
use Symfony\Component\Console\Input\InputArgument;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Output\OutputInterface;

class DownloadCommand extends Command
{
    /**
     * @var \Symfony\Component\Console\Helper\ProgressBar
     */
    private $progressBar;

    /**
     * @var \Symfony\Component\Console\Output\OutputInterface
     */
    private $output;

    /**
     * @var \League\Flysystem\Filesystem
     */
    private $files;

    /**
     * @param \League\Flysystem\Filesystem $files
     */
    public function __construct(Filesystem $files)
    {
        $this->files = $files;
        parent::__construct();
    }

    /**
     * Configure method.
     */
    public function configure()
    {
        $this->setName('download:file');
        $this->setDescription('Give a source file name and a destination file name and we\'ll beam it over');
        $this->addArgument('source', InputArgument::REQUIRED, 'Source file');
        $this->addArgument('destination', InputArgument::REQUIRED, 'Destination file name');
    }

    /**
     * @param \Symfony\Component\Console\Input\InputInterface $input
     * @param \Symfony\Component\Console\Output\OutputInterface $output
     *
     * @return int|null
     */
    public function execute(InputInterface $input, OutputInterface $output)
    {
        $this->output = $output;

        // Create stream context.
        $context = stream_context_create([], ['notification' => [$this, 'progress']]);

        // Pipe file.
        $resource = fopen($input->getArgument('source'), 'r', null, $context);
        $this->files->putStream($input->getArgument('destination'), $resource);

        // End output.
        $this->progressBar->finish();
        $this->output->writeln('finished');
    }

    /**
     * @param int $notificationCode
     * @param int $severity
     * @param string $message
     * @param int $messageCode
     * @param int $bytesTransferred
     * @param int $bytesMax
     */
    public function progress($notificationCode, $severity, $message, $messageCode, $bytesTransferred, $bytesMax)
    {
        if (STREAM_NOTIFY_REDIRECTED === $notificationCode) {
            $this->progressBar->clear();
            $this->progressBar = null;
            return;
        }

        if (STREAM_NOTIFY_FILE_SIZE_IS === $notificationCode) {
            if ($this->progressBar) {
                $this->progressBar->clear();
            }
            $this->progressBar = new ProgressBar($this->output, $bytesMax);
        }

        if (STREAM_NOTIFY_PROGRESS === $notificationCode) {
            if (is_null($this->progressBar)) {
                $this->progressBar = new ProgressBar($this->output);
            }
            $this->progressBar->setProgress($bytesTransferred);
        }

        if (STREAM_NOTIFY_COMPLETED === $notificationCode) {
            $this->finish($bytesTransferred);
        }
    }
}

There are a few caveats though…

First: stream contexts must be created before the resource is opened. This means one cannot use the notification callback for existing resources.

That is why in the previous example I only used Flysystem to write the stream to a file, but not to initiate the read stream. FilesystemInterface::readStream returns a resoure but it doesn’t allow you to set a stream context.
By the way, some Flysystem adapters implement the readStream in a slightly inexpected way. To comply with the interface, the FTP adapter and the Dropbox adapter readStream methods need to return a resource. They don’t have a native way to open a stream and return that, so they need to read the entire source file content and append it to the php://temp (in memory) stream. Then the memory stream is rewound, so it is ready to read, and ultimately returned.

Second caveat: using resources and a notification callback doesn’t mean the transfer is asynchronous. To create truly asynchronous transfers I would recommend taking a look at ReactPHP.

Finally the true benefits of resources and the notification callback are:

Large files aren’t stored in memory while transferring. Only buffered chunks are kept in memory. To give an example: I managed to transfer an entire Ubuntu iso file over HTTP to the local file system while the process was using merely 2.2MB of memory. And last but not least: the issuer of the taks sees how the task is progressing. This greatly improves the usability of the task.

Happy streaming!

This is a repost from this blog post.