Merge pull request #15 from xp-forge/feature/watch

Change streams: Watch databases and collections for changes

Author: thekid

src/main/php/com/mongodb/Collection.class.php

@@ -1,7 +1,7 @@
 <?php namespace com\mongodb;
 
 use com\mongodb\io\Protocol;
-use com\mongodb\result\{Insert, Update, Delete, Cursor};
+use com\mongodb\result\{Insert, Update, Delete, Cursor, ChangeStream};
 
 /**
  * A collection inside a database.
@@ -195,4 +195,24 @@ public function aggregate(array $pipeline= [], Session $session= null): Cursor {
 
     return new Cursor($this->proto, $session, $result['body']['cursor']);
   }
+
+  /**
+   * Watch for changes in this collection
+   *
+   * @param  [:var][] $pipeline
+   * @param  [:var] $options
+   * @param  ?com.mongodb.Session $session
+   * @return com.mongodb.result.ChangeStream
+   * @throws com.mongodb.Error
+   */
+  public function watch(array $pipeline= [], array $options= [], Session $session= null): ChangeStream {
+    array_unshift($pipeline, ['$changeStream' => (object)$options]);
+    $result= $this->proto->read($session, [
+      'aggregate' => $this->name,
+      'pipeline'  => $pipeline,
+      'cursor'    => (object)[],
+      '$db'       => $this->database,
+    ]);
+    return new ChangeStream($this->proto, $session, $result['body']['cursor']);
+  }
 }

src/main/php/com/mongodb/Database.class.php

@@ -1,7 +1,7 @@
 <?php namespace com\mongodb;
 
 use com\mongodb\io\Protocol;
-use com\mongodb\result\Cursor;
+use com\mongodb\result\{Cursor, ChangeStream};
 
 class Database {
   private $proto, $name;
@@ -34,4 +34,24 @@ public function collections($session= null) {
     ]);
     return new Cursor($this->proto, $session, $result['body']['cursor']);
   }
+
+  /**
+   * Watch for changes in this database
+   *
+   * @param  [:var][] $pipeline
+   * @param  [:var] $options
+   * @param  ?com.mongodb.Session $session
+   * @return com.mongodb.result.ChangeStream
+   * @throws com.mongodb.Error
+   */
+  public function watch(array $pipeline= [], array $options= [], Session $session= null): ChangeStream {
+    array_unshift($pipeline, ['$changeStream' => (object)$options]);
+    $result= $this->proto->read($session, [
+      'aggregate' => 1,
+      'pipeline'  => $pipeline,
+      'cursor'    => (object)[],
+      '$db'       => $this->name,
+    ]);
+    return new ChangeStream($this->proto, $session, $result['body']['cursor']);
+  }
 }

src/main/php/com/mongodb/MongoConnection.class.php

@@ -1,6 +1,7 @@
 <?php namespace com\mongodb;
 
 use com\mongodb\io\Protocol;
+use com\mongodb\result\ChangeStream;
 use lang\{IllegalArgumentException, Value};
 use peer\AuthenticationException;
 use util\{Bytes, UUID, Objects};
@@ -106,6 +107,28 @@ public function databases($filter= null, $session= null) {
     }
   }
 
+  /**
+   * Watch for changes in all databases.
+   *
+   * @param  [:var][] $pipeline
+   * @param  [:var] $options
+   * @param  ?com.mongodb.Session $session
+   * @return com.mongodb.result.ChangeStream
+   * @throws com.mongodb.Error
+   */
+  public function watch(array $pipeline= [], array $options= [], Session $session= null): ChangeStream {
+    $this->proto->connect();
+
+    array_unshift($pipeline, ['$changeStream' => ['allChangesForCluster' => true] + $options]);
+    $result= $this->proto->read($session, [
+      'aggregate' => 1,
+      'pipeline'  => $pipeline,
+      'cursor'    => (object)[],
+      '$db'       => 'admin',
+    ]);
+    return new ChangeStream($this->proto, $session, $result['body']['cursor']);
+  }
+
   /**
    * Close connection
    *

src/main/php/com/mongodb/result/ChangeStream.class.php

@@ -0,0 +1,9 @@
+<?php namespace com\mongodb\result;
+
+class ChangeStream extends Cursor {
+
+  /** @return ?[:var] */
+  public function resumeToken() {
+    return $this->current['postBatchResumeToken'] ?? null;
+  }
+}