<commit_before><?php

declare(strict_types=1);

namespace Smartbox\Integration\FrameworkBundle\Components\Queues\Handler;

use JMS\Serializer\SerializerBuilder;
use JMS\Serializer\SerializerInterface;
use Psr\Log\LoggerAwareInterface;
use Psr\Log\LoggerAwareTrait;
use Smartbox\CoreBundle\Type\SerializableInterface;
use Smartbox\Integration\FrameworkBundle\Components\Queues\QueueMessageHandlerInterface;
use Smartbox\Integration\FrameworkBundle\Components\Queues\QueueMessageInterface;
use Smartbox\Integration\FrameworkBundle\Core\Endpoints\EndpointFactory;
use Smartbox\Integration\FrameworkBundle\Core\Endpoints\EndpointInterface;
use Smartbox\Integration\FrameworkBundle\DependencyInjection\Traits\UsesSmartesbHelper;
use Smartbox\Integration\FrameworkBundle\Exceptions\Handler\UsesExceptionHandlerTrait;
use Smartbox\Integration\FrameworkBundle\Exceptions\QueueDeserializationException;

class AmqpQueueHandler implements LoggerAwareInterface
{
    use LoggerAwareTrait;
    use UsesSmartesbHelper;
    use UsesExceptionHandlerTrait;

    /**
     * @var EndpointInterface
     */
    private $endpoint;

    /**
     * @var int
     */
    private $max;

    /**
     * @var string
     */
    private $format;

    /**
     * @var SerializerInterface
     */
    private $serializer;

    private $stopped = false;

    /**
     * AmqpQueueHandler constructor.
     *
     * @param EndpointInterface        $endpoint   The endpoint used to consume
     * @param int                      $max        Maximum amount of message to consume before stopping
     * @param string                   $format     json|xml|php
     * @param SerializerInterface|null $serializer Serializer to use for deserialization
     */
    public function __construct(EndpointInterface $endpoint, int $max = -1, string $format = 'json', SerializerInterface $serializer = null)
    {
        $this->endpoint = $endpoint;
        $this->max = $max;
        $this->format = $format;
        $this->serializer = $serializer ?? SerializerBuilder::create()->build();
    }

    public function __invoke(\AMQPEnvelope $envelope, \AMQPQueue $queue)
    {
        if ($this->shouldStop()) {
            $queue->nack($envelope->getDeliveryTag(), AMQP_REQUEUE);
            $this->log('Handler stopped on {time}');

            return false;
        }

        $this->log('A message was received on {time}');

        try {
            /** @var \Smartbox\Integration\FrameworkBundle\Core\Messages\MessageInterface|QueueMessageInterface $message */
            $message = $this->serializer->deserialize($envelope->getBody(), SerializableInterface::class, $this->format);
        } catch (\Exception $exception) {
            $this->getExceptionHandler()($exception, $envelope->getBody());
            $queue->ack($envelope->getDeliveryTag());
            return false;
        }

        if ($this->isQueueMessage($message) && null !== $this->smartesbHelper) {
            $endpoint = $this->smartesbHelper->getEndpointFactory()->createEndpoint($message->getDestinationURI(), EndpointFactory::MODE_CONSUME);
            $endpoint->getHandler()->handle($message->getBody(), $endpoint);
        } else {
            $this->endpoint->handle($message);
        }

        $queue->ack($envelope->getDeliveryTag());
        $this->log('A message was consumed on {time}');
        --$this->max;

        if ($this->shouldStop()) {
            $this->log('Handler stopped on {time}');

            return false;
        }
    }

    /**
     * @return bool
     */
    public function isStopped(): bool
    {<commit_msg>amqp return true, so that it can continue to consume messages and not die<commit_after><?php

declare(strict_types=1);

namespace Smartbox\Integration\FrameworkBundle\Components\Queues\Handler;

use JMS\Serializer\SerializerBuilder;
use JMS\Serializer\SerializerInterface;
use Psr\Log\LoggerAwareInterface;
use Psr\Log\LoggerAwareTrait;
use Smartbox\CoreBundle\Type\SerializableInterface;
use Smartbox\Integration\FrameworkBundle\Components\Queues\QueueMessageHandlerInterface;
use Smartbox\Integration\FrameworkBundle\Components\Queues\QueueMessageInterface;
use Smartbox\Integration\FrameworkBundle\Core\Endpoints\EndpointFactory;
use Smartbox\Integration\FrameworkBundle\Core\Endpoints\EndpointInterface;
use Smartbox\Integration\FrameworkBundle\DependencyInjection\Traits\UsesSmartesbHelper;
use Smartbox\Integration\FrameworkBundle\Exceptions\Handler\UsesExceptionHandlerTrait;
use Smartbox\Integration\FrameworkBundle\Exceptions\QueueDeserializationException;

class AmqpQueueHandler implements LoggerAwareInterface
{
    use LoggerAwareTrait;
    use UsesSmartesbHelper;
    use UsesExceptionHandlerTrait;

    /**
     * @var EndpointInterface
     */
    private $endpoint;

    /**
     * @var int
     */
    private $max;

    /**
     * @var string
     */
    private $format;

    /**
     * @var SerializerInterface
     */
    private $serializer;

    private $stopped = false;

    /**
     * AmqpQueueHandler constructor.
     *
     * @param EndpointInterface        $endpoint   The endpoint used to consume
     * @param int                      $max        Maximum amount of message to consume before stopping
     * @param string                   $format     json|xml|php
     * @param SerializerInterface|null $serializer Serializer to use for deserialization
     */
    public function __construct(EndpointInterface $endpoint, int $max = -1, string $format = 'json', SerializerInterface $serializer = null)
    {
        $this->endpoint = $endpoint;
        $this->max = $max;
        $this->format = $format;
        $this->serializer = $serializer ?? SerializerBuilder::create()->build();
    }

    public function __invoke(\AMQPEnvelope $envelope, \AMQPQueue $queue)
    {
        if ($this->shouldStop()) {
            $queue->nack($envelope->getDeliveryTag(), AMQP_REQUEUE);
            $this->log('Handler stopped on {time}');

            return false;
        }

        $this->log('A message was received on {time}');

        try {
            /** @var \Smartbox\Integration\FrameworkBundle\Core\Messages\MessageInterface|QueueMessageInterface $message */
            $message = $this->serializer->deserialize($envelope->getBody(), SerializableInterface::class, $this->format);
        } catch (\Exception $exception) {
            $this->getExceptionHandler()($exception, $envelope->getBody());
            $queue->ack($envelope->getDeliveryTag());
            return true;
        }

        if ($this->isQueueMessage($message) && null !== $this->smartesbHelper) {
            $endpoint = $this->smartesbHelper->getEndpointFactory()->createEndpoint($message->getDestinationURI(), EndpointFactory::MODE_CONSUME);
            $endpoint->getHandler()->handle($message->getBody(), $endpoint);
        } else {
            $this->endpoint->handle($message);
        }

        $queue->ack($envelope->getDeliveryTag());
        $this->log('A message was consumed on {time}');
        --$this->max;

        if ($this->shouldStop()) {
            $this->log('Handler stopped on {time}');

            return false;
        }
    }

    /**
     * @return bool
     */
    public function isStopped(): bool
    {
{'fraction_non_alphanumeric': 0.12100677637947725, 'fraction_numerical': 0.0005531738348776103, 'mean_word_length': 2.9823788546255505, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 73, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3622612', 'n_tokens_mistral': 1984, 'n_tokens_neox': 1941, 'n_words': 379}