Conditional and Boolean Operators

This section explains operators with which you conditionally emit or transform Observables, or can do boolean evaluations of them:

Conditional Operators

• amb( ) — given two or more source Observables, emits all of the items from the first of these Observables to emit an item
• defaultIfEmpty( ) — emit items from the source Observable, or emit a default item if the source Observable completes after emitting no items
• skipWhile( ) and skipWhileWithIndex( ) — discard items emitted by an Observable until a specified condition is false, then emit the remainder
• takeUntil( ) — emits the items from the source Observable until a second Observable emits an item
• takeWhile( ) and takeWhileWithIndex( ) — emit items emitted by an Observable as long as a specified condition is true, then skip the remainder

Boolean Operators

---

amb( )

given two or more source Observables, emits all of the items from the first of these Observables to emit an item

When you pass a number of source Observables to amb( ), it will pass through the emissions and messages of exactly one of these Observables: the first one that emits an item to amb( ). It will ignore and discard the emissions of all of the other source Observables.

see also:

• javadoc: amb() (several varieties)
• RxJS: amb
• Linq: Amb

---

defaultIfEmpty( )

emit items from the source Observable, or emit a default item if the source Observable completes after emitting no items

see also:

• javadoc: defaultIfEmpty(default)
• RxJS: defaultIfEmpty
• Linq: DefaultIfEmpty
• Introduction to Rx: DefaultIfEmpty

---

skipWhile( ) and skipWhileWithIndex( )

discard items emitted by an Observable until a specified condition is false, then emit the remainder

The skipWhile( ) method returns an Observable that discards items emitted by the source Observable until such time as a function applied to an item emitted by that Observable returns false, whereupon the new Observable emits that item and the remainder of the items emitted by the source Observable.

numbers = Observable.from( [1, 2, 3, 4, 5, 6, 7, 8, 9] );

numbers.skipWhile({ (5 != it) }).subscribe(
  { println(it); },                          // onNext
  { println("Error: " + it.getMessage()); }, // onError
  { println("Sequence complete"); }          // onCompleted
);

5
6
7
8
9
Sequence complete

The skipWhileWithIndex( ) method is similar, but your function takes an additional parameter: the (zero-based) index of the item being emitted by the source Observable.

numbers = Observable.from( [1, 2, 3, 4, 5, 6, 7, 8, 9] );

numbers.skipWhileWithIndex({ it, index -> ((it < 6) || (index < 5)) }).subscribe(
  { println(it); },                          // onNext
  { println("Error: " + it.getMessage()); }, // onError
  { println("Sequence complete"); }          // onCompleted
);

6
7
8
9
Sequence complete

see also:

• javadoc: skipWhile(predicate)
• javadoc: skipWhileWithIndex(predicate)
• Linq: SkipWhile
• RxJS: skipWhile
• Introduction to Rx: SkipWhile and TakeWhile

---

takeUntil( )

emits the items from the source Observable until another Observable emits an item

see also:

• javadoc: takeUntil(other)
• RxJS: takeUntil
• Linq: TakeUntil

---

takeWhile( ) and takeWhileWithIndex( )

emit items emitted by an Observable as long as a specified condition is true, then skip the remainder

The takeWhile( ) method returns an Observable that mirrors the behavior of the source Observable until such time as a function applied to an item emitted by that Observable returns false, whereupon the new Observable invokes onCompleted( ).

numbers = Observable.from( [1, 2, 3, 4, 5, 6, 7, 8, 9] );

numbers.takeWhile({ ((it < 6) || (0 == (it % 2))) }).subscribe(
  { println(it); },                          // onNext
  { println("Error: " + it.getMessage()); }, // onError
  { println("Sequence complete"); }          // onCompleted
);

1
2
3
4
5
6
Sequence complete

The takeWhileWithIndex( ) method is similar, but your function takes an additional parameter: the (zero-based) index of the item being emitted by the source Observable.

numbers = Observable.from( [1, 2, 3, 4, 5, 6, 7, 8, 9] );

numbers.takeWhileWithIndex({ it, index -> ((it < 6) || (index < 5)) }).subscribe(
  { println(it); },                          // onNext
  { println("Error: " + it.getMessage()); }, // onError
  { println("Sequence complete"); }          // onCompleted
);

1
2
3
4
5
Sequence complete

see also:

• javadoc: takeWhile(predicate)
• javadoc: takeWhileWithIndex(predicate)
• Linq: TakeWhile
• RxJS: takeWhile
• Introduction to Rx: SkipWhile and TakeWhile