WIP

src/Client/TextDocument.php

@@ -36,7 +36,7 @@ class TextDocument
      * @param Diagnostic[] $diagnostics
      * @return Promise <void>
      */
-    public function publishDiagnostics(string $uri, array $diagnostics): Promise
+    public function publishDiagnostics(string $uri, array $diagnostics): Observable
     {
         return $this->handler->notify('textDocument/publishDiagnostics', [
             'uri' => $uri,
@@ -51,13 +51,11 @@ class TextDocument
      * @param TextDocumentIdentifier $textDocument The document to get the content for
      * @return Promise <TextDocumentItem> The document's current content
      */
-    public function xcontent(TextDocumentIdentifier $textDocument): Promise
+    public function xcontent(TextDocumentIdentifier $textDocument): Observable
     {
         return $this->handler->request(
             'textDocument/xcontent',
             ['textDocument' => $textDocument]
-        )->then(function ($result) {
-            return $this->mapper->map($result, new TextDocumentItem);
-        });
+        );
     }
 }

src/Client/Window.php

@@ -30,7 +30,7 @@ class Window
      * @param string $message
      * @return Promise <void>
      */
-    public function showMessage(int $type, string $message): Promise
+    public function showMessage(int $type, string $message): Observable
     {
         return $this->handler->notify('window/showMessage', ['type' => $type, 'message' => $message]);
     }
@@ -42,7 +42,7 @@ class Window
      * @param string $message
      * @return Promise <void>
      */
-    public function logMessage(int $type, string $message): Promise
+    public function logMessage(int $type, string $message): Observable
     {
         return $this->handler->notify('window/logMessage', ['type' => $type, 'message' => $message]);
     }

src/Client/Workspace.php

@@ -33,15 +33,13 @@ class Workspace
      * Returns a list of all files in a directory
      *
      * @param string $base The base directory (defaults to the workspace)
-     * @return Promise <TextDocumentIdentifier[]> Array of documents
+     * @return Observable Emits JSON Patches that eventually result in TextDocumentIdentifier[]
      */
-    public function xfiles(string $base = null): Promise
+    public function xfiles(string $base = null): Observable
     {
         return $this->handler->request(
             'workspace/xfiles',
             ['base' => $base]
-        )->then(function (array $textDocuments) {
-            return $this->mapper->mapArray($textDocuments, [], TextDocumentIdentifier::class);
-        });
+        );
     }
 }

src/ClientHandler.php

@@ -3,8 +3,8 @@ declare(strict_types = 1);
 
 namespace LanguageServer;
 
-use AdvancedJsonRpc;
-use Sabre\Event\Promise;
+use AdvancedJsonRpc as JsonRpc;
+use Rx\Observable;
 
 class ClientHandler
 {
@@ -35,31 +35,38 @@ class ClientHandler
      *
      * @param string $method The method to call
      * @param array|object $params The method parameters
-     * @return Promise <mixed> Resolved with the result of the request or rejected with an error
+     * @return Observable Emits JSON Patch operations for the result
      */
-    public function request(string $method, $params): Promise
+    public function request(string $method, $params): Observable
     {
         $id = $this->idGenerator->generate();
-        return $this->protocolWriter->write(
-            new Protocol\Message(
-                new AdvancedJsonRpc\Request($id, $method, (object)$params)
-            )
-        )->then(function () use ($id) {
-            $promise = new Promise;
-            $listener = function (Protocol\Message $msg) use ($id, $promise, &$listener) {
-                if (AdvancedJsonRpc\Response::isResponse($msg->body) && $msg->body->id === $id) {
-                    // Received a response
-                    $this->protocolReader->removeListener('message', $listener);
-                    if (AdvancedJsonRpc\SuccessResponse::isSuccessResponse($msg->body)) {
-                        $promise->fulfill($msg->body->result);
-                    } else {
-                        $promise->reject($msg->body->error);
-                    }
+        return Observable::defer(function () {
+            return $this->protocolWriter->write(
+                new Protocol\Message(
+                    new AdvancedJsonRpc\Request($id, $method, (object)$params)
+                )
+            );
+        })
+            // Wait for completion
+            ->toArray()
+            // Subscribe to message events
+            ->flatMap(function () {
+                return observableFromEvent($this->protocolReader, 'message');
+            })
+            ->flatMap(function (JsonRpc\Message $msg) {
+                if (JsonRpc\Request::isRequest($msg->body) && $msg->body->method === '$/partialResult' && $msg->body->params->id === $id) {
+                    return Observable::fromArray($msg->body->params->patch)->map(function ($operation) {
+                        return Operation::fromDecodedJson($operation);
+                    });
                 }
-            };
-            $this->protocolReader->on('message', $listener);
-            return $promise;
-        });
+                if (AdvancedJsonRpc\Response::isResponse($msg->body) && $msg->body->id === $id) {
+                    if (AdvancedJsonRpc\SuccessResponse::isSuccessResponse($msg->body)) {
+                        return Observable::just(new Operation\Replace('/', $msg->body->result));
+                    }
+                    return Observable::error($msg->body->error);
+                }
+                return Observable::emptyObservable();
+            });
     }
 
     /**
@@ -67,9 +74,9 @@ class ClientHandler
      *
      * @param string $method The method to call
      * @param array|object $params The method parameters
-     * @return Promise <null> Will be resolved as soon as the notification has been sent
+     * @return Observable Will complete as soon as the notification has been sent
      */
-    public function notify(string $method, $params): Promise
+    public function notify(string $method, $params): Observable
     {
         $id = $this->idGenerator->generate();
         return $this->protocolWriter->write(

src/ContentRetriever/ClientContentRetriever.php

@@ -24,12 +24,12 @@ class ClientContentRetriever implements ContentRetriever
      * Retrieves the content of a text document identified by the URI through a textDocument/xcontent request
      *
      * @param string $uri The URI of the document
-     * @return Promise <string> Resolved with the content as a string
+     * @return Observable <string> Emits the content as a string
      */
-    public function retrieve(string $uri): Promise
+    public function retrieve(string $uri): Observable
     {
         return $this->client->textDocument->xcontent(new TextDocumentIdentifier($uri))
-            ->then(function (TextDocumentItem $textDocument) {
+            ->map(function (TextDocumentItem $textDocument) {
                 return $textDocument->text;
             });
     }

src/ContentRetriever/ContentRetriever.php

@@ -4,6 +4,7 @@ declare(strict_types = 1);
 namespace LanguageServer\ContentRetriever;
 
 use Sabre\Event\Promise;
+use Rx\Observable;
 
 /**
  * Interface for retrieving the content of a text document
@@ -14,7 +15,7 @@ interface ContentRetriever
      * Retrieves the content of a text document identified by the URI
      *
      * @param string $uri The URI of the document
-     * @return Promise <string> Resolved with the content as a string
+     * @return Observable <string> Emits the content as a string
      */
-    public function retrieve(string $uri): Promise;
+    public function retrieve(string $uri): Observable;
 }

src/ContentRetriever/FileSystemContentRetriever.php

@@ -3,7 +3,7 @@ declare(strict_types = 1);
 
 namespace LanguageServer\ContentRetriever;
 
-use Sabre\Event\Promise;
+use Rx\Observable;
 use function LanguageServer\uriToPath;
 
 /**
@@ -15,10 +15,10 @@ class FileSystemContentRetriever implements ContentRetriever
      * Retrieves the content of a text document identified by the URI from the file system
      *
      * @param string $uri The URI of the document
-     * @return Promise <string> Resolved with the content as a string
+     * @return Observable Emits the content as a string
      */
-    public function retrieve(string $uri): Promise
+    public function retrieve(string $uri): Observable
     {
-        return Promise\resolve(file_get_contents(uriToPath($uri)));
+        return Observable::just(file_get_contents(uriToPath($uri)));
     }
 }

src/FilesFinder/ClientFilesFinder.php

@@ -31,11 +31,14 @@ class ClientFilesFinder implements FilesFinder
      * If the client does not support workspace/files, it falls back to searching the file system directly.
      *
      * @param string $glob
-     * @return Promise <string[]> The URIs
+     * @return Observable Emits each URI
      */
-    public function find(string $glob): Promise
+    public function find(string $glob): Observable
     {
-        return $this->client->workspace->xfiles()->then(function (array $textDocuments) use ($glob) {
+        return $this->client->workspace->xfiles()
+            ->flatMap(function (Operation $operation) use ($glob) {
+
+
             $uris = [];
             foreach ($textDocuments as $textDocument) {
                 $path = Uri\parse($textDocument->uri)['path'];

src/JsonPatch/Operation.php

@@ -0,0 +1,17 @@
+<?php
+declare(strict_types = 1);
+
+namespace JsonPatch;
+
+class Operation
+{
+    /**
+     * @var Pointer
+     */
+    public $path;
+
+    public function __construct(Pointer $path)
+    {
+        $this->path = $path;
+    }
+}

src/JsonPatch/Operation/Add.php

@@ -0,0 +1,39 @@
+<?php
+declare(strict_types = 1);
+
+namespace JsonPatch\Operation;
+
+/**
+ * Adds a value to an object or inserts it into an array.
+ * In the case of an array the value is inserted before the given index.
+ * The - character can be used instead of an index to insert at the end of an array.
+ */
+class Add extends Operation
+{
+    /**
+     * @var mixed
+     */
+    public $value;
+
+    public function __construct(Pointer $path, $value)
+    {
+        parent::__construct($path);
+        $this->value = $value;
+    }
+
+    /**
+     * @param mixed $target
+     * @return void
+     */
+    public function apply(&$target)
+    {
+        if (is_array($this->path->parent->at($target))) {
+            // Numeric key
+            if ($this->path->key === 0) {
+                throw new \Exception('Cannot add before 0');
+            }
+            $this->path->parent->go($this->path->key - 1)->at($target) = $this->value;
+        }
+        $this->path->at($target);
+    }
+}

src/JsonPatch/Patch.php

@@ -0,0 +1,16 @@
+<?php
+declare(strict_types = 1);
+
+class Patch
+{
+    public function __construct(Operation[] $operations)
+    {
+
+    }
+
+    public function apply(mixed $target)
+    {
+
+    }
+}
+

src/JsonPatch/Pointer.php

@@ -0,0 +1,97 @@
+<?php
+declare(strict_types = 1);
+
+namespace JsonPatch;
+
+class Pointer
+{
+    /**
+     * @var self|null
+     */
+    public $parent;
+
+    /**
+     * The property name or array index. The root pointer has an empty key
+     *
+     * @var string
+     */
+    public $key;
+
+    /**
+     * @var string $pointer A full JSON Pointer string
+     * @return self
+     */
+    public static function parse(string $pointer)
+    {
+        $p = new self(null, '');
+        // TODO unescape
+        foreach (explode('/', $pointer) as $key) {
+            if ((string)(int)$key === $key) {
+                $key = (int)$key;
+            }
+            $p = new self($p, $key);
+        }
+        return $p;
+    }
+
+    /**
+     * @param self $parent
+     * @param string|number $key
+     */
+    public function __construct($parent, $key)
+    {
+        if (!is_int($key) && !is_string($key)) {
+            throw new \IllegalArgumentException('Key must be string or int');
+        }
+        $this->parent = $parent;
+        $this->key = $key;
+    }
+
+    /**
+     * Returns a reference to the value the pointer points to at the target
+     *
+     * @param object|array $target
+     */
+    public function &at($target)
+    {
+        if ($this->parent !== null) {
+            $target = $this->parent->at($target);
+        }
+        $key = $this->key;
+        if ($key === '') {
+            return $target;
+        }
+        if ($key === '-') {
+            if (!is_array($target)) {
+                throw new \Exception('Trying to apply "-" on a non-array');
+            }
+            $key = count($target);
+        }
+        if (is_array($target)) {
+            return $target[$key];
+        }
+        return &$target->$key;
+    }
+
+    /**
+     * @param string|int $key
+     * @return self
+     */
+    public function go($key)
+    {
+        if (!is_int($key) && !is_string($key)) {
+            throw new \IllegalArgumentException('Key must be string or int');
+        }
+        return new self($this, $key);
+    }
+
+    /**
+     * @return string
+     */
+    public function __toString()
+    {
+        if ($this->parent !== null && $this->parent->key !== '') {
+            return (string)($this->parent ?? '') . '/' . $this->key;
+        }
+    }
+}

src/LanguageServer.php

@@ -336,7 +336,10 @@ class LanguageServer extends JsonRpc\Dispatcher
         return coroutine(function () use ($rootPath) {
 
             $pattern = Path::makeAbsolute('**/*.php', $rootPath);
-            $uris = yield $this->filesFinder->find($pattern);
+            $this->filesFinder->find($pattern)
+                ->flatMap(function (Operation $op) {
+                    if ($op instanceof Operation\Add && ($op->getPath() === '/' || $op->getPath() )
+                });
 
             $count = count($uris);
 

src/PhpDocumentLoader.php

@@ -80,11 +80,11 @@ class PhpDocumentLoader
      * If the document is not open, loads it.
      *
      * @param string $uri
-     * @return Promise <PhpDocument>
+     * @return Observable PhpDocument
      */
-    public function getOrLoad(string $uri): Promise
+    public function getOrLoad(string $uri): Observable
     {
-        return isset($this->documents[$uri]) ? Promise\resolve($this->documents[$uri]) : $this->load($uri);
+        return isset($this->documents[$uri]) ? Observable::just($this->documents[$uri]) : $this->load($uri);
     }
 
     /**
@@ -93,14 +93,12 @@ class PhpDocumentLoader
      * The document is NOT added to the list of open documents, but definitions are registered.
      *
      * @param string $uri
-     * @return Promise <PhpDocument>
+     * @return Observable <PhpDocument>
      */
-    public function load(string $uri): Promise
+    public function load(string $uri): Observable
     {
-        return coroutine(function () use ($uri) {
-
+        return $this->contentRetriever->retrieve($uri)->map(function (string $content) {
             $limit = 150000;
-            $content = yield $this->contentRetriever->retrieve($uri);
             $size = strlen($content);
             if ($size > $limit) {
                 throw new ContentTooLargeException($uri, $size, $limit);

src/ProtocolStreamWriter.php

@@ -33,7 +33,7 @@ class ProtocolStreamWriter implements ProtocolWriter
     /**
      * {@inheritdoc}
      */
-    public function write(Message $msg): Promise
+    public function write(Message $msg): Observable
     {
         // if the message queue is currently empty, register a write handler.
         if (empty($this->messages)) {
@@ -42,12 +42,12 @@ class ProtocolStreamWriter implements ProtocolWriter
             });
         }
 
-        $promise = new Promise();
+        $subject = new Subject;
         $this->messages[] = [
             'message' => (string)$msg,
-            'promise' => $promise
+            'subject' => $subject
         ];
-        return $promise;
+        return $subject->asObservable();
     }
 
     /**
@@ -60,7 +60,7 @@ class ProtocolStreamWriter implements ProtocolWriter
         $keepWriting = true;
         while ($keepWriting) {
             $message = $this->messages[0]['message'];
-            $promise = $this->messages[0]['promise'];
+            $subject = $this->messages[0]['subject'];
 
             $bytesWritten = @fwrite($this->output, $message);
 
@@ -78,7 +78,7 @@ class ProtocolStreamWriter implements ProtocolWriter
                     $keepWriting = false;
                 }
 
-                $promise->fulfill();
+                $subject->onComplete();
             } else {
                 $this->messages[0]['message'] = $message;
                 $keepWriting = false;

src/ProtocolWriter.php

@@ -4,7 +4,7 @@ declare(strict_types = 1);
 namespace LanguageServer;
 
 use LanguageServer\Protocol\Message;
-use Sabre\Event\Promise;
+use Rx\Observable;
 
 interface ProtocolWriter
 {
@@ -12,7 +12,7 @@ interface ProtocolWriter
      * Sends a Message to the client
      *
      * @param Message $msg
-     * @return Promise Resolved when the message has been fully written out to the output stream
+     * @return Observable Resolved when the message has been fully written out to the output stream
      */
-    public function write(Message $msg): Promise;
+    public function write(Message $msg): Observable;
 }

src/Rx/Operator/ApplyJsonPatchesOperator.php

@@ -0,0 +1,50 @@
+<?php
+declare(strict_types = 1);
+
+namespace Rx\Operator;
+
+use gamringer\JSONPatch\Patch;
+
+class ApplyJsonPatchesOperator extends Operator
+{
+    private $classType;
+    private $isArray;
+    private $mapper;
+
+    public function __construct(JsonMapper $mapper, string $classType, bool $isArray = false)
+    {
+        $this->classType = $classType;
+        $this->isArray = $isArray;
+        $this->mapper = $mapper;
+    }
+
+    /**
+     * @param ObservableInterface $observable
+     * @param ObserverInterface $observer
+     * @param SchedulerInterface|null $scheduler
+     * @return \Rx\DisposableInterface
+     */
+    public function __invoke(ObservableInterface $observable, ObserverInterface $observer, SchedulerInterface $scheduler = null)
+    {
+        $result = null;
+        $pointer = new Pointer($result);
+
+        return $observable->subscribe(new CallbackObserver(
+            function (JsonPatch $patch) use ($pointer) {
+                $patch->apply($pointer);
+                
+                if ($this->isArray) {
+                    $result = [];
+                } else {
+                    $classType = $this->classType;
+                    $result = new $classType;
+                }
+            }),
+            [$observer, 'onError'],
+            function () use (&$result) {
+                $observer->onNext($result);
+                $observer->onComplete();
+            }
+        );
+    }
+}

src/Rx/Operator/BuildJsonPatchResultOperator.php

@@ -0,0 +1,32 @@
+<?php
+declare(strict_types = 1);
+
+namespace Rx\Operator;
+
+use gamringer\JSONPatch\Patch;
+
+class ApplyJsonPatchesOperator extends Operator
+{
+    /**
+     * @param ObservableInterface $observable
+     * @param ObserverInterface $observer
+     * @param SchedulerInterface|null $scheduler
+     * @return \Rx\DisposableInterface
+     */
+    public function __invoke(ObservableInterface $observable, ObserverInterface $observer, SchedulerInterface $scheduler = null)
+    {
+        $result = null;
+        $pointer = new Pointer($result);
+
+        return $observable->subscribe(new CallbackObserver(
+            function (JsonPatch $patch) use ($pointer) {
+                $patch->apply($pointer);
+            }),
+            [$observer, 'onError'],
+            function () use (&$result) {
+                $observer->onNext($result);
+                $observer->onComplete();
+            }
+        );
+    }
+}

src/Server/TextDocument.php

@@ -385,52 +385,53 @@ class TextDocument
      */
     public function xdefinition(TextDocumentIdentifier $textDocument, Position $position): Promise
     {
-        return coroutine(function () use ($textDocument, $position) {
-            $document = yield $this->documentLoader->getOrLoad($textDocument->uri);
-            $node = $document->getNodeAtPosition($position);
-            if ($node === null) {
-                return [];
-            }
-            // Handle definition nodes
-            while (true) {
-                if ($fqn) {
-                    $def = $this->index->getDefinition($definedFqn);
-                } else {
-                    // Handle reference nodes
-                    $def = $this->definitionResolver->resolveReferenceNodeToDefinition($node);
+        return $this->documentLoader->getOrLoad($textDocument->uri)
+            ->flatMap(function (PhpDocument $document) {
+                $node = $document->getNodeAtPosition($position);
+                if ($node === null) {
+                    return Observable::empty();
                 }
-                // If no result was found and we are still indexing, try again after the index was updated
-                if ($def !== null || $this->index->isComplete()) {
-                    break;
-                }
-                yield waitForEvent($this->index, 'definition-added');
-            }
-            if (
-                $def === null
-                || $def->symbolInformation === null
-                || Uri\parse($def->symbolInformation->location->uri)['scheme'] === 'phpstubs'
-            ) {
-                return [];
-            }
-            $symbol = new SymbolDescriptor;
-            foreach (get_object_vars($def->symbolInformation) as $prop => $val) {
-                $symbol->$prop = $val;
-            }
-            $symbol->fqsen = $def->fqn;
-            if (preg_match('/\/vendor\/([^\/]+\/[^\/]+)\//', $def->symbolInformation->location->uri, $matches) && $this->composerLock !== null) {
-                // Definition is inside a dependency
-                $packageName = $matches[1];
-                foreach ($this->composerLock->packages as $package) {
-                    if ($package->name === $packageName) {
-                        $symbol->package = $package;
+                // Handle definition nodes
+                while (true) {
+                    if ($fqn) {
+                        $def = $this->index->getDefinition($definedFqn);
+                    } else {
+                        // Handle reference nodes
+                        $def = $this->definitionResolver->resolveReferenceNodeToDefinition($node);
+                    }
+                    // If no result was found and we are still indexing, try again after the index was updated
+                    if ($def !== null || $this->index->isComplete()) {
                         break;
                     }
+                    yield waitForEvent($this->index, 'definition-added');
                 }
-            } else if ($this->composerJson !== null) {
-                // Definition belongs to a root package
-                $symbol->package = $this->composerJson;
-            }
-            return [new SymbolLocationInformation($symbol, $symbol->location)];
+                if (
+                    $def === null
+                    || $def->symbolInformation === null
+                    || Uri\parse($def->symbolInformation->location->uri)['scheme'] === 'phpstubs'
+                ) {
+                    return [];
+                }
+                $symbol = new SymbolDescriptor;
+                foreach (get_object_vars($def->symbolInformation) as $prop => $val) {
+                    $symbol->$prop = $val;
+                }
+                $symbol->fqsen = $def->fqn;
+                if (preg_match('/\/vendor\/([^\/]+\/[^\/]+)\//', $def->symbolInformation->location->uri, $matches) && $this->composerLock !== null) {
+                    // Definition is inside a dependency
+                    $packageName = $matches[1];
+                    foreach ($this->composerLock->packages as $package) {
+                        if ($package->name === $packageName) {
+                            $symbol->package = $package;
+                            break;
+                        }
+                    }
+                } else if ($this->composerJson !== null) {
+                    // Definition belongs to a root package
+                    $symbol->package = $this->composerJson;
+                }
+                return [new SymbolLocationInformation($symbol, $symbol->location)];
+            })
         });
     }
 }