ADD RxCollection.awaitPersistence()

Author: pubkey

CHANGELOG.md

@@ -13,6 +13,7 @@ Breaking:
 
 Features:
   - Added `RxDocument.atomicSet()`
+  - Added `RxCollection.awaitPersistence()` for in-memory-collections
 
 Bugfixes:
   - checkAdapter doesn't cleanup test databases [#714](https://github.com/pubkey/rxdb/issues/714)

docs-src/in-memory.md

@@ -1,6 +1,25 @@
-# in-memory
+# InMemory Collections
 
-When you do a heavy amount of operations on a `RxCollection`, you might want to optimize this by using the in-memory-replication of the collection. The in-memory-replication behaves equal to the original collection but is stored in the Ram of your computer.
+When you do a heavy amount of operations on a `RxCollection`, you might want to optimize this by using the in-memory-replication of the collection. The in-memory-replication behaves equal to the original collection but is stored in the Memory of your computer instead of the hard drive
+
+## Pros:
+
+- Faster queries
+- Faster writes
+- Querying works over encrypted fields
+
+## Cons:
+
+- The original collection has to be small enough to fit into the memory
+- No attachment-support
+- Initial creation takes longer (all data is loaded from disc into the memory)
+
+
+### encryption
+Encrypted fields are automatically decrypted inside of the memory-collection. This means you can do queries over encrypted fields.
+
+### replication
+The memory-collection is two-way-replicated with its original collection. This means when you change documents on one of them, the update and the change-event will fire on both.
 
 ## RxCollection().inMemory();
 
@@ -20,24 +39,18 @@ const docs = await memCol.find().exec(); // has same result as on the original c
 
 ```
 
-### encryption
-Encrypted fields are automatically decrypted inside of the memory-collection. This means you can do queries over encrypted fields.
+## RxCollection().awaitPersistence()
 
-### replication
-The memory-collection is two-way-replicated with its original collection. This means when you change documents on one of them, the update and the change-event will fire on both.
+When you do a write into the inMemoryCollection, it takes some time until the change is replicated at the parents collections.
+To know when you can be sure that all writes have been replicated, call `awaitPersistence()`
 
-## Pros:
-
-- Faster queries
-- Faster writes
-- Querying works over encrypted fields
-
-## Cons:
+```js
+const memCol = await myCollection.inMemory();
 
-- The original collection has to be small enough to fit into the memory
-- No attachment-support
-- Initial creation takes longer (all data is loaded from disc into the memory)
+await memCol.insert({foo: 'bar'});
 
+await memCol.awaitPersistence(); // after this you can be sure that everything is replicated
+```
 
 --------------------------------------------------------------------------------
 

orga/releases/8.0.0.md

@@ -39,4 +39,4 @@ Because the fields of an `RxDocument` are defined dynamically by the schema and
 Once per collection, a custom RxDocument-protoptype and constructor is created and each RxDocument of this collection is created with the constructor.
 
 ## Rewritten the inMemory-plugin
-The inMemory-plugin was written with some wrong estimations. I rewrote it and added much more tests.
+The inMemory-plugin was written with some wrong estimations. I rewrote it and added much more tests. Also `awaitPersistence()` can now be used to check if all writes have already been replicated into the parent collection.

src/plugins/in-memory.js

@@ -12,7 +12,9 @@ import {
 
 import {
     filter,
-    map
+    map,
+    mergeMap,
+    first
 } from 'rxjs/operators';
 
 import RxCollection from '../rx-collection';
@@ -70,6 +72,12 @@ export class InMemoryRxCollection extends RxCollection.RxCollection {
 
         this._observable$ = new Subject();
         this._changeEventBuffer = ChangeEventBuffer.create(this);
+
+        const parentProto = Object.getPrototypeOf(parentCollection);
+        this._oldPouchPut = parentProto._pouchPut.bind(this);
+
+        this._nonPersistentRevisions = new Set();
+        this._nonPersistentRevisionsSubject = new Subject(); // emits Set.size() when Set is changed
     }
 
     async prepare() {
@@ -103,14 +111,51 @@ export class InMemoryRxCollection extends RxCollection.RxCollection {
          * create an ongoing replications between both sides
          */
         const thisToParentSub = streamChangedDocuments(this)
-            .subscribe(doc => applyChangedDocumentToPouch(this._parentCollection, doc));
+            .pipe(
+                mergeMap(doc => applyChangedDocumentToPouch(this._parentCollection, doc)
+                    .then(() => doc._rev)
+                )
+            )
+            .subscribe(changeRev => {
+                this._nonPersistentRevisions.delete(changeRev);
+                this._nonPersistentRevisionsSubject.next(this._nonPersistentRevisions.size);
+            });
         this._subs.push(thisToParentSub);
 
         const parentToThisSub = streamChangedDocuments(this._parentCollection)
             .subscribe(doc => applyChangedDocumentToPouch(this, doc));
         this._subs.push(parentToThisSub);
     }
 
+    /**
+     * waits until all writes are persistent
+     * in the parent collection
+     * @return {Promise}
+     */
+    awaitPersistence() {
+        if (this._nonPersistentRevisions.size === 0) return Promise.resolve();
+        return this._nonPersistentRevisionsSubject.pipe(
+            filter(() => this._nonPersistentRevisions.size === 0),
+            first()
+        ).toPromise();
+    }
+
+    /**
+     * To know which events are replicated and which are not,
+     * the _pouchPut is wrapped
+     * @overwrite
+     */
+    async _pouchPut(obj, overwrite) {
+        console.log('new pouch put');
+        const ret = await this._oldPouchPut(obj, overwrite);
+        console.log('Ret');
+        console.dir(ret);
+
+        this._nonPersistentRevisions.add(ret.rev);
+        return ret;
+    }
+
+
     /**
      * @overwrite
      */

src/rx-collection.js

@@ -231,9 +231,11 @@ export class RxCollection {
     }
 
     /**
-     * [overwrite description]
+     * every write on the pouchdb
+     * is tunneld throught this function
      * @param {object} obj
      * @param {boolean} [overwrite=false] if true, it will overwrite existing document
+     * @return {Promise}
      */
     async _pouchPut(obj, overwrite = false) {
         obj = this._handleToPouch(obj);

test/unit/in-memory.test.js

@@ -451,6 +451,26 @@ config.parallel('in-memory.test.js', () => {
             col.database.destroy();
         });
     });
+    describe('.awaitPersistence()', () => {
+        it('should resolve the promise after some time', async () => {
+            const col = await humansCollection.create(0);
+            const memCol = await col.inMemory();
+
+            await memCol.awaitPersistence();
+
+            await memCol.insert(schemaObjects.simpleHuman());
+            await memCol.awaitPersistence();
+
+            const doc = await memCol.findOne().exec();
+            await doc.atomicSet('age', 6);
+            await memCol.awaitPersistence();
+
+            await doc.remove();
+            await memCol.awaitPersistence();
+
+            col.database.destroy();
+        });
+    });
     describe('other', () => {
         it('should work with many documents', async () => {
             const amount = 100;

typings/rx-collection.d.ts

@@ -104,7 +104,6 @@ export declare class RxCollection<RxDocumentType, OrmMethods = {}> {
     postRemove(fun: RxCollectionHookCallback<RxDocumentType, OrmMethods>, parallel: boolean): void;
     postCreate(fun: RxCollectionHookCallback<RxDocumentType, OrmMethods>): void;
 
-
     // migration
     migrationNeeded(): Promise<boolean>;
     migrate(batchSize: number): Observable<{
@@ -130,6 +129,9 @@ export declare class RxCollection<RxDocumentType, OrmMethods = {}> {
     upsertLocal(id: string, data: any): Promise<RxLocalDocument<RxCollection<RxDocumentType, OrmMethods>>>;
     getLocal(id: string): Promise<RxLocalDocument<RxCollection<RxDocumentType, OrmMethods>>>;
 
+    // only inMemory-collections
+    awaitPersistence(): Promise<void>;
+
     destroy(): Promise<boolean>;
     remove(): Promise<any>;
 }