rst(O.all)
for i in 9, 10:
    d = subs(O.range(10, 20).all(lambda v: v > i))

"""
Explanation: A Decision Tree of Observable Operators
Part 6: Looking at Entire Streams

source: http://reactivex.io/documentation/operators.html#tree.
(transcribed to RxPY 1.5.7, Py2.7 / 2016-12, Gunther Klessinger, axiros)  

This tree can help you find the ReactiveX Observable operator you’re looking for.
See Part 1 for Usage and Output Instructions.  
We also require acquaintance with the marble diagrams feature of RxPy.
<h2 id="tocheading">Table of Contents</h2>
<div id="toc"></div>

I want to evaluate the entire sequence of items emitted by an Observable
... and emit a single boolean indicating if all of the items pass some test all
End of explanation
"""


rst(O.contains)
d = subs(O.range(10, 20).contains(11))
header("equality operation")
d = subs(O.range(10, 20).contains(11, comparer=lambda x, y: y == x + 1))

def comparer(*a):
    ''' find_index: you get value, index and the observable itself as argument
        some: you get only the value
    '''
    log('comparer args:', *a)
    l.append(1)
    if len(l) > 3:
        # => index will be 0
        return True
    
stream = O.from_((4, 1, 2, 1, 3))
for name in 'find_index', 'some':    
    l = []
    operator = getattr(stream, name)
    rst(operator) # output documentation, reset timer
    d = subs(operator(lambda *a: comparer(*a)))

    

"""
Explanation: ... and emit a single boolean indicating if the Observable emitted any item (that passes some test) contains, find_index, some
End of explanation
"""


rst(O.is_empty)
d = subs(O.from_([]).is_empty())

"""
Explanation: ... and emit a single boolean indicating if the Observable emitted no items is_empty
End of explanation
"""


rst(O.sequence_equal)
def f(x, y):
    log('got', x, y)
    return str(y) == str(x).upper()
d = subs(O.from_('rxpy rocks').sequence_equal(
         O.from_('RXPY ROCKS'), lambda x, y: f(x, y)))

header("there is no order guarantee of arguments in the comparer:")
d = subs(marble_stream('a-b------c-d|').sequence_equal(
         marble_stream('A-B----C-D|'), lambda x, y: f(x, y)))



"""
Explanation: ... and emit a single boolean indicating if the sequence is identical to one emitted by a second Observable sequence_equal
End of explanation
"""


rst(O.average)
d = subs(O.from_('1199').average(lambda x: int(x)))

"""
Explanation: ... and emit the average of all of their values average
End of explanation
"""


rst(O.sum)
d = subs(O.from_('1199').sum(lambda x: int(x)))

"""
Explanation: ... and emit the sum of all of their values sum
End of explanation
"""


rst(O.count)
d = subs(O.from_('1199').count(lambda x: int(x) > 2))

"""
Explanation: ... and emit a number indicating how many items were in the sequence count
End of explanation
"""


rst(O.max)
d = subs(O.from_('1199').max(lambda x, y: int(x) + int(y) < 5))

rst(O.max_by)
d = subs(O.from_('1271246').max_by(lambda x: int(x)  < 5))

"""
Explanation: ...and emit the item with the maximum value max, max_by
End of explanation
"""


rst(O.min)
d = subs(O.from_('1199').min(lambda x, y: int(x) + int(y) > 5))

rst(O.min_by)
d = subs(O.from_('1271246').min_by(lambda x: int(x)  < 5))

"""
Explanation: ... and emit the item with the minimum value min, min_by
End of explanation
"""


rst(O.scan)
# printing original value next to the aggregate:
d = subs(O.from_('12345').scan(
        lambda x, y: [y, int(x[1]) + int(y)],
        seed=[0, 0]))

rst(O.expand)
# printing original value next to the aggregate:
d = subs(O.just(42).expand(lambda x: O.just(42 + x)).take(5))

"""
Explanation: ... by applying an aggregation function to each item in turn (or recursing into its result) and emitting the result scan, expand
End of explanation
"""


