Use RxJS

config/rollup.config.js

@@ -21,7 +21,8 @@ const globals = {
   '@wry/equality': 'wryEquality',
   graphql: 'graphql',
   react: 'React',
-  'zen-observable': 'Observable',
+  rxjs: 'rxjs',
+  'rxjs/operators': 'rxjs.operators',
 };
 
 const hasOwn = Object.prototype.hasOwnProperty;

package.json

@@ -61,15 +61,13 @@
     }
   },
   "dependencies": {
-    "@types/zen-observable": "^0.8.0",
     "@wry/equality": "^0.1.9",
     "fast-json-stable-stringify": "^2.0.0",
     "graphql-tag": "^2.10.1",
     "optimism": "^0.11.3",
-    "symbol-observable": "^1.2.0",
+    "rxjs": "^6.5.4",
     "ts-invariant": "^0.4.4",
-    "tslib": "^1.10.0",
-    "zen-observable": "^0.8.14"
+    "tslib": "^1.10.0"
   },
   "devDependencies": {
     "@testing-library/react": "9.1.4",
@@ -97,7 +95,6 @@
     "rollup-plugin-invariant": "0.5.6",
     "rollup-plugin-node-resolve": "5.2.0",
     "rollup-plugin-terser": "5.1.3",
-    "rxjs": "6.5.3",
     "ts-jest": "24.0.2",
     "tsc-watch": "3.0.1",
     "typescript": "3.7.4"

src/core/QueryManager.ts

@@ -1,5 +1,7 @@
 import { ExecutionResult, DocumentNode } from 'graphql';
 import { invariant, InvariantError } from 'ts-invariant';
+import {of} from 'rxjs';
+import {map} from 'rxjs/operators';
 
 import { ApolloLink } from '../link/core/ApolloLink';
 import { execute } from '../link/core/execute';
@@ -986,7 +988,7 @@ export class QueryManager<TStore> {
         {},
         variables,
         false,
-      ).map(result => {
+      ).pipe(map(result => {
         if (!fetchPolicy || fetchPolicy !== 'no-cache') {
           // the subscription interface should handle not sending us results we no longer subscribe to.
           // XXX I don't think we ever send in an object with errors, but we might in the future...
@@ -1009,7 +1011,7 @@ export class QueryManager<TStore> {
         }
 
         return result;
-      });
+      }));
 
     if (this.transform(query).hasClientExports) {
       const observablePromise = this.localState.addExportedVariables(
@@ -1194,7 +1196,7 @@ export class QueryManager<TStore> {
         observable = multiplex(execute(link, operation) as Observable<FetchResult<T>>);
       }
     } else {
-      observable = Observable.of({ data: {} } as FetchResult<T>);
+      observable = of({ data: {} } as FetchResult<T>);
       context = this.prepareContext(context);
     }
 
@@ -1250,7 +1252,7 @@ export class QueryManager<TStore> {
         });
       };
 
-      const subscription = observable.map((result: ExecutionResult) => {
+      const subscription = observable.pipe(map((result: ExecutionResult) => {
         if (requestId >= this.getQuery(queryId).lastRequestId) {
           this.markQueryResult(
             queryId,
@@ -1298,7 +1300,7 @@ export class QueryManager<TStore> {
             resultFromStore = result;
           }
         }
-      }).subscribe({
+      })).subscribe({
         error(error: ApolloError) {
           cleanup();
           reject(error);

src/link/core/ApolloLink.ts

@@ -1,6 +1,7 @@
 import { InvariantError, invariant } from 'ts-invariant';
 
 import { Observable } from '../../utilities/observables/Observable';
+import { of } from 'rxjs'
 import {
   NextLink,
   Operation,
@@ -13,7 +14,7 @@ import { createOperation } from '../utils/createOperation';
 import { transformOperation } from '../utils/transformOperation';
 
 function passthrough(op: Operation, forward: NextLink) {
-  return forward ? forward(op) : Observable.of();
+  return forward ? forward(op) : of();
 }
 
 function toLink(handler: RequestHandler | ApolloLink) {
@@ -34,7 +35,7 @@ class LinkError extends Error {
 
 export class ApolloLink {
   public static empty(): ApolloLink {
-    return new ApolloLink(() => Observable.of());
+    return new ApolloLink(() => of());
   }
 
   public static from(links: ApolloLink[]): ApolloLink {
@@ -53,14 +54,14 @@ export class ApolloLink {
     if (isTerminating(leftLink) && isTerminating(rightLink)) {
       return new ApolloLink(operation => {
         return test(operation)
-          ? leftLink.request(operation) || Observable.of()
-          : rightLink.request(operation) || Observable.of();
+          ? leftLink.request(operation) || of()
+          : rightLink.request(operation) || of();
       });
     } else {
       return new ApolloLink((operation, forward) => {
         return test(operation)
-          ? leftLink.request(operation, forward) || Observable.of()
-          : rightLink.request(operation, forward) || Observable.of();
+          ? leftLink.request(operation, forward) || of()
+          : rightLink.request(operation, forward) || of();
       });
     }
   }
@@ -75,7 +76,7 @@ export class ApolloLink {
           operation.context,
           transformOperation(validateOperation(operation)),
         ),
-      ) || Observable.of()
+      ) || of()
     );
   }
 
@@ -100,15 +101,15 @@ export class ApolloLink {
         operation =>
           firstLink.request(
             operation,
-            op => nextLink.request(op) || Observable.of(),
-          ) || Observable.of(),
+            op => nextLink.request(op) || of(),
+          ) || of(),
       );
     } else {
       return new ApolloLink((operation, forward) => {
         return (
           firstLink.request(operation, op => {
-            return nextLink.request(op, forward) || Observable.of();
-          }) || Observable.of()
+            return nextLink.request(op, forward) || of();
+          }) || of()
         );
       });
     }

src/react/types/types.ts

@@ -1,7 +1,7 @@
 import { ReactNode } from 'react';
 import { DocumentNode } from 'graphql';
 
-import { Observable } from '../../utilities/observables/Observable';
+import { Observable, Subscription } from '../../utilities/observables/Observable';
 import { FetchResult } from '../../link/core/types';
 import { ApolloClient } from '../../ApolloClient';
 import {
@@ -115,7 +115,7 @@ export interface QueryPreviousData<TData, TVariables> {
 
 export interface QueryCurrentObservable<TData, TVariables> {
   query?: ObservableQuery<TData, TVariables> | null;
-  subscription?: ZenObservable.Subscription;
+  subscription?: Subscription;
 }
 
 export interface QueryLazyOptions<TVariables> {
@@ -245,5 +245,5 @@ export interface SubscriptionDataOptions<
 
 export interface SubscriptionCurrentObservable {
   query?: Observable<any>;
-  subscription?: ZenObservable.Subscription;
+  subscription?: Subscription;
 }

src/utilities/observables/Observable.ts

@@ -1,21 +1 @@
-import Observable from 'zen-observable';
-
-// This simplified polyfill attempts to follow the ECMAScript Observable
-// proposal (https://github.com/zenparsing/es-observable)
-import 'symbol-observable';
-
-export type Subscription = ZenObservable.Subscription;
-export type Observer<T> = ZenObservable.Observer<T>;
-
-// Use global module augmentation to add RxJS interop functionality. By
-// using this approach (instead of subclassing `Observable` and adding an
-// ['@@observable']() method), we ensure the exported `Observable` retains all
-// existing type declarations from `@types/zen-observable` (which is important
-// for projects like `apollo-link`).
-declare global {
-  interface Observable<T> {
-    ['@@observable'](): Observable<T>;
-  }
-}
-(Observable.prototype as any)['@@observable'] = function () { return this; };
-export { Observable };
+export {Observable, Subscription, PartialObserver as Observer} from 'rxjs';