RxPY – Conditional and Boolean Operators

all

This operator checks whether all values of a source observable satisfy a given condition.

Syntax

all(predicate)

Parameters

predicate: Boolean. This function will be applied to all values from the source observable and will return true or false based on the given condition.

Return Value

The return value is an observable that will have a Boolean value of true or false, based on the condition applied to all values of the source observable.

Example 1

from rx import of, operators as op
test = of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
sub1 = test.pipe(
   op.all(lambda a: a<10)
)
sub1.subscribe(lambda x: print("The result is {0}".format(x)))

Output

E:pyrx>python testrx.py
The result is False

Example 2

from rx import of, operators as op
test = of(1, 2, 3, 4, 5, 6, 7, 8, 9)
sub1 = test.pipe(
   op.all(lambda a: a<10)
)
sub1.subscribe(lambda x: print("The result is {0}".format(x)))

Output

E:pyrx>python testrx.py
The result is True

contains

This operator returns an observer whose value is true or false if the given value exists in the source observer’s value.

Syntax

contains(value, comparer=None)

Parameters

value: The value to check if it exists in the source observer.

comparer: Optional. This is a comparator function that is applied to the value in the source observer.

Examples

from rx import of, operators as op
test = of(17, 25, 34, 56, 78)
sub1 = test.pipe(
   op.contains(34)
)
sub1.subscribe(lambda x: print("The value is {0}".format(x)))

Output

E:pyrx>python testrx.py
The value is True

Example 2: Using a comparator

from rx import of, operators as op
test = of(17, 25, 34, 56, 78)
sub1 = test.pipe( op.contains(34, lambda x, y: x == y)
)
sub1.subscribe(lambda x: print("The valus is {0}".format(x)))

Output

E:pyrx>python testrx.py
The value is True

default_if_empty

This operator returns a default value if the source observation is empty.

Syntax

default_if_empty(default_value=None)

Parameters

default_value: Optional. It will return the output, or None if nothing is passed as default_value; otherwise, it will return whatever value is passed.

Return Value

If the source observer is null, this returns an observer with the default value.

Example 1

from rx import of, operators as op
test = of()
sub1 = test.pipe(
   op.default_if_empty()
)
sub1.subscribe(lambda x: print("The value is {0}".format(x)))

Output

E:pyrx>python testrx.py
The value is None

Example 2: Pass default_value

from rx import of, operators as op
test = of()
sub1 = test.pipe(
   op.default_if_empty("Empty!")
)
sub1.subscribe(lambda x: print("The value is {0}".format(x)))

Output

E:pyrx>python testrx.py
The value is Empty!

sequence_equal

This operator compares two observable sequences, or arrays of values, and returns an observable with a true or false value.

Syntax

sequence_equal(second_seq, comparer=None)

Parameters

second_seq: The observable or array to be compared with the first observable.

comparer: Optional. Comparator function used to compare the values in the two sequences.

Example

from rx import of, operators as op
test = of(1,2,3)
test1 = of(1,2,3)
sub1 = test.pipe(
   op.sequence_equal(test1)
)
sub1.subscribe(lambda x: print("The value is {0}".format(x)))

Output

E:pyrx>python testrx.py
The value is True

Example: Using a comparator function

from rx import of, operators as op
test = of(1,2,3)
test1 = of(1,2,3)
sub1 = test.pipe( op.sequence_equal(test1, lambda x, y : x == y)
)
sub1.subscribe(lambda x: print("The value is {0}".format(x)))

Output

E:pyrx>python testrx.py
The value is True

skip_until

This operator discards the value of the source observable until the second observable emits a value.

Syntax

skip_until(observable)

Parameters

observable: The second observable. When it emits a value, the source observable is triggered.

Return Value

This will return an observer that will hold the value of the source observer until the second observer emits a value.

Example

from rx import interval,range, operators as op
from datetime import date
test = interval(0)
test1 = range(10)
sub1 = test1.pipe(
   op.skip_until(test)
)
sub1.subscribe(lambda x: print("The value is {0}".format(x)))

Output

E:pyrx>python testrx.py
The value is 0
The value is 1
The value is 2
The value is 3
The value is 4
The value is 5
The value is 6
The value is 7
The value is 8
The value is 9

skip_while

This operator returns an observer containing the values of the source observer that satisfy the passed condition.

Syntax

skip_while(predicate_func)

Parameters

predicate_func: This function will be applied to all values of the source observer and return the values that satisfy the condition.

Return Value

It returns an observer containing the values of the source observer that satisfy the condition.

Example

from rx import of, operators as op
from datetime import date
test = of(1,2,3,4,5,6,7,8,9,10)
sub1 = test.pipe(
   op.skip_while(lambda x : x < 5)
)
sub1.subscribe(lambda x: print("The value is {0}".format(x)))

Output

E:pyrx>python testrx.py
The value is 5
The value is 6
The value is 7
The value is 8
The value is 9
The value is 10

take_until

This operator discards the value of the source observer after the second observer emits a value or terminates.

Syntax

take_until(observable)

Parameters

observable: The second observable, which will terminate the source observable when it emits a value.

Return Value

Returns an observer that only receives the value of the source observable when the second observer emits a value.

Example

from rx import timer,range, operators as op
from datetime import date
test = timer(0.01)
test1 = range(500)
sub1 = test1.pipe(
op.take_until(test)
)
sub1.subscribe(lambda x: print("The value is {0}".format(x)))

In this example, you will get the values emitted from the range. However, once the timer completes, it will stop further emissions from the source observer.

Output

E:pyrx>python testrx.py
The value is 0
The value is 1
The value is 2
The value is 3
The value is 4
The value is 5
The value is 6
The value is 7
The value is 8
The value is 9
The value is 10
The value is 11
The value is 12
The value is 13
The value is 14
The value is 15
The value is 16
The value is 17
The value is 18
The value is 19
The value is 20
The value is 21
The value is 22
The value is 23
The value is 24
The value is 25
The value is 26

take_while

This operator discards the value of the source observable when the condition fails.

Syntax

take_while(predicate_func)

Parameters

predicate_func: The function to be evaluated for each value of the source observable.

Return Value

This operator returns an observable with a value until the predicate function is satisfied.

Example

from rx import of, operators as op
from datetime import date
test = of(1,2,3,4,5,6,7,8,9,10)
sub1 = test.pipe(
   op.take_while(lambda a : a < 5)
)
sub1.subscribe(lambda x: print("The value is {0}".format(x)))

Output

E:pyrx>python testrx.py
The value is 1
The value is 2
The value is 3
The value is 4