swarrot / swarrot

examples/01-consumer-with-stacked-processor.php

<?php

require_once __DIR__.'/../vendor/autoload.php';

use Swarrot\Broker\Message;
use Swarrot\Broker\PeclPackageMessageProvider;

The type Swarrot\Broker\PeclPackageMessageProvider was not found. Maybe you did not declare it correctly or list all dependencies?

use Swarrot\Consumer;
use Swarrot\Processor\ProcessorInterface;

class Processor implements ProcessorInterface
{
    protected $processor;

    protected $num;

    public function __construct($processor, $num = 1)
    {
        $this->processor = $processor;
        $this->num = (int) $num;
    }

    public function process(Message $message, array $options)
    {
        printf("Start processing message #%d in processor #%d\n", $message->getId(), $this->num);

The property num does not exist on Processor. Did you maybe forget to declare it?

        $return = $this->processor->process($message, $options);

The property processor does not exist on Processor. Did you maybe forget to declare it?

        printf("End processing message #%d in processor #%d\n", $message->getId(), $this->num);

        return $return;
    }
}

class FinalProcessor implements ProcessorInterface
{
    public function process(Message $message, array $options)
    {
        printf("Consume message #%d\n", $message->getId());
    }
}

$connection = new \AMQPConnection();
$connection->connect();
$channel = new \AMQPChannel($connection);
$queue = new \AMQPQueue($channel);
$queue->setName('global');

$messageProvider = new PeclPackageMessageProvider($queue);

$stack = (new \Swarrot\Processor\Stack\Builder())
    ->push('Processor', 1)
    ->push('Processor', 2)
;
$processor = $stack->resolve(new FinalProcessor());

$consumer = new Consumer($messageProvider, $processor);
$consumer->consume();