feat(AsyncPipe): allow onError argument

Closes #7990

modules/@angular/common/src/pipes/async_pipe.ts

@@ -6,8 +6,9 @@ import {ObservableWrapper, Observable, EventEmitter} from '../../src/facade/asyn
 import {InvalidPipeArgumentException} from './invalid_pipe_argument_exception';
 
 class ObservableStrategy {
-  createSubscription(async: any, updateLatestValue: any): any {
+  createSubscription(async: any, updateLatestValue: any,
-    return ObservableWrapper.subscribe(async, updateLatestValue, e => { throw e; });
+                     onError: (v: any) => any = e => { throw e; }): any {
+    return ObservableWrapper.subscribe(async, updateLatestValue, onError);
   }
 
   dispose(subscription: any): void { ObservableWrapper.dispose(subscription); }
@@ -16,8 +17,9 @@ class ObservableStrategy {
 }
 
 class PromiseStrategy {
-  createSubscription(async: Promise<any>, updateLatestValue: (v: any) => any): any {
+  createSubscription(async: Promise<any>, updateLatestValue: (v: any) => any,
-    return async.then(updateLatestValue);
+                     onError: (v: any) => any = e => { throw e; }): any {
+    return async.then(updateLatestValue, onError);
   }
 
   dispose(subscription: any): void {}
@@ -70,10 +72,10 @@ export class AsyncPipe implements OnDestroy {
     }
   }
 
-  transform(obj: Observable<any>| Promise<any>| EventEmitter<any>): any {
+  transform(obj: Observable<any>| Promise<any>| EventEmitter<any>, onError?: (v: any) => any): any {
     if (isBlank(this._obj)) {
       if (isPresent(obj)) {
-        this._subscribe(obj);
+        this._subscribe(obj, onError);
       }
       this._latestReturnedValue = this._latestValue;
       return this._latestValue;
@@ -81,7 +83,7 @@ export class AsyncPipe implements OnDestroy {
 
     if (obj !== this._obj) {
       this._dispose();
-      return this.transform(obj);
+      return this.transform(obj, onError);
     }
 
     if (this._latestValue === this._latestReturnedValue) {
@@ -93,11 +95,11 @@ export class AsyncPipe implements OnDestroy {
   }
 
   /** @internal */
-  _subscribe(obj: Observable<any>| Promise<any>| EventEmitter<any>): void {
+  _subscribe(obj: Observable<any>| Promise<any>| EventEmitter<any>, onError?: any): void {
     this._obj = obj;
     this._strategy = this._selectStrategy(obj);
     this._subscription = this._strategy.createSubscription(
-        obj, (value: Object) => this._updateLatestValue(obj, value));
+        obj, (value: Object) => this._updateLatestValue(obj, value), onError);
   }
 
   /** @internal */

modules/@angular/common/test/pipes/async_pipe_spec.ts

@@ -52,10 +52,9 @@ export function main() {
              TimerWrapper.setTimeout(() => {
                expect(pipe.transform(emitter)).toEqual(new WrappedValue(message));
                async.done();
-             }, 0)
+             }, 0);
            }));
 
-
         it("should return same value when nothing has changed since the last call",
            inject([AsyncTestCompleter], (async) => {
              pipe.transform(emitter);
@@ -65,7 +64,23 @@ export function main() {
                pipe.transform(emitter);
                expect(pipe.transform(emitter)).toBe(message);
                async.done();
-             }, 0)
+             }, 0);
+           }));
+
+        it("should invoke onError of when a function is provided",
+           inject([AsyncTestCompleter], (async) => {
+             var error = false;
+             function onError() { error = true; }
+             expect(() => pipe.transform(emitter, onError)).not.toThrow();
+
+             expect(error).toBe(false);
+             // this should not affect the pipe
+             ObservableWrapper.callError(emitter, message);
+
+             TimerWrapper.setTimeout(() => {
+               expect(error).toBe(true);
+               async.done();
+             }, 0);
            }));
 
         it("should dispose of the existing subscription when subscribing to a new observable",
@@ -81,7 +96,7 @@ export function main() {
              TimerWrapper.setTimeout(() => {
                expect(pipe.transform(newEmitter)).toBe(null);
                async.done();
-             }, 0)
+             }, 0);
            }));
 
         it("should request a change detection check upon receiving a new value",
@@ -92,7 +107,7 @@ export function main() {
              TimerWrapper.setTimeout(() => {
                expect(ref.spy('markForCheck')).toHaveBeenCalled();
                async.done();
-             }, 10)
+             }, 10);
            }));
       });
 
@@ -109,7 +124,7 @@ export function main() {
              TimerWrapper.setTimeout(() => {
                expect(pipe.transform(emitter)).toBe(null);
                async.done();
-             }, 0)
+             }, 0);
            }));
       });
     });
@@ -140,7 +155,23 @@ export function main() {
              TimerWrapper.setTimeout(() => {
                expect(pipe.transform(completer.promise)).toEqual(new WrappedValue(message));
                async.done();
-             }, timer)
+             }, timer);
+           }));
+
+        it("should invoke onError of when a function is provided",
+           inject([AsyncTestCompleter], (async) => {
+             var error = false;
+             function onError() { error = true; }
+             expect(() => pipe.transform(completer.promise, onError)).not.toThrow();
+
+             expect(error).toBe(false);
+             // this should not affect the pipe
+             completer.reject('rejection');
+
+             TimerWrapper.setTimeout(() => {
+               expect(error).toBe(true);
+               async.done();
+             }, 0);
            }));
 
         it("should return unwrapped value when nothing has changed since the last call",
@@ -152,7 +183,7 @@ export function main() {
                pipe.transform(completer.promise);
                expect(pipe.transform(completer.promise)).toBe(message);
                async.done();
-             }, timer)
+             }, timer);
            }));
 
         it("should dispose of the existing subscription when subscribing to a new promise",
@@ -168,7 +199,7 @@ export function main() {
              TimerWrapper.setTimeout(() => {
                expect(pipe.transform(newCompleter.promise)).toBe(null);
                async.done();
-             }, timer)
+             }, timer);
            }));
 
         it("should request a change detection check upon receiving a new value",
@@ -180,7 +211,7 @@ export function main() {
              TimerWrapper.setTimeout(() => {
                expect(markForCheck).toHaveBeenCalled();
                async.done();
-             }, timer)
+             }, timer);
            }));
 
         describe("ngOnDestroy", () => {
@@ -190,15 +221,15 @@ export function main() {
           it("should dispose of the existing source", inject([AsyncTestCompleter], (async) => {
                pipe.transform(completer.promise);
                expect(pipe.transform(completer.promise)).toBe(null);
-               completer.resolve(message)
+               completer.resolve(message);
 
 
-                   TimerWrapper.setTimeout(() => {
+               TimerWrapper.setTimeout(() => {
-                     expect(pipe.transform(completer.promise)).toEqual(new WrappedValue(message));
+                 expect(pipe.transform(completer.promise)).toEqual(new WrappedValue(message));
-                     pipe.ngOnDestroy();
+                 pipe.ngOnDestroy();
-                     expect(pipe.transform(completer.promise)).toBe(null);
+                 expect(pipe.transform(completer.promise)).toBe(null);
-                     async.done();
+                 async.done();
-                   }, timer);
+               }, timer);
              }));
         });
       });