<commit_before><?php

namespace Honeybee\Infrastructure\Migration;

use Assert\Assertion;

abstract class RabbitMqMigration extends Migration
{
    const WAIT_SUFFIX = '.waiting';

    const UNROUTED_SUFFIX = '.unrouted';

    const REPUB_SUFFIX = '.repub';

    const QUEUE_SUFFIX = '.q';

    const REPUB_INTERVAL = 30000; //30 seconds

    protected function createQueue(MigrationTarget $migration_target, $exchange_name, $queue_name, $routing_key)
    {
        Assertion::string($exchange_name);
        Assertion::string($routing_key);
        Assertion::string($queue_name);

        $channel = $this->getConnection($migration_target)->channel();
        $channel->queue_declare($queue_name, false, true, false, false);
        $channel->queue_bind($queue_name, $exchange_name, $routing_key);
    }

    protected function createVersionList(MigrationTarget $migration_target, $exchange_name)
    {
        Assertion::string($exchange_name);

        $channel = $this->getConnection($migration_target)->channel();
        $channel->exchange_declare($exchange_name, 'topic', false, true, false, true);
    }

    protected function createExchangePipeline(MigrationTargetInterface $migration_target, $exchange_name)
    {
        Assertion::string($exchange_name);

        $wait_exchange_name = $exchange_name . self::WAIT_SUFFIX;
        $wait_queue_name = $wait_exchange_name . self::QUEUE_SUFFIX;
        $unrouted_exchange_name = $exchange_name . self::UNROUTED_SUFFIX;
        $unrouted_queue_name = $unrouted_exchange_name . self::QUEUE_SUFFIX;
        $repub_exchange_name = $exchange_name . self::REPUB_SUFFIX;
        $repub_queue_name = $repub_exchange_name . self::QUEUE_SUFFIX;

        $channel = $this->getConnection($migration_target)->channel();

        // Setup the default exchange and queue pipelines
        $channel->exchange_declare($unrouted_exchange_name, 'fanout', false, true, false, true); //internal
        $channel->exchange_declare($repub_exchange_name, 'fanout', false, true, false, true); //internal
        $channel->exchange_declare($wait_exchange_name, 'fanout', false, true, false);
        $channel->exchange_declare($exchange_name, 'direct', false, true, false, false, false, [
            'alternate-exchange' => [ 'S', $unrouted_exchange_name ]
        ]);
        $channel->queue_declare($wait_queue_name, false, true, false, false, false, [
            'x-dead-letter-exchange' => [ 'S', $exchange_name ]
        ]);
        $channel->queue_bind($wait_queue_name, $wait_exchange_name);
        $channel->queue_declare($unrouted_queue_name, false, true, false, false, false, [
            'x-dead-letter-exchange' => [ 'S', $repub_exchange_name ],
            'x-message-ttl' => [ 'I', self::REPUB_INTERVAL ]
        ]);
        $channel->queue_bind($unrouted_queue_name, $unrouted_exchange_name);
        $channel->queue_declare($repub_queue_name, false, true, false, false);
        $channel->queue_bind($repub_queue_name, $repub_exchange_name);

        $this->createShovel($migration_target, $repub_exchange_name, $exchange_name, $repub_queue_name);
    }

    protected function createShovel(
        MigrationTargetInterface $migration_target,
        $src_exchange_name,
        $dest_exchange_name,
        $src_queue
    ) {
        $connector = $migration_target->getTargetConnector();

        $endpoint = sprintf(
            '/api/parameters/shovel/%s/%s.shovel',
            $connector->getConfig()->get('vhost', '%2f'),
            $src_exchange_name
        );

        $body = [
            'value' => [
                'src-uri' => 'amqp://',
                'src-queue' => $src_queue,
                'dest-uri' => 'amqp://',
                'dest-exchange' => $dest_exchange_name,
                'add-forward-headers' => false,
                'ack-mode' => 'on-confirm',
                'delete-after' => 'never'
            ]
        ];

        $connector->putToAdminApi($endpoint, $body);
    }
}
<commit_msg>add renameQueue method to rabbit base migration<commit_after><?php

namespace Honeybee\Infrastructure\Migration;

use Assert\Assertion;

abstract class RabbitMqMigration extends Migration
{
    const WAIT_SUFFIX = '.waiting';

    const UNROUTED_SUFFIX = '.unrouted';

    const REPUB_SUFFIX = '.repub';

    const QUEUE_SUFFIX = '.q';

    const REPUB_INTERVAL = 30000; //30 seconds

    protected function createQueue(MigrationTarget $migration_target, $exchange_name, $queue_name, $routing_key)
    {
        Assertion::string($exchange_name);
        Assertion::string($routing_key);
        Assertion::string($queue_name);

        $channel = $this->getConnection($migration_target)->channel();
        $channel->queue_declare($queue_name, false, true, false, false);
        $channel->queue_bind($queue_name, $exchange_name, $routing_key);
    }

    protected function renameQueue(
        MigrationTarget $migration_target,
        $exchange_name,
        array $old_queue,
        array $new_queue
    ) {
        Assertion::string($exchange_name);
        Assertion::count($old_queue, 2);
        Assertion::count($new_queue, 2);
        list($prev_name, $prev_routing) = $old_queue;
        list($new_name, $new_routing) = $new_queue;
        Assertion::allString([ $prev_name, $prev_routing, $new_name, $new_routing ]);

        $channel = $this->getConnection($migration_target)->channel();
        // unbind and delete the previous queue
        $channel->queue_unbind($prev_name, $exchange_name, $prev_routing);
        $channel->queue_delete($prev_name);
        // declare and bind new queue
        $channel->queue_declare($new_name, false, true, false, false);
        $channel->queue_bind($new_name, $exchange_name, $new_routing);
    }

    protected function createVersionList(MigrationTarget $migration_target, $exchange_name)
    {
        Assertion::string($exchange_name);

        $channel = $this->getConnection($migration_target)->channel();
        $channel->exchange_declare($exchange_name, 'topic', false, true, false, true);
    }

    protected function createExchangePipeline(MigrationTargetInterface $migration_target, $exchange_name)
    {
        Assertion::string($exchange_name);

        $wait_exchange_name = $exchange_name . self::WAIT_SUFFIX;
        $wait_queue_name = $wait_exchange_name . self::QUEUE_SUFFIX;
        $unrouted_exchange_name = $exchange_name . self::UNROUTED_SUFFIX;
        $unrouted_queue_name = $unrouted_exchange_name . self::QUEUE_SUFFIX;
        $repub_exchange_name = $exchange_name . self::REPUB_SUFFIX;
        $repub_queue_name = $repub_exchange_name . self::QUEUE_SUFFIX;

        $channel = $this->getConnection($migration_target)->channel();

        // Setup the default exchange and queue pipelines
        $channel->exchange_declare($unrouted_exchange_name, 'fanout', false, true, false, true); //internal
        $channel->exchange_declare($repub_exchange_name, 'fanout', false, true, false, true); //internal
        $channel->exchange_declare($wait_exchange_name, 'fanout', false, true, false);
        $channel->exchange_declare($exchange_name, 'direct', false, true, false, false, false, [
            'alternate-exchange' => [ 'S', $unrouted_exchange_name ]
        ]);
        $channel->queue_declare($wait_queue_name, false, true, false, false, false, [
            'x-dead-letter-exchange' => [ 'S', $exchange_name ]
        ]);
        $channel->queue_bind($wait_queue_name, $wait_exchange_name);
        $channel->queue_declare($unrouted_queue_name, false, true, false, false, false, [
            'x-dead-letter-exchange' => [ 'S', $repub_exchange_name ],
            'x-message-ttl' => [ 'I', self::REPUB_INTERVAL ]
        ]);
        $channel->queue_bind($unrouted_queue_name, $unrouted_exchange_name);
        $channel->queue_declare($repub_queue_name, false, true, false, false);
        $channel->queue_bind($repub_queue_name, $repub_exchange_name);

        $this->createShovel($migration_target, $repub_exchange_name, $exchange_name, $repub_queue_name);
    }

    protected function createShovel(
        MigrationTargetInterface $migration_target,
        $src_exchange_name,
        $dest_exchange_name,
        $src_queue
    ) {
        $connector = $migration_target->getTargetConnector();

        $endpoint = sprintf(
            '/api/parameters/shovel/%s/%s.shovel',
            $connector->getConfig()->get('vhost', '%2f'),
            $src_exchange_name
        );

        $body = [
            'value' => [
                'src-uri' => 'amqp://',
                'src-queue' => $src_queue,
                'dest-uri' => 'amqp://',
                'dest-exchange' => $dest_exchange_name,
                'add-forward-headers' => false,
                'ack-mode' => 'on-confirm',
                'delete-after' => 'never'
            ]
        ];

        $connector->putToAdminApi($endpoint, $body);
    }
}

{'fraction_non_alphanumeric': 0.13366281028866675, 'fraction_numerical': 0.002021790407727732, 'mean_word_length': 3.086278109224415, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 81, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, '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': '6882811', 'n_tokens_mistral': 2868, 'n_tokens_neox': 2691, 'n_words': 443}