Use case for possible improvement

[seyfer]

Hello

I want to show you one use case which out team have used for our project. I hope it will show ways to improve current approach.
So, we believe, that current Consumer::consume() method which use get() method to get messages from queue isn't the best solution. Php AMQP extension object AMQPQueue has method consume() and it's work like a daemon.
We haven't found any possible solution to create Consumer with consume() method instead of default with get() method. And we did this.

In our config we have

'plugin_managers' => [
            'consumer' => [
                'abstract_factories' => [
                    \Application\Core\Infrastructure\Humus\Consumer\ConsumerAbstractServiceFactory::class
                ]
            ],
            'callback' => [
                'factories' => [
                    'import.core.application.console.history.consumer.HistoryConsumer' =>
//for example                        \Application\Core\Application\Console\History\Consumer\HistoryConsumerFactory::class,
                ],
            ],
        ],

We copied all code from default ConsumerAbstractServiceFactory and have changed only one line to use our Consumer.

$consumer = new Consumer($queues, $idleTimeout, $waitTimeout);

So, there is first bad place. Redefining of consumer should be easier.
And there is our Consumer code

class Consumer extends \HumusAmqpModule\Consumer
{
    /**
     * @param int $msgAmount
     */
    public function consume($msgAmount = 0)
    {
        $this->target = $msgAmount;
        /** @var \AMQPQueue $queue */
        foreach($this->queues as $queue) {
            $queue->consume(function ($message, $queue) {
                if (!$this->timestampLastAck) {
                    $this->timestampLastAck = microtime(1);
                }

                try {
                    $processFlag = $this->handleDelivery($message, $queue);
                } catch (\Exception $e) {
                    $this->handleDeliveryException($e);
                    $processFlag = false;
                }
                $this->handleProcessFlag($message, $processFlag);

                $now = microtime(1);

                if ($this->countMessagesUnacked > 0
                    && ($this->countMessagesUnacked == $this->blockSize
                        || ($now - $this->timestampLastAck) > $this->idleTimeout
                    )
                ) {
                    $this->ackOrNackBlock();
                }

                if ($this->usePcntlSignalDispatch) {
                    // Check for signals
                    pcntl_signal_dispatch();
                }

                if (!$this->keepAlive || (0 != $this->target && $this->countMessagesConsumed >= $this->target)) {
                    return;
                }
            });
        }
    }
}

As you can see we just moved original consumer code to callback function in consume() method.

This approach gives us better usage of module for our needs.
It would be great, if there will be easier way to overwrite Consumer or possibility to choose between two options - use get() method or consume() for running messages.

[seyfer]

Version 0.1.6 - 0.1.8

[seyfer]

It could be option in 'consumers' config part with values get or consume.

[seyfer]

With consume() run method actually consumers will be registered in AMQP and it's possible then to see them registered in RabbitMQ web interface and get this information from API.
With get() method there is no consumers registered. And impossible to check status.

[prolic]

Das you already noticed, this is in 0.1.6 - 0.1.8. use the most up to date
version instead, which is 1.0.0.

I don't think I like to update pre-final releases.

Am 18.08.2016 18:51 schrieb "Oleg Abrazhaev" notifications@github.com:

With consume() run method actually consumers will be registered in AMQP
and it's possible then to see them registered in RabbitMQ web interface and
get this information from API.
With get() method there is no consumers registered.

—
You are receiving this because you are subscribed to this thread.
Reply to this email directly, view it on GitHub
#41 (comment),
or mute the thread
https://github.com/notifications/unsubscribe-auth/AAYEvPMEzaxS14xwjKcN3AyX2IBlM2L6ks5qhDksgaJpZM4JnXRx
.

[seyfer]

Have you changed approach to consume() method in 1.0.0 ?

We have php 5.6, so we haven't choice.

[prolic]

1.0.0 uses consumer() besides lots of other improvements.

Am 18.08.2016 21:39 schrieb "Oleg Abrazhaev" notifications@github.com:

Did you changed approach to consume() method in 1.0.0 ?

We have php 5.6, so we haven't choice.

—
You are receiving this because you commented.
Reply to this email directly, view it on GitHub
#41 (comment),
or mute the thread
https://github.com/notifications/unsubscribe-auth/AAYEvHACtNWbPN1lL-T92NwwE30kGQksks5qhGB7gaJpZM4JnXRx
.

[seyfer]

You probably could just take code provided above for Consumer and release 1.8.1 version with consume() method for php 5.6 users.
In real world there is still a lot of php 5.6 projects that should be supported.

[prolic]

Please provide a PR with a tag as base branch. That would help a lot.
Btw. 5.6 must die quick! ;-)

Am 18.08.2016 22:41 schrieb "Oleg Abrazhaev" notifications@github.com:

You probably could just take code provided above for Consumer and release
1.8.1 version with consume() method for php 5.6 users.
In real world there is still a lot of php 5.6 projects that should be
supported.

—
You are receiving this because you commented.
Reply to this email directly, view it on GitHub
#41 (comment),
or mute the thread
https://github.com/notifications/unsubscribe-auth/AAYEvE_bBqWbxBID-lApsmwdImjcnThUks5qhG8egaJpZM4JnXRx
.

[seyfer]

It will be a long way
https://seld.be/notes/php-versions-stats-2016-1-edition

[seyfer]

I have replaced method in Consumer and created new branch
https://github.com/seyfer/HumusAmqpModule/tree/v0.1.8.1

How to create PR properly?

[prolic]

I just created a new branch "old" which is based on v0.3.1, you can submit your PR here.

[seyfer]

Why v0.3.1 ? I have forked from v0.1.8. v0.3.1 is the last version compatible with ZF2 ?

[prolic]

v0.3.1 is the latest version compatible with PHP 5.x series. ZF2 is supported by all versions.

[seyfer]

I have tried v0.3.1 and I see there is major differences

HumusAmqpModule\Consumer::setLogger() must be an instance of Psr\Log\LoggerInterface, instance of Zend\Log\Logger

But in ZF2 log configuration Zend\Log\Logger returns by default.
Could I find somewhere documentation or example how to properly configure consumers and producers with v0.3.1 version?

PS. I have solved this with https://github.com/michaelmoussa/zend-psr-log

[prolic]

In ZF2 there are delegators in the service manager. f.e. you're service manager config can look like this:

return [
    'factories' => [
        // your factories
    ],
    'delegators' => [
        'event_consumer_logger' => [
            PsrLoggerDelegator::class,
        ],
    ]
]

and your PsrLoggerDelegator class:

final class PsrLoggerDelegator implements DelegatorFactoryInterface
{
    /**
     * @param ContainerInterface $container
     * @param string $name
     * @param callable $callback
     * @param array|null $options
     * @return PsrLoggerAdapter
     */
    public function __invoke(ContainerInterface $container, $name, callable $callback, array $options = null)
    {
        return new PsrLoggerAdapter($callback());
    }
}