Pheanstalk\Pheanstalk

Complexity

Total Complexity

53

Size/Duplication

Total Lines	443
Duplicated Lines	0 %

Test Coverage

Coverage

88.67%

Importance

Changes	6
Bugs	2	Features	0

33 Methods

Rating	Name	Duplication	Size	Complexity
A	bury()	0	3	1
A	reconnect()	0	17	5
A	touch()	0	3	1
A	release()	0	7	1
A	pauseTube()	0	3	1
A	delete()	0	3	1
A	kickJob()	0	3	1
A	__construct()	0	3	1
A	resumeTube()	0	4	1
A	put()	0	11	1
A	statsJob()	0	3	1
A	kick()	0	5	1
A	useTube()	0	8	2
A	dispatch()	0	10	2
A	reserve()	0	7	1
A	peekDelayed()	0	10	2
A	peek()	0	7	1
A	reserveJob()	0	16	3
A	listTubeUsed()	0	10	2
A	createWithFactory()	0	3	1
A	listTubesWatched()	0	10	2
A	listTubes()	0	4	1
A	withWatchedTube()	0	12	3
A	withUsedTube()	0	8	1
A	stats()	0	3	1
A	create()	0	3	1
A	peekBuried()	0	10	2
A	statsTube()	0	3	1
A	peekReady()	0	10	2
A	ignore()	0	8	2
A	watch()	0	8	2
A	reserveWithTimeout()	0	15	3
A	watchOnly()	0	10	2

<?php

declare(strict_types=1);

namespace Pheanstalk;

use Pheanstalk\Command\PeekCommand;
use Pheanstalk\Contract\CommandInterface;
use Pheanstalk\Contract\JobIdInterface;
use Pheanstalk\Contract\PheanstalkInterface;
use Pheanstalk\Contract\ResponseInterface;
use Pheanstalk\Contract\SocketFactoryInterface;
use Pheanstalk\Exception\DeadlineSoonException;

/**
 * Pheanstalk is a PHP client for the beanstalkd workqueue.
 */
class Pheanstalk implements PheanstalkInterface
{
    /**
     * @var Connection
     */
    private $connection;
    /**
     * @var ?string
     */
    private $using = PheanstalkInterface::DEFAULT_TUBE;
    /**
     * @var array<string,bool>
     */
    private $watching = [PheanstalkInterface::DEFAULT_TUBE => true];

    public function __construct(Connection $connection)
    {
        $this->connection = $connection;
    }

    /**
     * Static constructor that uses autodetection to choose an underlying socket implementation
     * @param string $host
     * @param int $port
     * @param int $connectTimeout
     * @return Pheanstalk
     */
    public static function create(string $host, int $port = 11300, int $connectTimeout = 10)
    {
        return self::createWithFactory(new SocketFactory($host, $port, $connectTimeout));
    }

    /**
     * Static constructor that uses a given socket factory for underlying connections
     * @param SocketFactoryInterface $factory
     * @return Pheanstalk
     */
    public static function createWithFactory(SocketFactoryInterface $factory)
    {
        return new self(new Connection($factory));
    }

    // ----------------------------------------

    /**
     * {@inheritdoc}
     */
    public function bury(JobIdInterface $job, int $priority = PheanstalkInterface::DEFAULT_PRIORITY): void
    {
        $this->dispatch(new Command\BuryCommand($job, $priority));
    }

    /**
     * {@inheritdoc}
     */
    public function delete(JobIdInterface $job): void
    {
        $this->dispatch(new Command\DeleteCommand($job));
    }

    /**
     * {@inheritdoc}
     */
    public function ignore(string $tube): PheanstalkInterface
    {
        if (isset($this->watching[$tube])) {
            $this->dispatch(new Command\IgnoreCommand($tube));
            unset($this->watching[$tube]);
        }

        return $this;
    }

    /**
     * {@inheritdoc}
     */
    public function kick(int $max): int
    {
        $response = $this->dispatch(new Command\KickCommand($max));

        return $response['kicked'];
    }

    /**
     * {@inheritdoc}
     */
    public function kickJob(JobIdInterface $job): void
    {
        $this->dispatch(new Command\KickJobCommand($job));
    }

    /**
     * {@inheritdoc}
     */
    public function listTubes(): array
    {
        return (array)$this->dispatch(
            new Command\ListTubesCommand()
        );
    }

    /**
     * {@inheritdoc}
     */
    public function listTubesWatched(bool $askServer = false): array
    {
        if ($askServer) {
            $response = (array)$this->dispatch(
                new Command\ListTubesWatchedCommand()
            );
            $this->watching = array_fill_keys($response, true);
        }

        return array_keys($this->watching);
    }

    /**
     * {@inheritdoc}
     */
    public function listTubeUsed(bool $askServer = false): string
    {
        if ($askServer) {
            $response = $this->dispatch(
                new Command\ListTubeUsedCommand()
            );
            $this->using = $response['tube'];
        }

        return $this->using;

The expression return $this->using could return the type null which is incompatible with the type-hinted return string. Consider adding an additional type-check to rule them out.

    }

    /**
     * {@inheritdoc}
     */
    public function pauseTube(string $tube, int $delay): void
    {
        $this->dispatch(new Command\PauseTubeCommand($tube, $delay));
    }

    /**
     * {@inheritdoc}
     */
    public function resumeTube(string $tube): void
    {
        // Pause a tube with zero delay will resume the tube
        $this->pauseTube($tube, 0);
    }

    /**
     * {@inheritdoc}
     */
    public function peek(JobIdInterface $job): Job
    {
        $response = $this->dispatch(
            new Command\PeekJobCommand($job)
        );

        return new Job($response['id'], $response['jobdata']);
    }

    /**
     * {@inheritdoc}
     */
    public function peekReady(): ?Job
    {
        $response = $this->dispatch(
            new Command\PeekCommand(PeekCommand::TYPE_READY)
        );
        if ($response->getResponseName() === ResponseInterface::RESPONSE_NOT_FOUND) {
            return null;
        }

        return new Job($response['id'], $response['jobdata']);
    }

    /**
     * {@inheritdoc}
     */
    public function peekDelayed(): ?Job
    {
        $response = $this->dispatch(
            new Command\PeekCommand(Command\PeekCommand::TYPE_DELAYED)
        );
        if ($response->getResponseName() === ResponseInterface::RESPONSE_NOT_FOUND) {
            return null;
        }

        return new Job($response['id'], $response['jobdata']);
    }

    /**
     * {@inheritdoc}
     */
    public function peekBuried(): ?Job
    {
        $response = $this->dispatch(
            new Command\PeekCommand(Command\PeekCommand::TYPE_BURIED)
        );
        if ($response->getResponseName() === ResponseInterface::RESPONSE_NOT_FOUND) {
            return null;
        }

        return new Job($response['id'], $response['jobdata']);
    }

    /**
     * {@inheritdoc}
     */
    public function put(
        string $data,
        int $priority = PheanstalkInterface::DEFAULT_PRIORITY,
        int $delay = PheanstalkInterface::DEFAULT_DELAY,
        int $ttr = PheanstalkInterface::DEFAULT_TTR
    ): Job {
        $response = $this->dispatch(
            new Command\PutCommand($data, $priority, $delay, $ttr)
        );

        return new Job($response['id'], $data);
    }

    /**
     * {@inheritdoc}
     */
    public function release(
        JobIdInterface $job,
        int $priority = PheanstalkInterface::DEFAULT_PRIORITY,
        int $delay = PheanstalkInterface::DEFAULT_DELAY
    ): void {
        $this->dispatch(
            new Command\ReleaseCommand($job, $priority, $delay)
        );
    }

    /**
     * {@inheritdoc}
     */
    public function reserve(): Job
    {
        $response = $this->dispatch(
            new Command\ReserveCommand()
        );

        return new Job($response['id'], $response['jobdata']);
    }

    /**
     * {@inheritdoc}
     */
    public function reserveJob(JobIdInterface $job): Job
    {
        // New in 1.12, beanstalkd returns BadFormat instead of UnknownCommand
        try {
            $response = $this->dispatch(
                new Command\ReserveJobCommand($job)
            );
        } catch (Exception\ServerBadFormatException $e) {
            throw new Exception\ServerUnknownCommandException();
        }

        if ($response->getResponseName() === ResponseInterface::RESPONSE_BAD_FORMAT) {
            throw new Exception\ServerUnknownCommandException();
        }

        return new Job($response['id'], $response['jobdata']);
    }

    /**
     * {@inheritdoc}
     */
    public function reserveWithTimeout(int $timeout): ?Job
    {
        $response = $this->dispatch(
            new Command\ReserveWithTimeoutCommand($timeout)
        );

        if ($response->getResponseName() === ResponseInterface::RESPONSE_DEADLINE_SOON) {
            throw new DeadlineSoonException();
        }

        if ($response->getResponseName() === ResponseInterface::RESPONSE_TIMED_OUT) {
            return null;
        }

        return new Job($response['id'], $response['jobdata']);
    }

    /**
     * {@inheritdoc}
     */
    public function statsJob(JobIdInterface $job): ResponseInterface
    {
        return $this->dispatch(new Command\StatsJobCommand($job));
    }

    /**
     * {@inheritdoc}
     */
    public function statsTube(string $tube): ResponseInterface
    {
        return $this->dispatch(new Command\StatsTubeCommand($tube));
    }

    /**
     * {@inheritdoc}
     */
    public function stats(): ResponseInterface
    {
        return $this->dispatch(new Command\StatsCommand());
    }

    /**
     * {@inheritdoc}
     */
    public function touch(JobIdInterface $job): void
    {
        $this->dispatch(new Command\TouchCommand($job));
    }

    /**
     * {@inheritdoc}
     */
    public function useTube(string $tube): PheanstalkInterface
    {
        if ($this->using !== $tube) {
            $this->dispatch(new Command\UseCommand($tube));
            $this->using = $tube;
        }

        return $this;
    }

    /**
     * {@inheritdoc}
     */
    public function watch(string $tube): PheanstalkInterface
    {
        if (!isset($this->watching[$tube])) {
            $this->dispatch(new Command\WatchCommand($tube));
            $this->watching[$tube] = true;
        }

        return $this;
    }

    /**
     * {@inheritdoc}
     */
    public function watchOnly(string $tube): PheanstalkInterface
    {
        $this->watch($tube);

        $ignoreTubes = array_diff_key($this->watching, [$tube => true]);
        foreach ($ignoreTubes as $ignoreTube => $true) {
            $this->ignore($ignoreTube);
        }

        return $this;
    }

    // ----------------------------------------

    /**
     * Dispatches the specified command to the connection object.
     *
     * If a SocketException occurs, the connection is reset, and the command is
     * re-attempted once.
     *
     * @param CommandInterface $command
     *
     * @return ResponseInterface
     */
    private function dispatch($command)
    {
        try {
            $response = $this->connection->dispatchCommand($command);
        } catch (Exception\SocketException $e) {
            $this->reconnect();
            $response = $this->connection->dispatchCommand($command);
        }

        return $response;
    }

    /**
     * Creates a new connection object, based on the existing connection object,
     * and re-establishes the used tube and watchlist.
     */
    private function reconnect()
    {
        $this->connection->disconnect();

        if ($this->using !== PheanstalkInterface::DEFAULT_TUBE) {
            $this->dispatch(new Command\UseCommand($this->using));

It seems like $this->using can also be of type null; however, parameter $tube of Pheanstalk\Command\UseCommand::__construct() does only seem to accept string, maybe add an additional type check?

        }

        foreach ($this->watching as $tube => $true) {
            if ($tube != PheanstalkInterface::DEFAULT_TUBE) {
                unset($this->watching[$tube]);
                $this->watch($tube);
            }
        }

        if (!isset($this->watching[PheanstalkInterface::DEFAULT_TUBE])) {
            $this->ignore(PheanstalkInterface::DEFAULT_TUBE);
        }
    }

    /**
     * @param string $tube The tube to use during execution
     * @param \Closure $closure Closure to execute while using the specified tube
     * @return mixed the return value of the closure.
     * @internal This is marked as internal since it is not part of a stabilized interface.
     */
    public function withUsedTube(string $tube, \Closure $closure)
    {
        $used = $this->listTubeUsed();
        try {
            $this->useTube($tube);
            return $closure($this);
        } finally {
            $this->useTube($used);
        }
    }

    /**
     * @param string $tube The tube to watch during execution
     * @param \Closure $closure Closure to execute while using the specified tube
     * @return mixed the return value of the closure.
     * @internal This is marked as internal since it is not part of a stabilized interface.
     */
    public function withWatchedTube(string $tube, \Closure $closure)
    {
        $watched = $this->listTubesWatched();
        try {
            $this->watchOnly($tube);
            return $closure($this);
        } finally {
            foreach ($watched as $tube) {
                $this->watch($tube);
            }
            if (!in_array($tube, $watched)) {
                $this->ignore($tube);
            }
        }
    }
}