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