Use compression library

composer.json

@@ -9,6 +9,7 @@
     "xp-framework/core": "^12.0 | ^11.0 | ^10.0",
     "xp-framework/networking": "^10.0",
     "xp-framework/math": "^10.0 | ^9.1",
+    "xp-forge/compression": "^2.0",
     "php" : ">=7.4.0"
   },
   "require-dev" : {

src/main/php/com/mongodb/io/Compression.class.php

@@ -1,5 +1,6 @@
 <?php namespace com\mongodb\io;
 
+use io\streams\compress\{Gzip, ZStandard, Snappy};
 use lang\Value;
 use util\Comparison;
 
@@ -16,8 +17,21 @@ class Compression implements Value {
   private $compressors;
 
   static function __static() {
-    extension_loaded('zlib') && self::$negotiable['zlib']= fn($options) => new Zlib($options['zlibCompressionLevel'] ?? -1);
-    extension_loaded('zstd') && self::$negotiable['zstd']= fn($options) => new Zstd($options['zstdCompressionLevel'] ?? -1);
+    self::$negotiable['snappy']= fn($options) => new Compressor(
+      1,
+      new Snappy(),
+      null
+    );
+    extension_loaded('zlib') && self::$negotiable['zlib']= fn($options) => new Compressor(
+      2,
+      new Gzip(),
+      $options['zlibCompressionLevel'] ?? -1
+    );
+    extension_loaded('zstd') && self::$negotiable['zstd']= fn($options) => new Compressor(
+      3,
+      new ZStandard(),
+      $options['zstdCompressionLevel'] ?? -1
+    );
   }
 
   /** @param [:com.mongodb.io.Compressor] $compressors */

src/main/php/com/mongodb/io/Compressor.class.php

@@ -1,15 +1,18 @@
 <?php namespace com\mongodb\io;
 
+use io\streams\compress\Algorithm;
 use lang\Value;
 
-abstract class Compressor implements Value {
-  public $id;
+class Compressor implements Value {
+  public $id, $algorithm, $options;
 
-  public abstract function compress($data);
+  public function __construct(int $id, Algorithm $algorithm, $options= null) {
+    $this->id= $id;
+    $this->algorithm= $algorithm;
+    $this->options= $options;
+  }
 
-  public abstract function decompress($compressed);
-
-  public function toString() { return nameof($this).'(id: '.$this->id.')'; }
+  public function toString() { return nameof($this).'(id: '.$this->id.', options: '.$this->options.')'; }
 
   public function hashCode() { return 'C'.$this->id; }
 

src/main/php/com/mongodb/io/Connection.class.php

@@ -239,7 +239,7 @@ public function send($operation, $header, $sections, $readPreference= null) {
     $length= strlen($body);
 
     if ($this->compression && $compressor= $this->compression->for($sections, $length)) {
-      $compressed= $compressor->compress($body);
+      $compressed= $compressor->algorithm->compress($body, $compressor->options);
       $this->socket->write(pack(
         'VVVVVVCa*',
         strlen($compressed) + 25,
@@ -284,7 +284,7 @@ public function send($operation, $header, $sections, $readPreference= null) {
         $compressed= unpack('VoriginalOpcode/VuncompressedSize/CcompressorId', $response);
 
         if ($this->compression && $compressor= $this->compression->select($compressed['compressorId'] ?? null)) {
-          $response= $compressor->decompress(substr($response, 9));
+          $response= $compressor->algorithm->decompress(substr($response, 9));
           $meta['opCode']= $compressed['originalOpcode'];
           goto opcode;
         }

src/test/php/com/mongodb/unittest/CompressionTest.class.php

@@ -1,6 +1,7 @@
 <?php namespace com\mongodb\unittest;
 
-use com\mongodb\io\{Compression, Compressor, Zlib, Zstd};
+use com\mongodb\io\{Compression, Compressor};
+use io\streams\compress\{None, Gzip, Snappy, ZStandard};
 use test\verify\Runtime;
 use test\{Assert, Before, Test, Values};
 
@@ -9,11 +10,7 @@ class CompressionTest {
 
   #[Before]
   public function compressor() {
-    $this->compressor= new class() extends Compressor {
-      public $id= 9;
-      public function compress($data) { /** Not implemented */ }
-      public function decompress($compressed) { /** Not implemented */ }
-    };
+    $this->compressor= new Compressor(0, new None());
   }
 
 
@@ -57,29 +54,34 @@ public function negotiate_unsupported() {
     Assert::null(Compression::negotiate(['unsupported']));
   }
 
+  #[Test]
+  public function negotiate_snappy() {
+    Assert::instance(Snappy::class, Compression::negotiate(['unsupported', 'snappy'])->select(1)->algorithm);
+  }
+
   #[Test, Runtime(extensions: ['zlib'])]
   public function negotiate_zlib() {
-    Assert::instance(Zlib::class, Compression::negotiate(['unsupported', 'zlib'])->select(2));
+    Assert::instance(Gzip::class, Compression::negotiate(['unsupported', 'zlib'])->select(2)->algorithm);
   }
 
   #[Test, Runtime(extensions: ['zlib']), Values([[[], -1], [['zlibCompressionLevel' => 6], 6]])]
   public function negotiate_zlib_with($options, $level) {
     $compressor= Compression::negotiate(['zlib'], $options)->select(2);
 
-    Assert::instance(Zlib::class, $compressor);
-    Assert::equals($level, $compressor->level);
+    Assert::instance(Gzip::class, $compressor->algorithm);
+    Assert::equals($level, $compressor->options);
   }
 
   #[Test, Runtime(extensions: ['zstd'])]
   public function negotiate_zstd() {
-    Assert::instance(Zstd::class, Compression::negotiate(['unsupported', 'zstd'])->select(3));
+    Assert::instance(ZStandard::class, Compression::negotiate(['unsupported', 'zstd'])->select(3)->algorithm);
   }
 
   #[Test, Runtime(extensions: ['zstd']), Values([[[], -1], [['zstdCompressionLevel' => 6], 6]])]
   public function negotiate_zstd_with($options, $level) {
     $compressor= Compression::negotiate(['zstd'], $options)->select(3);
 
-    Assert::instance(Zstd::class, $compressor);
-    Assert::equals($level, $compressor->level);
+    Assert::instance(ZStandard::class, $compressor->algorithm);
+    Assert::equals($level, $compressor->options);
   }
 }

src/test/php/com/mongodb/unittest/ConnectionTest.class.php

@@ -1,7 +1,8 @@
 <?php namespace com\mongodb\unittest;
 
 use com\mongodb\Int64;
-use com\mongodb\io\{BSON, Connection, Compression, Compressor, Zlib, Zstd};
+use com\mongodb\io\{BSON, Connection, Compression, Compressor};
+use io\streams\compress\{Gzip, ZStandard};
 use peer\ConnectException;
 use test\verify\Runtime;
 use test\{Assert, Before, Expect, Test, Values};
@@ -31,7 +32,7 @@ private function msg(array $document): array {
   /** Creates an OP_COMPRESSED message with an embedded OP_MSG opcode */
   private function compressed(Compressor $compressor, array $document): array {
     $payload= pack('VC', 0, 0).$this->bson->sections($document);
-    $compressed= $compressor->compress($payload);
+    $compressed= $compressor->algorithm->compress($payload, $compressor->options);
     return [
       pack('VVVV', strlen($compressed) + 25, 0, 0, Connection::OP_COMPRESSED),
       pack('VVC', Connection::OP_MSG, strlen($payload), $compressor->id).$compressed
@@ -147,7 +148,7 @@ public function send_and_receive_zlib() {
     $documents= [['_id' => 'one']];
     $c= new Connection(new TestingSocket([
       ...$this->reply(['ok' => 1.0, 'compression' => ['zlib']]),
-      ...$this->compressed(new Zlib(), [
+      ...$this->compressed(new Compressor(2, new Gzip()), [
         'cursor' => ['firstBatch' => $documents, 'id' => new Int64(0), 'ns' => 'test.entries'],
         'ok'     => 1,
       ]),
@@ -163,7 +164,7 @@ public function send_and_receive_zstd() {
     $documents= [['_id' => 'one']];
     $c= new Connection(new TestingSocket([
       ...$this->reply(['ok' => 1.0, 'compression' => ['zstd']]),
-      ...$this->compressed(new Zstd(), [
+      ...$this->compressed(new Compressor(3, new ZStandard()), [
         'cursor' => ['firstBatch' => $documents, 'id' => new Int64(0), 'ns' => 'test.entries'],
         'ok'     => 1,
       ]),