feat(pluck): add pluck operator

Pretty similar to RxJS 4 implementation.

Closes ReactiveX#1134

MIGRATION.md

@@ -90,7 +90,6 @@ enabling "composite" subscription behavior.
 |`onErrorResumeNext`|`-`|
 |`pausable`|`-`|
 |`pausableBuffered`|`-`|
-|`pluck`|`-`|
 |`shareReplay`|`-`|
 |`shareValue`|`-`|
 |`selectConcatObserver` or `concatMapObserver`|`-`|

spec/operators/pluck-spec.js

@@ -0,0 +1,152 @@
+/* globals describe, it, expect, hot, cold, expectObservable, expectSubscriptions */
+var Rx = require('../../dist/cjs/Rx');
+var Observable = Rx.Observable;
+
+describe('Observable.prototype.pluck()', function () {
+ it('should work for one object', function () {
+ var a = cold('--x--|', {x: {prop: 42}});
+ var asubs = '^ !';
+ var expected = '--y--|';
+
+ var r = a.pluck('prop');
+ expectObservable(r).toBe(expected, {y: 42});
+ expectSubscriptions(a.subscriptions).toBe(asubs);
+ });
+
+ it('should work for multiple objects', function () {
+ var inputs = {
+ a: {prop: '1'},
+ b: {prop: '2'},
+ c: {prop: '3'},
+ d: {prop: '4'},
+ e: {prop: '5'},
+ };
+ var a = cold('--a-b--c-d---e-|', inputs);
+ var asubs = '^ !';
+ var expected = '--1-2--3-4---5-|';
+
+ var r = a.pluck('prop');
+ expectObservable(r).toBe(expected);
+ expectSubscriptions(a.subscriptions).toBe(asubs);
+ });
+
+ it('should work with deep nested properties', function () {
+ var inputs = {
+ a: {a: {b: {c: '1'}}},
+ b: {a: {b: {c: '2'}}},
+ c: {a: {b: {c: '3'}}},
+ d: {a: {b: {c: '4'}}},
+ e: {a: {b: {c: '5'}}},
+ };
+ var a = cold('--a-b--c-d---e-|', inputs);
+ var asubs = '^ !';
+ var expected = '--1-2--3-4---5-|';
+
+ var r = a.pluck('a', 'b', 'c');
+ expectObservable(r).toBe(expected);
+ expectSubscriptions(a.subscriptions).toBe(asubs);
+ });
+
+ it('should work with edge cases of deep nested properties', function () {
+ var inputs = {
+ a: {a: {b: {c: 1}}},
+ b: {a: {b: 2}},
+ c: {a: {c: {c: 3}}},
+ d: {},
+ e: {a: {b: {c: 5}}},
+ };
+ var a = cold('--a-b--c-d---e-|', inputs);
+ var asubs = '^ !';
+ var expected = '--r-x--y-z---w-|';
+ var values = {r: 1, x: undefined, y: undefined, z: undefined, w: 5};
+
+ var r = a.pluck('a', 'b', 'c');
+ expectObservable(r).toBe(expected, values);
+ expectSubscriptions(a.subscriptions).toBe(asubs);
+ });
+
+ it('should throw an error if not property is passed', function () {
+ expect(function () {
+ Observable.of({prop: 1}, {prop: 2}).pluck();
+ }).toThrow(new Error('List of properties cannot be empty.'));
+ });
+
+ it('should propagate errors from observable that emits only errors', function () {
+ var a = cold('#');
+ var asubs = '(^!)';
+ var expected = '#';
+
+ var r = a.pluck('whatever');
+ expectObservable(r).toBe(expected);
+ expectSubscriptions(a.subscriptions).toBe(asubs);
+ });
+
+ it('should propagate errors from observable that emit values', function () {
+ var a = cold('--a--b--#', {a: {prop: '1'}, b: {prop: '2'}}, 'too bad');
+ var asubs = '^ !';
+ var expected = '--1--2--#';
+
+ var r = a.pluck('prop');
+ expectObservable(r).toBe(expected, undefined, 'too bad');
+ expectSubscriptions(a.subscriptions).toBe(asubs);
+ });
+
+ it('should not pluck an empty observable', function () {
+ var a = cold('|');
+ var asubs = '(^!)';
+ var expected = '|';
+
+ var invoked = 0;
+ var r = a
+ .pluck('whatever')
+ .do(null, null, function () {
+ expect(invoked).toBe(0);
+ });
+
+ expectObservable(r).toBe(expected);
+ expectSubscriptions(a.subscriptions).toBe(asubs);
+ });
+
+ it('should allow unsubscribing explicitly and early', function () {
+ var a = cold('--a--b--c--|', {a: {prop: '1'}, b: {prop: '2'}});
+ var unsub = ' ! ';
+ var asubs = '^ ! ';
+ var expected = '--1--2- ';
+
+ var r = a.pluck('prop');
+ expectObservable(r, unsub).toBe(expected);
+ expectSubscriptions(a.subscriptions).toBe(asubs);
+ });
+
+ it('should pluck twice', function () {
+ var inputs = {
+ a: {a: {b: {c: '1'}}},
+ b: {a: {b: {c: '2'}}},
+ c: {a: {b: {c: '3'}}},
+ d: {a: {b: {c: '4'}}},
+ e: {a: {b: {c: '5'}}},
+ };
+ var a = cold('--a-b--c-d---e-|', inputs);
+ var asubs = '^ !';
+ var expected = '--1-2--3-4---5-|';
+
+ var r = a.pluck('a', 'b').pluck('c');
+ expectObservable(r).toBe(expected);
+ expectSubscriptions(a.subscriptions).toBe(asubs);
+ });
+
+ it('should not break unsubscription chain when unsubscribed explicitly', function () {
+ var a = cold('--a--b--c--|', {a: {prop: '1'}, b: {prop: '2'}});
+ var unsub = ' ! ';
+ var asubs = '^ ! ';
+ var expected = '--1--2- ';
+
+ var r = a
+ .mergeMap(function (x) { return Observable.of(x); })
+ .pluck('prop')
+ .mergeMap(function (x) { return Observable.of(x); });
+
+ expectObservable(r, unsub).toBe(expected);
+ expectSubscriptions(a.subscriptions).toBe(asubs);
+ });
+});

src/Observable.ts

@@ -236,6 +236,7 @@ export class Observable<T> implements CoreOperators<T> {
  multicast: (subjectOrSubjectFactory: Subject<T>|(() => Subject<T>)) => ConnectableObservable<T>;
  observeOn: (scheduler: Scheduler, delay?: number) => Observable<T>;
  partition: (predicate: (x: T) => boolean) => Observable<T>[];
+ pluck: (...properties: string[]) => Observable<any>;
  publish: () => ConnectableObservable<T>;
  publishBehavior: (value: any) => ConnectableObservable<T>;
  publishReplay: (bufferSize?: number, windowTime?: number, scheduler?: Scheduler) => ConnectableObservable<T>;

src/Rx.KitchenSink.ts

@@ -93,6 +93,7 @@ import './add/operator/multicast';
 import './add/operator/observeOn';
 import './add/operator/pairwise';
 import './add/operator/partition';
+import './add/operator/pluck';
 import './add/operator/publish';
 import './add/operator/publishBehavior';
 import './add/operator/publishReplay';

src/Rx.ts

@@ -68,6 +68,7 @@ import './add/operator/mergeMapTo';
 import './add/operator/multicast';
 import './add/operator/observeOn';
 import './add/operator/partition';
+import './add/operator/pluck';
 import './add/operator/publish';
 import './add/operator/publishBehavior';
 import './add/operator/publishReplay';

src/add/operator/pluck.ts

@@ -0,0 +1,10 @@
+/**
+ * Everything in this file is generated by the 'tools/generate-operator-patches.ts' script.
+ * Any manual edits to this file will be lost next time the script is run.
+ **/
+import {Observable} from '../../Observable';
+import {pluck} from '../../operator/pluck';
+
+Observable.prototype.pluck = pluck;
+
+export var _void: void;

src/operator/pluck.ts

@@ -0,0 +1,35 @@
+import {Observable} from '../Observable';
+import {map} from './map';
+
+/**
+ * Retrieves the value of a specified nested property from all elements in
+ * the Observable sequence. If a property can't be resolved, it will return
+ * `undefined` for that value.
+ *
+ * @param {...args} properties the nested properties to pluck
+ * @returns {Observable} Returns a new Observable sequence of property values
+ */
+export function pluck(...properties: string[]): Observable<any> {
+ const length = properties.length;
+ if (length === 0) {
+ throw new Error('List of properties cannot be empty.');
+ }
+ return map.call(this, plucker(properties, length));
+}
+
+function plucker(props: string[], length: number): (x: string) => any {
+ const mapper = (x: string) => {
+ let currentProp = x;
+ for (let i = 0; i < length; i++) {
+ const p = currentProp[props[i]];
+ if (typeof p !== 'undefined') {
+ currentProp = p;
+ } else {
+ return undefined;
+ }
+ }
+ return currentProp;
+ };
+
+ return mapper;
+}