public function listen(Queue ...$queues): Promise
    {
        /** Hack for phpunit tests */
        $isTestCall = 'phpunitTests' === (string) \getenv('SERVICE_BUS_TESTING');

        /**
         * @psalm-suppress InvalidArgument Incorrect psalm unpack parameters (...$args)
         * @psalm-suppress MixedArgument
         */
        return call(
            function(array $queues) use ($isTestCall): \Generator
            {
                /** @psalm-suppress TooManyTemplateParams */
                yield $this->transport->consume(
                    function(IncomingPackage $package) use ($isTestCall): \Generator
                    {
                        $this->currentTasksInProgressCount++;

                        /** Hack for phpUnit */
                        if (true === $isTestCall)
                        {
                            $this->currentTasksInProgressCount--;

                            yield $this->processor->handle($package);
                            yield $this->transport->stop();

                            Loop::stop();
                        }

                        /** Handle incoming package */
                        $this->deferExecution($package);

                        /** Limit the maximum number of concurrently running tasks */
                        while ($this->maxConcurrentTaskCount <= $this->currentTasksInProgressCount)
                        {
                            yield new Delayed($this->awaitDelay);
                        }
                    },
                    ...$queues
                );
            },
            $queues
        );
    }
public function listen(Queue ...$queues): Promise
    {
        /** Hack for phpunit tests */
        $isTestCall = 'phpunitTests' === (string) \getenv('SERVICE_BUS_TESTING');

        /**
         * @psalm-suppress InvalidArgument Incorrect psalm unpack parameters (...$args)
         * @psalm-suppress MixedArgument
         */
        return call(
            function(array $queues) use ($isTestCall): \Generator
            {
                /** @psalm-suppress TooManyTemplateParams */
                yield $this->transport->consume(
                    function(IncomingPackage $package) use ($isTestCall): \Generator
                    {
                        $this->currentTasksInProgressCount++;

                        /** Hack for phpUnit */
                        if (true === $isTestCall)
                        {
                            $this->currentTasksInProgressCount--;

                            yield $this->processor->handle($package);
                            yield $this->transport->stop();

                            Loop::stop();
                        }

                        /** Handle incoming package */
                        $this->deferExecution($package);

                        /** Limit the maximum number of concurrently running tasks */
                        while ($this->maxConcurrentTaskCount <= $this->currentTasksInProgressCount)
                        {
                            yield new Delayed($this->awaitDelay);
                        }
                    },
                    ...$queues
                );
            },
            $queues
        );
    }
public function listen ( Queue ... $ queues ) : Promise { /** Hack for phpunit tests */ $ isTestCall = 'phpunitTests' === ( string ) \ getenv ( 'SERVICE_BUS_TESTING' ) ; /**
         * @psalm-suppress InvalidArgument Incorrect psalm unpack parameters (...$args)
         * @psalm-suppress MixedArgument
         */ return call ( function ( array $ queues ) use ( $ isTestCall ) : \ Generator { /** @psalm-suppress TooManyTemplateParams */ yield $ this -> transport -> consume ( function ( IncomingPackage $ package ) use ( $ isTestCall ) : \ Generator { $ this -> currentTasksInProgressCount ++ ; /** Hack for phpUnit */ if ( true === $ isTestCall ) { $ this -> currentTasksInProgressCount -- ; yield $ this -> processor -> handle ( $ package ) ; yield $ this -> transport -> stop ( ) ; Loop :: stop ( ) ; } /** Handle incoming package */ $ this -> deferExecution ( $ package ) ; /** Limit the maximum number of concurrently running tasks */ while ( $ this -> maxConcurrentTaskCount <= $ this -> currentTasksInProgressCount ) { yield new Delayed ( $ this -> awaitDelay ) ; } } , ... $ queues ) ; } , $ queues ) ; }
Start queues listen.

@param Queue ...$queues

@throws \ServiceBus\Transport\Common\Exceptions\ConnectionFail Connection refused

@return Promise It does not return any result
Start queues listen .