Dazzle\MySQL\Database

Complexity

Total Complexity

25

Size/Duplication

Total Lines	276
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	13

Test Coverage

Coverage

0%

Importance

Changes

0

20 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	6	1
A	isPaused()	0	4	1
A	pause()	0	4	1
A	resume()	0	4	1
A	isStarted()	0	4	1
B	start()	0	41	1
A	stop()	0	14	1
A	query()	0	22	1
A	ping()	0	12	1
A	selectDB()	0	4	1
A	setOption()	0	6	1
A	getOption()	0	9	2
A	getState()	0	4	1
A	handleConnectionError()	0	4	1
A	handleConnectionClosed()	0	8	2
A	doCommand()	0	15	4
A	getServerOptions()	0	4	1
A	connect()	0	5	1
A	createExecutor()	0	4	1
A	createConfig()	0	10	1

<?php

namespace Dazzle\MySQL;

use Dazzle\Event\BaseEventEmitter;
use Dazzle\Loop\LoopAwareTrait;
use Dazzle\Loop\LoopInterface;
use Dazzle\MySQL\Command\Command;
use Dazzle\MySQL\Command\CommandInterface;
use Dazzle\MySQL\Command\Concrete\AuthCommand;
use Dazzle\MySQL\Command\Concrete\PingCommand;
use Dazzle\MySQL\Command\Concrete\QueryCommand;
use Dazzle\MySQL\Command\Concrete\QuitCommand;
use Dazzle\MySQL\Protocol\ProtocolParser;
use Dazzle\Promise\Promise;
use Dazzle\Promise\PromiseInterface;
use Dazzle\Socket\Socket;
use Dazzle\Socket\SocketInterface;
use Exception;
use InvalidArgumentException;
use RuntimeException;

class Database extends BaseEventEmitter implements DatabaseInterface
{
    use LoopAwareTrait;

    const STATE_INIT                = 0;
    const STATE_CONNECT_PENDING     = 3;
    const STATE_CONNECT_FAILED      = 1;
    const STATE_CONNECT_SUCCEEDED   = 4;
    const STATE_AUTH_PENDING        = 8;
    const STATE_AUTH_FAILED         = 2;
    const STATE_AUTH_SUCCEEDED      = 5;
    const STATE_CLOSEING            = 6;
    const STATE_STOPPED             = 7;

    protected $config;

    protected $serverOptions;

    protected $executor;

    protected $state = self::STATE_INIT;

    protected $stream;

    protected $parser;

    /**
     * @param LoopInterface $loop
     * @param mixed[] $config
     */
    public function __construct(LoopInterface $loop, $config = [])
    {
        $this->loop = $loop;
        $this->executor = $this->createExecutor();
        $this->config = $this->createConfig($config);
    }

    /**
     * @override
     * @inheritDoc
     */
    public function isPaused()
    {
        // TODO
    }

    /**
     * @override
     * @inheritDoc
     */
    public function pause()
    {
        // TODO
    }

    /**
     * @override
     * @inheritDoc
     */
    public function resume()
    {
        // TODO
    }

    /**
     * @override
     * @inheritDoc
     */
    public function isStarted()
    {
        // TODO
    }

    /**
     * @override
     * @inheritDoc
     */
    public function start()
    {
        return new Promise(function($resolve, $reject) {

            $this->state = self::STATE_CONNECT_PENDING;
            $options = $this->config;
            $streamRef = $this->stream;

            $errorHandler = function ($reason) use ($reject) {
                $this->state = self::STATE_AUTH_FAILED;
                return $reject($reason);
            };

            $connectedHandler = function ($serverOptions) use ($resolve) {
                $this->state = self::STATE_AUTH_SUCCEEDED;
                $this->serverOptions = $serverOptions;
                return $resolve($serverOptions);
            };

            $this
                ->connect()
                ->then(function ($stream) use (&$streamRef, $options, $errorHandler, $connectedHandler) {
                    $streamRef = $stream;

                    $stream->on('error', [ $this, 'handleConnectionError' ]);
                    $stream->on('close', [ $this, 'handleConnectionClosed' ]);

                    $parser = $this->parser = new ProtocolParser($stream, $this->executor);

                    $parser->setOptions($options);

                    $command = $this->doCommand(new AuthCommand($this));
                    $command->on('authenticated', $connectedHandler);
                    $command->on('error', $errorHandler);

                    //$parser->on('close', $closeHandler);

80% of this comment could be valid code. Did you maybe forget this after debugging?

                    $parser->start();

                }, [ $this, 'handleConnectionError' ]);
        });
    }

    /**
     * @override
     * @inheritDoc
     */
    public function stop()
    {
        return new Promise(function($resolve, $reject) {

The parameter $reject is not used and could be removed.

            $this
                ->doCommand(new QuitCommand($this))
                ->on('success', function() use($resolve) {
                    $this->state = self::STATE_STOPPED;
                    $this->emit('end', [ $this ]);
                    $this->emit('close', [ $this ]);
                    $resolve($this);
                });
            $this->state = self::STATE_CLOSEING;
        });
    }

    /**
     * Do a async query.
     *
     * @param string $sql
     * @param mixed[] $sqlParams
     * @return PromiseInterface
     */
    public function query($sql, $sqlParams = [])
    {
        $promise = new Promise();
        $query   = new Query($sql);
        $command = new QueryCommand($this);
        $command->setQuery($query);

        $query->bindParamsFromArray($sqlParams);
        $this->doCommand($command);

        $command->on('results', function ($rows, $command) use ($promise) {
            return $promise->resolve($command);
        });
        $command->on('error', function ($err, $command) use ($promise) {

The parameter $command is not used and could be removed.

            return $promise->reject($err);
        });
        $command->on('success', function ($command) use ($promise) {
            return $promise->resolve($command);
        });

        return $promise;
    }

    public function ping()
    {
        $promise = new Promise();

        $this->doCommand(new PingCommand($this))
            ->on('error', function ($reason) use ($promise) {
                return $promise->reject($reason);
            })
            ->on('success', function () use ($promise) {
                return $promise->resolve();
            });
    }

    public function selectDB($dbname)
    {
        return $this->query(sprintf('USE `%s`', $dbname));
    }

    public function setOption($name, $value)
    {
        $this->config[$name] = $value;

        return $this;
    }

    public function getOption($name, $default = null)
    {
        if (isset($this->config[$name]))
        {
            return $this->config[$name];
        }

        return $default;
    }

    public function getState()
    {
        return $this->state;
    }

    public function handleConnectionError($err)
    {
        $this->emit('error', [ $err, $this ]);
    }

    public function handleConnectionClosed()
    {
        if ($this->state < self::STATE_CLOSEING)
        {
            $this->state = self::STATE_STOPPED;
            $this->emit('error', [ new RuntimeException('mysql server has gone away'), $this ]);
        }
    }

    protected function doCommand(CommandInterface $command)
    {
        if ($command->equals(Command::INIT_AUTHENTICATE))
        {
            return $this->executor->undequeue($command);
        }
        elseif ($this->state >= self::STATE_CONNECT_PENDING && $this->state <= self::STATE_AUTH_SUCCEEDED)
        {
            return $this->executor->enqueue($command);
        }
        else
        {
            throw new Exception("Cann't send command");
        }
    }

    public function getServerOptions()
    {
        return $this->serverOptions;
    }

    protected function connect()
    {
        $socket = new Socket($this->config['endpoint'], $this->getLoop());
        return Promise::doResolve($socket);
    }

    /**
     * Create executor.
     *
     * @return Executor
     */
    protected function createExecutor()
    {
        return new Executor($this);
    }

    /**
     * Create configuration file.
     *
     * @param mixed[] $config
     * @return mixed[]
     */
    protected function createConfig($config = [])
    {
        $default = [
            'endpoint' => 'tcp://127.0.0.1:3306',
            'user'     => 'root',
            'pass'     => '',
            'dbname'   => '',
        ];
        return array_merge($default, $config);
    }
}