chore(dependency): 升级 rxjs 6

package.json

@@ -66,14 +66,14 @@
     "moment": "^2.18.1",
     "node-watch": "^0.5.8",
     "nyc": "^11.2.1",
-    "reactivedb": "~0.10.2",
+    "reactivedb": "~0.11.0",
     "rollup": "^0.60.5",
     "rollup-plugin-alias": "^1.3.1",
     "rollup-plugin-commonjs": "^8.4.1",
     "rollup-plugin-node-resolve": "^3.0.0",
     "rrule": "2.2.0",
-    "rxjs": "^5.4.3",
-    "rxjs-marbles": "^2.4.1",
+    "rxjs": "6",
+    "rxjs-marbles": "^4.3.1",
     "semver": "^5.5.0",
     "sinon": "^4.0.0",
     "sinon-chai": "^2.14.0",

src/Net/Http.ts

@@ -1,12 +1,6 @@
-import 'rxjs/add/observable/throw'
-import 'rxjs/add/observable/dom/ajax'
-import 'rxjs/add/observable/empty'
-import 'rxjs/add/operator/catch'
-import 'rxjs/add/operator/map'
-import { AjaxError } from 'rxjs/observable/dom/AjaxObservable'
-import { Observable } from 'rxjs/Observable'
-import { Observer } from 'rxjs/Observer'
-import { Subject } from 'rxjs/Subject'
+import { empty, throwError, Observable, Observer, Subject } from 'rxjs'
+import { ajax, AjaxError } from 'rxjs/ajax'
+import { map, catchError, publishReplay, refCount } from 'rxjs/operators'
 import { parseHeaders } from '../utils/index'
 import { testable } from '../testable'
 import { forEach } from '../utils'
@@ -40,9 +34,9 @@ const rxAjaxDefaultHeaderKey2NormKey = {
 }
 
 /**
- * Observable.ajax 目前的实现，对请求头字段的已有设置检查没有遵循
+ * rxjs ajax 目前的实现，对请求头字段的已有设置检查没有遵循
  * 头字段 key 不区分大小写的原则，比如：如果用户已經设置 `content-type`，
- * Observable.ajax 内部会发现 `Content-Type` 没有设置，结果会
+ * rxjs ajax 内部会发现 `Content-Type` 没有设置，结果会
  * 额外添加一个 `Content-Type` 字段，结果导致浏览器发现请求头字段里
  * 既有 `content-type` 又有 `Content-Type`，出现问题。
  */
@@ -63,36 +57,38 @@ export const createMethod = (method: AllowedHttpMethod) => (params: MethodParams
   /* istanbul ignore if */
   if (testable.UseXMLHTTPRequest && typeof window !== 'undefined') {
     coverRxAjaxHeadersBug(_opts.headers)
-    return Observable.ajax({
+    return ajax({
       url, body, method,
       headers: _opts.headers,
       withCredentials: _opts.credentials === 'include',
       responseType: _opts.responseType || 'json',
       crossDomain: typeof _opts.crossDomain !== 'undefined' ? !!_opts.crossDomain : true
     })
-      .map(value => {
-        const respBody = value.response
-        if (!includeHeaders) {
-          return respBody
-        }
-        const respHeaders = parseHeaders(value.xhr.getAllResponseHeaders())
-        return { headers: respHeaders, body: respBody }
-      })
-      .catch((e: AjaxError) => {
-        const headers = e.xhr.getAllResponseHeaders()
-        const errorResponse = new Response(new Blob([JSON.stringify(e.xhr.response)]), {
-          status: e.xhr.status,
-          statusText: e.xhr.statusText,
-          headers: headers.length ? parseHeaders(headers) : new Headers()
+      .pipe(
+        map(value => {
+          const respBody = value.response
+          if (!includeHeaders) {
+            return respBody
+          }
+          const respHeaders = parseHeaders(value.xhr.getAllResponseHeaders())
+          return { headers: respHeaders, body: respBody }
+        }),
+        catchError((e: AjaxError) => {
+          const headers = e.xhr.getAllResponseHeaders()
+          const errorResponse = new Response(new Blob([JSON.stringify(e.xhr.response)]), {
+            status: e.xhr.status,
+            statusText: e.xhr.statusText,
+            headers: headers.length ? parseHeaders(headers) : new Headers()
+          })
+          const requestInfo = { method, url, body }
+          const errorResponseClone = errorResponse.clone()
+
+          setTimeout(() => {
+            errorAdapter$.next({ ...requestInfo, error: errorResponseClone })
+          }, 10)
+          return throwError({ ...requestInfo, error: errorResponse })
         })
-        const requestInfo = { method, url, body }
-        const errorResponseClone = errorResponse.clone()
-
-        setTimeout(() => {
-          errorAdapter$.next({ ...requestInfo, error: errorResponseClone })
-        }, 10)
-        return Observable.throw({ ...requestInfo, error: errorResponse })
-      })
+      )
   } else { // 测试用分支
     return Observable.create((observer: Observer<any>) => {
       const _options = {
@@ -226,13 +222,13 @@ export class Http<T> {
   }
 
   send(): Observable<T> {
-    return this.request ? this.mapFn(this.request) : Observable.empty()
+    return this.request ? this.mapFn(this.request) : empty()
   }
 
   clone() {
     const result = new Http<T>(this.url, this.errorAdapter$)
     if (!this.cloned && this.request) {
-      this.request = this.request.publishReplay(1).refCount()
+      this.request = this.request.pipe(publishReplay(1), refCount())
       this.cloned = true
       result.cloned = true
     }

src/Net/Net.ts

@@ -1,13 +1,5 @@
-import 'rxjs/add/observable/forkJoin'
-import 'rxjs/add/observable/of'
-import 'rxjs/add/operator/concatAll'
-import 'rxjs/add/operator/do'
-import 'rxjs/add/operator/mapTo'
-import 'rxjs/add/operator/mergeMap'
-import 'rxjs/add/operator/switchMap'
-import 'rxjs/add/operator/filter'
-import { Observable } from 'rxjs/Observable'
-import { BehaviorSubject } from 'rxjs/BehaviorSubject'
+import { Observable, BehaviorSubject, of, from } from 'rxjs'
+import { concatAll, concatMap, filter, mapTo, mergeMap, switchMap, tap } from 'rxjs/operators'
 import { QueryToken, SelectorMeta, ProxySelector } from 'reactivedb/proxy'
 import { JoinMode } from 'reactivedb/interface'
 import { Database, Query, Predicate, ExecutorResult } from 'reactivedb'
@@ -155,25 +147,24 @@ export class Net {
     const hasPaddingFunction = typeof padding === 'function'
     const pk = this.primaryKeys.get(tableName)
 
-    const fn = (stream$: Observable<T[]>) =>
-      stream$.switchMap(data => !data.length
-        ? Observable.of(data)
-        : Observable.forkJoin(
-          Observable.from(data)
-            .mergeMap(datum => {
-              if (!hasRequiredFields || !hasPaddingFunction || !pk ||
-                required!.every(k => typeof datum[k] !== 'undefined')
-              ) {
-                return Observable.of(datum)
-              }
-              const patch = padding!(datum[pk]).filter(r => r != null) as Observable<T>
-              return patch
-                .concatMap(r => this.database!.upsert(tableName, r).mapTo(r))
-                .do(r => Object.assign(datum, r))
-            })
-        )
-          .mapTo(data)
-      )
+    const fn = switchMap<T[], T[]>(data => !data.length
+      ? of(data)
+      : from(data).pipe(
+        mergeMap(datum => {
+          if (!hasRequiredFields || !hasPaddingFunction || !pk ||
+              required!.every(k => typeof datum[k] !== 'undefined')
+             ) {
+            return of(datum)
+          }
+          const patch = padding!(datum[pk])
+          return patch.pipe(
+            filter((r): r is T => r != null),
+            concatMap(r => this.database!.upsert(tableName, r).pipe(mapTo(r))),
+            tap(r => Object.assign(datum, r))
+          )
+        })
+      ).pipe(mapTo(data))
+    )
     fn.toString = () => 'SDK_VALIDATE'
     return fn
   }
@@ -232,25 +223,24 @@ export class Net {
 
     const database = this.database!
 
-    const { request, method, tableName } = result as CUDApiResult<T>
-    let destination: Observable<ExecutorResult> | Observable<T | T[]>
-    return request
-      .concatMap(v => {
-        switch (method) {
-          case 'create':
-            destination = database.upsert<T>(tableName, v)
-            break
-          case 'update':
-            destination = database.upsert(tableName, v)
-            break
-          case 'delete':
-            destination = database.delete<T>(tableName, (result as UDResult<T>).clause)
-            break
-          default:
-            throw new Error()
-        }
-        return destination.mapTo<ExecutorResult | T | T[], T>(v)
-      })
+    const { request, method, tableName } = result
+    let destination: Observable<ExecutorResult>
+    return request.pipe(concatMap(v => {
+      switch (method) {
+        case 'create':
+          destination = database.upsert<T>(tableName, v)
+          break
+        case 'update':
+          destination = database.upsert(tableName, v)
+          break
+        case 'delete':
+          destination = database.delete<T>(tableName, (result as UDResult<T>).clause)
+          break
+        default:
+          throw new Error()
+      }
+      return destination.pipe(mapTo(v))
+    }))
   }
 
   persist(database: Database) {
@@ -277,12 +267,11 @@ export class Net {
           const token = this.handleRequestCache(v.realSelectorInfo)
           const selector$ = token.selector$
 
-          p = selector$
-            .do({
-              next(selector) {
-                cacheControl$.next(selector)
-              }
-            })
+          p = selector$.pipe(tap({
+            next(selector) {
+              cacheControl$.next(selector)
+            }
+          }))
           break
         default:
           break
@@ -295,12 +284,15 @@ export class Net {
 
     this.persistedDataBuffer.length = 0
 
-    return Observable.from(asyncQueue).concatAll().do({
-      error: async (err: Observable<Error>) => {
-        const errObj = await err.toPromise()
-        SDKLogger.error(errObj.message)
-      }
-    })
+    return from(asyncQueue).pipe(
+      concatAll(),
+      tap({
+        error: async (err: Observable<Error>) => {
+          const errObj = await err.toPromise()
+          SDKLogger.error(errObj.message)
+        }
+      })
+    )
   }
 
   bufferResponse<T>(result: ApiResult<T, CacheStrategy>) {
@@ -323,23 +315,22 @@ export class Net {
 
   bufferCUDResponse<T>(result: CUDApiResult<T>) {
     const { request, method, tableName } = result as CUDApiResult<T>
-    return request
-      .do((v: T | T[]) => {
+    return request.pipe(tap((v: T | T[]) => {
       this.persistedDataBuffer.push({
         kind: 'CUD',
         tableName,
         method: (method === 'create' || method === 'update') ? 'upsert' : method,
         value: method === 'delete' ? (result as UDResult<T>).clause : v
       })
-    })
+    }))
   }
 
   bufferSocketPush(socketMessage: ParsedWSMsg) {
     this.persistedDataBuffer.push({
       kind: 'SocketCUD',
       socketMessage
     })
-    return Observable.of(null)
+    return of(null)
   }
 
   private genCacheKey<T>(tableName: string, q: Readonly<Query<T>>) {
@@ -366,20 +357,22 @@ export class Net {
       case CacheStrategy.Request:
         if (!requestCache) {
           /*tslint:disable no-shadowed-variable*/
-          const selector$ = response$
-            .concatMap(v => database.upsert(tableName, v))
-            .do(() => this.requestMap.set(cacheKey, true))
-            .concatMap(() => dbGetWithSelfJoinEnabled<T>(database, tableName, q).selector$)
+          const selector$ = response$.pipe(
+            concatMap(v => database.upsert(tableName, v)),
+            tap(() => this.requestMap.set(cacheKey, true)),
+            concatMap(() => dbGetWithSelfJoinEnabled<T>(database, tableName, q).selector$)
+          )
           token = new QueryToken(selector$)
         } else {
           token = dbGetWithSelfJoinEnabled<T>(database, tableName, q)
         }
         token.map(this.validate(result))
         break
       case CacheStrategy.Cache:
-        const selector$ = response$
-          .concatMap(v => database.upsert(tableName, v))
-          .concatMap(() => dbGetWithSelfJoinEnabled<T>(database, tableName, q).selector$)
+        const selector$ = response$.pipe(
+          concatMap(v => database.upsert(tableName, v)),
+          concatMap(() => dbGetWithSelfJoinEnabled<T>(database, tableName, q).selector$)
+        )
         return new QueryToken(selector$)
       default:
         throw new TypeError('unreachable code path')

src/Net/Pagination.ts

@@ -1,9 +1,5 @@
-import 'rxjs/add/observable/throw'
-import 'rxjs/add/operator/startWith'
-import 'rxjs/add/operator/withLatestFrom'
-import { Observable } from 'rxjs/Observable'
-import { Observer } from 'rxjs/Observer'
-import { OperatorFunction } from 'rxjs/interfaces'
+import { empty, Observable, Observer, OperatorFunction, throwError } from 'rxjs'
+import { startWith, mergeAll, map, tap, catchError, finalize } from 'rxjs/operators'
 
 export type PageToken = string & { kind: 'PageToken' }
 
@@ -72,11 +68,14 @@ export const accumulateResultByConcat = <T>(state: State<T>, resp: OriginalRespo
 export const loadAndExpand = <T>(
   step: (curr: State<T>) => Observable<OriginalResponse<T>>,
   initState: State<T>,
-  loadMore$: Observable<{}> = Observable.empty()
+  loadMore$: Observable<{}> = empty()
 ): Observable<State<T>> => {
-  return loadMore$.startWith({})
-    .pipe(expand(step, accumulateResultByConcat, initState))
-    .mergeAll()
+  return loadMore$
+    .pipe(
+      startWith({}),
+      expand(step, accumulateResultByConcat, initState),
+      mergeAll()
+    )
 }
 
 export const expand = <T>(
@@ -99,10 +98,12 @@ export const expand = <T>(
         if (!isLoading) {
           isLoading = true
           observer.next(step(state)
-            .map((stepResult) => accumulator(state, stepResult))
-            .do((expanded) => Object.assign(state, expanded))
-            .catch((err) => Observable.throw(err))
-            .finally(() => { isLoading = false })
+            .pipe(
+              map((stepResult) => accumulator(state, stepResult)),
+              tap((expanded) => Object.assign(state, expanded)),
+              catchError((err) => throwError(err)),
+              finalize(() => { isLoading = false })
+            )
           )
         }
       },

src/SDK.ts

@@ -1,4 +1,4 @@
-import { Observable } from 'rxjs/Observable'
+import { Observable } from 'rxjs'
 import { Database } from 'reactivedb'
 import { Net } from './Net'
 import { forEach } from './utils'