WIP: added pdo proxy

src/ConfigProvider.php

@@ -19,6 +19,7 @@
  */
 namespace Hyperf\Seata;
 
+use Hyperf\Database\Connectors\Connector;
 use Hyperf\Seata\Annotation\GlobalTransactionScanner;
 use Hyperf\Seata\Core\Model\ResourceManagerInterface;
 use Hyperf\Seata\Listener\InitListener;
@@ -40,7 +41,7 @@ public function __invoke(): array
                 ResourceManagerInterface::class => DefaultResourceManager::class,
                 //                GlobalTransactionScanner::class => GlobalTransactionScannerFactory::class,
                 LoggerInterface::class => StdoutLogger::class,
-                //                DataSourceProxy::class => DataSourceProxyFactory::class,
+//                DataSourceProxy::class => DataSourceProxyFactory::class,
             ],
         ];
     }

src/Core/Context/GlobalLockConfigHolder.php

@@ -20,7 +20,7 @@
 namespace Hyperf\Seata\Core\Context;
 
 use Hyperf\Seata\Core\Model\GlobalLockConfig;
-use Hyperf\Utils\Context;
+use Hyperf\Context\Context;
 
 class GlobalLockConfigHolder
 {

src/Core/Context/RootContext.php

@@ -24,7 +24,7 @@
 use Hyperf\Seata\Logger\LoggerFactory;
 use Hyperf\Seata\Logger\LoggerInterface;
 use Hyperf\Utils\ApplicationContext;
-use Hyperf\Utils\Context;
+use Hyperf\Context\Context;
 use RuntimeException;
 
 class RootContext extends Context

src/Core/Rpc/Runtime/Swow/SocketChannel.php

@@ -27,6 +27,8 @@
 use Hyperf\Seata\Core\Rpc\Runtime\SocketChannelInterface;
 use Hyperf\Seata\Core\Rpc\Runtime\V1\ProtocolV1Decoder;
 use Hyperf\Seata\Core\Rpc\Runtime\V1\ProtocolV1Encoder;
+use Hyperf\Seata\Logger\LoggerFactory;
+use Hyperf\Seata\Logger\LoggerInterface;
 use Hyperf\Seata\Utils\Buffer\ByteBuffer;
 use Hyperf\Seata\Utils\Protocol\RpcMessageUtils;
 use Hyperf\Utils\ApplicationContext;
@@ -49,13 +51,16 @@ class SocketChannel implements SocketChannelInterface
 
     protected Channel $sendChannel;
 
+    private LoggerInterface $logger;
+
     public function __construct(Socket $socket, Address $address)
     {
         $this->socket = $socket;
         $this->address = $address;
         $container = ApplicationContext::getContainer();
         $this->protocolEncoder = $container->get(ProtocolV1Encoder::class);
         $this->protocolDecoder = $container->get(ProtocolV1Decoder::class);
+        $this->logger = $container->get(LoggerFactory::class)->create(static::class);
         $this->sendChannel = new Channel();
         $this->createRecvLoop();
         // $this->createSendLoop();
@@ -64,7 +69,7 @@ public function __construct(Socket $socket, Address $address)
     public function sendSyncWithResponse(RpcMessage $rpcMessage, int $timeoutMillis)
     {
         $channel = new Channel();
-        echo 'Ready to send the rpc message #' . RpcMessageUtils::toLogString($rpcMessage) . PHP_EOL;
+        $this->logger->debug('Ready to send the rpc message #' . RpcMessageUtils::toLogString($rpcMessage));
         $this->responses[$rpcMessage->getId()] = $channel;
         $this->sendSyncWithoutResponse($rpcMessage, $timeoutMillis);
         return $channel->pop();
@@ -96,7 +101,7 @@ protected function createRecvLoop()
                     $rpcMessage = $this->protocolDecoder->decode($byteBuffer);
                     $processorManger->dispatch($this, $rpcMessage);
 
-                    echo 'Recieved a rpc message #' . RpcMessageUtils::toLogString($rpcMessage) . PHP_EOL;
+                    $this->logger->debug('Recieved a rpc message #' . RpcMessageUtils::toLogString($rpcMessage));
                     if (isset($this->responses[$rpcMessage->getId()])) {
                         $responseChannel = $this->responses[$rpcMessage->getId()];
                         $responseChannel->push($rpcMessage);
@@ -107,10 +112,10 @@ protected function createRecvLoop()
                     // //                        var_dump('heartbeat', $rpcMessage);
 //                    }
                 } catch (\InvalidArgumentException $exception) {
-                    echo 'Recieved a rpc message fail error:' . $exception->getMessage() . PHP_EOL;
+                    $this->logger->debug('Recieved a rpc message fail error:' . $exception->getMessage());
                     break;
                 } catch (\Throwable $exception) {
-                    echo 'Recieved a rpc message fail error:' . $exception->getMessage() . PHP_EOL;
+                    $this->logger->debug( 'Recieved a rpc message fail error:' . $exception->getMessage());
                     break;
                 } finally {
                     isset($rpcMessage) && $this->responses[$rpcMessage->getId()]->close();

src/Exception/ExecutionException.php

@@ -65,18 +65,10 @@ public function setTransaction(GlobalTransaction $transaction): void
         $this->transaction = $transaction;
     }
 
-    /**
-     * @return int
-     */
-    public function getCode()
-    {
-        return $this->code;
-    }
-
     /**
      * @param int $code
      */
-    public function setCode($code): void
+    public function setCode(int $code): void
     {
         $this->code = $code;
     }

src/Rm/DataSource/ClassMap/Connector.php

@@ -0,0 +1,148 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * Copyright 2019-2022 Seata.io Group.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+namespace Hyperf\Database\Connectors;
+
+use Exception;
+use Hyperf\Database\DetectsLostConnections;
+use Hyperf\Seata\Rm\PDOProxy;
+use PDO;
+use Throwable;
+
+class Connector
+{
+    use DetectsLostConnections;
+
+    /**
+     * The default PDO connection options.
+     *
+     * @var array
+     */
+    protected $options = [
+        PDO::ATTR_CASE => PDO::CASE_NATURAL,
+        PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION,
+        PDO::ATTR_ORACLE_NULLS => PDO::NULL_NATURAL,
+        PDO::ATTR_STRINGIFY_FETCHES => false,
+        PDO::ATTR_EMULATE_PREPARES => false,
+    ];
+
+    /**
+     * Create a new PDO connection.
+     *
+     * @param string $dsn
+     * @throws \Exception
+     * @return \PDO
+     */
+    public function createConnection($dsn, array $config, array $options)
+    {
+        [$username, $password] = [
+                $config['username'] ?? null, $config['password'] ?? null,
+        ];
+
+        try {
+            return $this->createPdoConnection(
+                $dsn,
+                $username,
+                $password,
+                $options
+            );
+        } catch (Exception $e) {
+            return $this->tryAgainIfCausedByLostConnection(
+                $e,
+                $dsn,
+                $username,
+                $password,
+                $options
+            );
+        }
+    }
+
+    /**
+     * Get the PDO options based on the configuration.
+     *
+     * @return array
+     */
+    public function getOptions(array $config)
+    {
+        return array_replace($this->options, $config['options'] ?? []);
+    }
+
+    /**
+     * Get the default PDO connection options.
+     *
+     * @return array
+     */
+    public function getDefaultOptions()
+    {
+        return $this->options;
+    }
+
+    /**
+     * Set the default PDO connection options.
+     */
+    public function setDefaultOptions(array $options)
+    {
+        $this->options = $options;
+    }
+
+    /**
+     * Create a new PDO connection instance.
+     *
+     * @param string $dsn
+     * @param string $username
+     * @param string $password
+     * @param array $options
+     * @return \PDO
+     */
+    protected function createPdoConnection($dsn, $username, $password, $options)
+    {
+        return new PDOProxy($dsn, $username, $password, $options);
+    }
+
+    /**
+     * Determine if the connection is persistent.
+     *
+     * @param array $options
+     * @return bool
+     */
+    protected function isPersistentConnection($options)
+    {
+        return isset($options[PDO::ATTR_PERSISTENT])
+            && $options[PDO::ATTR_PERSISTENT];
+    }
+
+    /**
+     * Handle an exception that occurred during connect execution.
+     *
+     * @param string $dsn
+     * @param string $username
+     * @param string $password
+     * @param array $options
+     * @throws \Exception
+     * @return \PDO
+     */
+    protected function tryAgainIfCausedByLostConnection(Throwable $e, $dsn, $username, $password, $options)
+    {
+        if ($this->causedByLostConnection($e)) {
+            return $this->createPdoConnection($dsn, $username, $password, $options);
+        }
+
+        throw $e;
+    }
+}

src/Rm/DataSource/ConnectionContext.php

@@ -21,7 +21,7 @@
 
 use Hyperf\Seata\Exception\ShouldNeverHappenException;
 use Hyperf\Seata\Rm\DataSource\Undo\SQLUndoLog;
-use Hyperf\Utils\Context;
+use Hyperf\Context\Context;
 
 class ConnectionContext
 {

src/Rm/PDOStatementProxy.php

@@ -50,6 +50,18 @@ public function __call(string $name, array $arguments)
         return $this->__object->{$name}(...$arguments);
     }
 
+    public function fetchAll($mode = PDO::FETCH_BOTH, $fetch_argument = null, ...$args)
+    {
+        $args = func_get_args();
+        return $this->__object->fetchAll(...$args);
+    }
+
+    public function setFetchMode($mode, $className = null, ...$params)
+    {
+        $args = func_get_args();
+        return $this->__object->setFetchMode(...$args);
+    }
+
     public function bindParam(int|string $param, mixed &$var, int $type = PDO::PARAM_INT, int $maxLength = null, mixed $driverOptions = null)
     {
         $this->bindParamContext[$param] = [$var, $type, $maxLength, $driverOptions];
@@ -70,6 +82,7 @@ public function bindValue(int|string $param, mixed $value, int $type = PDO::PARA
 
     public function execute(?array $params = null)
     {
+
         if ($this->sqlParser->isDelete()) {
             $deleteExecutor = new DeleteExecutor($this->sqlParser, $this->PDOProxy, $this->bindParamContext, $this->bindColumnContext, $this->bindValueContext);
             $deleteExecutor->execute($params);
@@ -80,6 +93,6 @@ public function execute(?array $params = null)
 
         if ($this->sqlParser->isInsert()) {
         }
-        return parent::execute($params);
+        return $this->__object->execute($params);
     }
 }

src/Tm/Api/Transaction/TransactionHookManager.php

@@ -19,7 +19,7 @@
  */
 namespace Hyperf\Seata\Tm\Api\Transaction;
 
-use Hyperf\Utils\Context;
+use Hyperf\Context\Context;
 
 class TransactionHookManager
 {

src/Tm/Api/Transaction/TransactionInfo.php

@@ -23,35 +23,21 @@
 
 class TransactionInfo
 {
-    /**
-     * @var int
-     */
-    private $timeOut;
+    private int $timeOut;
 
-    /**
-     * @var string
-     */
-    private $name;
+    private string $name;
 
     /**
      * @var RollbackRule[]
      */
-    private $rollbackRules;
+    private array $rollbackRules = [];
 
-    /**
-     * @var Propagation
-     */
-    private $propagation;
+    private ?Propagation $propagation = null;
 
-    /**
-     * @var int
-     */
-    private $lockRetryInterval;
+    private int $lockRetryInterval;
+
+    private int $lockRetryTimes;
 
-    /**
-     * @var int
-     */
-    private $lockRetryTimes;
 
     public function getTimeOut(): int
     {

src/Tm/Api/TransactionalTemplate.php

@@ -27,19 +27,21 @@
 use Hyperf\Seata\Exception\ExecutionException;
 use Hyperf\Seata\Exception\ShouldNeverHappenException;
 use Hyperf\Seata\Exception\TransactionException;
+use Hyperf\Seata\Logger\LoggerFactory;
 use Hyperf\Seata\Tm\Api\Transaction\Propagation;
 use Hyperf\Seata\Tm\Api\Transaction\TransactionHook;
 use Hyperf\Seata\Tm\Api\Transaction\TransactionHookManager;
 use Hyperf\Seata\Tm\Api\Transaction\TransactionInfo;
+use Psr\Log\LoggerInterface;
 use Throwable;
 
 class TransactionalTemplate
 {
-    private StdoutLoggerInterface $logger;
+    private LoggerInterface $logger;
 
-    public function __construct(StdoutLoggerInterface $logger)
+    public function __construct(LoggerFactory $loggerFactory)
     {
-        $this->logger = $logger;
+        $this->logger = $loggerFactory->create(static::class);
     }
 
     public function execute(TransactionalExecutor $business)
@@ -113,7 +115,6 @@ public function execute(TransactionalExecutor $business)
                 // 2. If the tx role is 'GlobalTransactionRole.Launcher', send the request of beginTransaction to TC,
                 //    else do nothing. Of course, the hooks will still be triggered.
                 $this->beginTransaction($txInfo, $tx);
-
                 $rs = null;
                 try {
                     // Do Your Business