Bolt driver

src/main/php/com/neo4j/BoltProtocol.class.php

@@ -0,0 +1,132 @@
+<?php namespace com\neo4j;
+
+use peer\Socket;
+use peer\URL;
+
+class BoltProtocol extends Protocol {
+  private $sock, $init, $serialization;
+
+  const EOR                 = "\x00\x00";
+  const PREAMBLE            = "\x60\x60\xb0\x17";
+
+  const INIT                = "\x01";
+  const ACK_FAILURE         = "\x0e";
+  const RESET               = "\x0f";
+  const RUN                 = "\x10";
+  const PULL_ALL            = "\x3f";
+  const NODE                = "\x4e";
+  const PATH                = "\x50";
+  const RELATIONSHIP        = "\x52";
+  const SUCCESS             = "\x70";
+  const RECORD              = "\x71";
+  const UNBOUNDRELATIONSHIP = "\x72";
+  const FAILURE             = "\x7f";
+  const IGNORE              = "\x7e";
+
+  /**
+   * Creates a new Neo4J graph connection
+   *
+   * @param  peer.URL $endpoint
+   */
+  public function __construct(URL $endpoint) {
+    $this->sock= new Socket($endpoint->getHost(), $endpoint->getPort(7687));
+    if ($user= $endpoint->getUser()) {
+      $this->init= ['scheme' => 'basic', 'principal' => $user, 'credentials' => $endpoint->getPassword()];
+    } else {
+      $this->init= ['scheme' => 'none'];
+    }
+    $this->serialization= new Serialization();
+  }
+
+  /** Sends a message */
+  private function send($signature, ... $args) {
+    $s= pack('ca', 0xb0 + sizeof($args), $signature);
+    foreach ($args as $arg) {
+      $s.= $this->serialization->serialize($arg);
+    }
+
+    for ($l= strlen($s), $o= 0; $o < $l; $o+= 65536) {
+      $p= min($l - $o, 65535);
+      $this->sock->write(pack('n', $p).substr($s, $o, $p));
+    }
+    $this->sock->write(self::EOR);
+  }
+
+  /** Receives one receive at a time */
+  private function receive() {
+    $r= '';
+    while (self::EOR !== ($length= $this->sock->readBinary(2))) {
+      $r.= $this->sock->readBinary(unpack('n', $length)[1]);
+    }
+
+    return $r;
+  }
+
+  /**
+   * Initialize communication
+   *
+   * @see    https://boltprotocol.org/v1/#handshake
+   * @return void
+   * @throws com.neo4j.UnexpectedResponse
+   */
+  private function init() {
+    $this->sock->write(self::PREAMBLE.pack('NNNN', 1, 0, 0, 0));
+    $protocol= unpack('N', $this->sock->readBinary(4));
+    if (0 === $protocol[1]) {
+      throw new UnexpectedResponse(['Protocol handshake failed, server does not support protocol version']);
+    }
+
+    $this->send(self::INIT, nameof($this), $this->init);
+    $answer= $this->receive();
+    if (self::SUCCESS !== $answer{1}) {
+      throw new CannotAuthenticate([$this->serialization->unserialize($answer)]);
+    }
+  }
+
+  /** @return [:var][] */
+  private function records() {
+    $records= [];
+    do {
+      $answer= $this->receive();
+      if (self::RECORD !== $answer{1}) break;
+      $records[]= $this->serialization->unserialize($answer);
+    } while (true);
+
+    return $records;
+  }
+
+  /**
+   * Commits multiple statements
+   *
+   * @param  [:var][] $payload
+   * @return [:var] Results
+   * @throws com.neo4j.UnexpectedResponse
+   */
+  public function commit($payload) {
+    if (!$this->sock->isConnected()) {
+      $this->sock->connect();
+      $this->init();
+    }
+
+    $r= ['results' => [], 'errors' => []];
+    foreach ($payload['statements'] as $s) {
+      $this->send(self::RUN, $s['statement'], isset($s['parameters']) ? $s['parameters'] : []);
+
+      $answer= $this->receive();
+      $offset= 2;
+      if (self::SUCCESS === $answer{1}) {
+        $this->send(self::PULL_ALL);
+        $r['results'][]= ['columns' => $this->serialization->unserialize($answer, $offset)['fields'], 'data' => $this->records()];        
+      } else if (self::FAILURE === $answer{1}) {
+        $this->send(self::ACK_FAILURE);
+        $this->receive();
+        $r['errors'][]= $this->serialization->unserialize($answer, $offset);        
+      } else {
+        $this->send(self::RESET);
+        $this->receive();
+        $r['errors'][]= ['Ignored'];
+      }
+    }
+    return $r;
+  }
+}

src/main/php/com/neo4j/CannotAuthenticate.class.php

@@ -0,0 +1,8 @@
+<?php namespace com\neo4j;
+
+/**
+ * Indicates auhentication with the server failed
+ */
+class CannotAuthenticate extends UnexpectedResponse {
+
+}

src/main/php/com/neo4j/Graph.class.php

@@ -1,12 +1,5 @@
 <?php namespace com\neo4j;
 
-use peer\http\HttpConnection;
-use peer\http\HttpRequest;
-use peer\http\RequestData;
-use text\json\Json;
-use text\json\Format;
-use text\json\StreamInput;
-
 /**
  * Neo4J interface using its HTTP API
  * 
@@ -15,40 +8,16 @@
  * @test  xp://com.neo4j.unittest.GraphTest
  */
 class Graph implements \lang\Value {
-  private $conn, $cypher, $json, $base;
+  private $protocol, $cypher;
 
   /**
    * Creates a new Neo4J graph connection
    *
-   * @param  string|peer.URL|peer.http.HttpConnection $endpoint
+   * @param  string|peer.URL|com.neo4j.Protocol $endpoint
    */
   public function __construct($endpoint) {
-    $this->conn= $endpoint instanceof HttpConnection ? $endpoint : new HttpConnection($endpoint);
+    $this->protocol= $endpoint instanceof Protocol ? $endpoint : Protocol::forEndpoint($endpoint);
     $this->cypher= new Cypher();
-    $this->json= Format::dense();
-    $this->base= rtrim($this->conn->getURL()->getPath(), '/');
-  }
-
-  /**
-   * Commits multiple statements using `transaction/commit` endpoint.
-   *
-   * @param  [:var][] $payload
-   * @return [:var] Results
-   */
-  protected function commit($payload) {
-    $req= $this->conn->create(new HttpRequest());
-    $req->setMethod('POST');
-    $req->setTarget($this->base.'/transaction/commit');
-    $req->setHeader('X-Stream', 'true');
-    $req->setHeader('Content-Type', 'application/json');
-    $req->setParameters(new RequestData(Json::of($payload, $this->json)));
-
-    $res= $this->conn->send($req);
-    if (200 !== $res->statusCode()) {
-      throw new QueryFailed(['Unexpected HTTP response status '.$res->statusCode()]);
-    }
-
-    return Json::read(new StreamInput($res->in()));
   }
 
   /**
@@ -103,7 +72,7 @@ public function execute($statements) {
       $list[]= is_array($statement) ? $statement : ['statement' => $statement];
     }
 
-    $response= $this->commit(['statements' => $list]);
+    $response= $this->protocol->commit(['statements' => $list]);
     if (empty($response['errors'])) {
       return $response['results'];
     } else {

src/main/php/com/neo4j/HttpProtocol.class.php

@@ -0,0 +1,58 @@
+<?php namespace com\neo4j;
+
+use peer\URL;
+use peer\http\HttpConnection;
+use peer\http\HttpRequest;
+use peer\http\RequestData;
+use text\json\Json;
+use text\json\Format;
+use text\json\StreamInput;
+use io\IOException;
+
+class HttpProtocol extends Protocol {
+  private $conn, $json, $base;
+
+  /**
+   * Creates a new Neo4J graph connection
+   *
+   * @param  peer.URL $endpoint
+   */
+  public function __construct(URL $endpoint) {
+    $this->conn= new HttpConnection($endpoint
+      ->setPort($endpoint->getPort(7474))
+      ->setPath($endpoint->getPath('/db/data'))
+    );
+    $this->json= Format::dense();
+    $this->base= rtrim($this->conn->getURL()->getPath(), '/');
+  }
+
+  /**
+   * Commits multiple statements using `transaction/commit` endpoint.
+   *
+   * @param  [:var][] $payload
+   * @return [:var] Results
+   * @throws com.neo4j.UnexpectedResponse
+   */
+  public function commit($payload) {
+    $req= $this->conn->create(new HttpRequest());
+    $req->setMethod('POST');
+    $req->setTarget($this->base.'/transaction/commit');
+    $req->setHeader('X-Stream', 'true');
+    $req->setHeader('Content-Type', 'application/json');
+    $req->setParameters(new RequestData(Json::of($payload, $this->json)));
+
+    try {
+      $res= $this->conn->send($req);
+    } catch (IOException $e) {
+      throw new QueryFailed(['I/O error'], $e);
+    }
+
+    if (200 === $res->statusCode()) {
+      return Json::read(new StreamInput($res->in()));
+    } else if (401 === $res->statusCode()) {
+      throw new CannotAuthenticate([Json::read(new StreamInput($res->in()))]);
+    } else {
+      throw new UnexpectedResponse(['Unexpected HTTP response status '.$res->statusCode(), $res->readData().'...']);
+    }
+  }
+}

src/main/php/com/neo4j/Protocol.class.php

@@ -0,0 +1,30 @@
+<?php namespace com\neo4j;
+
+use lang\IllegalArgumentException;
+use peer\URL;
+
+/**
+ * Protocol base class
+ *
+ * @test  xp://com.neo4j.unittest.ProtocolTest
+ */
+abstract class Protocol {
+
+  /**
+   * Factory method
+   *
+   * @param  peer.URL|string $endpoint
+   * @return self
+   */
+  public static function forEndpoint($endpoint) {
+    $url= $endpoint instanceof URL ? $endpoint : new URL($endpoint);
+    switch ($url->getScheme()) {
+      case 'http': case 'https': return new HttpProtocol($url);
+      case 'bolt': return new BoltProtocol($url);
+      default: throw new IllegalArgumentException('Unsupported protocol "'.$url->getScheme().'"');
+    }
+  }
+
+  /** Commits a payload and returns records */
+  public abstract function commit($payload);
+}

src/main/php/com/neo4j/QueryFailed.class.php

@@ -1,13 +1,8 @@
 <?php namespace com\neo4j;
 
-use lang\XPException;
-use lang\Throwable;
-use util\Objects;
+/**
+ * Indicates a Cypher query failed on the server
+ */
+class QueryFailed extends UnexpectedResponse {
 
-class QueryFailed extends XPException {
-
-  /** Creates a new instance */
-  public function __construct(array $errors, Throwable $cause= null) {
-    parent::__construct('Query failed '.Objects::stringOf($errors), $cause);
-  }
 }