RxJava ReplaySubject Sınıfı

Giriş

Açıklaması şöyle

ReplaySubject emits all the items of the Observable, regardless of when the subscriber subscribes.

Örnek

Elimizde şöyle bir kod olsun

ReplaySubject<Integer> pSubject = ReplaySubject.create();
pSubject.onNext(0);


pSubject.subscribe(it -> System.out.println("Observer 1 onNext: " + it),
                  (Throwable error) -> { }, () -> {},
                  on1 -> System.out.println("Observer 1 onSubscribe"));

pSubject.onNext(1);
pSubject.onNext(2);


pSubject.subscribe(it -> System.out.println("Observer 2 onNext: " + it),
                  (Throwable error) -> { }, () -> {},
                  on1 -> System.out.println("Observer 2 onSubscribe"));

pSubject.onNext(3);
pSubject.onNext(4);

Çıktı olarak şunu alırız. Aslında bir anlamda Cold Observale'ın tüm çıktısı kaydediliyor ve kaydedilen şey tekrar en baştan oynatılıyor.

Observer 1 onSubscribe

Observer 1 onNext: 0

Observer 1 onNext: 1

Observer 1 onNext: 2

Observer 2 onSubscribe

Observer 2 onNext: 0

Observer 2 onNext: 1

Observer 2 onNext: 2

Observer 1 onNext: 3

Observer 2 onNext: 3

Observer 1 onNext: 4

Observer 2 onNext: 4

Yani ReplaySubject pahalı işlemleri bir şekilde cache'lemek için kullanılabilir

Örnek

Elimizde şöyle bir kod olsun

Observable<Integer> observable = Observable.range(1, 5)
                .subscribeOn(Schedulers.io());

//Record everyting
ReplaySubject<Integer> subject = ReplaySubject.create();
observable.subscribe(subject);

//Replay everything
subject.subscribe(s -> System.out.println("subscriber one: " + s));

//Replay everything
subject.subscribe(s -> System.out.println("subscriber two: " + s));

Çıktı olarak şunu alırız

subscriber one: 1

subscriber one: 2

subscriber one: 3

subscriber one: 4

subscriber one: 5

subscriber two: 1

subscriber two: 2

subscriber two: 3

subscriber two: 4

subscriber two: 5