Server push notification handling #46

.travis.yml

@@ -29,6 +29,8 @@ after_script:
  - php ./vendor/bin/coveralls -v
 
 env:
+  - ORIENTDB_VERSION=2.1.2
+  - ORIENTDB_VERSION=2.1.1
   - ORIENTDB_VERSION=2.1.0
   - ORIENTDB_VERSION=2.0.13
   - ORIENTDB_VERSION=1.7.10

composer.json

@@ -1,7 +1,7 @@
 {
     "name": "ostico/phporient",
     "description": "PHPOrient is a good php driver based on the binary protocol of OrientDB.",
-    "version": "v1.2.0",
+    "version": "v1.2.1",
     "keywords": [
         "orientdb",
         "orient db",

src/PhpOrient/Configuration/Constants.php

@@ -14,10 +14,10 @@ class Constants {
      * @const string the client version.
      *
      */
-    const VERSION = "v1.1.9";
+    const VERSION = "v1.2.1";
     const ID = "1";
-    const NAME = "PhpOrient - PHP binary client for OrientDB";
-    const SUPPORTED_PROTOCOL = 30;
+    const NAME = "PhpOrient - Php driver based on the binary protocol of OrientDB.";
+    const SUPPORTED_PROTOCOL = 32;
     public static $LOGGING = false;
     public static $LOG_FILE_PATH = false;
 

src/PhpOrient/Protocols/Binary/Abstracts/Operation.php

@@ -17,6 +17,7 @@
 use PhpOrient\Exceptions\PhpOrientException;
 use PhpOrient\Exceptions\PhpOrientBadMethodCallException;
 use PhpOrient\Protocols\Binary\OrientSocket;
+use PhpOrient\Protocols\Common\ClusterMap;
 use Closure;
 
 abstract class Operation implements ConfigurableInterface {
@@ -55,9 +56,18 @@ abstract class Operation implements ConfigurableInterface {
     protected $_transport;
 
     /**
+     * Callback function to apply on Async records when they are fetched
+     *
+     * @var Closure|string
+     */
+    protected $_callback;
+
+    /**
+     * Callback function to apply on push request when they are received
+     *
      * @var Closure|string
      */
-    public $_callback;
+    protected $_pushRequestCallback;
 
     /**
      * Class constructor
@@ -72,6 +82,7 @@ public function __construct( SocketTransport $_transport ) {
         $this->_transport = $_transport;
         $this->_socket    = $_transport->getSocket();
 
+        $this->_pushRequestCallback = 'Operation::_pushReceived';
         $this->_callback = function(){};
 
     }
@@ -145,6 +156,50 @@ protected function _readHeader() {
         } elseif( $status === 3 ){
             // server push data for nodes up/down update info,
             // needed for failover on cluster
+
+            # Push notification, Node cluster changed
+            #
+            # FIELD_BYTE (OChannelBinaryProtocol.PUSH_DATA);  # WRITE 3
+            # FIELD_INT (Integer.MIN_VALUE);  # SESSION ID = 2^-31
+            # 80: \x50 Request Push 1 byte: Push command id
+            $push_command_id = $this->_readByte();
+            list( , $payload ) = CSV::unserialize( $this->_readString() );
+            if ( !empty( $this->_pushRequestCallback ) && is_callable( $this->_pushRequestCallback ) ){
+                $this->{$this->_pushRequestCallback}( $push_command_id, $payload );
+            }
+
+            $end_flag = $this->_readByte();
+            # this flag can be set more than once
+            while ( $end_flag == 3 ) {
+                $this->_readInt();  # FAKE SESSION ID = 2^-31
+                $this->_readByte();  # 80: 0x50 Request Push
+                if ( !empty( $this->_pushRequestCallback ) && is_callable( $this->_pushRequestCallback ) ){
+                    $this->{$this->_pushRequestCallback}( $push_command_id, $payload );
+                }
+                $end_flag = $this->_readByte();
+            }
+
+        }
+
+    }
+
+    /**
+     * Default callback for received push Notices
+     *
+     * @param $command_id
+     * @param $payload
+     */
+    protected function _pushReceived( $command_id, $payload ){
+        # REQUEST_PUSH_RECORD	        79
+        # REQUEST_PUSH_DISTRIB_CONFIG	80
+        # REQUEST_PUSH_LIVE_QUERY	    81
+        # TODO: this logic must stay within Messages class here I just want to receive
+        # an object of something, like a new array of cluster.
+        # We should register a callback and then execute it
+        if ( $command_id == 80 && $payload ){
+            $this->_transport->setClusterMap(
+                ClusterMap::fromConfig( $this->_readString() )
+            ); # JSON WITH THE NEW CLUSTER CFG
         }
 
     }

src/PhpOrient/Protocols/Binary/Operations/RecordLoad.php

@@ -8,16 +8,10 @@
 use PhpOrient\Protocols\Common\Constants;
 use PhpOrient\Protocols\Binary\Data\ID;
 use PhpOrient\Protocols\Binary\Abstracts\NeedDBOpenedTrait;
-use Closure;
 
 class RecordLoad extends Operation {
     use NeedDBOpenedTrait;
 
-    /**
-     * @var Closure|string
-     */
-    public $_callback;
-
     /**
      * @var int The op code.
      */

tests/PhpOrient/PerformanceTest.php

@@ -17,12 +17,12 @@
 
 class PerformanceTest extends TestCase {
 
-    protected $db_name = 'stocazzo';
+    protected $db_name = 'emptiness_from_which_i_fed';
 
 
     public function testDateDeserialization() {
 
-        $this->markTestSkipped('Pupù');
+        $this->markTestSkipped('Nothing to do');
 
         $client = new PhpOrient( 'localhost', 2424 );
         $client->dbOpen( "GratefulDeadConcerts", 'admin', 'admin' );
@@ -36,7 +36,7 @@ public function testDateDeserialization() {
 
     public function testVersionProperty() {
 
-        $this->markTestSkipped('Pupù');
+        $this->markTestSkipped('Nothing to do');
 
         $recOrig = [ 'name' => 'foo', 'version' => '1.0.0' ];
         $rec = new Record();
@@ -54,7 +54,7 @@ public function testVersionProperty() {
 
     public function testRestrictedProperties() {
 
-        $this->markTestSkipped('Pupù');
+        $this->markTestSkipped('Nothing to do');
         $client = new PhpOrient( 'localhost', 2424 );
         $client->dbOpen( "GratefulDeadConcerts", 'admin', 'admin' );
         $result = $client->query('SELECT @rid, @class, @version FROM #11:0');