Merge branch 'krowinski:master' into master

src/MySQLReplication/Event/DTO/QueryDTO.php

@@ -12,18 +12,21 @@ class QueryDTO extends EventDTO
     private $query;
     private $database;
     private $type = ConstEventsNames::QUERY;
+    private $threadId;
 
     public function __construct(
         EventInfo $eventInfo,
         string $database,
         int $executionTime,
-        string $query
+        string $query,
+        int $threadId
     ) {
         parent::__construct($eventInfo);
 
         $this->executionTime = $executionTime;
         $this->query = $query;
         $this->database = $database;
+        $this->threadId = $threadId;
     }
 
     public function getDatabase(): string
@@ -46,6 +49,11 @@ public function getType(): string
         return $this->type;
     }
 
+    public function getThreadId(): int
+    {
+        return $this->threadId;
+    }
+
     public function __toString(): string
     {
         return PHP_EOL .

src/MySQLReplication/Event/QueryEvent.php

@@ -12,7 +12,7 @@ class QueryEvent extends EventCommon
 {
     public function makeQueryDTO(): QueryDTO
     {
-        $this->binaryDataReader->advance(4);
+        $threadId = $this->binaryDataReader->readUInt32();
         $executionTime = $this->binaryDataReader->readUInt32();
         $schemaLength = $this->binaryDataReader->readUInt8();
         $this->binaryDataReader->advance(2);
@@ -26,7 +26,8 @@ public function makeQueryDTO(): QueryDTO
             $this->eventInfo,
             $schema,
             $executionTime,
-            $query
+            $query,
+            $threadId
         );
     }
 }

src/MySQLReplication/JsonBinaryDecoder/JsonBinaryDecoderService.php

@@ -74,6 +74,11 @@ public static function makeJsonBinaryDecoder(string $data): self
      */
     public function parseToString(): string
     {
+        // Sometimes, we can insert a NULL JSON even we set the JSON field as NOT NULL.
+        // If we meet this case, we can return a 'null' value.
+        if($this->binaryDataReader->getBinaryDataLength() === 0) {
+            return 'null';
+        }
         $this->parseJson($this->binaryDataReader->readUInt8());
 
         return $this->jsonBinaryDecoderFormatter->getJsonString();
@@ -197,7 +202,7 @@ private function parseArrayOrObject(int $type, int $intSize): array
 
         $entries = [];
         for ($i = 0; $i !== $elementCount; ++$i) {
-            $entries[$i] = $this->getOffsetOrInLinedValue($bytes, $intSize);
+            $entries[$i] = $this->getOffsetOrInLinedValue($bytes, $intSize, $valueEntrySize);
         }
 
         $keys = [];
@@ -237,11 +242,21 @@ private static function valueEntrySize(bool $large): int
      * @throws BinaryDataReaderException
      * @throws JsonBinaryDecoderException
      */
-    private function getOffsetOrInLinedValue(int $bytes, int $intSize): JsonBinaryDecoderValue
+    private function getOffsetOrInLinedValue(int $bytes, int $intSize, int $valueEntrySize): JsonBinaryDecoderValue
     {
         $type = $this->binaryDataReader->readUInt8();
+
         if (self::isInLinedType($type, $intSize)) {
-            return $this->parseScalar($type);
+            $scalar = $this->parseScalar($type);
+
+            // In binlog format, JSON arrays are fixed width elements, even though type value can be smaller.
+            // In order to properly process this case, we need to move cursor to the next element, which is on position 1 + $valueEntrySize (1 is length of type)
+            if($type === self::UINT16 || $type === self::INT16) {
+                $readNextBytes = $valueEntrySize - 2 - 1;
+                $this->binaryDataReader->read($readNextBytes);
+            }
+
+            return $scalar;
         }
 
         $offset = $this->binaryDataReader->readUIntBySize($intSize);