Hi!,

I've been reading your book on reactive programming, and came to see cyclotron repo. I found that the documentation explains too little and I'm wondering if I can make it a bit better. This is also for the sub-repos (which I think should use the same docs url).

Nice work and thank you for the book!

I couldn't find your e-mail address, and the blog post in question does not have a comment section, so I'm commenting here instead. Sorry about the noise. Let me know if there are better channels for this discussion.

First of all, it is a very well-written and informative blog post. Thanks for sharing your insights.

I think that I found an issue with the to_agen async generator. Specifically, this section:

while True:
    i = await queue.get()
    if isinstance(i, OnNext):
        yield i.value
        queue.task_done()
    elif isinstance(i, OnError):
        disposable.dispose()
        raise(Exception(i.value))
    else:
        disposable.dispose()
        break

If the async generator object is not iterated until completion, then disposable.dispose() is never called.

It is up to the user code to iterate the async generator object. The user can even choose to just iterate the async generator a couple of times and then discard it. In this case, a GeneratorExit exception will be thrown when the async generator object is garbage collected. Consequently, dispose will never be called.

This can be easily fixed by calling dispose in a finally block instead:

try:
    while True:
        i = await queue.get()
        if isinstance(i, OnNext):
            yield i.value
            queue.task_done()
        elif isinstance(i, OnError):
            raise(Exception(i.value))
        else:
            break
finally:
    disposable.dispose()

I encountered this issue in production where it led to strange errors. Thought I should let you know.