refactor(QueryToken): 令 traces() 返回流在重新订阅时照样推值

...满足 Observable 订阅类似于函数调用，二次调用不应该出现与一次调用不
同的情况，否则行为难以预知。将 trace 的概念从 lastEmit 中区分开来，作
为 QueryToken 级的变量，而不受订阅影响。

Author: chuan6

src/storage/modules/QueryToken.ts

@@ -1,5 +1,17 @@
 import { Observable, OperatorFunction, from } from 'rxjs'
-import { combineAll, filter, map, publishReplay, refCount, skipWhile, switchMap, take, tap } from 'rxjs/operators'
+import {
+  combineAll,
+  filter,
+  map,
+  pairwise,
+  publishReplay,
+  refCount,
+  skipWhile,
+  switchMap,
+  startWith,
+  take,
+  tap,
+} from 'rxjs/operators'
 import { Selector } from './Selector'
 import { ProxySelector } from './ProxySelector'
 import { assert } from '../../utils/assert'
@@ -14,7 +26,7 @@ function initialTraceResult<T>(list: ReadonlyArray<T>): TraceResult<T> {
   return {
     type: OpsType.Success,
     ops: list.map((_value, index) => ({ type: OpType.New, index })),
-    result: list
+    result: list,
   }
 }
 
@@ -28,20 +40,19 @@ export class QueryToken<T> {
   selector$: Observable<SelectorMeta<T>>
 
   private consumed = false
+  private lastEmit: ReadonlyArray<T> | undefined
+  private trace: ReadonlyArray<T> | undefined
 
-  constructor(
-    selector$: Observable<SelectorMeta<T>>,
-    private lastEmit?: ReadonlyArray<T>
-  ) {
+  constructor(selector$: Observable<SelectorMeta<T>>, trace?: ReadonlyArray<T>) {
     this.selector$ = selector$.pipe(
       publishReplay(1),
       refCount(),
     )
-    this.lastEmit = lastEmit
+    this.trace = trace
   }
 
-  setLastEmit(data: T[]) {
-    this.lastEmit = data
+  setTrace(data: T[]) {
+    this.trace = data
   }
 
   map<K>(fn: OperatorFunction<T[], K[]>) {
@@ -68,11 +79,14 @@ export class QueryToken<T> {
 
   traces(pk?: string): Observable<TraceResult<T>> {
     return this.changes().pipe(
-      map((result: T[]) => {
-        if (!this.lastEmit) {
+      startWith<undefined | ReadonlyArray<T>>(this.trace),
+      pairwise(),
+      map(([prev, curr]) => {
+        const result = curr!
+        if (!prev) {
           return initialTraceResult(result)
         }
-        const ops = diff(this.lastEmit, result, pk)
+        const ops = diff(prev, result, pk)
         return { result, ...ops }
       }),
       filter(({ type }) => type !== OpsType.ShouldSkip),

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

@@ -112,33 +112,17 @@ export default describe('QueryToken Testcase', () => {
           note: newNote,
         })
       })
-    })
 
-    describe('Method: traces', () => {
-      it('should get traces when updated', (done) => {
-        const task = tasks[0]
-        const newNote = 'new task note'
+      it('should emit result when the returning observable is re-subscribed', function*() {
+        const data$ = queryToken.traces('_id')
 
-        queryToken
-          .traces('_id')
-          .pipe(skip(1))
-          .subscribe((r) => {
-            const { result, type, ops } = r
-            expect(result[0].note).to.equal(newNote)
-            expect(type).to.equal(1)
-            ops.forEach((op: Op, index: number) => {
-              if (index === 0) {
-                expect(op.type).to.equal(1)
-              } else {
-                expect(op.type).to.equal(0)
-              }
-            })
-            done()
-          })
+        yield data$.pipe(take(1)).toPromise()
 
-        MockSelector.update(task._id as string, {
-          note: newNote,
-        })
+        let emittedOnResubscribe = false
+
+        yield data$.pipe(take(1)).subscribe(() => (emittedOnResubscribe = true))
+
+        expect(emittedOnResubscribe).to.be.true
       })
     })