Remote procedure call (RPC) about laravel-amqp HOT 10 OPEN

public function rpc($routing, $message, $timeout = 0, $queueOptions, $exchangeOptions)
{
//@todo check if message is instance of Message class
//@todo use queueOptions options for queue declaration
//@todo use exchangeOptions for exchange declaration if needed, but optional
//@notice multiple messages per temp queue will not be supported

    /* @var Bschmitt\Amqp\Publisher $publisher */
    $publisher = app()->make('Bschmitt\Amqp\Publisher');
    $publisher->connect();
    $publisher->getConnection()->set_close_on_destruct();
    $replyTo = $publisher->getChannel()->queue_declare(
        '',
        false,
        false,
        true,
        true
    );
    $replyTo = $replyTo[0];
    $publisher->getChannel()->queue_declare(
        $queue,
        false,
        false,
        false,
        false
    );
    $response = false;
    $publisher->getChannel()->basic_consume(
        $replyTo,
        '',
        false,
        false,
        false,
        false,
        function ($message) use (&$response) {
            $response = $message;
        }
    );
    $publisher->getChannel()->queue_bind($queue, 'amq.direct', $queue);
    $publisher->getChannel()->basic_publish(
        new \Bschmitt\Amqp\Message(
            $message,
            [
                'content_type' => 'text/plain',
                'delivery_mode' => 1,
                'reply_to' => $replyTo,
            ]
        ),
        'amq.direct',
        $queue
    );
    $publisher->getChannel()->wait(null, false, $timeout);
    return $response;
}

function queue_rpc($queue, $message, $timeout = 0) {
    /* @var Bschmitt\Amqp\Publisher $publisher */
    $publisher = app()->make('Bschmitt\Amqp\Publisher');
    $publisher->connect();
    $publisher->getConnection()->set_close_on_destruct();
    $replyTo = $publisher->getChannel()->queue_declare(
        '',
        false,
        false,
        true,
        true
    );
    $replyTo = $replyTo[0];
    $publisher->getChannel()->queue_declare(
        $queue,
        false,
        false,
        false,
        false
    );
    $response = false;
    $publisher->getChannel()->basic_consume(
        $replyTo,
        '',
        false,
        false,
        false,
        false,
        function ($message) use (&$response) {
            $response = $message;
        }
    );
    $publisher->getChannel()->queue_bind($queue, 'amq.direct', $queue);
    $publisher->getChannel()->basic_publish(
        new \Bschmitt\Amqp\Message(
            $message,
            [
                'content_type' => 'text/plain',
                'delivery_mode' => 1,
                'reply_to' => $replyTo,
            ]
        ),
        'amq.direct',
        $queue
    );
    $publisher->getChannel()->wait(null, false, $timeout);
    return $response;
}

queue_rpc('ping-queue', 'ping', 10); // -> pong

\Amqp::consume('worker', function (AMQPMessage $message, Consumer $consumer) {
            $this->info($message->getBody(), 'v');
   
            $correlationId = $message->has('correlation_id') ? $message->get('correlation_id') : null;
            $consumer->getChannel()->basic_publish(
                new AMQPMessage(
                    $response,
                    [
                        'content_type' => 'text/plain',
                        'delivery_mode' => 1,
                        'correlation_id' => $correlationId
                    ]
                    ),
                '',
                $message->get('reply_to')
            );
 }, [
            'exchange' => 'amq.direct',
            'exchange_type' => 'direct',
            'queue_force_declare' => true,
            'queue_durable' => false,
            'consumer_no_ack' => true,
 ]);