README

如何使用。

创建发送器。例如，使用无缓冲的 PDO 查询和 symfony 中的发送器

return new \Symfony\Component\HttpFoundation\StreamedResponse(function() use ($pdo) {
    $tm = new \RV\TProto\Proto\Transmitter\Transmitter(function($data) {
        echo $data;
    }, null, true);
    
    $pdo->setAttribute(\PDO::MYSQL_ATTR_USE_BUFFERED_QUERY, false);
    $res = $pdo->query("SELECT id, status FROM tblclients");
    while (false !== ($row = $res->fetch(\PDO::FETCH_ASSOC))) {
        $tm->send($row);
    }
});

创建接收器。例如（目前只有 guzzle 适配器，您可以使用自己的适配器）

$stream = $client->get('https://test.com/stream')->getBody();
$update = new \RV\TProto\Proto\Receiver\Receiver($stream, function($data) {
    print_r($data);
});
$update->run();

如果您需要批量接收对象（例如每次 200 个）而不是逐个接收，则可以使用 BatchReceiver

$stream = $client->get('https://test.com/stream')->getBody();
$update = new \RV\TProto\Proto\Receiver\BatchReceiver($stream, function($data) {
    print count($data);
}, null, 200);
$update->run();

此外，您还可以使用自定义序列化器和 DTO

class Client
{
    public $id;
    public $status;
    
    public function __construct($id, $status)
    {
        $this->id = $id;
        $this->status = $status;
    }
}

class CustomSerializer implements \RV\TProto\Serializer\SerializerInterface
{
    public function serialize($data)
    {
        return $data['id'].";".$data['status'];
    }
    
    public function deserialize($data)
    {
        return new Client($data['id'], $data['status']);
    }
    
    public function getDelimiter()
    {
        return ":";
    }
}

然后在发送器和接收器上使用它，例如接收器

$update = new \RV\TProto\Proto\Receiver\Receiver($stream, function($data) {
    var_dump($data);
}, new CustomSerializer());

它打印对象

object(Client)#1 (2) {
  ["id"]=>
  int(123)
  ["status"]=>
  int(0)
}