Merge pull request #8 from krowinski/event-dispacher

Event dispacher

Author: krowinski

.gitignore

@@ -1,7 +1,6 @@
 language: php
 
 php:
-  - 5.4
   - 5.5
   - 5.6
   - 7.0

.travis.yml

@@ -9,10 +9,12 @@
   ],
   "type": "library",
   "require": {
-    "php": ">=5.4",
+    "php": ">=5.5.9",
     "doctrine/dbal": "^2.5",
     "doctrine/collections": "^1.3",
-    "ext-sockets": "*"
+    "ext-sockets": "*",
+    "symfony/event-dispatcher": "^3.1",
+    "symfony/dependency-injection": "^3.1"
   },
   "require-dev": {
     "phpunit/phpunit": "*"

LICENSE

@@ -1,17 +1,57 @@
 <?php
+
+namespace example;
+
 error_reporting(E_ALL);
 ini_set('display_errors', 1);
 date_default_timezone_set('UTC');
 include __DIR__ . '/../vendor/autoload.php';
 
+use Doctrine\DBAL\Connection;
+use Doctrine\DBAL\DBALException;
 use Doctrine\DBAL\DriverManager;
+use MySQLReplication\BinLog\Exception\BinLogException;
+use MySQLReplication\Config\Exception\ConfigException;
+use MySQLReplication\Event\EventSubscribers;
+use MySQLReplication\Exception\MySQLReplicationException;
 use MySQLReplication\MySQLReplicationFactory;
 use MySQLReplication\Config\ConfigService;
 use MySQLReplication\Definitions\ConstEventType;
 use MySQLReplication\Event\DTO\UpdateRowsDTO;
 
+/**
+ * Class BenchmarkEventSubscribers
+ * @package example
+ */
+class BenchmarkEventSubscribers extends EventSubscribers
+{
+    /**
+     * @var int
+     */
+    private $start = 0;
+    /**
+     * @var int
+     */
+    private $counter = 0;
+
+    public function __construct()
+    {
+        $this->start = microtime(true);
+    }
+
+    public function onUpdate(UpdateRowsDTO $event)
+    {
+        ++$this->counter;
+        if (0 === ($this->counter % 1000))
+        {
+            echo ((int)($this->counter / (microtime(true) - $this->start)) . ' event by seconds (' . $this->counter . ' total)') . PHP_EOL;
+        }
+    }
+}
+
 /**
  * Class Benchmark
+ * @package example
  */
 class Benchmark
 {
@@ -22,33 +62,41 @@ class Benchmark
 
     /**
      * Benchmark constructor.
+     * @throws DBALException
+     * @throws ConfigException
+     * @throws BinLogException
+     * @throws MySQLReplicationException
      */
     public function __construct()
     {
         $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");
+        $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 MySQLReplicationFactory(
             (new ConfigService())->makeConfigFromArray([
                 'user' => 'root',
                 'ip' => '127.0.0.1',
                 'password' => 'root',
+                // we only interest in update row event
                 'eventsOnly' => [ConstEventType::UPDATE_ROWS_EVENT_V1, ConstEventType::UPDATE_ROWS_EVENT_V2],
                 'slaveId' => 9999
             ])
         );
+
+        // register benchmark subscriber handler
+        $this->binLogStream->registerSubscriber(new BenchmarkEventSubscribers());
     }
 
     /**
-     * @return \Doctrine\DBAL\Connection
-     * @throws \Doctrine\DBAL\DBALException
+     * @return Connection
+     * @throws DBALException
      */
     private function getConnection()
     {
@@ -63,12 +111,13 @@ private function getConnection()
     }
 
     /**
-     *
+     * @throws MySQLReplicationException
+     * @throws DBALException
      */
     public function run()
     {
         $pid = pcntl_fork();
-        if ($pid == -1)
+        if ($pid === -1)
         {
             die('could not fork');
         }
@@ -84,29 +133,17 @@ public function run()
     }
 
     /**
-     *
+     * @throws MySQLReplicationException
      */
     private function consume()
     {
-        $start = microtime(true);
-        $i = 0;
-
-        while (1)
-        {
-            $result = $this->binLogStream->getBinLogEvent();
-            if ($result instanceof UpdateRowsDTO)
-            {
-                ++$i;
-                if (0 === ($i % 1000))
-                {
-                    echo ((int)($i / (microtime(true) - $start)) . ' event by seconds (' . $i . ' total)') . PHP_EOL;
-                }
-            }
+        while(1) {
+            $this->binLogStream->binLogEvent();
         }
     }
 
     /**
-     * @throws \Doctrine\DBAL\DBALException
+     * @throws DBALException
      */
     private function produce()
     {
@@ -115,8 +152,8 @@ private function produce()
         echo 'Start insert data' . PHP_EOL;
         while (1)
         {
-            $conn->exec("UPDATE test SET i = i + 1;");
-            $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();

README.md

@@ -1,32 +1,52 @@
-<?php
-error_reporting(E_ALL);
-ini_set('display_errors', 1);
-date_default_timezone_set('UTC');
-include __DIR__ . '/../vendor/autoload.php';
-
-use MySQLReplication\MySQLReplicationFactory;
-use MySQLReplication\Config\ConfigService;
-
-$binLogStream = new MySQLReplicationFactory(
-    (new ConfigService())->makeConfigFromArray([
-        'user' => 'root',
-        'ip' => '127.0.0.1',
-        'password' => 'root',
-        //'mariaDbGtid' => '1-1-3,0-1-88',
-        //'gtid' => '9b1c8d18-2a76-11e5-a26b-000c2976f3f3:1-177592',
-    ])
-);
-while (1)
-{
-    $result = $binLogStream->getBinLogEvent();
-    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;
-    }
+<?php
+
+namespace example;
+
+error_reporting(E_ALL);
+ini_set('display_errors', 1);
+date_default_timezone_set('UTC');
+include __DIR__ . '/../vendor/autoload.php';
+
+use MySQLReplication\Config\ConfigService;
+use MySQLReplication\Event\DTO\EventDTO;
+use MySQLReplication\Event\EventSubscribers;
+use MySQLReplication\MySQLReplicationFactory;
+
+$binLogStream = new MySQLReplicationFactory(
+    (new ConfigService())->makeConfigFromArray([
+        'user' => 'root',
+        'ip' => '127.0.0.1',
+        'password' => 'root',
+        //'mariaDbGtid' => '1-1-3,0-1-88',
+        //'gtid' => '9b1c8d18-2a76-11e5-a26b-000c2976f3f3:1-177592',
+    ])
+);
+
+/**
+ * Class BenchmarkEventSubscribers
+ * @package example
+ */
+class MyEventSubscribers extends EventSubscribers
+{
+    /**
+     * @param EventDTO $event (your own handler more in EventSubscribers class )
+     */
+    public function allEvents(EventDTO $event)
+    {
+        // all events got __toString() implementation
+        echo $event;
+
+        // all events got JsonSerializable implementation
+        //echo json_encode($result, JSON_PRETTY_PRINT);
+
+        echo 'Memory usage ' . round(memory_get_usage() / 1048576, 2) . ' MB' . PHP_EOL;
+    }
+}
+
+// register your events handler here
+$binLogStream->registerSubscriber(new MyEventSubscribers());
+
+// start consuming events
+while (1) {
+    $binLogStream->binLogEvent();
 }