Loại Subject và Quản Lý Vòng Đời

Subject là thành phần cốt lõi trong ReactiveX, hoạt động như cả Observable và Observer.

Mã Dùng Chung

Rx.NET

private static void InRaConsole(IObservable<string> chuoi)
{
    chuoi.Subscribe(Console.WriteLine);
}

RxSwift

extension ObservableType {
    func themObserver(_ id: String) -> Disposable {
        return subscribe { print("Theo dõi:", id, "Sự kiện:", $0) }
    }
}

func inChuoiRaConsole<O: ObservableType>(ten: String, chuoi: O) -> Disposable {
    return chuoi.subscribe { suKien in
        print("Theo dõi: \(ten), sự kiện: \(suKien)")
    }
}

Subject<T>

Triển khai cả IObservable<T> và IObserver<T>.

Rx.NET

var subj = new Subject<string>();
InRaConsole(subj);
subj.OnNext("X");
subj.OnNext("Y");
// Kết quả: X Y

RxJava

val subj = PublishSubject.create<Int>()
subj.onNext(10)
subj.subscribe { println(it) }
subj.onNext(20)
// Kết quả: 20

RxSwift

let bag = DisposeBag()
let subj = PublishSubject<String>()
subj.themObserver("A").disposed(by: bag)
subj.onNext("Một")
subj.themObserver("B").disposed(by: bag)
subj.onNext("Hai")
// Kết quả: 
// Theo dõi: A Sự kiện: Một
// Theo dõi: A Sự kiện: Hai
// Theo dõi: B Sự kiện: Hai

ReplaySubject<T>

Lưu trữ lịch sử sự kiện theo bộ đệm hoặc thời gian.

Rx.NET

var subj = new ReplaySubject<string>(2);
subj.OnNext("A");
subj.OnNext("B");
subj.Subscribe(Console.WriteLine);
subj.OnNext("C");
// Kết quả: B C

RxJava

val subj = ReplaySubject.createWithSize<Int>(1)
subj.onNext(100)
subj.onNext(200)
subj.subscribe { println("Muộn: $it") }
// Kết quả: Muộn: 200

RxSwift

let subj = ReplaySubject<String>.create(bufferSize: 1)
subj.themObserver("X").disposed(by: bag)
subj.onNext("Alpha")
subj.themObserver("Y").disposed(by: bag)
// Kết quả: 
// Theo dõi: X Sự kiện: Alpha
// Theo dõi: Y Sự kiện: Alpha

BehaviorSubject<T>

Luôn phát giá trị cuối cùng cho subscriber mới.

Rx.NET

var subj = new BehaviorSubject<string>("Khởi tạo");
subj.Subscribe(Console.WriteLine);
// Kết quả: Khởi tạo

RxJava

val subj = BehaviorSubject.createDefault(999)
subj.subscribe { println(it) }
// Kết quả: 999

RxSwift

let subj = BehaviorSubject(value: "🔥")
subj.themObserver("Mới").disposed(by: bag)
// Kết quả: Theo dõi: Mới Sự kiện: 🔥

AsyncSubject<T>

Chỉ phát giá trị cuối cùng khi hoàn thành.

Rx.NET

var subj = new AsyncSubject<string>();
subj.OnNext("Tạm");
subj.OnCompleted();
subj.Subscribe(Console.WriteLine);
// Kết quả: Tạm

RxJava

val subj = AsyncSubject.create<Int>()
subj.onNext(99)
subj.onComplete()
subj.subscribe { println(it) }
// Kết quả: 99

RxSwift

let subj = AsyncSubject<String>()
subj.onNext("Tạm")
subj.onCompleted()
subj.themObserver("Cuối").disposed(by: bag)
// Kết quả: Theo dõi: Cuối Sự kiện: Tạm

Quy Ước Gọi Hàm

Dữ liệu phát sau khi hoàn thành bị bỏ qua.

Rx.NET

var subj = new Subject<string>();
subj.Subscribe(Console.WriteLine);
subj.OnNext("Hiệu lực");
subj.OnCompleted();
subj.OnNext("Vô hiệu");
// Kết quả: Hiệu lực

Quản Lý Đăng Ký

Hủy đăng ký bằng Dispose()/dispose().

RxJava

val stream = ReplaySubject.create<Int>()
val sub = stream.subscribe { println(it) }
stream.onNext(1)
sub.dispose()
stream.onNext(2)
// Kết quả: 1