fuitad\LaravelCassandra\Connection

Complexity

Total Complexity

39

Size/Duplication

Total Lines	306
Duplicated Lines	10.46 %

Coupling/Cohesion

Components	1
Dependencies	4

Importance

Changes

0

17 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	24	4
A	table()	0	8	1
A	getCassandraCluster()	0	4	1
A	getCassandraSession()	0	4	1
A	getKeyspace()	0	4	1
C	createCluster()	6	41	15
A	disconnect()	0	4	1
A	getDriverName()	0	4	1
A	select()	0	12	2
A	insertBulk()	0	4	1
A	batchStatement()	0	17	3
A	statement()	12	12	2
A	affectingStatement()	14	14	2
A	getDefaultPostProcessor()	0	4	1
A	getDefaultQueryGrammar()	0	4	1
A	getDefaultSchemaGrammar()	0	4	1
A	__call()	0	4	1

<?php

namespace fuitad\LaravelCassandra;

use Cassandra;
use Cassandra\BatchStatement;

class Connection extends \Illuminate\Database\Connection
{
    const DEFAULT_PAGE_SIZE = 5000;
    
    /**
     * The Cassandra keyspace
     *
     * @var string
     */
    protected $keyspace;

    /**
     * The Cassandra cluster
     *
     * @var \Cassandra\Cluster
     */
    protected $cluster;

    /**
     * The Cassandra connection handler.
     *
     * @var \Cassandra\Session
     */
    protected $session;

    /**
     * Create a new database connection instance.
     *
     * @param  array   $config
     */
    public function __construct(array $config)
    {
        $this->config = $config;
        if (empty($this->config['page_size'])) {
            $this->config['page_size'] = self::DEFAULT_PAGE_SIZE;
        }
        
        // You can pass options directly to the Cassandra constructor
        $options = array_get($config, 'options', []);

        // Create the connection
        $this->cluster = $this->createCluster(null, $config, $options);

        if (isset($options['database']) || isset($config['database'])) {
            $this->keyspace = $config['database'];
            $this->session = $this->cluster->connect($config['database']);
        }

        $this->useDefaultPostProcessor();

        $this->useDefaultSchemaGrammar();

        $this->setQueryGrammar($this->getDefaultQueryGrammar());
    }

    /**
     * Begin a fluent query against a database table.
     *
     * @param  string  $table
     * @return Query\Builder
     */
    public function table($table)
    {
        $processor = $this->getPostProcessor();

        $query = new Query\Builder($this, null, $processor);

$processor is of type object<Illuminate\Database\Query\Processors\Processor>, but the function expects a null|object<fuitad\LaravelCassandra\Query\Processor>.

        return $query->from($table);
    }

    /**
     * return Cassandra cluster.
     *
     * @return \Cassandra\Cluster
     */
    public function getCassandraCluster()
    {
        return $this->cluster;
    }

    /**
     * return Cassandra Session.
     *
     * @return \Cassandra\Session
     */
    public function getCassandraSession()
    {
        return $this->session;
    }

    /**
     * Return the Cassandra keyspace
     *
     * @return string
     */
    public function getKeyspace()
    {
        return $this->keyspace;
    }

    /**
     * Create a new Cassandra cluster object.
     *
     * @param  string  $dsn
     * @param  array   $config
     * @param  array   $options
     * @return \Cassandra\Cluster
     */
    protected function createCluster($dsn, array $config, array $options)

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

    {
        // By default driver options is an empty array.
        $driverOptions = [];

$driverOptions is not used, you could remove the assignment.

        if (isset($config['driver_options']) && is_array($config['driver_options'])) {
            $driverOptions = $config['driver_options'];

$driverOptions is not used, you could remove the assignment.

        }

        $cluster = Cassandra::cluster();

        // Check if the credentials are not already set in the options
        if (!isset($options['username']) && !empty($config['username'])) {
            $options['username'] = $config['username'];
        }
        if (!isset($options['password']) && !empty($config['password'])) {
            $options['password'] = $config['password'];
        }

        // Authentication
        if (isset($options['username']) && isset($options['password'])) {
            $cluster->withCredentials($options['username'], $options['password']);
        }

        // Contact Points/Host
        if (isset($options['contactpoints']) || (isset($config['host']) && !empty($config['host']))) {
            $contactPoints = $config['host'];

            if (isset($options['contactpoints'])) {
                $contactPoints = $options['contactpoints'];
            }

            call_user_func_array([$cluster, 'withContactPoints'], $contactPoints);
        }

        if (!isset($options['port']) && !empty($config['port'])) {
            $cluster->withPort((int) $config['port']);
        }

        return $cluster->build();
    }

    /**
     * Disconnect from the underlying Cassandra connection.
     */
    public function disconnect()
    {
        unset($this->connection);
    }

    /**
     * Get the PDO driver name.
     *
     * @return string
     */
    public function getDriverName()
    {
        return 'cassandra';
    }

    /**
     * Run a select statement against the database.
     *
     * @param  string  $query
     * @param  array  $bindings
     * @param  bool  $useReadPdo
     * @return array
     */
    public function select($query, $bindings = [], $useReadPdo = true)
    {
        return $this->run($query, $bindings, function ($query, $bindings) use ($useReadPdo) {
            if ($this->pretending()) {
                return [];
            }

            $preparedStatement = $this->session->prepare($query);

            return $this->session->execute($preparedStatement, ['arguments' => $bindings, 'page_size' => (int)$this->config['page_size']]);
        });
    }

    /**
     * Run an bulk insert statement against the database.
     *
     * @param  array  $queries
     * @param  array  $bindings
     * @return bool
     */
    public function insertBulk($queries = [], $bindings = [], $type = Cassandra::BATCH_LOGGED)
    {
        return $this->batchStatement($queries, $bindings, $type);
    }

    /**
     * Execute a group of queries inside a batch statement against the database.
     *
     * @param  array  $queries
     * @param  array  $bindings
     * @return bool
     */
    public function batchStatement($queries = [], $bindings = [], $type = Cassandra::BATCH_LOGGED)

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

    {
        return $this->run($queries, $bindings, function ($queries, $bindings) {

$queries is of type array, but the function expects a string.

            if ($this->pretending()) {
                return [];
            }

            $batch = new BatchStatement(Cassandra::BATCH_LOGGED);

            foreach ($queries as $k => $query) {
                $preparedStatement = $this->session->prepare($query);
                $batch->add($preparedStatement, $bindings[$k]);
            }

            return $this->session->execute($batch);
        });
    }

    /**
     * Execute an SQL statement and return the boolean result.
     *
     * @param  string  $query
     * @param  array   $bindings
     * @return bool
     */
    public function statement($query, $bindings = [])
    {
        return $this->run($query, $bindings, function ($query, $bindings) {
            if ($this->pretending()) {
                return [];
            }

            $preparedStatement = $this->session->prepare($query);

            return $this->session->execute($preparedStatement, ['arguments' => $bindings]);
        });
    }

    /**
     * Because Cassandra is an eventually consistent database, it's not possible to obtain
     * the affected count for statements so we're just going to return 0, based on the idea
     * that if the query fails somehow, an exception will be thrown
     *
     * @param  string  $query
     * @param  array   $bindings
     * @return int
     */
    public function affectingStatement($query, $bindings = [])
    {
        return $this->run($query, $bindings, function ($query, $bindings) {
            if ($this->pretending()) {
                return 0;
            }

            $preparedStatement = $this->session->prepare($query);

            $this->session->execute($preparedStatement, ['arguments' => $bindings]);

            return 1;
        });
    }

    /**
     * @inheritdoc
     */
    protected function getDefaultPostProcessor()
    {
        return new Query\Processor();
    }

    /**
     * @inheritdoc
     */
    protected function getDefaultQueryGrammar()
    {
        return new Query\Grammar();
    }

    /**
     * @inheritdoc
     */
    protected function getDefaultSchemaGrammar()
    {
        //return new Schema\Grammar();
    }

    /**
     * Dynamically pass methods to the connection.
     *
     * @param  string  $method
     * @param  array   $parameters
     * @return mixed
     */
    public function __call($method, $parameters)
    {
        return call_user_func_array([$this->cluster, $method], $parameters);
    }
}