Các toán tử điều kiện trong ReactiveX

Toán tử điều kiện và boolean

Bài viết này tập trung vào các toán tử điều kiện và boolean dùng để xử lý Observable, vốn là luồng dữ liệu có thể quan sát.

Mã chung

fun <T> Observable<T>.hienThi() =
    this.subscribe({ println("onNext: $it") },
        { println("onError: $it: ${it.message}") },
        { println("onComplete") })
fun <T> Single<T>.hienThi() =
    this.subscribe({ println("onNext: $it") },
        { println("onError: $it: ${it.message}") })

All

Toán tử All kiểm tra xem tất cả dữ liệu trong luồng có thỏa mãn điều kiện nào đó hay không.

  • RxNET
var subject = new Subject<int>();
subject.Subscribe(Console.WriteLine, () => Console.WriteLine("Subject completed"));
var all = subject.All(i => i < 5);
all.Subscribe(b => Console.WriteLine("All values less than 5? {0}", b));
subject.OnNext(1);
subject.OnNext(2);
subject.OnNext(6);
subject.OnNext(2);
subject.OnNext(1);
subject.OnCompleted();
/*
1
2
6
All values less than 5? False
2
1
Subject completed
*/
  • RxJava
val values = Observable.create<Int> { o ->
    o.onNext(0)
    o.onNext(10)
    o.onNext(10)
    o.onNext(2)
    o.onComplete()
}
values
    .all { i -> i % 2 == 0 }
    .hienThi()
/*
onSuccess: true
*/

val values = Observable.interval(150, TimeUnit.MILLISECONDS).take(5)
val subscription = values
    .all { i -> i < 3 }
    .hienThi()
val subscription2 = values
    .hienThi()
readLine()
subscription.dispose()
subscription2.dispose()
/*
onNext: 0
onNext: 1
onNext: 2
onNext: 3
onSuccess: false
onNext: 4
onComplete
*/

val values = Observable.create<Int> { o ->
    o.onNext(0)
    o.onNext(2)
    o.onError(Exception())
}
values
    .all { i -> i % 2 == 0 }
    .hienThi()
/*
onError: java.lang.Exception: null
*/
  • RxJS

RxJs - every learn-rxjs - every

Amb

Toán tử Amb chỉ gửi dữ liệu từ luồng nào phát ra dữ liệu đầu tiên khi có nhiều luồng.

  • RxNET
/*
    s1 -1--1--|
    s2 --2--2--|
    s3 ---3--3--|
    r  -1--1--|
*/
var s1 = new Subject<int>();
var s2 = new Subject<int>();
var s3 = new Subject<int>();
var result = Observable.Amb(s1, s2, s3);
result.Subscribe(
Console.WriteLine,
() => Console.WriteLine("Completed"));
s1.OnNext(1);
s2.OnNext(2);
s3.OnNext(3);
s1.OnNext(1);
s2.OnNext(2);
s3.OnNext(3);
s1.OnCompleted();
s2.OnCompleted();
s3.OnCompleted();
/*
1
1
Completed
*/

/*
    s1 ---1--|
    s2 -2--2--|
    s3 --3--3--|
    r  -2--2--|
*/
var s1 = new Subject<int>();
var s2 = new Subject<int>();
var s3 = new Subject<int>();
var result = Observable.Amb(s1, s2, s3);
result.Subscribe(
Console.WriteLine,
() => Console.WriteLine("Completed"));
// s1.OnNext(1);
s2.OnNext(2);
s3.OnNext(3);
s1.OnNext(1);
s2.OnNext(2);
s3.OnNext(3);
s1.OnCompleted();
s2.OnCompleted();
s3.OnCompleted();
/*
2
2
Completed
*/
  • RxJava
Observable.amb(listOf(
    Observable.timer(100, TimeUnit.MILLISECONDS).map { i -> "First" },
    Observable.timer(50, TimeUnit.MILLISECONDS).map { i -> "Second" },
    Observable.timer(70, TimeUnit.MILLISECONDS).map { i -> "Third" }))
    .hienThi()
/*
onNext: Second
onComplete
*/

Observable.ambArray(
    Observable.timer(100, TimeUnit.MILLISECONDS).map { i -> "First" },
    Observable.timer(50, TimeUnit.MILLISECONDS).map { i -> "Second" },
    Observable.timer(70, TimeUnit.MILLISECONDS).map { i -> "Third" })
    .hienThi()
/*
onNext: Second
onComplete
*/

Observable.timer(100, TimeUnit.MILLISECONDS).map { i -> "First" }
    .ambWith(Observable.timer(50, TimeUnit.MILLISECONDS).map { i -> "Second" })
    .ambWith(Observable.timer(70, TimeUnit.MILLISECONDS).map { i -> "Third" })
    .hienThi()
/*
onNext: Second
onComplete
*/
  • RxJS

RxJs - race learn-rxjs - race

Contains / Any / Exists / IsEmpty

Contains kiểm tra xem luồng có chứa một giá trị cụ thể hay không. Any / Exists kiểm tra xem luồng có gửi dữ liệu thỏa mãn điều kiện nào đó hay không. IsEmpty kiểm tra xem luồng có rỗng hay không.

  • RxNET
var subject = new Subject<int>();
subject.Subscribe(
Console.WriteLine,
() => Console.WriteLine("Subject completed"));
var contains = subject.Contains(2);
contains.Subscribe(
b => Console.WriteLine("Contains the value 2? {0}", b),
() => Console.WriteLine("contains completed"));
subject.OnNext(1);
subject.OnNext(2);
subject.OnNext(3);
subject.OnCompleted();
/*
1
2
Contains the value 2? True
contains completed
3
Subject completed
*/

var subject = new Subject<int>();
subject.Subscribe(Console.WriteLine, () => Console.WriteLine("Subject completed"));
var any = subject.Any();
any.Subscribe(b => Console.WriteLine("The subject has any values? {0}", b));
subject.OnNext(1);
subject.OnCompleted();
/*
1
The subject has any values? True
Subject completed
*/
  • RxJava
  • exists đã được đổi tên thành any
val values = Observable.interval(100, TimeUnit.MILLISECONDS)
values
    .contains(4L)
    .hienThi()
/*
onSuccess: true
*/

val values = Observable.range(0, 2)
values
    .any { i -> i > 2 }
    .hienThi()
/*
onSuccess: false
*/

val values = Observable.range(0, 2)
values
    .any { i -> i > 2 }
    .hienThi()
/*
onSuccess: false
*/

val values = Observable.timer(1000, TimeUnit.MILLISECONDS)
values
    .isEmpty
    .hienThi()
/*
onSuccess: false
*/

DefaultIfEmpty

DefaultIfEmpty gửi tất cả dữ liệu từ luồng nguồn, và nếu luồng nguồn rỗng thì gửi một giá trị mặc định.

  • RxNET
var subject = new Subject<int>();
subject.Subscribe(
Console.WriteLine,
() => Console.WriteLine("Subject completed"));
var defaultIfEmpty = subject.DefaultIfEmpty();
defaultIfEmpty.Subscribe(
b => Console.WriteLine("defaultIfEmpty value: {0}", b),
() => Console.WriteLine("defaultIfEmpty completed"));
subject.OnNext(1);
subject.OnNext(2);
subject.OnNext(3);
subject.OnCompleted();
/*
1
defaultIfEmpty value: 1
2
defaultIfEmpty value: 2
3
defaultIfEmpty value: 3
Subject completed
defaultIfEmpty completed
*/

var subject = new Subject<int>();
subject.Subscribe(
Console.WriteLine,
() => Console.WriteLine("Subject completed"));
var defaultIfEmpty = subject.DefaultIfEmpty();
defaultIfEmpty.Subscribe(
b => Console.WriteLine("defaultIfEmpty value: {0}", b),
() => Console.WriteLine("defaultIfEmpty completed"));
var default42IfEmpty = subject.DefaultIfEmpty(42);
default42IfEmpty.Subscribe(
b => Console.WriteLine("default42IfEmpty value: {0}", b),
() => Console.WriteLine("default42IfEmpty completed"));
subject.OnCompleted();
/*
Subject completed
defaultIfEmpty value: 0
defaultIfEmpty completed
default42IfEmpty value: 42
default42IfEmpty completed
*/
  • RxJava
val values = Observable.empty<Int>()
values
    .defaultIfEmpty(2)
    .hienThi()
/*
onNext: 2
onComplete
*/

val values = Observable.error<Int>(Exception())
values
    .defaultIfEmpty(2)
    .hienThi()
/*
onError: java.lang.Exception: null
*/

SequenceEqual

SequenceEqual kiểm tra xem hai luồng có gửi các dữ liệu giống hệt nhau hay không.

  • RxNET
var subject1 = new Subject<int>();
subject1.Subscribe(
i => Console.WriteLine("subject1.OnNext({0})", i),
() => Console.WriteLine("subject1 completed"));
var subject2 = new Subject<int>();
subject2.Subscribe(
i => Console.WriteLine("subject2.OnNext({0})", i),
() => Console.WriteLine("subject2 completed"));
var areEqual = subject1.SequenceEqual(subject2);
areEqual.Subscribe(
i => Console.WriteLine("areEqual.OnNext({0})", i),
() => Console.WriteLine("areEqual completed"));
subject1.OnNext(1);
subject1.OnNext(2);
subject2.OnNext(1);
subject2.OnNext(2);
subject2.OnNext(3);
subject1.OnNext(3);
subject1.OnCompleted();
subject2.OnCompleted();
/*
subject1.OnNext(1)
subject1.OnNext(2)
subject2.OnNext(1)
subject2.OnNext(2)
subject2.OnNext(3)
subject1.OnNext(3)
subject1 completed
subject2 completed
areEqual.OnNext(True)
areEqual completed
*/
  • RxJava
val strings = Observable.just("1", "2", "3")
val ints = Observable.just(1, 2, 3)
Observable.sequenceEqual(strings, ints, BiPredicate { s, i -> s == i.toString() } )
    .hienThi()
/*
onSuccess: true
*/

val strings = Observable.just("1", "2", "3")
val ints = Observable.just(1, 2, 3)
Observable.sequenceEqual(strings, ints)
    .hienThi()
/*
onSuccess: false
*/

val values = Observable.create<Int> { o ->
    o.onNext(1)
    o.onNext(2)
    o.onError(Exception())
}
Observable.sequenceEqual(values, values)
    .hienThi()
/*
onError: java.lang.Exception: null
*/

SkipUntil

SkipUntil bỏ qua dữ liệu từ luồng nguồn cho đến khi một luồng khác bắt đầu gửi dữ liệu.

  • RxNET
var subject = new Subject<int>();
var otherSubject = new Subject<Unit>();
subject
.SkipUntil(otherSubject)
.Subscribe(Console.WriteLine, () => Console.WriteLine("Completed"));
subject.OnNext(1);
subject.OnNext(2);
subject.OnNext(3);
otherSubject.OnNext(Unit.Default);
subject.OnNext(4);
subject.OnNext(5);
subject.OnNext(6);
subject.OnNext(7);
subject.OnNext(8);
subject.OnCompleted();
/*
4
5
6
7
8
Completed
*/
  • RxJava
val values = Observable.interval(100, TimeUnit.MILLISECONDS)
val cutoff = Observable.timer(250, TimeUnit.MILLISECONDS)
val subscription = values
    .skipUntil(cutoff)
    .hienThi()
readLine()
subscription.dispose()
/*
onNext: 2
onNext: 3
onNext: 4
...
*/
  • RxSwift
let disposeBag = DisposeBag()
Observable.of(1, 2, 3, 4, 5, 6)
    .takeWhile { $0 < 4 }
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)
/*
1
2
3
*/

SkipWhile

SkipWhile bỏ qua dữ liệu từ luồng nguồn cho đến khi một điều kiện không còn thỏa mãn.

  • RxNET
var subject = new Subject<int>();
subject
.SkipWhile(i => i < 4)
.Subscribe(Console.WriteLine, () => Console.WriteLine("Completed"));
subject.OnNext(1);
subject.OnNext(2);
subject.OnNext(3);
subject.OnNext(4);
subject.OnNext(3);
subject.OnNext(2);
subject.OnNext(1);
subject.OnNext(0);
subject.OnCompleted();
/*
4
3
2
1
0
Completed
*/
  • RxJava
val values = Observable.interval(100, TimeUnit.MILLISECONDS)
val subscription = values
    .skipWhile { v -> v < 2 }
    .hienThi()
readLine()
subscription.dispose()
/*
onNext: 2
onNext: 3
onNext: 4
...
*/
  • RxSwift
let disposeBag = DisposeBag()
Observable.of(1, 2, 3, 4, 5, 6)
    .skipWhile { $0 < 4 }
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)
/*
4
5
6
*/

let disposeBag = DisposeBag()
Observable.of("🐱", "🐰", "🐶", "🐸", "🐷", "🐵")
    .skipWhileWithIndex { element, index in
        index < 3
    }
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)
/*
🐸
🐷
🐵
*/

TakeUntil

TakeUntil gửi dữ liệu từ luồng nguồn cho đến khi một luồng khác bắt đầu gửi dữ liệu.

  • RxNET
var subject = new Subject<int>();
var otherSubject = new Subject<Unit>();
subject
.TakeUntil(otherSubject)
.Subscribe(Console.WriteLine, () => Console.WriteLine("Completed"));
subject.OnNext(1);
subject.OnNext(2);
subject.OnNext(3);
otherSubject.OnNext(Unit.Default);
subject.OnNext(4);
subject.OnNext(5);
subject.OnNext(6);
subject.OnNext(7);
subject.OnNext(8);
subject.OnCompleted();
/*
1
2
3
Completed
*/
  • RxJava
val values = Observable.interval(100, TimeUnit.MILLISECONDS)
val cutoff = Observable.timer(250, TimeUnit.MILLISECONDS)
val subscription = values
    .takeUntil(cutoff)
    .hienThi()
readLine()
subscription.dispose()
/*
onNext: 0
onNext: 1
onComplete
*/
  • RxSwift
let disposeBag = DisposeBag()
let sourceSequence = PublishSubject<String>()
let referenceSequence = PublishSubject<String>()
sourceSequence
    .takeUntil(referenceSequence)
    .subscribe { print($0) }
    .disposed(by: disposeBag)
sourceSequence.onNext("🐱")
sourceSequence.onNext("🐰")
sourceSequence.onNext("🐶")
referenceSequence.onNext("🔴")
sourceSequence.onNext("🐸")
sourceSequence.onNext("🐷")
sourceSequence.onNext("🐵")
/*
next(🐱)
next(🐰)
next(🐶)
completed
*/

TakeWhile

TakeWhile gửi dữ liệu từ luồng nguồn cho đến khi một điều kiện không còn thỏa mãn.

  • RxNET
var subject = new Subject<int>();
subject
.TakeWhile(i => i < 4)
.Subscribe(Console.WriteLine, () => Console.WriteLine("Completed"));
subject.OnNext(1);
subject.OnNext(2);
subject.OnNext(3);
subject.OnNext(4);
subject.OnNext(3);
subject.OnNext(2);
subject.OnNext(1);
subject.OnNext(0);
subject.OnCompleted();
/*
1
2
3
Completed
*/
  • RxJava
val values = Observable.interval(100, TimeUnit.MILLISECONDS)
val subscription = values
    .takeWhile { v -> v < 2 }
    .hienThi()
readLine()
subscription.dispose()
/*
onNext: 0
onNext: 1
onComplete
*/
  • RxSwift
let disposeBag = DisposeBag()
Observable.of(1, 2, 3, 4, 5, 6)
    .takeWhile { $0 < 4 }
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)
/*
1
2
3
*/

Thẻ: RxJava RxJS RxNET RxSwift Observable

Đăng vào ngày 6 tháng 7 lúc 04:07

Thẻ Phổ Biến

©2026 Thành phố Cuồng loạn