60 lines
1.6 KiB
PHP
60 lines
1.6 KiB
PHP
|
<?php
|
||
|
|
|
||
|
|
namespace Amp\Parallel\Sync;
|
||
|
|
|
||
|
|
use Amp\Parser\Parser;
|
||
|
|
use Amp\Serialization\NativeSerializer;
|
||
|
|
use Amp\Serialization\Serializer;
|
||
|
|
use function Amp\Serialization\encodeUnprintableChars;
|
||
|
|
|
||
|
|
final class ChannelParser extends Parser
|
||
|
|
{
|
||
|
|
const HEADER_LENGTH = 5;
|
||
|
|
|
||
|
|
/** @var Serializer */
|
||
|
|
private $serializer;
|
||
|
|
|
||
|
|
/**
|
||
|
|
* @param callable(mixed $data) Callback invoked when data is parsed.
|
||
|
|
*/
|
||
|
|
public function __construct(callable $callback, ?Serializer $serializer = null)
|
||
|
|
{
|
||
|
|
$this->serializer = $serializer ?? new NativeSerializer;
|
||
|
|
parent::__construct(self::parser($callback, $this->serializer));
|
||
|
|
}
|
||
|
|
|
||
|
|
/**
|
||
|
|
* @param mixed $data Data to encode to send over a channel.
|
||
|
|
*
|
||
|
|
* @return string Encoded data that can be parsed by this class.
|
||
|
|
*
|
||
|
|
* @throws SerializationException
|
||
|
|
*/
|
||
|
|
public function encode($data): string
|
||
|
|
{
|
||
|
|
$data = $this->serializer->serialize($data);
|
||
|
|
return \pack("CL", 0, \strlen($data)) . $data;
|
||
|
|
}
|
||
|
|
|
||
|
|
/**
|
||
|
|
* @throws ChannelException
|
||
|
|
* @throws SerializationException
|
||
|
|
*/
|
||
|
|
private static function parser(callable $push, Serializer $serializer): \Generator
|
||
|
|
{
|
||
|
|
while (true) {
|
||
|
|
$header = yield self::HEADER_LENGTH;
|
||
|
|
$data = \unpack("Cprefix/Llength", $header);
|
||
|
|
|
||
|
|
if ($data["prefix"] !== 0) {
|
||
|
|
$data = $header . yield;
|
||
|
|
throw new ChannelException("Invalid packet received: " . encodeUnprintableChars($data));
|
||
|
|
}
|
||
|
|
|
||
|
|
$data = yield $data["length"];
|
||
|
|
|
||
|
|
$push($serializer->unserialize($data));
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|