/* With share */

var interval = Rx.Observable.interval(1000);

​

var source = interval

    .take(2)

    .do(

        function (x) {

            console.log('Side effect');

        });

​

var published = source.share();

​

// When the number of observers subscribed to published observable goes from

// 0 to 1, we connect to the underlying observable sequence.

published.subscribe(createObserver('SourceA'));

// When the second subscriber is added, no additional subscriptions are added to the

// underlying observable sequence. As a result the operations that result in side

// effects are not repeated per subscriber.

published.subscribe(createObserver('SourceB'));

​

function createObserver(tag) {

    return Rx.Observer.create(

        function (x) {

            console.log('Next: ' + tag + x);

        },

        function (err) {

            console.log('Error: ' + err);

        },

        function () {

            console.log('Completed');

        });

}

​

// => Side effect

// => Next: SourceA0

// => Next: SourceB0

// => Side effect

// => Next: SourceA1

// => Next: SourceB1

// => Completed

// => Completed