[Feature]retry connection.

.gitattributes

@@ -0,0 +1,8 @@
+/.coveralls.yml export-ignore
+/.gitattributes export-ignore
+/.github/ export-ignore
+/.gitignore export-ignore
+/.travis.yml export-ignore
+/.travis/ export-ignore
+/phpunit.xml export-ignore
+/tests/ export-ignore

composer.json

@@ -18,10 +18,12 @@
     "ext-sockets": "*",
     "doctrine/collections": "^1.3",
     "doctrine/dbal": "^3.0",
+    "psr/log": "^1.0 | ^2.0 | ^3.0",
     "psr/simple-cache": "^1.0 | ^3.0",
     "symfony/event-dispatcher": "^3.1|^4.0|^5.0|^6.0"
   },
   "require-dev": {
+    "monolog/monolog": "^2.8",
     "phpunit/phpunit": "^9.0"
   },
   "license": "MIT",
@@ -44,5 +46,8 @@
   "minimum-stability": "stable",
   "config": {
     "sort-packages": true
+  },
+  "conflict": {
+    "krowinski/php-mysql-replication": "*"
   }
 }

example/resuming.php

@@ -7,12 +7,16 @@
 date_default_timezone_set('UTC');
 include __DIR__ . '/../vendor/autoload.php';
 
+use Monolog\Handler\StreamHandler;
+use Monolog\Logger;
 use MySQLReplication\BinLog\BinLogCurrent;
 use MySQLReplication\Config\ConfigBuilder;
 use MySQLReplication\Event\DTO\EventDTO;
 use MySQLReplication\Event\EventSubscribers;
 use MySQLReplication\MySQLReplicationFactory;
 
+$logger = new Logger("app", [new StreamHandler(STDOUT)]);
+BinLogBootstrap::clear();
 /**
  * Your db configuration @see ConfigBuilder for more options
  */
@@ -22,7 +26,10 @@
         ->withHost('127.0.0.1')
         ->withPort(3306)
         ->withPassword('root')
-        ->build()
+        ->withRetry(3)
+        ->build(),
+    null, null, null, null,
+    $logger
 );
 
 /**
@@ -66,11 +73,15 @@ class BinLogBootstrap
     private static function getFileAndPath(): string
     {
         if (null === self::$fileAndPath) {
-            self::$fileAndPath = sys_get_temp_dir() . '/bin-log-replicator-last-position';
+            self::$fileAndPath = '/tmp/bin-log-replicator-last-position';
         }
         return self::$fileAndPath;
     }
 
+    public static function clear() {
+        file_exists(self::getFileAndPath()) && unlink(self::getFileAndPath());
+    }
+
     /**
      * @param BinLogCurrent $binLogCurrent
      */

phpunit.xml

@@ -1,24 +1,16 @@
 <?xml version="1.0" encoding="UTF-8"?>
-<phpunit backupGlobals="false"
-         backupStaticAttributes="false"
-         bootstrap="tests/bootstrap.php"
-         colors="true"
-         convertErrorsToExceptions="true"
-         convertNoticesToExceptions="true"
-         convertWarningsToExceptions="true"
-         processIsolation="false"
-         stopOnFailure="false">
-    <testsuites>
-        <testsuite name="Integration Suite">
-            <directory>./tests/Integration</directory>
-        </testsuite>
-        <testsuite name="Unit Suite">
-            <directory>./tests/Unit</directory>
-        </testsuite>
-    </testsuites>
-    <filter>
-        <whitelist>
-            <directory suffix=".php">src/</directory>
-        </whitelist>
-    </filter>
-</phpunit>
+<phpunit xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" backupGlobals="false" backupStaticAttributes="false" bootstrap="tests/bootstrap.php" colors="true" convertErrorsToExceptions="true" convertNoticesToExceptions="true" convertWarningsToExceptions="true" processIsolation="false" stopOnFailure="false" xsi:noNamespaceSchemaLocation="https://schema.phpunit.de/9.3/phpunit.xsd">
+  <coverage>
+    <include>
+      <directory suffix=".php">src/</directory>
+    </include>
+  </coverage>
+  <testsuites>
+    <testsuite name="Integration Suite">
+      <directory>./tests/Integration</directory>
+    </testsuite>
+    <testsuite name="Unit Suite">
+      <directory>./tests/Unit</directory>
+    </testsuite>
+  </testsuites>
+</phpunit>

src/MySQLReplication/BinLog/BinLogCurrent.php

@@ -10,11 +10,11 @@ class BinLogCurrent implements JsonSerializable
     /**
      * @var int
      */
-    private $binLogPosition;
+    private $binLogPosition = 0;
     /**
      * @var string
      */
-    private $binFileName;
+    private $binFileName = '';
     /**
      * @var string
      */

src/MySQLReplication/BinLog/BinLogServerInfo.php

@@ -17,6 +17,7 @@ public static function parsePackage(string $data, string $version): BinLogServer
     {
         $i = 0;
         $length = strlen($data);
+        $serverInfo = [];
         $serverInfo['protocol_version'] = ord($data[$i]);
         ++$i;
 

src/MySQLReplication/BinLog/BinLogSocketConnect.php

@@ -52,7 +52,16 @@ public function __construct(
         $this->socket = $socket;
         $this->binLogCurrent = new BinLogCurrent();
 
-        $this->socket->connectToStream($config->getHost(), $config->getPort());
+    }
+
+    public function isConnected(): bool
+    {
+        return $this->socket->isConnected();
+    }
+
+    public function connect(): void
+    {
+        $this->socket->connectToStream($this->config->getHost(), $this->config->getPort());
         $this->binLogServerInfo = BinLogServerInfo::parsePackage(
             $this->getResponse(false),
             $this->repository->getVersion()
@@ -275,8 +284,9 @@ private function setBinLogDumpGtid(): void
      */
     private function setBinLogDump(): void
     {
-        $binFilePos = $this->config->getBinLogPosition();
-        $binFileName = $this->config->getBinLogFileName();
+        $binLogCurrent = $this->binLogCurrent;
+        $binFilePos = $binLogCurrent->getBinLogPosition() ?: $this->config->getBinLogPosition();
+        $binFileName = $binLogCurrent->getBinFileName() ?: $this->config->getBinLogFileName();
         if (0 === $binFilePos && '' === $binFileName) {
             $masterStatusDTO = $this->repository->getMasterStatus();
             $binFilePos = $masterStatusDTO->getPosition();
@@ -292,8 +302,8 @@ private function setBinLogDump(): void
         $this->socket->writeToSocket($data);
         $this->getResponse();
 
-        $this->binLogCurrent->setBinLogPosition($binFilePos);
-        $this->binLogCurrent->setBinFileName($binFileName);
+        $binLogCurrent->setBinLogPosition($binFilePos);
+        $binLogCurrent->setBinFileName($binFileName);
     }
 
     public function getBinLogCurrent(): BinLogCurrent

src/MySQLReplication/Config/Config.php

@@ -24,6 +24,7 @@ class Config implements JsonSerializable
     private $tableCacheSize;
     private $custom;
     private $heartbeatPeriod;
+    private $retry;
 
     public function __construct(
         string $user,
@@ -42,7 +43,8 @@ public function __construct(
         array $databasesOnly,
         int $tableCacheSize,
         array $custom,
-        float $heartbeatPeriod
+        float $heartbeatPeriod,
+        int $retry = 0
     ) {
         $this->user = $user;
         $this->host = $host;
@@ -61,6 +63,7 @@ public function __construct(
         $this->tableCacheSize = $tableCacheSize;
         $this->custom = $custom;
         $this->heartbeatPeriod = $heartbeatPeriod;
+        $this->retry = $retry;
     }
 
     /**
@@ -220,6 +223,11 @@ public function getHeartbeatPeriod(): float
         return $this->heartbeatPeriod;
     }
 
+    public function getRetry(): int
+    {
+        return $this->retry;
+    }
+
     public function jsonSerialize()
     {
         return get_object_vars($this);

src/MySQLReplication/Config/ConfigBuilder.php

@@ -22,6 +22,7 @@ class ConfigBuilder
     private $tableCacheSize = 128;
     private $custom = [];
     private $heartbeatPeriod = 0.0;
+    private $retry = 0;
 
     public function withUser(string $user): self
     {
@@ -147,6 +148,13 @@ public function withHeartbeatPeriod(float $heartbeatPeriod): self
         return $this;
     }
 
+    public function withRetry(int $retry): self
+    {
+        $this->retry = $retry;
+
+        return $this;
+    }
+
     public function build(): Config
     {
         return new Config(
@@ -166,7 +174,8 @@ public function build(): Config
             $this->databasesOnly,
             $this->tableCacheSize,
             $this->custom,
-            $this->heartbeatPeriod
+            $this->heartbeatPeriod,
+            $this->retry
         );
     }
 }

src/MySQLReplication/Event/Event.php

@@ -6,7 +6,6 @@
 use MySQLReplication\BinaryDataReader\BinaryDataReader;
 use MySQLReplication\BinaryDataReader\BinaryDataReaderException;
 use MySQLReplication\BinLog\BinLogException;
-use MySQLReplication\BinLog\BinLogServerInfo;
 use MySQLReplication\BinLog\BinLogSocketConnect;
 use MySQLReplication\Config\Config;
 use MySQLReplication\Definitions\ConstEventType;
@@ -35,7 +34,6 @@ class Event
      * @var Config
      */
     private $config;
-    private $binLogServerInfo;
 
     public function __construct(
         Config $config,
@@ -45,13 +43,17 @@ public function __construct(
         CacheInterface $cache
     ) {
         $this->config = $config;
-        $this->binLogServerInfo = $binLogSocketConnect->getBinLogServerInfo();
         $this->binLogSocketConnect = $binLogSocketConnect;
         $this->rowEventFactory = $rowEventFactory;
         $this->eventDispatcher = $eventDispatcher;
         $this->cache = $cache;
     }
 
+    public function connect(): void
+    {
+        $this->binLogSocketConnect->connect();
+    }
+
     /**
      * @throws BinaryDataReaderException
      * @throws BinLogException
@@ -62,6 +64,7 @@ public function __construct(
      */
     public function consume(): void
     {
+        $binLogServerInfo = $this->binLogSocketConnect->getBinLogServerInfo();
         $binaryDataReader = new BinaryDataReader($this->binLogSocketConnect->getResponse());
 
         // check EOF_Packet -> https://dev.mysql.com/doc/internals/en/packet-EOF_Packet.html
@@ -80,7 +83,7 @@ public function consume(): void
             $eventDTO = $this->rowEventFactory->makeRowEvent($binaryDataReader, $eventInfo)->makeTableMapDTO();
         } else if (ConstEventType::ROTATE_EVENT === $eventInfo->getType()) {
             $this->cache->clear();
-            $eventDTO = (new RotateEvent($this->binLogServerInfo, $eventInfo, $binaryDataReader))->makeRotateEventDTO();
+            $eventDTO = (new RotateEvent($binLogServerInfo, $eventInfo, $binaryDataReader))->makeRotateEventDTO();
         } else if (ConstEventType::GTID_LOG_EVENT === $eventInfo->getType()) {
             $eventDTO = (new GtidEvent($eventInfo, $binaryDataReader))->makeGTIDLogDTO();
         } else if (ConstEventType::HEARTBEAT_LOG_EVENT === $eventInfo->getType()) {
@@ -127,7 +130,8 @@ private function createEventInfo(BinaryDataReader $binaryDataReader): EventInfo
 
     private function filterDummyMariaDbEvents(QueryDTO $queryDTO): ?QueryDTO
     {
-        if ($this->binLogServerInfo->isMariaDb() && false !== strpos($queryDTO->getQuery(), self::MARIADB_DUMMY_QUERY)) {
+        if ($this->binLogSocketConnect->getBinLogServerInfo()->isMariaDb() &&
+            false !== strpos($queryDTO->getQuery(), self::MARIADB_DUMMY_QUERY)) {
             return null;
         }