Home Manual Reference Source Test Repository

spec-js/operators/multicast-spec.js

"use strict";
var chai_1 = require('chai');
var Rx = require('../../dist/package/Rx');
var Observable = Rx.Observable;
var Subject = Rx.Subject;
var ReplaySubject = Rx.ReplaySubject;
/** @test {multicast} */
describe('Observable.prototype.multicast', function () {
    asDiagram('multicast(() => new Subject())')('should mirror a simple source Observable', function () {
        var source = cold('--1-2---3-4--5-|');
        var sourceSubs = '^              !';
        var multicasted = source.multicast(function () { return new Subject(); });
        var expected = '--1-2---3-4--5-|';
        expectObservable(multicasted).toBe(expected);
        expectSubscriptions(source.subscriptions).toBe(sourceSubs);
        multicasted.connect();
    });
    it('should accept Subjects', function (done) {
        var expected = [1, 2, 3, 4];
        var connectable = Observable.of(1, 2, 3, 4).multicast(new Subject());
        connectable.subscribe(function (x) { chai_1.expect(x).to.equal(expected.shift()); }, function (x) {
            done(new Error('should not be called'));
        }, function () {
            done();
        });
        connectable.connect();
    });
    it('should multicast a ConnectableObservable', function (done) {
        var expected = [1, 2, 3, 4];
        var source = new Subject();
        var connectable = source.multicast(new Subject());
        var replayed = connectable.multicast(new ReplaySubject());
        connectable.connect();
        replayed.connect();
        source.next(1);
        source.next(2);
        source.next(3);
        source.next(4);
        source.complete();
        replayed.do({
            next: function (x) {
                chai_1.expect(x).to.equal(expected.shift());
            },
            complete: function () {
                chai_1.expect(expected.length).to.equal(0);
            }
        })
            .subscribe(null, done, done);
    });
    it('should accept Subject factory functions', function (done) {
        var expected = [1, 2, 3, 4];
        var connectable = Observable.of(1, 2, 3, 4).multicast(function () { return new Subject(); });
        connectable.subscribe(function (x) { chai_1.expect(x).to.equal(expected.shift()); }, function (x) {
            done(new Error('should not be called'));
        }, function () {
            done();
        });
        connectable.connect();
    });
    it('should accept a multicast selector and connect to a hot source for each subscriber', function () {
        var source = hot('-1-2-3----4-|');
        var sourceSubs = ['^           !',
            '    ^       !',
            '        ^   !'];
        var multicasted = source.multicast(function () { return new Subject(); }, function (x) { return x.zip(x, function (a, b) { return (parseInt(a) + parseInt(b)).toString(); }); });
        var subscriber1 = hot('a|           ').mergeMapTo(multicasted);
        var expected1 = '-2-4-6----8-|';
        var subscriber2 = hot('    b|       ').mergeMapTo(multicasted);
        var expected2 = '    -6----8-|';
        var subscriber3 = hot('        c|   ').mergeMapTo(multicasted);
        var expected3 = '        --8-|';
        expectObservable(subscriber1).toBe(expected1);
        expectObservable(subscriber2).toBe(expected2);
        expectObservable(subscriber3).toBe(expected3);
        expectSubscriptions(source.subscriptions).toBe(sourceSubs);
    });
    it('should accept a multicast selector and connect to a cold source for each subscriber', function () {
        var source = cold('-1-2-3----4-|');
        var sourceSubs = ['^           !',
            '    ^           !',
            '        ^           !'];
        var multicasted = source.multicast(function () { return new Subject(); }, function (x) { return x.zip(x, function (a, b) { return (parseInt(a) + parseInt(b)).toString(); }); });
        var expected1 = '-2-4-6----8-|';
        var expected2 = '    -2-4-6----8-|';
        var expected3 = '        -2-4-6----8-|';
        var subscriber1 = hot('a|           ').mergeMapTo(multicasted);
        var subscriber2 = hot('    b|       ').mergeMapTo(multicasted);
        var subscriber3 = hot('        c|   ').mergeMapTo(multicasted);
        expectObservable(subscriber1).toBe(expected1);
        expectObservable(subscriber2).toBe(expected2);
        expectObservable(subscriber3).toBe(expected3);
        expectSubscriptions(source.subscriptions).toBe(sourceSubs);
    });
    it('should accept a multicast selector and respect the subject\'s messaging semantics', function () {
        var source = cold('-1-2-3----4-|');
        var sourceSubs = ['^           !',
            '    ^           !',
            '        ^           !'];
        var multicasted = source.multicast(function () { return new ReplaySubject(1); }, function (x) { return x.concat(x.takeLast(1)); });
        var expected1 = '-1-2-3----4-(4|)';
        var expected2 = '    -1-2-3----4-(4|)';
        var expected3 = '        -1-2-3----4-(4|)';
        var subscriber1 = hot('a|           ').mergeMapTo(multicasted);
        var subscriber2 = hot('    b|       ').mergeMapTo(multicasted);
        var subscriber3 = hot('        c|   ').mergeMapTo(multicasted);
        expectObservable(subscriber1).toBe(expected1);
        expectObservable(subscriber2).toBe(expected2);
        expectObservable(subscriber3).toBe(expected3);
        expectSubscriptions(source.subscriptions).toBe(sourceSubs);
    });
    it('should do nothing if connect is not called, despite subscriptions', function () {
        var source = cold('--1-2---3-4--5-|');
        var sourceSubs = [];
        var multicasted = source.multicast(function () { return new Subject(); });
        var expected = '-';
        expectObservable(multicasted).toBe(expected);
        expectSubscriptions(source.subscriptions).toBe(sourceSubs);
    });
    it('should multicast the same values to multiple observers', function () {
        var source = cold('-1-2-3----4-|');
        var sourceSubs = '^           !';
        var multicasted = source.multicast(function () { return new Subject(); });
        var subscriber1 = hot('a|           ').mergeMapTo(multicasted);
        var expected1 = '-1-2-3----4-|';
        var subscriber2 = hot('    b|       ').mergeMapTo(multicasted);
        var expected2 = '    -3----4-|';
        var subscriber3 = hot('        c|   ').mergeMapTo(multicasted);
        var expected3 = '        --4-|';
        expectObservable(subscriber1).toBe(expected1);
        expectObservable(subscriber2).toBe(expected2);
        expectObservable(subscriber3).toBe(expected3);
        expectSubscriptions(source.subscriptions).toBe(sourceSubs);
        multicasted.connect();
    });
    it('should multicast an error from the source to multiple observers', function () {
        var source = cold('-1-2-3----4-#');
        var sourceSubs = '^           !';
        var multicasted = source.multicast(function () { return new Subject(); });
        var subscriber1 = hot('a|           ').mergeMapTo(multicasted);
        var expected1 = '-1-2-3----4-#';
        var subscriber2 = hot('    b|       ').mergeMapTo(multicasted);
        var expected2 = '    -3----4-#';
        var subscriber3 = hot('        c|   ').mergeMapTo(multicasted);
        var expected3 = '        --4-#';
        expectObservable(subscriber1).toBe(expected1);
        expectObservable(subscriber2).toBe(expected2);
        expectObservable(subscriber3).toBe(expected3);
        expectSubscriptions(source.subscriptions).toBe(sourceSubs);
        multicasted.connect();
    });
    it('should multicast the same values to multiple observers, ' +
        'but is unsubscribed explicitly and early', function () {
        var source = cold('-1-2-3----4-|');
        var sourceSubs = '^        !   ';
        var multicasted = source.multicast(function () { return new Subject(); });
        var unsub = '         u   ';
        var subscriber1 = hot('a|           ').mergeMapTo(multicasted);
        var expected1 = '-1-2-3----   ';
        var subscriber2 = hot('    b|       ').mergeMapTo(multicasted);
        var expected2 = '    -3----   ';
        var subscriber3 = hot('        c|   ').mergeMapTo(multicasted);
        var expected3 = '        --   ';
        expectObservable(subscriber1).toBe(expected1);
        expectObservable(subscriber2).toBe(expected2);
        expectObservable(subscriber3).toBe(expected3);
        expectSubscriptions(source.subscriptions).toBe(sourceSubs);
        // Set up unsubscription action
        var connection;
        expectObservable(hot(unsub).do(function () {
            connection.unsubscribe();
        })).toBe(unsub);
        connection = multicasted.connect();
    });
    it('should not break unsubscription chains when result is unsubscribed explicitly', function () {
        var source = cold('-1-2-3----4-|');
        var sourceSubs = '^        !   ';
        var multicasted = source
            .mergeMap(function (x) { return Observable.of(x); })
            .multicast(function () { return new Subject(); });
        var subscriber1 = hot('a|           ').mergeMapTo(multicasted);
        var expected1 = '-1-2-3----   ';
        var subscriber2 = hot('    b|       ').mergeMapTo(multicasted);
        var expected2 = '    -3----   ';
        var subscriber3 = hot('        c|   ').mergeMapTo(multicasted);
        var expected3 = '        --   ';
        var unsub = '         u   ';
        expectObservable(subscriber1).toBe(expected1);
        expectObservable(subscriber2).toBe(expected2);
        expectObservable(subscriber3).toBe(expected3);
        expectSubscriptions(source.subscriptions).toBe(sourceSubs);
        // Set up unsubscription action
        var connection;
        expectObservable(hot(unsub).do(function () {
            connection.unsubscribe();
        })).toBe(unsub);
        connection = multicasted.connect();
    });
    it('should multicast an empty source', function () {
        var source = cold('|');
        var sourceSubs = '(^!)';
        var multicasted = source.multicast(function () { return new Subject(); });
        var expected = '|';
        expectObservable(multicasted).toBe(expected);
        expectSubscriptions(source.subscriptions).toBe(sourceSubs);
        multicasted.connect();
    });
    it('should multicast a never source', function () {
        var source = cold('-');
        var sourceSubs = '^';
        var multicasted = source.multicast(function () { return new Subject(); });
        var expected = '-';
        expectObservable(multicasted).toBe(expected);
        expectSubscriptions(source.subscriptions).toBe(sourceSubs);
        multicasted.connect();
    });
    it('should multicast a throw source', function () {
        var source = cold('#');
        var sourceSubs = '(^!)';
        var multicasted = source.multicast(function () { return new Subject(); });
        var expected = '#';
        expectObservable(multicasted).toBe(expected);
        expectSubscriptions(source.subscriptions).toBe(sourceSubs);
        multicasted.connect();
    });
    describe('with refCount() and subject factory', function () {
        it('should connect when first subscriber subscribes', function () {
            var source = cold('-1-2-3----4-|');
            var sourceSubs = '   ^           !';
            var multicasted = source.multicast(function () { return new Subject(); }).refCount();
            var subscriber1 = hot('   a|           ').mergeMapTo(multicasted);
            var expected1 = '   -1-2-3----4-|';
            var subscriber2 = hot('       b|       ').mergeMapTo(multicasted);
            var expected2 = '       -3----4-|';
            var subscriber3 = hot('           c|   ').mergeMapTo(multicasted);
            var expected3 = '           --4-|';
            expectObservable(subscriber1).toBe(expected1);
            expectObservable(subscriber2).toBe(expected2);
            expectObservable(subscriber3).toBe(expected3);
            expectSubscriptions(source.subscriptions).toBe(sourceSubs);
        });
        it('should disconnect when last subscriber unsubscribes', function () {
            var source = cold('-1-2-3----4-|');
            var sourceSubs = '   ^        !   ';
            var multicasted = source.multicast(function () { return new Subject(); }).refCount();
            var subscriber1 = hot('   a|           ').mergeMapTo(multicasted);
            var unsub1 = '          !     ';
            var expected1 = '   -1-2-3--     ';
            var subscriber2 = hot('       b|       ').mergeMapTo(multicasted);
            var unsub2 = '            !   ';
            var expected2 = '       -3----   ';
            expectObservable(subscriber1, unsub1).toBe(expected1);
            expectObservable(subscriber2, unsub2).toBe(expected2);
            expectSubscriptions(source.subscriptions).toBe(sourceSubs);
        });
        it('should be retryable when cold source is synchronous', function () {
            function subjectFactory() { return new Subject(); }
            var source = cold('(123#)');
            var multicasted = source.multicast(subjectFactory).refCount();
            var subscribe1 = 's               ';
            var expected1 = '(123123123123#) ';
            var subscribe2 = ' s              ';
            var expected2 = ' (123123123123#)';
            var sourceSubs = ['(^!)',
                '(^!)',
                '(^!)',
                '(^!)',
                ' (^!)',
                ' (^!)',
                ' (^!)',
                ' (^!)'];
            expectObservable(hot(subscribe1).do(function () {
                expectObservable(multicasted.retry(3)).toBe(expected1);
            })).toBe(subscribe1);
            expectObservable(hot(subscribe2).do(function () {
                expectObservable(multicasted.retry(3)).toBe(expected2);
            })).toBe(subscribe2);
            expectSubscriptions(source.subscriptions).toBe(sourceSubs);
        });
        it('should be retryable with ReplaySubject and cold source is synchronous', function () {
            function subjectFactory() { return new Rx.ReplaySubject(1); }
            var source = cold('(123#)');
            var multicasted = source.multicast(subjectFactory).refCount();
            var subscribe1 = 's               ';
            var expected1 = '(123123123123#) ';
            var subscribe2 = ' s              ';
            var expected2 = ' (123123123123#)';
            var sourceSubs = ['(^!)',
                '(^!)',
                '(^!)',
                '(^!)',
                ' (^!)',
                ' (^!)',
                ' (^!)',
                ' (^!)'];
            expectObservable(hot(subscribe1).do(function () {
                expectObservable(multicasted.retry(3)).toBe(expected1);
            })).toBe(subscribe1);
            expectObservable(hot(subscribe2).do(function () {
                expectObservable(multicasted.retry(3)).toBe(expected2);
            })).toBe(subscribe2);
            expectSubscriptions(source.subscriptions).toBe(sourceSubs);
        });
        it('should be repeatable when cold source is synchronous', function () {
            function subjectFactory() { return new Subject(); }
            var source = cold('(123|)');
            var multicasted = source.multicast(subjectFactory).refCount();
            var subscribe1 = 's                  ';
            var expected1 = '(123123123123123|) ';
            var subscribe2 = ' s                 ';
            var expected2 = ' (123123123123123|)';
            var sourceSubs = ['(^!)',
                '(^!)',
                '(^!)',
                '(^!)',
                '(^!)',
                ' (^!)',
                ' (^!)',
                ' (^!)',
                ' (^!)',
                ' (^!)'];
            expectObservable(hot(subscribe1).do(function () {
                expectObservable(multicasted.repeat(5)).toBe(expected1);
            })).toBe(subscribe1);
            expectObservable(hot(subscribe2).do(function () {
                expectObservable(multicasted.repeat(5)).toBe(expected2);
            })).toBe(subscribe2);
            expectSubscriptions(source.subscriptions).toBe(sourceSubs);
        });
        it('should be repeatable with ReplaySubject and cold source is synchronous', function () {
            function subjectFactory() { return new Rx.ReplaySubject(1); }
            var source = cold('(123|)');
            var multicasted = source.multicast(subjectFactory).refCount();
            var subscribe1 = 's                  ';
            var expected1 = '(123123123123123|) ';
            var subscribe2 = ' s                 ';
            var expected2 = ' (123123123123123|)';
            var sourceSubs = ['(^!)',
                '(^!)',
                '(^!)',
                '(^!)',
                '(^!)',
                ' (^!)',
                ' (^!)',
                ' (^!)',
                ' (^!)',
                ' (^!)'];
            expectObservable(hot(subscribe1).do(function () {
                expectObservable(multicasted.repeat(5)).toBe(expected1);
            })).toBe(subscribe1);
            expectObservable(hot(subscribe2).do(function () {
                expectObservable(multicasted.repeat(5)).toBe(expected2);
            })).toBe(subscribe2);
            expectSubscriptions(source.subscriptions).toBe(sourceSubs);
        });
        it('should be retryable', function () {
            function subjectFactory() { return new Subject(); }
            var source = cold('-1-2-3----4-#                        ');
            var sourceSubs = ['^           !                        ',
                '            ^           !            ',
                '                        ^           !'];
            var multicasted = source.multicast(subjectFactory).refCount();
            var subscribe1 = 's                                    ';
            var expected1 = '-1-2-3----4--1-2-3----4--1-2-3----4-#';
            var subscribe2 = '    s                                ';
            var expected2 = '    -3----4--1-2-3----4--1-2-3----4-#';
            expectObservable(hot(subscribe1).do(function () {
                expectObservable(multicasted.retry(2)).toBe(expected1);
            })).toBe(subscribe1);
            expectObservable(hot(subscribe2).do(function () {
                expectObservable(multicasted.retry(2)).toBe(expected2);
            })).toBe(subscribe2);
            expectSubscriptions(source.subscriptions).toBe(sourceSubs);
        });
        it('should be retryable using a ReplaySubject', function () {
            function subjectFactory() { return new Rx.ReplaySubject(1); }
            var source = cold('-1-2-3----4-#                        ');
            var sourceSubs = ['^           !                        ',
                '            ^           !            ',
                '                        ^           !'];
            var multicasted = source.multicast(subjectFactory).refCount();
            var expected1 = '-1-2-3----4--1-2-3----4--1-2-3----4-#';
            var subscribe2 = time('----|                                ');
            var expected2 = '    23----4--1-2-3----4--1-2-3----4-#';
            expectObservable(multicasted.retry(2)).toBe(expected1);
            rxTestScheduler.schedule(function () {
                return expectObservable(multicasted.retry(2)).toBe(expected2);
            }, subscribe2);
            expectSubscriptions(source.subscriptions).toBe(sourceSubs);
        });
        it('should be repeatable', function () {
            function subjectFactory() { return new Subject(); }
            var source = cold('-1-2-3----4-|                        ');
            var sourceSubs = ['^           !                        ',
                '            ^           !            ',
                '                        ^           !'];
            var multicasted = source.multicast(subjectFactory).refCount();
            var subscribe1 = 's                                    ';
            var expected1 = '-1-2-3----4--1-2-3----4--1-2-3----4-|';
            var subscribe2 = '    s                                ';
            var expected2 = '    -3----4--1-2-3----4--1-2-3----4-|';
            expectObservable(hot(subscribe1).do(function () {
                expectObservable(multicasted.repeat(3)).toBe(expected1);
            })).toBe(subscribe1);
            expectObservable(hot(subscribe2).do(function () {
                expectObservable(multicasted.repeat(3)).toBe(expected2);
            })).toBe(subscribe2);
            expectSubscriptions(source.subscriptions).toBe(sourceSubs);
        });
        it('should be repeatable using a ReplaySubject', function () {
            function subjectFactory() { return new Rx.ReplaySubject(1); }
            var source = cold('-1-2-3----4-|                        ');
            var sourceSubs = ['^           !                        ',
                '            ^           !            ',
                '                        ^           !'];
            var multicasted = source.multicast(subjectFactory).refCount();
            var subscribe1 = 's                                    ';
            var expected1 = '-1-2-3----4--1-2-3----4--1-2-3----4-|';
            var subscribe2 = '    s                                ';
            var expected2 = '    23----4--1-2-3----4--1-2-3----4-|';
            expectObservable(hot(subscribe1).do(function () {
                expectObservable(multicasted.repeat(3)).toBe(expected1);
            })).toBe(subscribe1);
            expectObservable(hot(subscribe2).do(function () {
                expectObservable(multicasted.repeat(3)).toBe(expected2);
            })).toBe(subscribe2);
            expectSubscriptions(source.subscriptions).toBe(sourceSubs);
        });
    });
    it('should multicast one observable to multiple observers', function (done) {
        var results1 = [];
        var results2 = [];
        var subscriptions = 0;
        var source = new Observable(function (observer) {
            subscriptions++;
            observer.next(1);
            observer.next(2);
            observer.next(3);
            observer.next(4);
            observer.complete();
        });
        var connectable = source.multicast(function () {
            return new Subject();
        });
        connectable.subscribe(function (x) {
            results1.push(x);
        });
        connectable.subscribe(function (x) {
            results2.push(x);
        });
        chai_1.expect(results1).to.deep.equal([]);
        chai_1.expect(results2).to.deep.equal([]);
        connectable.connect();
        chai_1.expect(results1).to.deep.equal([1, 2, 3, 4]);
        chai_1.expect(results2).to.deep.equal([1, 2, 3, 4]);
        chai_1.expect(subscriptions).to.equal(1);
        done();
    });
    it('should remove all subscribers from the subject when disconnected', function () {
        var subject = new Subject();
        var expected = [1, 2, 3, 4];
        var i = 0;
        var source = Observable.from([1, 2, 3, 4]).multicast(subject);
        source.subscribe(function (x) {
            chai_1.expect(x).to.equal(expected[i++]);
        });
        source.connect();
        chai_1.expect(subject.observers.length).to.equal(0);
    });
    describe('when given a subject factory', function () {
        it('should allow you to reconnect by subscribing again', function (done) {
            var expected = [1, 2, 3, 4];
            var i = 0;
            var source = Observable.of(1, 2, 3, 4).multicast(function () { return new Subject(); });
            source.subscribe(function (x) {
                chai_1.expect(x).to.equal(expected[i++]);
            }, null, function () {
                i = 0;
                source.subscribe(function (x) {
                    chai_1.expect(x).to.equal(expected[i++]);
                }, null, done);
                source.connect();
            });
            source.connect();
        });
        it('should not throw ObjectUnsubscribedError when used in ' +
            'a switchMap', function (done) {
            var source = Observable.of(1, 2, 3)
                .multicast(function () { return new Subject(); })
                .refCount();
            var expected = ['a1', 'a2', 'a3', 'b1', 'b2', 'b3', 'c1', 'c2', 'c3'];
            Observable.of('a', 'b', 'c')
                .switchMap(function (letter) { return source.map(function (n) { return String(letter + n); }); })
                .subscribe(function (x) {
                chai_1.expect(x).to.equal(expected.shift());
            }, function (x) {
                done(new Error('should not be called'));
            }, function () {
                chai_1.expect(expected.length).to.equal(0);
                done();
            });
        });
    });
    describe('when given a subject', function () {
        it('should not throw ObjectUnsubscribedError when used in ' +
            'a switchMap', function (done) {
            var source = Observable.of(1, 2, 3)
                .multicast(new Subject())
                .refCount();
            var expected = ['a1', 'a2', 'a3'];
            Observable.of('a', 'b', 'c')
                .switchMap(function (letter) { return source.map(function (n) { return String(letter + n); }); })
                .subscribe(function (x) {
                chai_1.expect(x).to.equal(expected.shift());
            }, function (x) {
                done(new Error('should not be called'));
            }, function () {
                chai_1.expect(expected.length).to.equal(0);
                done();
            });
        });
    });
    describe('typings', function () {
        type('should infer the type', function () {
            /* tslint:disable:no-unused-variable */
            var source = Rx.Observable.of(1, 2, 3);
            var result = source.multicast(function () { return new Subject(); });
            /* tslint:enable:no-unused-variable */
        });
        type('should infer the type with a selector', function () {
            /* tslint:disable:no-unused-variable */
            var source = Rx.Observable.of(1, 2, 3);
            var result = source.multicast(function () { return new Subject(); }, function (s) { return s.map(function (x) { return x; }); });
            /* tslint:enable:no-unused-variable */
        });
        type('should infer the type with a type-changing selector', function () {
            /* tslint:disable:no-unused-variable */
            var source = Rx.Observable.of(1, 2, 3);
            var result = source.multicast(function () { return new Subject(); }, function (s) { return s.map(function (x) { return x + '!'; }); });
            /* tslint:enable:no-unused-variable */
        });
        type('should infer the type for the pipeable operator', function () {
            /* tslint:disable:no-unused-variable */
            var source = Rx.Observable.of(1, 2, 3);
            // TODO: https://github.com/ReactiveX/rxjs/issues/2972
            var result = Rx.operators.multicast(function () { return new Subject(); })(source);
            /* tslint:enable:no-unused-variable */
        });
        type('should infer the type for the pipeable operator with a selector', function () {
            /* tslint:disable:no-unused-variable */
            var source = Rx.Observable.of(1, 2, 3);
            var result = source.pipe(Rx.operators.multicast(function () { return new Subject(); }, function (s) { return s.map(function (x) { return x; }); }));
            /* tslint:enable:no-unused-variable */
        });
        type('should infer the type for the pipeable operator with a type-changing selector', function () {
            /* tslint:disable:no-unused-variable */
            var source = Rx.Observable.of(1, 2, 3);
            var result = source.pipe(Rx.operators.multicast(function () { return new Subject(); }, function (s) { return s.map(function (x) { return x + '!'; }); }));
            /* tslint:enable:no-unused-variable */
        });
    });
});
//# sourceMappingURL=multicast-spec.js.map