diff --git a/src/Connection.php b/src/Connection.php
index 3935ebe..e4fae4c 100644
--- a/src/Connection.php
+++ b/src/Connection.php
@@ -23,37 +23,13 @@
use Doctrine\DBAL\Query\QueryBuilder;
use Doctrine\DBAL\Schema\Schema;
use Drift\DBAL\Driver\Driver;
-use Drift\DBAL\Mock\MockedDBALConnection;
-use Drift\DBAL\Mock\MockedDriver;
-use function React\Promise\map;
use React\Promise\PromiseInterface;
/**
* Class Connection.
*/
-class Connection
+interface Connection
{
- private Driver $driver;
- private Credentials $credentials;
- private AbstractPlatform $platform;
-
- /**
- * Connection constructor.
- *
- * @param Driver $driver
- * @param Credentials $credentials
- * @param AbstractPlatform $platform
- */
- private function __construct(
- Driver $driver,
- Credentials $credentials,
- AbstractPlatform $platform
- ) {
- $this->driver = $driver;
- $this->credentials = $credentials;
- $this->platform = $platform;
- }
-
/**
* Create new connection.
*
@@ -67,9 +43,7 @@ public static function create(
Driver $driver,
Credentials $credentials,
AbstractPlatform $platform
- ) {
- return new self($driver, $credentials, $platform);
- }
+ ): Connection;
/**
* Create new connection.
@@ -84,32 +58,22 @@ public static function createConnected(
Driver $driver,
Credentials $credentials,
AbstractPlatform $platform
- ) {
- $connection = new self($driver, $credentials, $platform);
- $connection->connect();
+ ): Connection;
- return $connection;
- }
+ /**
+ * @return string
+ */
+ public function getDriverNamespace(): string;
/**
* Connect.
*/
- public function connect()
- {
- $this
- ->driver
- ->connect($this->credentials);
- }
+ public function connect();
/**
* Close.
*/
- public function close()
- {
- $this
- ->driver
- ->close();
- }
+ public function close();
/**
* Creates QueryBuilder.
@@ -118,14 +82,7 @@ public function close()
*
* @throws DBALException
*/
- public function createQueryBuilder(): QueryBuilder
- {
- return new QueryBuilder(
- new MockedDBALConnection([
- 'platform' => $this->platform,
- ], new MockedDriver())
- );
- }
+ public function createQueryBuilder(): QueryBuilder;
/**
* Query by query builder.
@@ -134,13 +91,7 @@ public function createQueryBuilder(): QueryBuilder
*
* @return PromiseInterface<Result>
*/
- public function query(QueryBuilder $queryBuilder): PromiseInterface
- {
- return $this->queryBySQL(
- $queryBuilder->getSQL(),
- $queryBuilder->getParameters()
- );
- }
+ public function query(QueryBuilder $queryBuilder): PromiseInterface;
/**
* Query by sql and parameters.
@@ -150,12 +101,7 @@ public function query(QueryBuilder $queryBuilder): PromiseInterface
*
* @return PromiseInterface<Result>
*/
- public function queryBySQL(string $sql, array $parameters = []): PromiseInterface
- {
- return $this
- ->driver
- ->query($sql, $parameters);
- }
+ public function queryBySQL(string $sql, array $parameters = []): PromiseInterface;
/**
* Execute, sequentially, an array of sqls.
@@ -164,16 +110,7 @@ public function queryBySQL(string $sql, array $parameters = []): PromiseInterfac
*
* @return PromiseInterface<Connection>
*/
- public function executeSQLs(array $sqls): PromiseInterface
- {
- return
- map($sqls, function (string $sql) {
- return $this->queryBySQL($sql);
- })
- ->then(function () {
- return $this;
- });
- }
+ public function executeSQLs(array $sqls): PromiseInterface;
/**
* Execute an schema.
@@ -182,14 +119,7 @@ public function executeSQLs(array $sqls): PromiseInterface
*
* @return PromiseInterface<Connection>
*/
- public function executeSchema(Schema $schema): PromiseInterface
- {
- return $this
- ->executeSQLs($schema->toSql($this->platform))
- ->then(function () {
- return $this;
- });
- }
+ public function executeSchema(Schema $schema): PromiseInterface;
/**
* Shortcuts.
@@ -208,13 +138,7 @@ public function executeSchema(Schema $schema): PromiseInterface
public function findOneBy(
string $table,
array $where
- ): PromiseInterface {
- return $this
- ->getResultByWhereClause($table, $where)
- ->then(function (Result $result) {
- return $result->fetchFirstRow();
- });
- }
+ ): PromiseInterface;
/**
* Find by.
@@ -229,13 +153,7 @@ public function findOneBy(
public function findBy(
string $table,
array $where = []
- ): PromiseInterface {
- return $this
- ->getResultByWhereClause($table, $where)
- ->then(function (Result $result) {
- return $result->fetchAllRows();
- });
- }
+ ): PromiseInterface;
/**
* @param string $table
@@ -246,11 +164,7 @@ public function findBy(
public function insert(
string $table,
array $values
- ): PromiseInterface {
- $queryBuilder = $this->createQueryBuilder();
-
- return $this->driver->insert($queryBuilder, $table, $values);
- }
+ ): PromiseInterface;
/**
* @param string $table
@@ -263,19 +177,7 @@ public function insert(
public function delete(
string $table,
array $values
- ): PromiseInterface {
- if (empty($values)) {
- throw InvalidArgumentException::fromEmptyCriteria();
- }
-
- $queryBuilder = $this
- ->createQueryBuilder()
- ->delete($table);
-
- $this->applyWhereClausesFromArray($queryBuilder, $values);
-
- return $this->query($queryBuilder);
- }
+ ): PromiseInterface;
/**
* @param string $table
@@ -290,25 +192,7 @@ public function update(
string $table,
array $id,
array $values
- ): PromiseInterface {
- if (empty($id)) {
- throw InvalidArgumentException::fromEmptyCriteria();
- }
-
- $queryBuilder = $this
- ->createQueryBuilder()
- ->update($table);
-
- $parameters = $queryBuilder->getParameters();
- foreach ($values as $field => $value) {
- $queryBuilder->set($field, '?');
- $parameters[] = $value;
- }
- $queryBuilder->setParameters($parameters);
- $this->applyWhereClausesFromArray($queryBuilder, $id);
-
- return $this->query($queryBuilder);
- }
+ ): PromiseInterface;
/**
* @param string $table
@@ -323,15 +207,7 @@ public function upsert(
string $table,
array $id,
array $values
- ) {
- return $this
- ->findOneBy($table, $id)
- ->then(function (?array $result) use ($table, $id, $values) {
- return is_null($result)
- ? $this->insert($table, array_merge($id, $values))
- : $this->update($table, $id, $values);
- });
- }
+ ): PromiseInterface;
/**
* Table related shortcuts.
@@ -359,38 +235,7 @@ public function createTable(
array $fields,
array $extra = [],
bool $autoincrementId = false
- ): PromiseInterface {
- if (empty($fields)) {
- throw InvalidArgumentException::fromEmptyCriteria();
- }
-
- $schema = new Schema();
- $table = $schema->createTable($name);
- foreach ($fields as $field => $type) {
- $extraField = (
- array_key_exists($field, $extra) &&
- is_array($extra[$field])
- ) ? $extra[$field] : [];
-
- if (
- 'string' == $type &&
- !array_key_exists('length', $extraField)
- ) {
- $extraField = array_merge(
- $extraField,
- ['length' => 255]
- );
- }
-
- $table->addColumn($field, $type, $extraField);
- }
-
- $id = array_key_first($fields);
- $table->setPrimaryKey([$id]);
- $table->getColumn($id)->setAutoincrement($autoincrementId);
-
- return $this->executeSchema($schema);
- }
+ ): PromiseInterface;
/**
* @param string $name
@@ -399,14 +244,7 @@ public function createTable(
*
* @throws TableNotFoundException
*/
- public function dropTable(string $name): PromiseInterface
- {
- return $this
- ->queryBySQL("DROP TABLE $name")
- ->then(function () {
- return $this;
- });
- }
+ public function dropTable(string $name): PromiseInterface;
/**
* @param string $name
@@ -415,79 +253,5 @@ public function dropTable(string $name): PromiseInterface
*
* @throws TableNotFoundException
*/
- public function truncateTable(string $name): PromiseInterface
- {
- $truncateTableQuery = $this
- ->platform
- ->getTruncateTableSQL($name);
-
- return $this
- ->queryBySQL($truncateTableQuery)
- ->then(function () {
- return $this;
- });
- }
-
- /**
- * Get result by where clause.
- *
- * @param string $table
- * @param array $where
- *
- * @return PromiseInterface<Result>
- */
- private function getResultByWhereClause(
- string $table,
- array $where
- ): PromiseInterface {
- $queryBuilder = $this
- ->createQueryBuilder()
- ->select('t.*')
- ->from($table, 't');
-
- $this->applyWhereClausesFromArray($queryBuilder, $where);
-
- return $this->query($queryBuilder);
- }
-
- /**
- * Apply where clauses.
- *
- * [
- * "id" => 1,
- * "name" => "Marc"
- * ]
- *
- * to
- *
- * [
- * [ "id = ?", "name = ?"],
- * [1, "Marc"]
- * ]
- *
- * @param QueryBuilder $queryBuilder
- * @param array $array
- */
- private function applyWhereClausesFromArray(
- QueryBuilder $queryBuilder,
- array $array
- ) {
- $params = $queryBuilder->getParameters();
- foreach ($array as $field => $value) {
- if (\is_null($value)) {
- $queryBuilder->andWhere(
- $queryBuilder->expr()->isNull($field)
- );
- continue;
- }
-
- $queryBuilder->andWhere(
- $queryBuilder->expr()->eq($field, '?')
- );
-
- $params[] = $value;
- }
-
- $queryBuilder->setParameters($params);
- }
+ public function truncateTable(string $name): PromiseInterface;
}
diff --git a/src/ConnectionPool.php b/src/ConnectionPool.php
new file mode 100644
index 0000000..466a57b
--- /dev/null
+++ b/src/ConnectionPool.php
@@ -0,0 +1,434 @@
+<?php
+
+/*
+ * This file is part of the DriftPHP Project
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ *
+ * Feel free to edit as you please, and have fun.
+ *
+ * @author Marc Morera <yuhu@mmoreram.com>
+ */
+
+declare(strict_types=1);
+
+namespace Drift\DBAL;
+
+use Doctrine\DBAL\Exception as DBALException;
+use Doctrine\DBAL\Exception\InvalidArgumentException;
+use Doctrine\DBAL\Exception\TableExistsException;
+use Doctrine\DBAL\Exception\TableNotFoundException;
+use Doctrine\DBAL\Platforms\AbstractPlatform;
+use Doctrine\DBAL\Query\QueryBuilder;
+use Doctrine\DBAL\Schema\Schema;
+use Drift\DBAL\Driver\Driver;
+use React\Promise\PromiseInterface;
+
+/**
+ * Class ConnectionPool.
+ */
+class ConnectionPool implements Connection, ConnectionPoolInterface
+{
+ private array $connections;
+
+ /**
+ * Connection constructor.
+ *
+ * @param ConnectionWorker[] $connections
+ */
+ private function __construct(array $connections)
+ {
+ $this->connections = $connections;
+ }
+
+ /**
+ * Create new connection.
+ *
+ * @param Driver $driver
+ * @param Credentials $credentials
+ * @param AbstractPlatform $platform
+ *
+ * @return Connection
+ */
+ public static function create(
+ Driver $driver,
+ Credentials $credentials,
+ AbstractPlatform $platform
+ ): Connection {
+ $numberOfConnections = $credentials->getConnections();
+ if ($numberOfConnections <= 1) {
+ return SingleConnection::create(
+ $driver,
+ $credentials,
+ $platform
+ );
+ }
+
+ $connections = [];
+ for ($i = 0; $i < $numberOfConnections; ++$i) {
+ $connections[] = new ConnectionWorker(
+ SingleConnection::create(
+ clone $driver,
+ $credentials,
+ $platform
+ ), $i
+ );
+ }
+
+ return new self($connections);
+ }
+
+ /**
+ * Create new connection.
+ *
+ * @param Driver $driver
+ * @param Credentials $credentials
+ * @param AbstractPlatform $platform
+ *
+ * @return Connection
+ */
+ public static function createConnected(
+ Driver $driver,
+ Credentials $credentials,
+ AbstractPlatform $platform
+ ): Connection {
+ $connection = self::create($driver, $credentials, $platform);
+ $connection->connect();
+
+ return $connection;
+ }
+
+ /**
+ * @return string
+ */
+ public function getDriverNamespace(): string
+ {
+ return $this
+ ->bestConnection()
+ ->getConnection()
+ ->getDriverNamespace();
+ }
+
+ /**
+ * @param bool $increaseJobs
+ *
+ * @return ConnectionWorker
+ */
+ private function bestConnection(bool $increaseJobs = false): ConnectionWorker
+ {
+ $minJobs = 1000000000;
+ $minJobsConnection = null;
+ foreach ($this->connections as $i => $connection) {
+ if ($connection->getJobs() < $minJobs) {
+ $minJobs = $connection->getJobs();
+ $minJobsConnection = $i;
+ }
+ }
+
+ if ($increaseJobs) {
+ $this->connections[$minJobsConnection]->startJob();
+ }
+
+ return $this->connections[$minJobsConnection];
+ }
+
+ /**
+ * @return ConnectionWorker
+ */
+ private function firstConnection(): ConnectionWorker
+ {
+ return $this->connections[0];
+ }
+
+ /**
+ * @param callable $callable
+ *
+ * @return PromiseInterface
+ */
+ private function executeInBestConnection(callable $callable): PromiseInterface
+ {
+ $connectionWorker = $this->bestConnection(true);
+
+ return $callable($connectionWorker->getConnection())
+ ->then(function ($whatever) use ($connectionWorker) {
+ $connectionWorker->stopJob();
+
+ return $whatever;
+ });
+ }
+
+ /**
+ * Connect.
+ */
+ public function connect()
+ {
+ foreach ($this->connections as $connection) {
+ $connection->getConnection()->connect();
+ }
+ }
+
+ /**
+ * Close.
+ */
+ public function close()
+ {
+ foreach ($this->connections as $connection) {
+ $connection->getConnection()->close();
+ }
+ }
+
+ /**
+ * Creates QueryBuilder.
+ *
+ * @return QueryBuilder
+ *
+ * @throws DBALException
+ */
+ public function createQueryBuilder(): QueryBuilder
+ {
+ return $this
+ ->firstConnection()
+ ->getConnection()
+ ->createQueryBuilder();
+ }
+
+ /**
+ * Query by query builder.
+ *
+ * @param QueryBuilder $queryBuilder
+ *
+ * @return PromiseInterface<Result>
+ */
+ public function query(QueryBuilder $queryBuilder): PromiseInterface
+ {
+ return $this->executeInBestConnection(function (Connection $connection) use ($queryBuilder) {
+ return $connection->query($queryBuilder);
+ });
+ }
+
+ /**
+ * Query by sql and parameters.
+ *
+ * @param string $sql
+ * @param array $parameters
+ *
+ * @return PromiseInterface<Result>
+ */
+ public function queryBySQL(string $sql, array $parameters = []): PromiseInterface
+ {
+ return $this->executeInBestConnection(function (Connection $connection) use ($sql, $parameters) {
+ return $connection->queryBySQL($sql, $parameters);
+ });
+ }
+
+ /**
+ * Execute, sequentially, an array of sqls.
+ *
+ * @param string[] $sqls
+ *
+ * @return PromiseInterface<Connection>
+ */
+ public function executeSQLs(array $sqls): PromiseInterface
+ {
+ return $this->executeInBestConnection(function (Connection $connection) use ($sqls) {
+ return $connection->executeSQLs($sqls);
+ });
+ }
+
+ /**
+ * Execute an schema.
+ *
+ * @param Schema $schema
+ *
+ * @return PromiseInterface<Connection>
+ */
+ public function executeSchema(Schema $schema): PromiseInterface
+ {
+ return $this->executeInBestConnection(function (Connection $connection) use ($schema) {
+ return $connection->executeSchema($schema);
+ });
+ }
+
+ /**
+ * Shortcuts.
+ */
+
+ /**
+ * Find one by.
+ *
+ * connection->findOneById('table', ['id' => 1]);
+ *
+ * @param string $table
+ * @param array $where
+ *
+ * @return PromiseInterface<array|null>
+ */
+ public function findOneBy(
+ string $table,
+ array $where
+ ): PromiseInterface {
+ return $this->executeInBestConnection(function (Connection $connection) use ($table, $where) {
+ return $connection->findOneBy($table, $where);
+ });
+ }
+
+ /**
+ * Find by.
+ *
+ * connection->findBy('table', ['id' => 1]);
+ *
+ * @param string $table
+ * @param array $where
+ *
+ * @return PromiseInterface<array>
+ */
+ public function findBy(
+ string $table,
+ array $where = []
+ ): PromiseInterface {
+ return $this->executeInBestConnection(function (Connection $connection) use ($table, $where) {
+ return $connection->findBy($table, $where);
+ });
+ }
+
+ /**
+ * @param string $table
+ * @param array $values
+ *
+ * @return PromiseInterface
+ */
+ public function insert(
+ string $table,
+ array $values
+ ): PromiseInterface {
+ return $this->executeInBestConnection(function (Connection $connection) use ($table, $values) {
+ return $connection->insert($table, $values);
+ });
+ }
+
+ /**
+ * @param string $table
+ * @param array $values
+ *
+ * @return PromiseInterface
+ *
+ * @throws InvalidArgumentException
+ */
+ public function delete(
+ string $table,
+ array $values
+ ): PromiseInterface {
+ return $this->executeInBestConnection(function (Connection $connection) use ($table, $values) {
+ return $connection->delete($table, $values);
+ });
+ }
+
+ /**
+ * @param string $table
+ * @param array $id
+ * @param array $values
+ *
+ * @return PromiseInterface
+ *
+ * @throws InvalidArgumentException
+ */
+ public function update(
+ string $table,
+ array $id,
+ array $values
+ ): PromiseInterface {
+ return $this->executeInBestConnection(function (Connection $connection) use ($table, $id, $values) {
+ return $connection->update($table, $id, $values);
+ });
+ }
+
+ /**
+ * @param string $table
+ * @param array $id
+ * @param array $values
+ *
+ * @return PromiseInterface
+ *
+ * @throws InvalidArgumentException
+ */
+ public function upsert(
+ string $table,
+ array $id,
+ array $values
+ ): PromiseInterface {
+ return $this->executeInBestConnection(function (Connection $connection) use ($table, $id, $values) {
+ return $connection->upsert($table, $id, $values);
+ });
+ }
+
+ /**
+ * Table related shortcuts.
+ */
+
+ /**
+ * Easy shortcut for creating tables. Fields is just a simple key value,
+ * being the key the name of the field, and the value the type. By default,
+ * Varchar types have length 255.
+ *
+ * First field is considered as primary key.
+ *
+ * @param string $name
+ * @param array $fields
+ * @param array $extra
+ * @param bool $autoincrementId
+ *
+ * @return PromiseInterface<Connection>
+ *
+ * @throws InvalidArgumentException
+ * @throws TableExistsException
+ */
+ public function createTable(
+ string $name,
+ array $fields,
+ array $extra = [],
+ bool $autoincrementId = false
+ ): PromiseInterface {
+ return $this->executeInBestConnection(function (Connection $connection) use ($name, $fields, $extra, $autoincrementId) {
+ return $connection->createTable($name, $fields, $extra, $autoincrementId);
+ });
+ }
+
+ /**
+ * @param string $name
+ *
+ * @return PromiseInterface<Connection>
+ *
+ * @throws TableNotFoundException
+ */
+ public function dropTable(string $name): PromiseInterface
+ {
+ return $this->executeInBestConnection(function (Connection $connection) use ($name) {
+ return $connection->dropTable($name);
+ });
+ }
+
+ /**
+ * @param string $name
+ *
+ * @return PromiseInterface<Connection>
+ *
+ * @throws TableNotFoundException
+ */
+ public function truncateTable(string $name): PromiseInterface
+ {
+ return $this->executeInBestConnection(function (Connection $connection) use ($name) {
+ return $connection->truncateTable($name);
+ });
+ }
+
+ /**
+ * Get the Pool's connection workers
+ *
+ * @return ConnectionWorker[]
+ */
+ public function getWorkers(): array
+ {
+ return $this->connections;
+ }
+}
diff --git a/src/ConnectionPoolInterface.php b/src/ConnectionPoolInterface.php
new file mode 100644
index 0000000..d90eb31
--- /dev/null
+++ b/src/ConnectionPoolInterface.php
@@ -0,0 +1,13 @@
+<?php
+
+namespace Drift\DBAL;
+
+interface ConnectionPoolInterface
+{
+ /**
+ * Get the Pool's connection workers
+ *
+ * @return ConnectionWorker[]
+ */
+ public function getWorkers(): array;
+}
diff --git a/src/ConnectionWorker.php b/src/ConnectionWorker.php
new file mode 100644
index 0000000..3537d7b
--- /dev/null
+++ b/src/ConnectionWorker.php
@@ -0,0 +1,71 @@
+<?php
+
+/*
+ * This file is part of the DriftPHP Project
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ *
+ * Feel free to edit as you please, and have fun.
+ *
+ * @author Marc Morera <yuhu@mmoreram.com>
+ */
+
+declare(strict_types=1);
+
+namespace Drift\DBAL;
+
+/**
+ * Class ConnectionWorker.
+ */
+class ConnectionWorker
+{
+ private Connection $connection;
+ private int $id;
+ private int $jobs;
+
+ /**
+ * @param Connection $connection
+ * @param int $id
+ */
+ public function __construct(Connection $connection, int $id)
+ {
+ $this->connection = $connection;
+ $this->id = $id;
+ $this->jobs = 0;
+ }
+
+ public function startJob()
+ {
+ ++$this->jobs;
+ }
+
+ public function stopJob()
+ {
+ --$this->jobs;
+ }
+
+ /**
+ * @return Connection
+ */
+ public function getConnection(): Connection
+ {
+ return $this->connection;
+ }
+
+ /**
+ * @return int
+ */
+ public function getId(): int
+ {
+ return $this->id;
+ }
+
+ /**
+ * @return int
+ */
+ public function getJobs(): int
+ {
+ return $this->jobs;
+ }
+}
diff --git a/src/Credentials.php b/src/Credentials.php
index 5771f5c..dcd9aec 100644
--- a/src/Credentials.php
+++ b/src/Credentials.php
@@ -20,35 +20,13 @@
*/
class Credentials
{
- /**
- * @var string
- */
- private $host;
-
- /**
- * @var string
- */
- private $port;
-
- /**
- * @var string
- */
- private $user;
-
- /**
- * @var string
- */
- private $password;
-
- /**
- * @var string
- */
- private $dbName;
-
- /**
- * @var array
- */
- private $options;
+ private string $host;
+ private string $port;
+ private string $user;
+ private string $password;
+ private string $dbName;
+ private array $options;
+ private int $connections;
/**
* Credentials constructor.
@@ -59,6 +37,7 @@ class Credentials
* @param string $password
* @param string $dbName
* @param array $options
+ * @param int $connections
*/
public function __construct(
string $host,
@@ -66,7 +45,8 @@ public function __construct(
string $user,
string $password,
string $dbName,
- array $options = []
+ array $options = [],
+ int $connections = 1
) {
$this->host = $host;
$this->port = $port;
@@ -74,6 +54,7 @@ public function __construct(
$this->password = $password;
$this->dbName = $dbName;
$this->options = $options;
+ $this->connections = $connections;
}
/**
@@ -124,6 +105,14 @@ public function getOptions(): array
return $this->options;
}
+ /**
+ * @return int
+ */
+ public function getConnections(): int
+ {
+ return $this->connections;
+ }
+
/**
* To string.
*/
diff --git a/src/SingleConnection.php b/src/SingleConnection.php
new file mode 100644
index 0000000..6b113f9
--- /dev/null
+++ b/src/SingleConnection.php
@@ -0,0 +1,498 @@
+<?php
+
+/*
+ * This file is part of the DriftPHP Project
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ *
+ * Feel free to edit as you please, and have fun.
+ *
+ * @author Marc Morera <yuhu@mmoreram.com>
+ */
+
+declare(strict_types=1);
+
+namespace Drift\DBAL;
+
+use Doctrine\DBAL\Exception as DBALException;
+use Doctrine\DBAL\Exception\InvalidArgumentException;
+use Doctrine\DBAL\Exception\TableExistsException;
+use Doctrine\DBAL\Exception\TableNotFoundException;
+use Doctrine\DBAL\Platforms\AbstractPlatform;
+use Doctrine\DBAL\Query\QueryBuilder;
+use Doctrine\DBAL\Schema\Schema;
+use Drift\DBAL\Driver\Driver;
+use Drift\DBAL\Mock\MockedDBALConnection;
+use Drift\DBAL\Mock\MockedDriver;
+use function React\Promise\map;
+use React\Promise\PromiseInterface;
+
+class SingleConnection implements Connection
+{
+ private Driver $driver;
+ private Credentials $credentials;
+ private AbstractPlatform $platform;
+
+ /**
+ * Connection constructor.
+ *
+ * @param Driver $driver
+ * @param Credentials $credentials
+ * @param AbstractPlatform $platform
+ */
+ private function __construct(
+ Driver $driver,
+ Credentials $credentials,
+ AbstractPlatform $platform
+ ) {
+ $this->driver = $driver;
+ $this->credentials = $credentials;
+ $this->platform = $platform;
+ }
+
+ /**
+ * Create new connection.
+ *
+ * @param Driver $driver
+ * @param Credentials $credentials
+ * @param AbstractPlatform $platform
+ *
+ * @return Connection
+ */
+ public static function create(
+ Driver $driver,
+ Credentials $credentials,
+ AbstractPlatform $platform
+ ): Connection {
+ return new self($driver, $credentials, $platform);
+ }
+
+ /**
+ * Create new connection.
+ *
+ * @param Driver $driver
+ * @param Credentials $credentials
+ * @param AbstractPlatform $platform
+ *
+ * @return Connection
+ */
+ public static function createConnected(
+ Driver $driver,
+ Credentials $credentials,
+ AbstractPlatform $platform
+ ): Connection {
+ $connection = self::create($driver, $credentials, $platform);
+ $connection->connect();
+
+ return $connection;
+ }
+
+ /**
+ * @return string
+ */
+ public function getDriverNamespace(): string
+ {
+ return get_class($this->driver);
+ }
+
+ /**
+ * Connect.
+ */
+ public function connect()
+ {
+ $this
+ ->driver
+ ->connect($this->credentials);
+ }
+
+ /**
+ * Close.
+ */
+ public function close()
+ {
+ $this
+ ->driver
+ ->close();
+ }
+
+ /**
+ * Creates QueryBuilder.
+ *
+ * @return QueryBuilder
+ *
+ * @throws DBALException
+ */
+ public function createQueryBuilder(): QueryBuilder
+ {
+ return new QueryBuilder(
+ new MockedDBALConnection([
+ 'platform' => $this->platform,
+ ], new MockedDriver())
+ );
+ }
+
+ /**
+ * Query by query builder.
+ *
+ * @param QueryBuilder $queryBuilder
+ *
+ * @return PromiseInterface<Result>
+ */
+ public function query(QueryBuilder $queryBuilder): PromiseInterface
+ {
+ return $this->queryBySQL(
+ $queryBuilder->getSQL(),
+ $queryBuilder->getParameters()
+ );
+ }
+
+ /**
+ * Query by sql and parameters.
+ *
+ * @param string $sql
+ * @param array $parameters
+ *
+ * @return PromiseInterface<Result>
+ */
+ public function queryBySQL(string $sql, array $parameters = []): PromiseInterface
+ {
+ return $this
+ ->driver
+ ->query($sql, $parameters);
+ }
+
+ /**
+ * Execute, sequentially, an array of sqls.
+ *
+ * @param string[] $sqls
+ *
+ * @return PromiseInterface<Connection>
+ */
+ public function executeSQLs(array $sqls): PromiseInterface
+ {
+ return
+ map($sqls, function (string $sql) {
+ return $this->queryBySQL($sql);
+ })
+ ->then(function () {
+ return $this;
+ });
+ }
+
+ /**
+ * Execute an schema.
+ *
+ * @param Schema $schema
+ *
+ * @return PromiseInterface<Connection>
+ */
+ public function executeSchema(Schema $schema): PromiseInterface
+ {
+ return $this
+ ->executeSQLs($schema->toSql($this->platform))
+ ->then(function () {
+ return $this;
+ });
+ }
+
+ /**
+ * Shortcuts.
+ */
+
+ /**
+ * Find one by.
+ *
+ * connection->findOneById('table', ['id' => 1]);
+ *
+ * @param string $table
+ * @param array $where
+ *
+ * @return PromiseInterface<array|null>
+ */
+ public function findOneBy(
+ string $table,
+ array $where
+ ): PromiseInterface {
+ return $this
+ ->getResultByWhereClause($table, $where)
+ ->then(function (Result $result) {
+ return $result->fetchFirstRow();
+ });
+ }
+
+ /**
+ * Find by.
+ *
+ * connection->findBy('table', ['id' => 1]);
+ *
+ * @param string $table
+ * @param array $where
+ *
+ * @return PromiseInterface<array>
+ */
+ public function findBy(
+ string $table,
+ array $where = []
+ ): PromiseInterface {
+ return $this
+ ->getResultByWhereClause($table, $where)
+ ->then(function (Result $result) {
+ return $result->fetchAllRows();
+ });
+ }
+
+ /**
+ * @param string $table
+ * @param array $values
+ *
+ * @return PromiseInterface
+ */
+ public function insert(
+ string $table,
+ array $values
+ ): PromiseInterface {
+ $queryBuilder = $this->createQueryBuilder();
+
+ return $this->driver->insert($queryBuilder, $table, $values);
+ }
+
+ /**
+ * @param string $table
+ * @param array $values
+ *
+ * @return PromiseInterface
+ *
+ * @throws InvalidArgumentException
+ */
+ public function delete(
+ string $table,
+ array $values
+ ): PromiseInterface {
+ if (empty($values)) {
+ throw InvalidArgumentException::fromEmptyCriteria();
+ }
+
+ $queryBuilder = $this
+ ->createQueryBuilder()
+ ->delete($table);
+
+ $this->applyWhereClausesFromArray($queryBuilder, $values);
+
+ return $this->query($queryBuilder);
+ }
+
+ /**
+ * @param string $table
+ * @param array $id
+ * @param array $values
+ *
+ * @return PromiseInterface
+ *
+ * @throws InvalidArgumentException
+ */
+ public function update(
+ string $table,
+ array $id,
+ array $values
+ ): PromiseInterface {
+ if (empty($id)) {
+ throw InvalidArgumentException::fromEmptyCriteria();
+ }
+
+ $queryBuilder = $this
+ ->createQueryBuilder()
+ ->update($table);
+
+ $parameters = $queryBuilder->getParameters();
+ foreach ($values as $field => $value) {
+ $queryBuilder->set($field, '?');
+ $parameters[] = $value;
+ }
+ $queryBuilder->setParameters($parameters);
+ $this->applyWhereClausesFromArray($queryBuilder, $id);
+
+ return $this->query($queryBuilder);
+ }
+
+ /**
+ * @param string $table
+ * @param array $id
+ * @param array $values
+ *
+ * @return PromiseInterface
+ *
+ * @throws InvalidArgumentException
+ */
+ public function upsert(
+ string $table,
+ array $id,
+ array $values
+ ): PromiseInterface {
+ return $this
+ ->findOneBy($table, $id)
+ ->then(function (?array $result) use ($table, $id, $values) {
+ return is_null($result)
+ ? $this->insert($table, array_merge($id, $values))
+ : $this->update($table, $id, $values);
+ });
+ }
+
+ /**
+ * Table related shortcuts.
+ */
+
+ /**
+ * Easy shortcut for creating tables. Fields is just a simple key value,
+ * being the key the name of the field, and the value the type. By default,
+ * Varchar types have length 255.
+ *
+ * First field is considered as primary key.
+ *
+ * @param string $name
+ * @param array $fields
+ * @param array $extra
+ * @param bool $autoincrementId
+ *
+ * @return PromiseInterface<Connection>
+ *
+ * @throws InvalidArgumentException
+ * @throws TableExistsException
+ */
+ public function createTable(
+ string $name,
+ array $fields,
+ array $extra = [],
+ bool $autoincrementId = false
+ ): PromiseInterface {
+ if (empty($fields)) {
+ throw InvalidArgumentException::fromEmptyCriteria();
+ }
+
+ $schema = new Schema();
+ $table = $schema->createTable($name);
+ foreach ($fields as $field => $type) {
+ $extraField = (
+ array_key_exists($field, $extra) &&
+ is_array($extra[$field])
+ ) ? $extra[$field] : [];
+
+ if (
+ 'string' == $type &&
+ !array_key_exists('length', $extraField)
+ ) {
+ $extraField = array_merge(
+ $extraField,
+ ['length' => 255]
+ );
+ }
+
+ $table->addColumn($field, $type, $extraField);
+ }
+
+ $id = array_key_first($fields);
+ $table->setPrimaryKey([$id]);
+ $table->getColumn($id)->setAutoincrement($autoincrementId);
+
+ return $this->executeSchema($schema);
+ }
+
+ /**
+ * @param string $name
+ *
+ * @return PromiseInterface<Connection>
+ *
+ * @throws TableNotFoundException
+ */
+ public function dropTable(string $name): PromiseInterface
+ {
+ return $this
+ ->queryBySQL("DROP TABLE $name")
+ ->then(function () {
+ return $this;
+ });
+ }
+
+ /**
+ * @param string $name
+ *
+ * @return PromiseInterface<Connection>
+ *
+ * @throws TableNotFoundException
+ */
+ public function truncateTable(string $name): PromiseInterface
+ {
+ $truncateTableQuery = $this
+ ->platform
+ ->getTruncateTableSQL($name);
+
+ return $this
+ ->queryBySQL($truncateTableQuery)
+ ->then(function () {
+ return $this;
+ });
+ }
+
+ /**
+ * Get result by where clause.
+ *
+ * @param string $table
+ * @param array $where
+ *
+ * @return PromiseInterface<Result>
+ */
+ private function getResultByWhereClause(
+ string $table,
+ array $where
+ ): PromiseInterface {
+ $queryBuilder = $this
+ ->createQueryBuilder()
+ ->select('t.*')
+ ->from($table, 't');
+
+ $this->applyWhereClausesFromArray($queryBuilder, $where);
+
+ return $this->query($queryBuilder);
+ }
+
+ /**
+ * Apply where clauses.
+ *
+ * [
+ * "id" => 1,
+ * "name" => "Marc"
+ * ]
+ *
+ * to
+ *
+ * [
+ * [ "id = ?", "name = ?"],
+ * [1, "Marc"]
+ * ]
+ *
+ * @param QueryBuilder $queryBuilder
+ * @param array $array
+ */
+ private function applyWhereClausesFromArray(
+ QueryBuilder $queryBuilder,
+ array $array
+ ) {
+ $params = $queryBuilder->getParameters();
+ foreach ($array as $field => $value) {
+ if (\is_null($value)) {
+ $queryBuilder->andWhere(
+ $queryBuilder->expr()->isNull($field)
+ );
+ continue;
+ }
+
+ $queryBuilder->andWhere(
+ $queryBuilder->expr()->eq($field, '?')
+ );
+
+ $params[] = $value;
+ }
+
+ $queryBuilder->setParameters($params);
+ }
+}
diff --git a/tests/ConnectionTest.php b/tests/ConnectionTest.php
index feae045..7c1c82c 100644
--- a/tests/ConnectionTest.php
+++ b/tests/ConnectionTest.php
@@ -559,7 +559,10 @@ public function testGetLastInsertedId()
*/
public function testAffectedRows()
{
- if ($this instanceof PostgreSQLConnectionTest) {
+ if (
+ $this instanceof PostgreSQLConnectionTest ||
+ $this instanceof PostgreSQLConnectionPoolTest
+ ) {
$this->markTestSkipped('This feature is not implemented in the Postgres client');
}
diff --git a/tests/Mysql5ConnectionPoolTest.php b/tests/Mysql5ConnectionPoolTest.php
new file mode 100644
index 0000000..e8816ac
--- /dev/null
+++ b/tests/Mysql5ConnectionPoolTest.php
@@ -0,0 +1,48 @@
+<?php
+
+/*
+ * This file is part of the DriftPHP Project
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ *
+ * Feel free to edit as you please, and have fun.
+ *
+ * @author Marc Morera <yuhu@mmoreram.com>
+ */
+
+declare(strict_types=1);
+
+namespace Drift\DBAL\Tests;
+
+use Doctrine\DBAL\Platforms\MySQLPlatform;
+use Drift\DBAL\Connection;
+use Drift\DBAL\ConnectionPool;
+use Drift\DBAL\Credentials;
+use Drift\DBAL\Driver\Mysql\MysqlDriver;
+use React\EventLoop\LoopInterface;
+
+/**
+ * Class Mysql5ConnectionPoolTest.
+ */
+class Mysql5ConnectionPoolTest extends ConnectionTest
+{
+ /**
+ * {@inheritdoc}
+ */
+ protected function getConnection(LoopInterface $loop): Connection
+ {
+ $mysqlPlatform = new MySQLPlatform();
+
+ return ConnectionPool::createConnected(new MysqlDriver(
+ $loop
+ ), new Credentials(
+ '127.0.0.1',
+ '3306',
+ 'root',
+ 'root',
+ 'test',
+ [], 10
+ ), $mysqlPlatform);
+ }
+}
diff --git a/tests/Mysql5ConnectionTest.php b/tests/Mysql5ConnectionTest.php
index 5258bc4..7eeee7c 100644
--- a/tests/Mysql5ConnectionTest.php
+++ b/tests/Mysql5ConnectionTest.php
@@ -19,6 +19,7 @@
use Drift\DBAL\Connection;
use Drift\DBAL\Credentials;
use Drift\DBAL\Driver\Mysql\MysqlDriver;
+use Drift\DBAL\SingleConnection;
use React\EventLoop\LoopInterface;
/**
@@ -33,7 +34,7 @@ protected function getConnection(LoopInterface $loop): Connection
{
$mysqlPlatform = new MySQLPlatform();
- return Connection::createConnected(new MysqlDriver(
+ return SingleConnection::createConnected(new MysqlDriver(
$loop
), new Credentials(
'127.0.0.1',
diff --git a/tests/PostgreSQLConnectionPoolTest.php b/tests/PostgreSQLConnectionPoolTest.php
new file mode 100644
index 0000000..a2472db
--- /dev/null
+++ b/tests/PostgreSQLConnectionPoolTest.php
@@ -0,0 +1,46 @@
+<?php
+
+/*
+ * This file is part of the DriftPHP Project
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ *
+ * Feel free to edit as you please, and have fun.
+ *
+ * @author Marc Morera <yuhu@mmoreram.com>
+ */
+
+declare(strict_types=1);
+
+namespace Drift\DBAL\Tests;
+
+use Doctrine\DBAL\Platforms\PostgreSQLPlatform;
+use Drift\DBAL\Connection;
+use Drift\DBAL\ConnectionPool;
+use Drift\DBAL\Credentials;
+use Drift\DBAL\Driver\PostgreSQL\PostgreSQLDriver;
+use React\EventLoop\LoopInterface;
+
+/**
+ * Class PostgreSQLConnectionPoolTest.
+ */
+class PostgreSQLConnectionPoolTest extends ConnectionTest
+{
+ /**
+ * {@inheritdoc}
+ */
+ protected function getConnection(LoopInterface $loop): Connection
+ {
+ $postgreSQLPlatform = new PostgreSQLPlatform();
+
+ return ConnectionPool::createConnected(new PostgreSQLDriver($loop), new Credentials(
+ '127.0.0.1',
+ '5432',
+ 'root',
+ 'root',
+ 'test',
+ [], 10
+ ), $postgreSQLPlatform);
+ }
+}
diff --git a/tests/PostgreSQLConnectionTest.php b/tests/PostgreSQLConnectionTest.php
index a864ada..e5e92ad 100644
--- a/tests/PostgreSQLConnectionTest.php
+++ b/tests/PostgreSQLConnectionTest.php
@@ -15,10 +15,11 @@
namespace Drift\DBAL\Tests;
-use Doctrine\DBAL\Platforms\PostgreSQL94Platform;
+use Doctrine\DBAL\Platforms\PostgreSQLPlatform;
use Drift\DBAL\Connection;
use Drift\DBAL\Credentials;
use Drift\DBAL\Driver\PostgreSQL\PostgreSQLDriver;
+use Drift\DBAL\SingleConnection;
use React\EventLoop\LoopInterface;
/**
@@ -31,9 +32,9 @@ class PostgreSQLConnectionTest extends ConnectionTest
*/
public function getConnection(LoopInterface $loop): Connection
{
- $postgreSQLPlatform = new PostgreSQL94Platform();
+ $postgreSQLPlatform = new PostgreSQLPlatform();
- return Connection::createConnected(new PostgreSQLDriver($loop), new Credentials(
+ return SingleConnection::createConnected(new PostgreSQLDriver($loop), new Credentials(
'127.0.0.1',
'5432',
'root',
diff --git a/tests/SQLiteConnectionTest.php b/tests/SQLiteConnectionTest.php
index 96200f3..c83503a 100644
--- a/tests/SQLiteConnectionTest.php
+++ b/tests/SQLiteConnectionTest.php
@@ -19,6 +19,7 @@
use Drift\DBAL\Connection;
use Drift\DBAL\Credentials;
use Drift\DBAL\Driver\SQLite\SQLiteDriver;
+use Drift\DBAL\SingleConnection;
use React\EventLoop\LoopInterface;
/**
@@ -33,7 +34,7 @@ public function getConnection(LoopInterface $loop): Connection
{
$mysqlPlatform = new SqlitePlatform();
- return Connection::createConnected(new SQLiteDriver(
+ return SingleConnection::createConnected(new SQLiteDriver(
$loop
), new Credentials(
'',