stream.read.php

<?php

// Returns a string or NULL on error
// Use this on a non-blocking stream to read all the data safely
// (in a blocking way which doesn't make much sense)
function readStream($stream)
{
	$buf = '';

	do {

		// check stream meta data for remaining buffer
		// yes, the manual explicitly says this value SHOULD NOT be used,
		// but this has been confirmed to work around blocking on blocking streams
		$meta = stream_get_meta_data($stream);
		$len = $meta['unread_bytes'];

		if ($len === 0) {
			// length unknown (usual case), read up to one chunk from incoming streams
			$len = $this->bufferSize;
		} else if ($len > $this->bufferSize) {
			// buffer length known and bigger than chunk, read exactly one chunk from buffer
			$len = $this->bufferSize;
		} else {
			// small buffer remaining, read EXACTLY remaining buffer. execeeding buffer length WILL block fread() when no more data is incoming
		}
		
		$chunk = fread($stream, $len);

		if($chunk === false) {
			$error = socket_last_error($stream);
			// 11 = EWOULDBLOCK and 115 = EINPROGRESS
			if($error != 11 && $error != 115) {
				print socket_strerror($error) . $error . "\n";
				return null;
			}
			break;
		} elseif($chunk == '') {
			break;
		} else { 
			$buf .= $chunk;
		}

		if (!is_resource($stream) || feof($stream)) {
			fclose($stream);
			break;
		}

	} while(true);

	return $buf;
}

StreamUnblock.php

<?php

// extend base Stream class by checking stream meta data for remaining stream buffer before reading from blocking streams
// heavily inspired by: https://github.com/clue/Worker/commit/8c6d7a4e733f0ee5aa431ffda1d4929229ae2336
class StreamUnblock extends \React\Stream\Stream
{
    public function handleData($stream)
    {
        // check stream meta data for remaining buffer
        // yes, the manual explicitly says this value SHOULD NOT be used,
        // but this has been confirmed to work around blocking on blocking streams
        $meta = stream_get_meta_data($stream);
        $len = $meta['unread_bytes'];
        
        if ($len === 0) {
            // length unknown (usual case), read up to one chunk from incoming streams
            $len = $this->bufferSize;
        } else if ($len > $this->bufferSize) {
            // buffer length known and bigger than chunk, read exactly one chunk from buffer
            $len = $this->bufferSize;
        } else {
            // small buffer remaining, read EXACTLY remaining buffer. execeeding buffer length WILL block fread() when no more data is incoming
        }
        // this above mess could probably be cleaned up, but I left it unchanged to make it more clear what's going on:
        // $len = $this->bufferSize;
        // if ($meta['unread_bytes'] !== 0 && $meta['unread_bytes'] < $len) { $len = $meta['unread_bytes']; }
        
        $data = fread($stream, $len);

        $this->emit('data', array($data, $this));

        if (!is_resource($stream) || feof($stream)) {
            $this->end();
        }
    }
}