krowinski · krowinski · Feb 27, 2016 · Feb 27, 2016
diff --git a/README.md b/README.md
@@ -35,34 +35,8 @@ Remember to change config for your user, host and password.
 
 User should have replication privileges [ REPLICATION CLIENT, SELECT]
 
-```php
-<?php
-error_reporting(E_ALL);
-date_default_timezone_set('UTC');
-ini_set('memory_limit', '8M');
-
-include __DIR__ . '/../vendor/autoload.php';
-
-use MySQLReplication\Service\BinLogStream;
-use MySQLReplication\Config\Config;
-
-$binLogStream = new BinLogStream(
-    new Config('root', '192.168.1.100', 3306, 'root')
-);
-while (1)
-{
-    $result = $binLogStream->analysisBinLog();
-    if (!is_null($result))
-    {
-        // all events got __toString() implementation
-        echo $result;
-
-        // all events got JsonSerializable implementation
-        //echo json_encode($result, JSON_PRETTY_PRINT);
-
-        //echo 'Memory usage ' . round(memory_get_usage() / 1048576, 2) . ' MB' . PHP_EOL;
-    }
-}
+```sh
+php example/dump_events.php
 ```
 
 For this SQL sessions:

diff --git a/composer.json b/composer.json
@@ -10,7 +10,8 @@
   "type": "library",
   "require": {
     "php": ">=5.4",
-    "doctrine/dbal": "^2.5"
+    "doctrine/dbal": "^2.5",
+    "doctrine/collections": "^1.3"
   },
   "require-dev": {
     "phpunit/phpunit": "*"

diff --git a/example/benchmark.php b/example/benchmark.php
@@ -4,65 +4,96 @@
 error_reporting(E_ALL);
 ini_set('display_errors', 1);
 
-use MySQLReplication\Config\Config;
-use MySQLReplication\DataBase\DBHelper;
+use Doctrine\DBAL\DriverManager;
+use MySQLReplication\BinLogStream;
+use MySQLReplication\Config\ConfigService;
 use MySQLReplication\Definitions\ConstEventType;
-use MySQLReplication\DTO\UpdateRowsDTO;
-use MySQLReplication\DTO\WriteRowsDTO;
-use MySQLReplication\Service\BinLogStream;
-
-
+use MySQLReplication\Event\DTO\UpdateRowsDTO;
 
 /**
- * Class Base
+ * Class Benchmark
  */
 class Benchmark
 {
     /**
      * @var string
      */
     private $database = 'mysqlreplication_test';
+
     /**
-     * @var \Doctrine\DBAL\Connection
-     */
-    private $conn;
-    /**
-     * @var Config
+     * Benchmark constructor.
      */
-    private $config;
-
     public function __construct()
     {
-        $this->config = new Config('root', '192.168.1.100', 3306, 'root');
-        $this->conn = (new DBHelper($this->config))->getConnection();
+        $conn = $this->getConnection();
+        $conn->exec("DROP DATABASE IF EXISTS " . $this->database);
+        $conn->exec("CREATE DATABASE " . $this->database);
+        $conn->exec("USE " . $this->database);
+        $conn->exec("CREATE TABLE test (i INT) ENGINE = MEMORY");
+        $conn->exec("INSERT INTO test VALUES(1)");
+        $conn->exec("CREATE TABLE test2 (i INT) ENGINE = MEMORY");
+        $conn->exec("INSERT INTO test2 VALUES(1)");
+        $conn->exec("RESET MASTER");
+
+        $this->binLogStream = new BinLogStream(
+            (new ConfigService())->makeConfigFromArray([
+                'user' => 'root',
+                'host' => '192.168.1.100',
+                'password' => 'root',
+                'eventsOnly' => [ConstEventType::UPDATE_ROWS_EVENT_V1, ConstEventType::UPDATE_ROWS_EVENT_V2],
+                'slaveId' => 9999
+            ])
+        );
+    }
 
-        $this->conn->exec("DROP DATABASE IF EXISTS " . $this->database);
-        $this->conn->exec("CREATE DATABASE " . $this->database);
-        $this->conn->exec("USE " . $this->database);
-        $this->conn->exec("CREATE TABLE test (i INT) ENGINE = MEMORY");
-        $this->conn->exec("INSERT INTO test VALUES(1)");
-        $this->conn->exec("CREATE TABLE test2 (i INT) ENGINE = MEMORY");
-        $this->conn->exec("INSERT INTO test2 VALUES(1)");
-        $this->conn->exec("RESET MASTER");
+    /**
+     * @return \Doctrine\DBAL\Connection
+     * @throws \Doctrine\DBAL\DBALException
+     */
+    private function getConnection()
+    {
+        return DriverManager::getConnection([
+            'user' => 'root',
+            'password' => 'root',
+            'host' => '192.168.1.100',
+            'port' => 3306,
+            'driver' => 'pdo_mysql',
+            'dbname' => $this->database
+        ]);
+    }
 
-        $this->binLogStream =  new BinLogStream(
-            $this->config,
-            '',
-            '',
-            '',
-            '',
-            [ConstEventType::UPDATE_ROWS_EVENT_V1, ConstEventType::UPDATE_ROWS_EVENT_V2]
-        );
+    /**
+     *
+     */
+    public function run()
+    {
+        $pid = pcntl_fork();
+        if ($pid == -1)
+        {
+            die('could not fork');
+        }
+        else if ($pid)
+        {
+            $this->consume();
+            pcntl_wait($status);
+        }
+        else
+        {
+            $this->produce();
+        }
     }
 
-    public function consume()
+    /**
+     *
+     */
+    private function consume()
     {
         $start = microtime(true);
         $i = 0;
 
         while (1)
         {
-            $result = $this->binLogStream->analysisBinLog();
+            $result = $this->binLogStream->getBinLogEvent();
             if ($result instanceof UpdateRowsDTO)
             {
                 $i += 1;
@@ -74,34 +105,22 @@ public function consume()
         }
     }
 
-    public function run()
-    {
-        $pid = pcntl_fork();
-        if ($pid == -1) {
-            die('could not fork');
-        } else if ($pid) {
-            $this->consume();
-            pcntl_wait($status);
-        } else {
-            $this->produce();
-        }
-
-    }
-
-    public function produce()
+    /**
+     * @throws \Doctrine\DBAL\DBALException
+     */
+    private function produce()
     {
-        $this->conn = (new DBHelper($this->config))->getConnection();
-        $this->conn->exec("USE " . $this->database);
+        $conn = $this->getConnection();
 
         echo 'Start insert data' . PHP_EOL;
         while (1)
         {
-
-            $this->conn->exec("UPDATE test SET i = i + 1;");
-            $this->conn->exec("UPDATE test2 SET i = i + 1;");
+            $conn->exec("UPDATE test SET i = i + 1;");
+            $conn->exec("UPDATE test2 SET i = i + 1;");
         }
-    }
 
+        $conn->close();
+    }
 }
 
 (new Benchmark())->run();
diff --git a/example/dump_events.php b/example/dump_events.php
@@ -5,15 +5,20 @@
 
 include __DIR__ . '/../vendor/autoload.php';
 
-use MySQLReplication\Service\BinLogStream;
-use MySQLReplication\Config\Config;
+use MySQLReplication\BinLogStream;
+use MySQLReplication\Config\ConfigService;
 
 $binLogStream = new BinLogStream(
-    new Config('root', '192.168.1.100', 3306, 'root')
+    (new ConfigService())->makeConfigFromArray([
+        'user' => 'root',
+        'host' => '192.168.1.100',
+        'password' => 'root',
+        //'gtid' => '9b1c8d18-2a76-11e5-a26b-000c2976f3f3:1-177592',
+    ])
 );
 while (1)
 {
-    $result = $binLogStream->analysisBinLog();
+    $result = $binLogStream->getBinLogEvent();
     if (!is_null($result))
     {
         // all events got __toString() implementation

diff --git a/src/MySQLReplication/Pack/PackAuth.php → src/MySQLReplication/BinLog/BinLogAuth.php b/src/MySQLReplication/Pack/PackAuth.php → src/MySQLReplication/BinLog/BinLogAuth.php
@@ -1,28 +1,44 @@
 <?php
 
-namespace MySQLReplication\Pack;
+namespace MySQLReplication\BinLog;
 
-use MySQLReplication\Definitions\ConstAuth;
 use MySQLReplication\Exception\BinLogException;
 
 /**
  * Class PackAuth
  */
-class PackAuth
+class BinLogAuth
 {
+    /**
+     * 2^24 - 1 16m
+     * @var int
+     */
+    public $packageMaxLength = 16777215;
+
+    /**
+     * http://dev.mysql.com/doc/internals/en/auth-phase-fast-path.html 00 FE
+     * @var array
+     */
+    public $packageOkHeader = [0, 254];
+
+    /**
+     * FF
+     * @var array
+     */
+    public $packageErrorHeader = [255];
+
     /**
      * @param $flag
      * @param $user
      * @param $pass
      * @param $salt
-     * @param string $db
      * @return string
      * @link http://dev.mysql.com/doc/internals/en/secure-password-authentication.html#packet-Authentication::Native41
      */
-    public static function initPack($flag, $user, $pass, $salt, $db = '')
+    public function createAuthenticationPacket($flag, $user, $pass, $salt)
     {
         $data = pack('L', $flag);
-        $data .= pack('L', ConstAuth::$PACK_MAX_LENGTH);
+        $data .= pack('L', $this->packageMaxLength);
         $data .= chr(33);
         for ($i = 0; $i < 23; $i++)
         {
@@ -31,10 +47,6 @@ public static function initPack($flag, $user, $pass, $salt, $db = '')
         $result = sha1($pass, true) ^ sha1($salt . sha1(sha1($pass, true), true), true);
 
         $data = $data . $user . chr(0) . chr(strlen($result)) . $result;
-        if ($db)
-        {
-            $data .= $db . chr(0);
-        }
         $str = pack('L', strlen($data));
         $s = $str[0] . $str[1] . $str[2];
         $data = $s . chr(1) . $data;
@@ -43,25 +55,26 @@ public static function initPack($flag, $user, $pass, $salt, $db = '')
     }
 
     /**
-     * @param $pack
+     * @param string $packet
      * @return array
      * @throws BinLogException
      */
-    public static function success($pack)
+    public function isWriteSuccessful($packet)
     {
-        $head = ord($pack[0]);
-        if (in_array($head, ConstAuth::$OK_PACK_HEAD))
+        $head = ord($packet[0]);
+        if (in_array($head, $this->packageOkHeader))
         {
             return ['status' => true, 'code' => 0, 'msg' => ''];
         }
         else
         {
-            $error_code = unpack('v', $pack[1] . $pack[2])[1];
+            $error_code = unpack('v', $packet[1] . $packet[2])[1];
             $error_msg = '';
-            for ($i = 9; $i < strlen($pack); $i++)
+            for ($i = 9; $i < strlen($packet); $i++)
             {
-                $error_msg .= $pack[$i];
+                $error_msg .= $packet[$i];
             }
+
             throw new BinLogException($error_msg, $error_code);
         }
     }