Rxの基本をまとめてみた(コードはRxJS)~Subjectの基本編~ ref: http://qiita.com/Sekky0905/items/a6534da15ce5f18e2c51

file0.txt

import {Subject, Observable} from "rxjs";

 // Subjectのインスタンスを生成
 let subject = new Subject;

 // 冗長だけれども理解の促進のためにObserverのオブジェクトをここで生成
 let observer1 = {next: (value) => console.log(`observer1ですよ : ${value}`)};
 let observer2 = {next: (value) => console.log(`observer2ですよ : ${value}`)};

 // ObservableとしてのSubjectのsubscribeメソッド
 // 引数にObserverオブジェクトを与えている
 // 2つのObserverオブジェクトをSubjectのObserverのリスト追加する。
 // これで、2つのObserver(実際は、Subject)にマルチキャストすることができるようになった
 subject.subscribe(observer1);
 subject.subscribe(observer2);

 // ObserverとしてのSubject
 subject.next('A');
 subject.next('B');

file1.txt

import {Subject, Observable} from "rxjs";

// subjectのインスタンスを生成
let subject = new Subject();

// subject.subscribe(observe)でSubjectのObserverリストに登録
subject.subscribe({
    next: (value) => console.log(`observerAですよ : ${value}`)
});

subject.subscribe({
    next: (value) => console.log(`observerBですよ : ${value}`)
});

// 1から5までの数字を作成
let observable = Observable.range(1, 5);

// SubjectはObserverでもあるんだ!
observable.subscribe(subject);

file2.txt

import {Observable, Subject} from "rxjs";

// 0~5のObservableを生成
let observable = Observable.range(0, 5);
// Subjectを生成
let subject = new Subject();
// Subjectを元にMulticasted Observablesを生成
let multiCastedObservable = observable.multicast(subject);

// 冗長だが、わかりやすくするためObserverのオブジェクトをここで生成
let observer1 = {
    next : (value) => {console.log(`observer1 : ${value}`)}
};

let observer2 = {
    next : (value) => {console.log(`observer2 : ${value}`)}
};

// 同一Multicasted Observablesを複数のObserver(observer1とobserver2)がsubscribe
// ここで内部的に実際に使用されているのは、Subjectのsubscribeメソッドである
multiCastedObservable.subscribe(observer1);
multiCastedObservable.subscribe(observer2);

// ここで内部的に実際に行われているのは、Observable.subscribe(subject);
multiCastedObservable.connect();

file3.txt

observer1 : 0
observer2 : 0
observer1 : 1
observer2 : 1
observer1 : 2
observer2 : 2
observer1 : 3
observer2 : 3
observer1 : 4
observer2 : 4

file4.txt

import {Observable, Subject} from "rxjs";

// 20ミリ秒感覚で、連続した数字のObservableを生成
let observable = Observable.interval(20);
// Subjectのインスタンスを生成
let subject = new Subject();
// Subjectを元にObservableを生成(Multicasted Observablesではないことに注意!)
let refCountedmultiCastedObservable = observable.multicast(subject).refCount();

// 冗長だが、わかりやすくするためObserverのオブジェクトをここで生成
let observer1 = {
    next: (value) => {
        console.log(`observer1 : ${value}`)
    }
};

// 冗長だが、わかりやすくするためObserverのオブジェクトをここで生成
let observer2 = {
    next: (value) => {
        console.log(`observer2 : ${value}`)
    }
};

console.log('observer1 subscribed');
console.log('最初のObserver(observer1)がsubscribeしたので、multicasted Observableの実行が開始された');
let subscription1 = refCountedmultiCastedObservable.subscribe(observer1);

console.log('observer2 subscribed');
let subscription2 = refCountedmultiCastedObservable.subscribe(observer2);

// 50ms後にunsubscribe
setTimeout(() => {
        console.log('observer1 unsubscribed');
        subscription1.unsubscribe();
    }, 50
);

// 100ms後にunsubscribe
setTimeout(() => {
        console.log('observer2 unsubscribed');
        subscription2.unsubscribe();
    }, 100
);

setTimeout(() => {
    console.log('最後のObserver(observer2)がunsubscribeしたので、multicasted Observableの実行が停止された');
}, 100);

file5.txt

observer1 subscribed
最初のObserver(observer1)がsubscribeしたので、multicasted Observableの実行が開始された
observer2 subscribed
observer1 : 0
observer2 : 0
observer1 : 1
observer2 : 1
observer1 unsubscribed
observer2 : 2
observer2 : 3
observer2 unsubscribed
最後のObserver(observer2)がunsubscribeしたので、multicasted Observableの実行が停止された