1 files changed, 126 insertions, 24 deletions

diff --git a/extension/background/query.ts b/extension/background/query.ts
index bfe3102f3..1a61c66ca 100644
--- a/extension/background/query.ts
+++ b/extension/background/query.ts
@@ -1,6 +1,6 @@
 /*
  This file is part of TALER
- (C) 2015 GNUnet e.V.
+ (C) 2016 GNUnet e.V.
 
  TALER is free software; you can redistribute it and/or modify it under the
  terms of the GNU General Public License as published by the Free Software
@@ -16,50 +16,147 @@
 
 /// <reference path="../decl/chrome/chrome.d.ts" />
 
-"use strict";
 
+/**
+ * Database query abstractions.
+ * @module Query
+ * @author Florian Dold
+ */
+
+"use strict";
 
 function Query(db) {
   return new QueryRoot(db);
 }
 
-class QueryStream  {
-  qr: QueryRoot;
-  storeName;
-  constructor(qr, storeName) {
-    this.qr = qr;
-    this.storeName = storeName;
+
+abstract class QueryStreamBase {
+  abstract subscribe(f: (isDone: boolean, value: any) => void);
+  root: QueryRoot;
+
+  constructor(root: QueryRoot) {
+    this.root = root;
   }
-  join(indexName: string, key: any) {
+
+  indexJoin(storeName: string, indexName: string, key: any): QueryStreamBase {
     // join on the source relation's key, which may be
     // a path or a transformer function
-    throw Error("Not implemented");
+    return new QueryStreamIndexJoin(this, storeName, indexName, key);
+  }
+
+  filter(f: (any) => boolean): QueryStreamBase {
+    return new QueryStreamFilter(this, f);
   }
-  reduce(f, acc): Promise<any> {
+
+  reduce(f, acc?): Promise<any> {
     let leakedResolve;
     let p = new Promise((resolve, reject) => {
       leakedResolve = resolve;
     });
-    let qr = this.qr;
-    let storeName = this.storeName;
 
-    function doReduce() {
-      let req = qr.tx.objectStore(storeName).openCursor();
+    this.subscribe((isDone, value) => {
+      if (isDone) {
+        leakedResolve(acc);
+        return;
+      }
+      acc = f(value, acc);
+    });
+
+    return Promise.resolve().then(() => this.root.finish().then(() => p));
+  }
+}
+
+
+class QueryStreamFilter extends QueryStreamBase {
+  s: QueryStreamBase;
+  filterFn;
+
+  constructor(s: QueryStreamBase, filterFn) {
+    super(s.root);
+    this.s = s;
+    this.filterFn = filterFn;
+  }
+
+  subscribe(f) {
+    this.s.subscribe((isDone, value) => {
+      if (isDone) {
+        f(true, undefined);
+        return;
+      }
+      if (this.filterFn(value)) {
+        f(false, value)
+      }
+    });
+  }
+}
+
+
+class QueryStreamIndexJoin extends QueryStreamBase {
+  s: QueryStreamBase;
+  storeName;
+  key;
+  constructor(s, storeName: string, indexName: string, key: any) {
+    super(s.root);
+    this.s = s;
+    this.storeName = storeName;
+    this.key = key;
+  }
+
+  subscribe(f) {
+    this.s.subscribe((isDone, value) => {
+      if (isDone) {
+        f(true, undefined);
+        return;
+      }
+
+      let s = this.root.tx.objectStore(this.storeName);
+      let req = s.openCursor(IDBKeyRange.only(value));
+      req.onsuccess = () => {
+        let cursor = req.result;
+        if (cursor) {
+          f(false, [value, cursor.value]);
+          cursor.continue();
+        } else {
+          f(true, undefined);
+        }
+      }
+    });
+  }
+
+}
+
+
+class IterQueryStream extends QueryStreamBase {
+  private qr: QueryRoot;
+  private storeName;
+  private options;
+
+  constructor(qr, storeName, options?) {
+    super(qr);
+    this.qr = qr;
+    this.options = options;
+    this.storeName = storeName;
+  }
+
+  subscribe(f) {
+    function doIt() {
+      let s = this.qr.tx.objectStore(this.storeName);
+      let kr = undefined;
+      if (this.options && ("only" in this.options)) {
+        kr = IDBKeyRange.only(this.options.only);
+      }
+      let req = s.openCursor(kr);
       req.onsuccess = (e) => {
         let cursor: IDBCursorWithValue = req.result;
         if (cursor) {
-          acc = f(acc, cursor.value);
+          f(false, cursor.value);
           cursor.continue();
         } else {
-          leakedResolve(acc);
+          f(true, undefined);
         }
       }
     }
-
-    this.qr.work.push(doReduce);
-    // We need this one level of indirection so that the kickoff
-    // is run asynchronously.
-    return Promise.resolve().then(() => this.qr.finish().then(() => p));
+    this.qr.work.push(doIt.bind(this));
   }
 }
 
@@ -75,9 +172,14 @@ class QueryRoot {
     this.db = db;
   }
 
-  iter(storeName): QueryStream {
+  iter(storeName): QueryStreamBase {
+    this.stores.add(storeName);
+    return new IterQueryStream(this, storeName);
+  }
+
+  iterOnly(storeName, key): QueryStreamBase {
     this.stores.add(storeName);
-    return new QueryStream(this, storeName);
+    return new IterQueryStream(this, storeName, {only: key});
   }
 
   put(storeName, val): QueryRoot {