FIX #2471 RxLocalDocument.$ emits to often

Author: pubkey

CHANGELOG.md

@@ -2,8 +2,11 @@
 
 ### X.X.X (comming soon)
 
+Bugfixes:
+  - `RxLocalDocument.$` emitted to often on changes [#2471](https://github.com/pubkey/rxdb/issues/2471)
+
 Other:
-  - Refactored GraphQL replication to run faster [#2524](https://github.com/pubkey/rxdb/pull/2524/) Thanks [@corinv](https://github.com/corinv)
+  - Refactored GraphQL replication to run faster [#2524](https://github.com/pubkey/rxdb/pull/2524/)` Thanks [@corinv](https://github.com/corinv)
 
 ### 9.6.0 (7 September 2020) 
 

src/plugins/local-documents.ts

@@ -21,7 +21,9 @@ import {
     newRxTypeError
 } from '../rx-error';
 import {
-    clone, now, LOCAL_PREFIX
+    clone,
+    now,
+    LOCAL_PREFIX
 } from '../util';
 
 import type {
@@ -114,7 +116,9 @@ const RxLocalDocumentPrototype: any = {
         this: any,
         changeEvent: RxChangeEvent
     ) {
-        if (changeEvent.documentId !== this.primary) return;
+        if (changeEvent.documentId !== this.primary) {
+            return;
+        }
         switch (changeEvent.operation) {
             case 'UPDATE':
                 const newData = clone(changeEvent.documentData);
@@ -202,12 +206,10 @@ const RxLocalDocumentPrototype: any = {
             .then((res: any) => {
                 const endTime = now();
                 newData._rev = res.rev;
-                this._dataSync$.next(newData);
-
                 const changeEvent = new RxChangeEvent(
                     'UPDATE',
                     this.id,
-                    clone(this._data),
+                    clone(newData),
                     isRxDatabase(this.parent) ? this.parent.token : this.parent.database.token,
                     isRxCollection(this.parent) ? this.parent.name : null,
                     true,
@@ -294,8 +296,9 @@ RxLocalDocument.create = (id: string, data: any, parent: any) => {
  * throws if already exists
  */
 function insertLocal(this: any, id: string, data: any): Promise<RxLocalDocument> {
-    if (isRxCollection(this) && this._isInMemory)
+    if (isRxCollection(this) && this._isInMemory) {
         return this._parentCollection.insertLocal(id, data);
+    }
 
     data = clone(data);
 
@@ -326,15 +329,16 @@ function insertLocal(this: any, id: string, data: any): Promise<RxLocalDocument>
  * overwrites existing if exists
  */
 function upsertLocal(this: any, id: string, data: any): Promise<RxLocalDocument> {
-    if (isRxCollection(this) && this._isInMemory)
+    if (isRxCollection(this) && this._isInMemory) {
         return this._parentCollection.upsertLocal(id, data);
+    }
 
     return this.getLocal(id)
-        .then((existing: any) => {
+        .then((existing: RxDocument) => {
             if (!existing) {
                 // create new one
-                const doc = this.insertLocal(id, data);
-                return doc;
+                const docPromise = this.insertLocal(id, data);
+                return docPromise;
             } else {
                 // update existing
                 data._rev = existing._data._rev;

test/unit/local-documents.test.ts

@@ -1,5 +1,5 @@
 import assert from 'assert';
-import AsyncTestUtil from 'async-test-util';
+import AsyncTestUtil, { wait } from 'async-test-util';
 
 import * as humansCollection from '../helper/humans-collection';
 import * as schemas from '../helper/schemas';
@@ -103,17 +103,44 @@ config.parallel('local-documents.test.js', () => {
                 c.database.destroy();
             });
             it('should update when exists', async () => {
-                const c = await humansCollection.create();
+                const c = await humansCollection.create(0);
                 await c.upsertLocal('foobar', {
                     foo: 'bar'
                 });
                 const doc = await c.upsertLocal('foobar', {
                     foo: 'bar2'
                 });
+                await wait(1000);
                 assert.ok(doc);
                 assert.strictEqual(doc.get('foo'), 'bar2');
                 c.database.destroy();
             });
+            /**
+             * @link https://github.com/pubkey/rxdb/issues/2471
+             */
+            it('should invoke subscription once', async () => {
+                const c = await humansCollection.create();
+                const emitted: any[] = [];
+                const doc = await c.upsertLocal('foobar', {
+                    foo: 'barOne',
+                });
+                await wait(50);
+                const docSub = doc.$.subscribe(x => {
+                    emitted.push(x);
+                });
+                await c.upsertLocal('foobar', {
+                    foo: 'barTwo',
+                });
+
+                assert.strictEqual(emitted.length, 2);
+                // first 'barOne' is emitted because.$ is a BehaviorSubject
+                assert.strictEqual(emitted[0].foo, 'barOne');
+                // second after the change, barTwo is emitted
+                assert.strictEqual(emitted[1].foo, 'barTwo');
+
+                docSub.unsubscribe();
+                c.database.destroy();
+            });
         });
         describe('negative', () => { });
     });