Merge branch 'master' of github.com:as2252258/kiri-databases

DatabasesProviders.php

@@ -47,8 +47,6 @@ class DatabasesProviders extends Providers
 		if (empty($databases)) {
 			return;
 		}
-		$this->provider->on(OnWorkerStart::class, [$this, 'check']);
-		$this->provider->on(OnTaskerStart::class, [$this, 'check']);
 		$this->provider->on(OnWorkerExit::class, [$this, 'exit'], 9999);
 		foreach ($databases as $key => $database) {
 			$application->set($key, $this->_settings($database));
@@ -85,61 +83,6 @@ class DatabasesProviders extends Providers
 	}
 
 
-	/**
-	 * @param OnTaskerStart|OnWorkerStart $start
-	 * @return void
-	 */
-	public function check(OnTaskerStart|OnWorkerStart $start): void
-	{
-		Timer::tick(10000, function ($timerId) {
-			$valid = $count = 0;
-			$logger = Kiri::getDi()->get(LoggerInterface::class);
-			$databases = Config::get('databases.connections', []);
-			if (!empty($databases)) {
-				[$valid, $count] = $this->each($databases, $logger);
-			}
-			$const = 'Worker %d db client has %d, valid %d';
-			$logger->alert(sprintf($const, env('environmental_workerId'), $count, $valid));
-			if ($this->container->get(WorkerStatus::class)->is(StatusEnum::EXIT)) {
-				Timer::clear($timerId);
-			}
-		});
-	}
-
-
-	/**
-	 * @param $databases
-	 * @param LoggerInterface $logger
-	 * @return array
-	 */
-	public function each($databases, LoggerInterface $logger): array
-	{
-		$connection = Kiri::getDi()->get(PoolConnection::class);
-		$valid = $count = 0;
-		foreach ($databases as $database) {
-			try {
-				[$total, $success] = $connection->check($database['cds']);
-
-				$count += $total;
-				$valid += $success;
-
-				if (isset($database['slaveConfig']) && isset($database['slaveConfig']['cds'])) {
-					if ($database['slaveConfig']['cds'] != $database['cds']) {
-						[$total, $success] = $connection->check($database['slaveConfig']['cds']);
-
-						$count += $total;
-						$valid += $success;
-					}
-				}
-			} catch (\Throwable $throwable) {
-				$logger->error($throwable->getMessage());
-			}
-		}
-
-		return [$valid, $count];
-	}
-
-
 	/**
 	 * @param $database
 	 * @return array

Mysql/PDO.php

@@ -8,6 +8,9 @@ use Kiri\Events\EventProvider;
 use Kiri\Pool\StopHeartbeatCheck;
 use Kiri\Server\Events\OnWorkerExit;
 use PDOStatement;
+use Kiri\Server\WorkerStatus;
+use Kiri\Server\Abstracts\StatusEnum;
+use Swoole\Timer;
 
 /**
  *
@@ -33,6 +36,8 @@ class PDO implements StopHeartbeatCheck
     public int $connect_timeout;
     public int $read_timeout;
 
+    private int $_timerId = -1;
+
 
     public array $attributes = [];
 
@@ -61,6 +66,7 @@ class PDO implements StopHeartbeatCheck
     {
         $eventProvider = Kiri::getDi()->get(EventProvider::class);
         $eventProvider->on(OnWorkerExit::class, [$this, 'onWorkerExit']);
+        $this->_timerId = Timer::tick(60000, [$this, 'check']);
     }
 
 
@@ -89,6 +95,10 @@ class PDO implements StopHeartbeatCheck
     public function stopHeartbeatCheck(): void
     {
         $this->pdo = null;
+        if ($this->_timerId > -1) {
+            Timer::clear($this->_timerId);
+            $this->_timerId = -1;
+        }
     }
 
 
@@ -200,8 +210,11 @@ class PDO implements StopHeartbeatCheck
      */
     private function queryPrev(string $sql, array $params = []): PDOStatement
     {
-		$this->_last = time();
         try {
+            if ($this->_timerId === -1) {
+                $this->_timerId = Timer::tick(6000, [$this, 'check']);
+            }
+            $this->_last = time();
             if (($statement = $this->_pdo()->query($sql)) === false) {
                 throw new Exception($this->_pdo()->errorInfo()[1]);
             }
@@ -225,9 +238,8 @@ class PDO implements StopHeartbeatCheck
         try {
             if ($this->_last == 0) $this->_last = time();
             if (time() - $this->_last >= 600) {
-				throw new Exception('Idle dis.');
+                return $result = false;
-			}
+            } else if (!($this->pdo instanceof \PDO)) {
-			if (!($this->pdo instanceof \PDO)) {
                 return $result = false;
             }
             $this->pdo->getAttribute(\PDO::ATTR_SERVER_INFO);
@@ -236,14 +248,29 @@ class PDO implements StopHeartbeatCheck
             if (!str_contains($throwable->getMessage(), 'Idle dis')) {
                 Kiri::getLogger()->error($throwable->getMessage());
             }
-			$this->pdo = null;
             $result = false;
         } finally {
-			return $result;
+            return $this->afterCheck($result);
         }
     }
 
 
+    /**
+     * @param bool $result
+     * @return bool
+     */
+    private function afterCheck(bool $result): bool
+    {
+        $container = Kiri::getDi()->get(WorkerStatus::class);
+        if (!$result || $container->is(StatusEnum::EXIT)) {
+            $this->pdo = null;
+            $result = Timer::clear($this->_timerId);
+            $this->_timerId = -1;
+        }
+        return $result;
+    }
+
+
     /**
      * @param PDOStatement $statement
      * @param array $params
@@ -267,20 +294,22 @@ class PDO implements StopHeartbeatCheck
      */
     public function execute(string $sql, array $params = []): int
     {
-		$this->_last = time();
         $pdo = $this->_pdo();
+        if ($this->_timerId === -1) {
+            $this->_timerId = Timer::tick(6000, [$this, 'check']);
+        }
+        $this->_last = time();
         if (!(($prepare = $pdo->prepare($sql)) instanceof PDOStatement)) {
             throw new Exception($prepare->errorInfo()[2] ?? static::DB_ERROR_MESSAGE);
         }
         if ($prepare->execute($params) === false) {
             throw new Exception($prepare->errorInfo()[2] ?? static::DB_ERROR_MESSAGE);
         }
+
         $result = (int)$pdo->lastInsertId();
         $prepare->closeCursor();
-		if ($result == 0) {
+
-			return true;
+        return $result == 0 ? true : $result;
-		}
-		return $result;
     }