umbrellio / php-table-sync

src/Rabbit/MessageBuilder.php

<?php

declare(strict_types=1);

namespace Umbrellio\TableSync\Rabbit;

use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPArray;
use PhpAmqpLib\Wire\AMQPTable;
use function Safe\json_decode;
use Umbrellio\TableSync\Messages\PublishMessage;
use Umbrellio\TableSync\Messages\ReceivedMessage;

class MessageBuilder
{
    public const EVENT_NAME = 'table_sync';
    private $publishMessageConfig;

    public function __construct(Config\PublishMessage $publishMessageConfig)
    {
        $this->publishMessageConfig = $publishMessageConfig;
    }

    public function buildReceivedMessage(AMQPMessage $message): ReceivedMessage
    {
        [
            'event' => $event,
            'model' => $model,
            'attributes' => $attributes,
            'version' => $version,
            'metadata' => $metadata,
        ] = json_decode($message->body, true);

        $appId = $message->get('app_id');
        $headers = $message->has('application_headers')
            ? $this->headersToArray($message->get('application_headers'))

It seems like $message->get('application_headers') can also be of type PhpAmqpLib\Channel\AMQPChannel; however, parameter $table of Umbrellio\TableSync\Rabbit\MessageBuilder::headersToArray() does only seem to accept PhpAmqpLib\Wire\AMQPTable, maybe add an additional type check?

            : [];

        return new ReceivedMessage($event, $model, $attributes, $version, $metadata, $appId, $headers);

It seems like $appId can also be of type PhpAmqpLib\Channel\AMQPChannel; however, parameter $appId of Umbrellio\TableSync\Messages\ReceivedMessage::__construct() does only seem to accept string, maybe add an additional type check?

    }

    public function buildForPublishing(PublishMessage $message): AMQPMessage
    {
        return new AMQPMessage(
            $this->buildBody($message),
            [
                'content_type' => 'application/json',
                'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,
                'app_id' => $this->publishMessageConfig->appId(),
                'type' => self::EVENT_NAME,
                'application_headers' => $this->publishMessageConfig->headers(),
            ]
        );
    }

    private function headersToArray(AMQPTable $table): array
    {
        $headers = [];
        foreach ($table as $key => $value) {
            if ($value[1] instanceof AMQPArray) {
                $headers[$key] = $value[1]->getNativeData();
            } else {
                $headers[$key] = $value[1];
            }
        }

        return $headers;
    }

    private function buildBody(PublishMessage $message): string
    {
        $event = $message->isDestroyed() ? 'destroy' : 'update';

        $data = [
            'event' => $event,
            'model' => $message->className(),
            'attributes' => $message->attributes(),
            'version' => microtime(true),
            'metadata' => [
                'created' => $message->isCreated(),
            ],
        ];
        return json_encode($data);
    }
}