fix(ApolloQueryObservable): Make it compatible with Rx.Observable

Author: Kamil Kisiela

CHANGELOG.md

@@ -5,6 +5,7 @@
 - Added `fetchMore` support ([PR #58](https://github.com/apollostack/angular2-apollo/pull/58))
 - Exposed `ApolloQueryObservable` in the index module ([PR #54](https://github.com/apollostack/angular2-apollo/pull/54))
 - Added support for getting `loading` state from ApolloQueryResult [Issue #36](https://github.com/apollostack/angular2-apollo/issues/36) ([PR #43](https://github.com/apollostack/angular2-apollo/pull/43))
+- Fixed `ApolloQueryObservable` incompatibility with `Rx.Observable` ([PR #59](https://github.com/apollostack/angular2-apollo/pull/59))
 
 ### v0.4.1
 

src/apolloQueryObservable.ts

@@ -1,18 +1,11 @@
-import {
-  Observable,
-} from 'rxjs/Observable';
+import { Observable } from 'rxjs/Observable';
+import { Subscriber } from 'rxjs/Subscriber';
+import { Subscription } from 'rxjs/Subscription';
+import { Operator } from 'rxjs/Operator';
+import { $$observable } from 'rxjs/symbol/observable';
 
-import {
-  Operator,
-} from 'rxjs/Operator';
-
-import {
-  Subscriber,
-} from 'rxjs/Subscriber';
-
-import {
-  Subscription,
-} from 'rxjs/Subscription';
+import { FetchMoreOptions } from 'apollo-client/ObservableQuery';
+import { FetchMoreQueryOptions } from 'apollo-client/watchQueryOptions';
 
 export class ApolloQueryObservable<T> extends Observable<T> {
   constructor(public apollo: any, subscribe?: <R>(subscriber: Subscriber<R>) => Subscription | Function | void) {
@@ -28,6 +21,8 @@ export class ApolloQueryObservable<T> extends Observable<T> {
     return observable;
   }
 
+  // apollo-specific methods
+
   public refetch(variables?: any): Promise<any> {
     return this.apollo.refetch(variables);
   }
@@ -39,4 +34,15 @@ export class ApolloQueryObservable<T> extends Observable<T> {
   public startPolling(p: number): void {
     return this.apollo.startPolling(p);
   }
+
+  public fetchMore(options: FetchMoreQueryOptions & FetchMoreOptions): Promise<any> {
+    return this.apollo.fetchMore(options);
+  }
+
+  // where magic happens
+
+  protected _subscribe(subscriber: Subscriber<T>) {
+    const apollo = this.apollo;
+    return apollo[$$observable]().subscribe(subscriber);
+  }
 }

tests/apolloQueryObservable.ts

@@ -1,49 +1,120 @@
-import {
-  Observable,
-} from 'rxjs/Observable';
-
 import {
   ApolloQueryObservable,
 } from '../src/apolloQueryObservable';
 
+import {
+  mockClient,
+} from './_mocks';
+
+import gql from 'graphql-tag';
+
 import 'rxjs/add/operator/map';
 
-class ObservableQuery<T> extends Observable<T> {
-  refetch(v?: any) {}
-  stopPolling() {}
-  startPolling(n?: any) {}
-}
+const query = gql`
+  query heroes {
+    allHeroes {
+      heroes {
+        name
+      }
+    }
+  }
+`;
+const data = {
+  allHeroes: {
+    heroes: [{ name: 'Mr Foo' }, { name: 'Mr Bar' }],
+  },
+};
 
 describe('ApolloQueryObservable', () => {
-  it('should be able to call refetch()', () => {
-    const obs = new ObservableQuery();
-    const res = new ApolloQueryObservable(obs);
-    const variables = { foo: true };
-
-    spyOn(obs, 'refetch').and.returnValue('refetch');
+  let obsApollo;
+  let obsQuery;
+  let client;
 
-    expect(res.refetch(variables)).toEqual('refetch');
-    expect(obs.refetch).toHaveBeenCalledWith(variables);
+  beforeEach(() => {
+    client = mockClient({
+      request: { query },
+      result: { data },
+    });
+    obsApollo = client.watchQuery({ query });
+    obsQuery = new ApolloQueryObservable(obsApollo);
   });
 
-  it('should be able to call stopPolling()', () => {
-    const obs = new ObservableQuery();
-    const res = new ApolloQueryObservable(obs);
+  describe('regular', () => {
+    it('should be able to subscribe', () => {
+      expect(() => {
+        obsQuery.subscribe({
+          next() {},
+        });
+      }).not.toThrow();
+    });
+
+    it('should be able to receive data', (done) => {
+      obsQuery.subscribe({
+        next(result) {
+          expect(result.data).toEqual(data);
+          done();
+        }
+      });
+    });
+
+    it('should be able to receive an error', (done) => {
+      obsQuery.subscribe({
+        next() {},
+      });
 
-    spyOn(obs, 'stopPolling').and.returnValue('stopPolling');
+      obsQuery.subscribe({
+        error(error) {
+          expect(error instanceof Error).toBe(true);
+          done();
+        },
+      })
+    });
 
-    expect(res.stopPolling()).toEqual('stopPolling');
-    expect(obs.stopPolling).toHaveBeenCalled();
+    it('should be able to use a operator', (done) => {
+      obsQuery.map(result => result.data).subscribe({
+        next(result) {
+          expect(result).toEqual(data);
+          done();
+        },
+      });
+    });
   });
 
-  it('should be able to call custom method after operator', () => {
-    const obs = new ObservableQuery();
-    spyOn(obs, 'refetch').and.returnValue('refetch');
+  describe('apollo-specific', () => {
+    it('should be able to refech', () => {
+      spyOn(obsApollo, 'refetch').and.returnValue('promise');
+
+      const arg = 'foo';
+      const result = obsQuery.refetch(arg);
+
+      expect(obsApollo.refetch).toHaveBeenCalledWith(arg);
+      expect(result).toBe('promise');
+    });
+
+    it('should be able to startPolling', () => {
+      spyOn(obsApollo, 'startPolling');
+
+      const arg = 200;
+      obsQuery.startPolling(arg);
+
+      expect(obsApollo.startPolling).toHaveBeenCalledWith(arg);
+    });
+
+    it('should be able to stopPolling', () => {
+      spyOn(obsApollo, 'stopPolling');
+
+      obsQuery.stopPolling();
+
+      expect(obsApollo.stopPolling).toHaveBeenCalled();
+    });
 
-    const res = new ApolloQueryObservable(obs).map((i) => i);
+    it('should be able to fetchMore', () => {
+      spyOn(obsApollo, 'fetchMore');
 
+      const arg = 200;
+      obsQuery.fetchMore(arg);
 
-    expect(res['refetch']()).toEqual('refetch');
-    expect(obs.refetch).toHaveBeenCalled();
+      expect(obsApollo.fetchMore).toHaveBeenCalledWith(arg);
+    });
   });
 });