This commit is contained in:
2021-08-09 11:12:01 +08:00
parent 27297e29fd
commit eaccb88e46
+189 -186
View File
@@ -3,6 +3,7 @@ declare(strict_types=1);
namespace Snowflake\Pool; namespace Snowflake\Pool;
use Closure;
use Exception; use Exception;
use HttpServer\Http\Context; use HttpServer\Http\Context;
use PDO; use PDO;
@@ -20,215 +21,217 @@ use Throwable;
class Connection extends Component class Connection extends Component
{ {
use Alias; use Alias;
/** /**
* @param $cds * @param $cds
* @return bool * @return bool
* *
* db is in transaction * db is in transaction
* @throws Exception * @throws Exception
*/ */
public function inTransaction($cds): bool public function inTransaction($cds): bool
{ {
$name = $this->name('Mysql:' . $cds, true); $name = $this->name('Mysql:' . $cds, true);
return Context::getContext('begin_' . $name) == 0; return Context::getContext('begin_' . $name) == 0;
} }
/** /**
* @param $coroutineName * @param $coroutineName
* @throws Exception * @throws Exception
*/ */
public function beginTransaction($coroutineName) public function beginTransaction($coroutineName)
{ {
$coroutineName = $this->name('Mysql:' . $coroutineName, true); $coroutineName = $this->name('Mysql:' . $coroutineName, true);
if (!Context::hasContext('begin_' . $coroutineName)) { if (!Context::hasContext('begin_' . $coroutineName)) {
Context::setContext('begin_' . $coroutineName, 0); Context::setContext('begin_' . $coroutineName, 0);
} }
$connection = Context::getContext($coroutineName); Context::increment('begin_' . $coroutineName, 1);
if ($connection instanceof PDO && !$connection->inTransaction()) {
$connection->beginTransaction();
}
}
/** $connection = Context::getContext($coroutineName);
* @param $coroutineName if ($connection instanceof PDO && !$connection->inTransaction()) {
* @throws Exception $connection->beginTransaction();
*/ }
public function commit($coroutineName) }
{
$coroutineName = $this->name('Mysql:' . $coroutineName, true); /**
if (Context::decrement('begin_' . $coroutineName) != 0) { * @param $coroutineName
return; * @throws Exception
} */
$connection = Context::getContext($coroutineName); public function commit($coroutineName)
if ($connection instanceof PDO && $connection->inTransaction()) { {
$connection->commit(); $coroutineName = $this->name('Mysql:' . $coroutineName, true);
} if (Context::decrement('begin_' . $coroutineName) != 0) {
} return;
}
$connection = Context::getContext($coroutineName);
if ($connection instanceof PDO && $connection->inTransaction()) {
$connection->commit();
}
}
/** /**
* @param $coroutineName * @param $coroutineName
* @throws Exception * @throws Exception
*/ */
public function rollback($coroutineName) public function rollback($coroutineName)
{ {
$coroutineName = $this->name('Mysql:' . $coroutineName, true); $coroutineName = $this->name('Mysql:' . $coroutineName, true);
if (Context::decrement('begin_' . $coroutineName) != 0) { if (Context::decrement('begin_' . $coroutineName) != 0) {
return; return;
} }
if (($connection = Context::getContext($coroutineName)) instanceof PDO) { if (($connection = Context::getContext($coroutineName)) instanceof PDO) {
if ($connection->inTransaction()) { if ($connection->inTransaction()) {
$connection->rollBack(); $connection->rollBack();
} }
} }
} }
/** /**
* @param mixed $config * @param mixed $config
* @param bool $isMaster * @param bool $isMaster
* @return mixed * @return PDO|null
* @throws Exception * @throws Exception
*/ */
public function get(mixed $config, bool $isMaster = false): ?PDO public function get(mixed $config, bool $isMaster = false): ?PDO
{ {
$coroutineName = $this->name('Mysql:' . $config['cds'], $isMaster); $coroutineName = $this->name('Mysql:' . $config['cds'], $isMaster);
if (($pdo = Context::getContext($coroutineName)) instanceof PDO) { if (($pdo = Context::getContext($coroutineName)) instanceof PDO) {
return $pdo; return $pdo;
} }
/** @var PDO $connections */ /** @var PDO $connections */
$connections = $this->getPool()->get($coroutineName, $this->create($coroutineName, $config)); $connections = $this->getPool()->get($coroutineName, $this->create($coroutineName, $config));
if (Context::hasContext('begin_' . $coroutineName)) { if (Context::hasContext('begin_' . $coroutineName)) {
$connections->beginTransaction(); $connections->beginTransaction();
} }
return Context::setContext($coroutineName, $connections); return Context::setContext($coroutineName, $connections);
} }
/** /**
* @param $coroutineName * @param $coroutineName
* @param $config * @param $config
* @return \Closure * @return Closure
*/ */
public function create($coroutineName, $config) public function create($coroutineName, $config): Closure
{ {
return static function () use ($coroutineName, $config) { return static function () use ($coroutineName, $config) {
if (Coroutine::getCid() === -1) { if (Coroutine::getCid() === -1) {
Runtime::enableCoroutine(false); Runtime::enableCoroutine(false);
} }
$cds = 'mysql:dbname=' . $config['database'] . ';host=' . $config['cds']; $cds = 'mysql:dbname=' . $config['database'] . ';host=' . $config['cds'];
$link = new PDO($cds, $config['username'], $config['password'], [ $link = new PDO($cds, $config['username'], $config['password'], [
PDO::ATTR_EMULATE_PREPARES => false, PDO::ATTR_EMULATE_PREPARES => false,
PDO::ATTR_CASE => PDO::CASE_NATURAL, PDO::ATTR_CASE => PDO::CASE_NATURAL,
PDO::ATTR_TIMEOUT => 60, PDO::ATTR_TIMEOUT => 60,
PDO::MYSQL_ATTR_USE_BUFFERED_QUERY => true, PDO::MYSQL_ATTR_USE_BUFFERED_QUERY => true,
PDO::MYSQL_ATTR_INIT_COMMAND => 'SET NAMES ' . ($config['charset'] ?? 'utf8mb4') PDO::MYSQL_ATTR_INIT_COMMAND => 'SET NAMES ' . ($config['charset'] ?? 'utf8mb4')
]); ]);
$link->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION); $link->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
$link->setAttribute(PDO::ATTR_STRINGIFY_FETCHES, false); $link->setAttribute(PDO::ATTR_STRINGIFY_FETCHES, false);
$link->setAttribute(PDO::ATTR_ORACLE_NULLS, PDO::NULL_EMPTY_STRING); $link->setAttribute(PDO::ATTR_ORACLE_NULLS, PDO::NULL_EMPTY_STRING);
return $link; return $link;
}; };
} }
/** /**
* @param $name * @param $name
* @param $isMaster * @param $isMaster
* @param $max * @param $max
* @throws Exception * @throws Exception
*/ */
public function initConnections($name, $isMaster, $max) public function initConnections($name, $isMaster, $max)
{ {
$this->getPool()->initConnections($name, $isMaster, $max); $this->getPool()->initConnections($name, $isMaster, $max);
} }
/** /**
* @param $coroutineName * @param $coroutineName
* @param $isMaster * @param $isMaster
* @throws Exception * @throws Exception
*/ */
public function release($coroutineName, $isMaster) public function release($coroutineName, $isMaster)
{ {
$coroutineName = $this->name('Mysql:' . $coroutineName, $isMaster); $coroutineName = $this->name('Mysql:' . $coroutineName, $isMaster);
/** @var PDO $client */ /** @var PDO $client */
if (!($client = Context::getContext($coroutineName)) instanceof PDO) { if (!($client = Context::getContext($coroutineName)) instanceof PDO) {
return; return;
} }
if ($client->inTransaction()) { if ($client->inTransaction()) {
return; return;
} }
$this->getPool()->push($coroutineName, $client); $this->getPool()->push($coroutineName, $client);
Context::remove($coroutineName); Context::remove($coroutineName);
} }
/** /**
* @param $coroutineName * @param $coroutineName
* @return bool * @return bool
*/ */
private function hasClient($coroutineName): bool private function hasClient($coroutineName): bool
{ {
return Context::hasContext($coroutineName); return Context::hasContext($coroutineName);
} }
/** /**
* batch release * batch release
* @throws Exception * @throws Exception
*/ */
public function connection_clear($name, $isMaster) public function connection_clear($name, $isMaster)
{ {
$this->getPool()->clean($this->name($name, $isMaster)); $this->getPool()->clean($this->name($name, $isMaster));
} }
/** /**
* @param string $name * @param string $name
* @param mixed $client * @param mixed $client
* @return bool * @return bool
* @throws Exception * @throws Exception
*/ */
public function checkCanUse(string $name, mixed $client): bool public function checkCanUse(string $name, mixed $client): bool
{ {
try { try {
if (empty($client) || !($client instanceof PDO)) { if (empty($client) || !($client instanceof PDO)) {
$result = false; $result = false;
} else { } else {
$result = true; $result = true;
} }
} catch (Error | Throwable $exception) { } catch (Error | Throwable $exception) {
$result = $this->addError($exception, 'mysql'); $result = $this->addError($exception, 'mysql');
} finally { } finally {
return $result; return $result;
} }
} }
/** /**
* @param $coroutineName * @param $coroutineName
* @param bool $isMaster * @param bool $isMaster
* @throws Exception * @throws Exception
*/ */
public function disconnect($coroutineName, bool $isMaster = false) public function disconnect($coroutineName, bool $isMaster = false)
{ {
Context::remove($coroutineName); Context::remove($coroutineName);
$coroutineName = $this->name('Mysql:' . $coroutineName, $isMaster); $coroutineName = $this->name('Mysql:' . $coroutineName, $isMaster);
$this->getPool()->clean($coroutineName); $this->getPool()->clean($coroutineName);
} }
/** /**
* @return Pool * @return Pool
* @throws Exception * @throws Exception
*/ */
public function getPool(): Pool public function getPool(): Pool
{ {
return Snowflake::getDi()->get(Pool::class); return Snowflake::getDi()->get(Pool::class);
} }
} }