WillRy / rabbitrun

src/Base.php

<?php

namespace WillRy\RabbitRun;

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Exception\AMQPIOException;
use PhpAmqpLib\Exception\AMQPRuntimeException;
use PhpAmqpLib\Exception\AMQPTimeoutException;
use WillRy\RabbitRun\Connections\Connect;

class Base
{


    /** @var AMQPStreamConnection Instância de conexão */
    protected $instance;

    /** @var \PhpAmqpLib\Channel\AMQPChannel Canal de comunicação */
    protected $channel;

    /** @var PDO Conexão do PDO */

The type WillRy\RabbitRun\PDO was not found. Did you mean PDO? If so, make sure to prefix the type with \.

    protected $db;

    /** @var bool */
    protected $executing = false;

    protected $confirmSelect = false;


    public function __construct()
    {

        /**
         * Graceful shutdown
         * Faz a execucao parar ao enviar um sinal do linux para matar o script
         */
        if (php_sapi_name() == "cli") {
            \pcntl_signal(SIGTERM, function ($signal) {
                $this->shutdown($signal);
            }, false);
            \pcntl_signal(SIGINT, function ($signal) {
                $this->shutdown($signal);
            }, false);
        }

        return $this;
    }

    /**
     * Garante o desligamento correto dos workers
     * via sinal no sistema operacional
     * eliminando loops e conexões
     * @param $signal
     */
    public function shutdown($signal)
    {
        $data = date('Y-m-d H:i:s');
        switch ($signal) {
            case SIGTERM:
                print "Caught SIGTERM {$data}" . PHP_EOL;
                exit;

Using exit here is not recommended.

            case SIGKILL:
                print "Caught SIGKILL {$data}" . PHP_EOL;
                exit;

Using exit here is not recommended.

            case SIGINT:
                print "Caught SIGINT {$data}" . PHP_EOL;
                exit;

Using exit here is not recommended.

        }
    }

    /**
     * Configura o rabbitmq e gera conexão ativa
     * @param $host
     * @param $port
     * @param $user
     * @param $pass
     * @param $vhost
     * @return $this
     */
    public function configRabbit($host, $port, $user, $pass, $vhost): Base
    {
        Connect::config($host, $port, $user, $pass, $vhost);

        $this->getConnection();

        return $this;
    }

    /**
     * Cria uma exchange
     *
     * @param string $exchange
     * @param string $type
     * @param bool $passive
     * @param bool $durable
     * @param bool $auto_delete
     * @return $this
     */
    public function exchange(
        string $exchange,
        string $type = "direct",
        bool   $passive = false,
        bool   $durable = true,
        bool   $auto_delete = false
    )
    {
        $this->channel->exchange_declare(
            $exchange,
            $type,
            $passive,
            $durable,
            $auto_delete,
        );
        return $this;
    }

    /**
     * Cria uma fila
     *
     * @param string $queue
     * @param false $passive
     * @param false $durable
     * @param false $exclusive
     * @param bool $auto_delete
     * @return array|null
     */
    public function queue(
        string $queue = '',
        bool   $passive = false,
        bool   $durable = true,
        bool   $exclusive = false,
        bool   $auto_delete = false
    )
    {
        return $this->channel->queue_declare(
            $queue,
            $passive,
            $durable,
            $exclusive,
            $auto_delete,
            false
        );
    }

    /**
     * Vincula a fila e a exchange criada
     * @return $this
     */
    public function bind(
        $queue,
        $exchange
    )
    {
        $this->channel->queue_bind($queue, $exchange);
        return $this;
    }

    public function randomConsumer($len = 30): string
    {
        $pool = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ';
        return substr(str_shuffle(str_repeat($pool, (int)ceil($len / strlen($pool)))), 0, $len);
    }

    /**
     * Mantém a conexão, mesmo em caso de erro
     * de rede ou conexão
     * @param $callback
     * @throws Exception
     */
    public function loopConnection($callback)
    {
        while (true) {
            try {
                $this->getConnection(true);
                echo ' [x] [  connected  ] ', "\n";
                $callback();
            } catch (\Exception $e) {
                echo get_class($e) . ':' . $e->getMessage() . " | file:" . $e->getFile() . " | line:" . $e->getLine() . PHP_EOL;
                $this->cleanConnection();
                sleep(2);
            }
        }
    }

    /**
     * Gera uma conexão no rabbitmq e gera um canal(opcional)
     * @param bool $createChannel
     * @return AMQPStreamConnection|void
     */
    public function getConnection(bool $createChannel = true)
    {
        $this->instance = Connect::getInstance();

        if ($createChannel) {
            $this->getChannel();
        }

        return $this->instance;
    }

    /**
     * Retorna canal ativo ou cria um novo
     * @return AMQPChannel|void

The type WillRy\RabbitRun\AMQPChannel was not found. Did you mean AMQPChannel? If so, make sure to prefix the type with \.

     */
    public function getChannel()
    {
        $this->channel = Connect::getChannel();

        if ($this->confirmSelect) {
            $this->channel->confirm_select();
        }

        return $this->channel;
    }

    /**
     * Fecha a conexão com o RabbitMQ
     * @throws Exception
     */
    public function cleanConnection()
    {
        try {
            Connect::closeChannel();
            Connect::closeInstance();
        } catch (\Exception $e) {
            echo '[ERROR CLOSE CHANNEL|INSTANCE]' . $e->getMessage() . "|file:" . $e->getFile() . "|line:" . $e->getLine() . PHP_EOL;
        }
    }

    public function configConfirmSelect(bool $confirmSelect = true)
    {
        $this->confirmSelect = $confirmSelect;
        return $this;
    }
}