Merge pull request #100 from ReactiveDB/fix/db-unobserve

fix(Selector): 避免在 db.observe 之前调用对应的 db.unobserve

Author: Brooooooklyn

src/storage/modules/Selector.ts

@@ -1,24 +1,31 @@
-import { Observable, Observer, OperatorFunction, from, asyncScheduler } from 'rxjs'
+import { Observable, OperatorFunction, concat, from, asyncScheduler } from 'rxjs'
 import {
-  filter,
   combineAll,
   debounceTime,
   map,
   mergeMap,
   publishReplay,
   reduce,
   refCount,
-  scan,
+  startWith,
   switchMap,
 } from 'rxjs/operators'
 import * as lf from 'lovefield'
 import * as Exception from '../../exception'
 import { predicatableQuery, graph } from '../helper'
-import { identity, forEach, assert, warn, isNonNullable } from '../../utils'
+import { identity, forEach, assert, warn } from '../../utils'
 import { PredicateProvider } from './PredicateProvider'
 import { ShapeMatcher, OrderInfo, StatementType } from '../../interface'
 import { mapFn } from './mapFn'
 
+const observeQuery = (db: lf.Database, query: lf.query.Select) => {
+  return new Observable<void>((observer) => {
+    const listener = () => observer.next()
+    db.observe(query, listener)
+    return () => db.unobserve(query, listener)
+  })
+}
+
 export class Selector<T> {
   private static concatFactory<U>(...metaDatas: Selector<U>[]) {
     const [meta] = metaDatas
@@ -105,24 +112,23 @@ export class Selector<T> {
     let { skip } = this
     skip = limit && !skip ? 0 : skip
 
-    const observeOn = (query: lf.query.Select) =>
-      Observable.create((observer: Observer<T[]>) => {
-        const listener = () => {
-          this.getValue(query)
-            .then((r) => observer.next(r as T[]))
-            .catch((e) => observer.error(e))
-        }
-        db.observe(query, listener)
-        listener()
-        return () => this.db.unobserve(query, listener)
-      }) as Observable<T[]>
+    const observeOn = (query: lf.query.Select) => {
+      const queryOnce = () => from(this.getValue(query))
+      // 下面的语句针对两个 lovefield issue 做了特殊调整：
+      // issue#209: 确保 db.observe 之后立即执行一次查询；
+      // issue#215: 确保 db.observe “不正确地”立即调用回调的行为，不会给消费方造成初始的重复推送。
+      return observeQuery(db, query).pipe(
+        startWith(void 0),
+        switchMap(queryOnce),
+      )
+    }
 
     const changesOnQuery =
       limit || skip
         ? this.buildPrefetchingObserve().pipe(switchMap((pks) => observeOn(this.getQuery(this.inPKs(pks)))))
         : observeOn(this.getQuery())
 
-    return lfIssueFix(changesOnQuery).pipe(
+    return changesOnQuery.pipe(
       publishReplay(1),
       refCount(),
     )
@@ -278,37 +284,9 @@ export class Selector<T> {
   }
 
   private buildPrefetchingObserve(): Observable<(string | number)[]> {
-    return Observable.create((observer: Observer<(string | number)[]>) => {
-      const { rangeQuery } = this
-      const listener = () => {
-        return rangeQuery
-          .exec()
-          .then((r) => {
-            observer.next(r.map((v) => v[this.shape.pk.name]))
-          })
-          .catch((e) => observer.error(e))
-      }
-
-      listener().then(() => {
-        this.db.observe(rangeQuery, listener)
-      })
-
-      return () => this.db.unobserve(rangeQuery, listener)
-    })
+    const { rangeQuery } = this
+    const queryOnce = () => from(rangeQuery.exec())
+    const update$ = observeQuery(this.db, rangeQuery).pipe(switchMap(queryOnce))
+    return concat(queryOnce(), update$).pipe(map((r) => r.map((v) => v[this.shape.pk.name])))
   }
 }
-
-/**
- * Lovefield observe 出来的推送，第一次和第二次在它们的值不为空
- * 的时候是重复的，这里做优化，省去重复；但不是简单的 skip(1)，因为
- * 那样会导致不能推出空结果集。详见：Lovefield issue#215
- */
-const lfIssueFix = <T>(changes: Observable<T[]>) => {
-  const doKeep = (prev: T[] | null, curr: T[] | null, idx: number) =>
-    idx === 1 && prev && prev.length && curr && curr.length ? null : curr
-
-  return changes.pipe(
-    scan(doKeep, null),
-    filter(isNonNullable),
-  )
-}

test/specs/storage/modules/Selector.spec.ts

@@ -1,4 +1,4 @@
-import { Subscription, of, defer, range, from, asyncScheduler, asapScheduler } from 'rxjs'
+import { Subscription, of, defer, range, from, asyncScheduler, asapScheduler, timer } from 'rxjs'
 import * as lf from 'lovefield'
 import { expect, use } from 'chai'
 import * as sinon from 'sinon'
@@ -199,7 +199,35 @@ export default describe('Selector test', () => {
   })
 
   describe('Selector.prototype.changes', () => {
-    it('observe should ok', (done) => {
+    it('should allow immediate unsubscribe', () => {
+      const selector = new Selector(
+        db,
+        db.select().from(table),
+        tableShape,
+        new PredicateProvider(table, { time: { $gte: 50 } }),
+      )
+
+      const signal = selector.changes().subscribe()
+
+      expect(() => signal.unsubscribe()).to.not.throw()
+    })
+
+    it('should allow immediate unsubscribe, for range query (with limit, skip)', () => {
+      const selector = new Selector(
+        db,
+        db.select().from(table),
+        tableShape,
+        new PredicateProvider(table, { time: { $gte: 50 } }),
+        10,
+        0,
+      )
+
+      const signal = selector.changes().subscribe()
+
+      expect(() => signal.unsubscribe()).to.not.throw()
+    })
+
+    it('observe should ok', function*() {
       const selector = new Selector(
         db,
         db.select().from(table),
@@ -209,18 +237,30 @@ export default describe('Selector test', () => {
 
       const newName = 'test name change'
 
-      subscription = selector
-        .changes()
-        .pipe(skip(1))
-        .subscribe((r: any[]) => {
-          expect(r[0].name).to.equal(newName)
-          done()
-        })
+      const signal = selector.changes()
+      subscription = signal.subscribe()
+
+      yield signal.pipe(
+        take(1),
+        subscribeOn(asapScheduler),
+        tap((r: any[]) => {
+          expect(r[0].name).to.equal('name:50') // before update
+        }),
+      )
 
-      db.update(table)
+      yield db
+        .update(table)
         .set(table['name'], newName)
         .where(table['_id'].eq('_id:50'))
         .exec()
+
+      yield signal.pipe(
+        take(1),
+        subscribeOn(asapScheduler),
+        tap((r: any[]) => {
+          expect(r[0].name).to.equal(newName) // after update
+        }),
+      )
     })
 
     it('observe asynchronous insertions completely', function*() {
@@ -307,14 +347,22 @@ export default describe('Selector test', () => {
 
       const newName = 'test name change'
 
-      const _subscription = selector.changes().subscribe(spy)
+      const _subscription = selector
+        .changes()
+        .pipe(subscribeOn(asapScheduler))
+        .subscribe(spy)
 
       yield db
         .update(table)
         .set(table['name'], newName)
         .where(table['_id'].eq('_id:50'))
         .exec()
 
+      yield timer(0) // 从 update 执行完成，到 changes 推出，是异步的，需要等待
+
+      const countFollowFirstUpdate = spy.callCount
+      expect(countFollowFirstUpdate).to.equal(1)
+
       _subscription.unsubscribe()
 
       yield db
@@ -323,7 +371,9 @@ export default describe('Selector test', () => {
         .where(table['_id'].eq('_id:50'))
         .exec()
 
-      expect(spy.callCount).to.equal(1)
+      yield timer(0) // 从 update 执行完成，到 changes 推出，是异步的，需要等待
+
+      expect(spy.callCount).to.equal(countFollowFirstUpdate)
     })
 
     it('predicate should be clone before use', function*() {