human
...from scratch, with custom logic and cleanup (calling a function again and again): create
 Give me a the Python code you requested?
assistant
#Here is the Python code you requested.
rx = O.create
rst(rx)

def f(obs):
    # this function is called for every observer
    obs.on_next(rand())
    obs.on_next(rand())
    obs.on_completed()
    def cleanup():
        log('cleaning up...')
    return cleanup
stream = O.create(f).delay(200) # the delay causes the cleanup called before the subs gets the vals
d = subs(stream)
d = subs(stream)



sleep(0.5)
rst(title='Exceptions are handled nicely')
l = []
def excepting_f(obs):
    for i in range(3):
        l.append(1)
        obs.on_next('%s %s (observer hash: %s)' % (i, 1. / (3 - len(l)), hash(obs) ))
    obs.on_completed()

stream = O.create(excepting_f)
d = subs(stream)
d = subs(stream)



rst(title='Feature or Bug?')
print('(where are the first two values?)')
l = []
def excepting_f(obs):
    for i in range(3):
        l.append(1)
        obs.on_next('%s %s (observer hash: %s)' % (i, 1. / (3 - len(l)), hash(obs) ))
    obs.on_completed()

stream = O.create(excepting_f).delay(100)
d = subs(stream)
d = subs(stream)
# I think its an (amazing) feature, preventing to process functions results of later(!) failing functions

rx = O.generate
rst(rx)
"""The basic form of generate takes four parameters:

the first item to emit
a function to test an item to determine whether to emit it (true) or terminate the Observable (false)
a function to generate the next item to test and emit based on the value of the previous item
a function to transform items before emitting them
"""
def generator_based_on_previous(x): return x + 1.1
def doubler(x): return 2 * x
d = subs(rx(0, lambda x: x < 4, generator_based_on_previous, doubler))

rx = O.generate_with_relative_time
rst(rx)
stream = rx(1, lambda x: x < 4, lambda x: x + 1, lambda x: x, lambda t: 100)
d = subs(stream)