feat: complete rewrite of result encoding algorithm

See rgraphql/magellan -> Design.md for the new design.

Author: paralin

package.json

@@ -6,12 +6,13 @@
   "main": "./lib/index.js",
   "types": "./lib/index.d.ts",
   "peerDependencies": {
-    "rgraphql": "^0.3.0",
+    "rgraphql": "^0.4.5",
     "rxjs": "^5.0.0"
   },
   "dependencies": {
-    "lodash": "^4.17.0",
     "graphql": "^0.9.0",
+    "lodash": "^4.17.0",
+    "lru_map": "^0.3.3",
     "rgraphql": "^0.3.0",
     "rxjs": "^5.0.0"
   },
@@ -22,6 +23,7 @@
     "lint": "tslint -c tslint.json --project tsconfig.json --type-check",
     "mocha": "ts-node node_modules/istanbul/lib/cli.js cover -e .ts  -x \"*.d.ts\" -x \"*.spec.ts\" test/run_tests.js",
     "mocha-nocover": "ts-node test/run_tests.js",
+    "mocha-nocover-debug": "ts-node debug test/run_tests.js",
     "semantic-release": "semantic-release pre && npm publish && semantic-release post"
   },
   "devDependencies": {

src/client.spec.ts

@@ -3,13 +3,18 @@ import { ITransport } from './transport';
 import {
   IRGQLServerMessage,
   IRGQLClientMessage,
+  CacheStrategy,
+  IRGQLValue,
+  RGQLValue,
+  Kind,
 } from 'rgraphql';
 import {
   parse,
 } from 'graphql';
 
 class MockTransport implements ITransport {
   public messageHandler: (mes: IRGQLServerMessage) => void;
+  private queryIdCtr = 1;
 
   public onMessage(cb: (mes: IRGQLServerMessage) => void) {
     this.messageHandler = cb;
@@ -18,6 +23,10 @@ class MockTransport implements ITransport {
   public send(msg: IRGQLClientMessage) {
     console.log(`Sending: ${JSON.stringify(msg)}`);
   }
+
+  public nextQueryId(): number {
+    return this.queryIdCtr++;
+  }
 }
 
 describe('SoyuzClient', () => {
@@ -49,24 +58,25 @@ query myQuery($age: Int) {
     });
     sub.subscribe((val) => {
       console.log(`Query returned value: ${JSON.stringify(val)}`);
-      if (val.data && val.data.allPeople && val.data.allPeople.length) {
+      if (val.data && val.data.allPeople && val.data.allPeople.length && val.data.allPeople[0].name === 'Test') {
         done();
       }
     });
     console.log('Setting transport.');
     client.setTransport(mt);
+    let batchValues: IRGQLValue[] = [
+      {queryNodeId: 1},
+      {arrayIndex: 1},
+      {queryNodeId: 2, value: {kind: Kind.PRIMITIVE_KIND_STRING, stringValue: 'Test'}},
+    ];
+    let encValues: Uint8Array[] = [];
+    for (let v of batchValues) {
+      encValues.push(RGQLValue.encode(v).finish());
+    }
     let msgs: IRGQLServerMessage[] = [
-      {mutateValue: {valueNodeId: 1, queryNodeId: 1, isArray: true}},
-      {mutateValue: {valueNodeId: 4, queryNodeId: 1, parentValueNodeId: 1, arrayIdx: 1}},
-      {
-        mutateValue: {
-          valueNodeId: 5,
-          parentValueNodeId: 4,
-          queryNodeId: 2,
-          valueJson: '"John"',
-          hasValue: true,
-        },
-      },
+      {valueInit: {queryId: 1, resultId: 1, cacheSize: 200, cacheStrategy: CacheStrategy.CACHE_LRU}},
+      {valueBatch: {resultId: 1, values: encValues}},
+      {valueFinalize: {resultId: 1}},
     ];
     for (let msg of msgs) {
       mt.messageHandler(msg);

src/client.ts

@@ -1,62 +1,73 @@
 import { QueryTreeNode } from './query-tree';
-import { ValueTreeNode } from './value-tree';
+import { ResultTree } from './result';
 import { ITransport } from './transport';
-import { ClientBus } from './client-bus';
+import { RunningQuery } from './running-query';
 import {
   ObservableQuery,
   IQueryOptions,
 } from './query';
 import {
-  Mutation,
   IMutationOptions,
 } from './mutation';
-import {
-  ISoyuzClientContext,
-  ISoyuzSerialOperation,
-} from './interfaces';
 import { parse, OperationDefinitionNode } from 'graphql';
 import { simplifyQueryAst } from './util/graphql';
 import { BehaviorSubject } from 'rxjs/BehaviorSubject';
+import { Subscription } from 'rxjs/Subscription';
 
 // Soyuz client.
 export class SoyuzClient {
-  private queryTree: QueryTreeNode;
-  private context = new BehaviorSubject<ISoyuzClientContext>(null);
+  // queryTree holds the global live query tree.
+  private queryTree: QueryTreeNode = new QueryTreeNode();
   private transportIdCounter = 0;
-
-  // Active serial operations in-flight.
-  private serialOperations: { [operationId: number]: ISoyuzSerialOperation } = {};
-  private serialOperationIdCounter = 0;
-
-  constructor() {
-    this.queryTree = new QueryTreeNode();
-    this.initHandlers();
-  }
+  private transport: ITransport;
+  private queries: { [id: number]: RunningQuery } = {};
+  private primaryQueryId: number;
+  private primaryResultTree: BehaviorSubject<ResultTree> = new BehaviorSubject<ResultTree>(null);
+  private transportSubs: Subscription[] = [];
 
   // Set transport causes the client to start using a new transport to talk to the server.
   // Pass null to stop using the previous transport.
   public setTransport(transport: ITransport) {
-    if (this.context.value && this.context.value.transport === transport) {
+    if (this.transport === transport) {
       return;
     }
 
+    if (this.transport) {
+      for (let queryId in this.queries) {
+        if (!this.queries.hasOwnProperty(queryId)) {
+          continue;
+        }
+        this.queries[queryId].dispose();
+      }
+      this.queries = {};
+      for (let sub of this.transportSubs) {
+        sub.unsubscribe();
+      }
+      this.transportSubs.length = 0;
+    }
+
     if (!transport) {
-      this.context.next(null);
       return;
     }
 
     let tid: number = this.transportIdCounter++;
-    let vtr = new ValueTreeNode(this.queryTree);
-    let clib = new ClientBus(transport, this.queryTree, vtr, this.serialOperations);
-    this.context.next({
-      transport: transport,
-      valueTree: vtr,
-      clientBus: clib,
-    });
+    // start the initial root query
+    let query = new RunningQuery(transport, this.queryTree, 'query');
+    this.transportSubs.push(query.resultTree.subscribe((tree) => {
+      this.primaryResultTree.next(tree);
+    }));
+    this.primaryQueryId = query.id;
+    this.queries = {};
+    this.queries[query.id] = query;
+    query.resultTree.subscribe({complete: () => {
+      if (this.queries[query.id] === query) {
+        delete this.queries[query.id];
+      }
+    }});
   }
 
   // Build a query against the system.
-  public query<T>(options: IQueryOptions): ObservableQuery<T> {
+  public query<T>(options: IQueryOptions): ObservableQuery {
     if (!options || !options.query) {
       throw new Error('You must specify a options object and query.');
     }
@@ -70,10 +81,7 @@ export class SoyuzClient {
     if (!odef) {
       throw new Error('Your provided query document did not contain a query definition.');
     }
-    return new ObservableQuery<T>(this.context,
-                                  this.queryTree,
-                                  odef,
-                                  options.variables);
+    return new ObservableQuery(this.primaryResultTree, this.queryTree, odef, options.variables);
   }
 
   // Execute a mutation against the system.
@@ -91,27 +99,29 @@ export class SoyuzClient {
     if (!odef) {
       throw new Error('Your provided mutation document did not contain a mutation definition.');
     }
-    let operationId = ++this.serialOperationIdCounter;
-    let mutation: ISoyuzSerialOperation = new Mutation<T>(operationId, this.context, odef, options.variables);
-    this.startSerialOperation(operationId, mutation);
-    return mutation.asPromise();
-  }
-
-  // startSerialOperation begins a already-built operation.
-  private startSerialOperation(id: number, operation: ISoyuzSerialOperation) {
-    this.serialOperations[id] = operation;
-    operation.init();
-  }
-
-  private initHandlers() {
-    let lastContext: ISoyuzClientContext;
-    this.context.subscribe((ctx) => {
-      if (lastContext) {
-        lastContext.valueTree.dispose();
-        lastContext.clientBus.dispose();
+    // start the query.
+    let qt = new QueryTreeNode();
+    let qr = new RunningQuery(this.transport, qt, 'mutation');
+    let uqr = qt.buildQuery(odef, options.variables || {});
+    this.queries[qr.id] = qr;
+    let lrt: any;
+    let data: any;
+    let rtsub = qr.resultTree.subscribe((rt) => {
+      if (!rt || rt === lrt) {
+        return;
       }
-
-      lastContext = ctx;
+      lrt = rt;
+      data = rt.addQuery(uqr.id, (id) => {});
+    });
+    return new Promise<T>((reject, resolve) => {
+      qr.resultTree.subscribe({
+        error: (err) => {
+          reject(err);
+        },
+        complete: () => {
+          resolve(data);
+        },
+      });
     });
   }
 }