Kunnu\RabbitMQ\RabbitMQPublisher::publish()

Complexity

Conditions	7
Paths	12

Size

Total Lines	69
Code Lines	34

Duplication

Lines	0
Ratio	0 %

Importance

Changes	3
Bugs	2	Features	1

<?php
    public function publish(
        $messages,
        string $routingKey = '',
        string $connectionName = null,
        PublishConfig $publishConfig = null
    ): void {
        $messages = !is_array($messages) ? [$messages] : $messages;
        $publishConfig = $publishConfig ?? new PublishConfig();

        $defaultConfig = new Collection($this->manager->getConfig()->get(RabbitMQManager::CONFIG_KEY . ".defaults"));

        $connectionName = $connectionName ?? $this->manager->resolveDefaultConfigName();
        $connection = $this->manager->resolveConnection();

        $channelId = $this->manager->resolveChannelId($publishConfig->get("channel_id"), $connectionName);
        $channel = $this->manager->resolveChannel($connectionName, $channelId, $connection);

        $connectionConfig = $this->manager->resolveConfig($connectionName);

It seems like $connectionName can also be of type null; however, parameter $connectionName of Kunnu\RabbitMQ\RabbitMQManager::resolveConfig() does only seem to accept string, maybe add an additional type check?

        // Merge/Override default connection configuration with
        // the configuration specified for this publishing.
        if ($publishConfig && $publishConfig->getConnectionConfig()) {
            // Publish config > Connection config
            $connectionConfig = $connectionConfig->merge($publishConfig->getConnectionConfig());
        }

        $readyMessages = [];

        foreach ($messages as $message) {
            // Merge message configuration
            // Message config > Publish config > Connection config > Default config
            $messageConfig = array_merge(
                $defaultConfig->get('message', []), // Default config
                $connectionConfig->get('message', []), // Connection config
                $publishConfig->get('message', []), // Publish config
                $message->getConfig()->toArray(), // Message config
            );
            // Override the message config
            $message->setConfig($messageConfig);

            // Merge the exchange properties
            // Publish config > Connection config > Default config
            $exchangeProperties = array_merge(
                $defaultConfig->get('exchange', ['properties' => []])['properties'] ?? [], // Default properties
                $connectionConfig->get('exchange', ['properties' => []])['properties'] ?? [], // Connection properties
                $publishConfig->get('exchange', ['properties' => []])['properties'] ?? [], // Publish properties
            );

            // Merge the exchange config
            // Exchange config > Publish config > Connection config > Default config
            $exchangeConfig = array_merge(
                $defaultConfig->get('exchange', []), // Default config
                $connectionConfig->get('exchange', []), // Connection config
                $publishConfig->get('exchange', ['properties' => $exchangeProperties]), // Publish config,
                $message->getExchange() ? $message->getExchange()->getConfig()->toArray() : [], // Exchange config
            );

            // Merge message exchange configuration
            if ($message->getExchange()) {
                $message->getExchange()->setConfig($exchangeConfig);
                $message->getExchange()->getConfig()->put('name', $message->getExchange()->getName());
            } else {
                $message->setExchange(new RabbitMQExchange($exchangeConfig['name'] ?? '', $exchangeConfig));
            }

            $readyMessages[] = $message;
        }

        $this->publishBulk($readyMessages, $channel, $routingKey);
    }