async-rx

Versions following Semantic Versioning

Overview

A free implementation of "rx" alias "react" alias "the power of observable pattern and his children" for application server side.

Implementation is based on:

• async function with curio framework
• user Protocol declaration for type checking
• our friends: poetry, flake8, black, isort, pytest, mypy, sphinx, tox, travis ...
• a taste of namedtuple
• a lot of closed variable, clojure function, inner function
• a taste of curiosity
• shake it, again a little bit and tadaa !

Installation

Install this library directly into an activated virtual environment:

$ pip install async-rx

or add it to your Poetry project:

$ poetry add async-rx

Note about Python version support:

• python code and test use python 3.8+
• typing extensions give us an optional python < 3.8 support

API and Usage

After installation, the package can imported:

$ python
>>> import async_rx
>>> async_rx.__version__

Take a look on documentation and API.

A short sample

With this amazing observer:

class ObserverCounterCollector:
    def __init__(self):
        self.on_next_count = 0
        self.on_completed_count = 0
        self.on_error_count = 0
        self.items: Any = list([]) # a bad idea isn't it

    async def on_next(self, item: Any) -> None:
        """Process item."""
        self.items.append(item)
        self.on_next_count += 1

    async def on_completed(self) -> None:
        """Signal completion of this observable."""
        self.on_completed_count += 1

    async def on_error(self, err: Any) -> None:
        self.on_error_count += 1

We will going to select odd number:

async def _predicate(item: int) -> bool:
    return item % 2 == 0

seeker = ObserverCounterCollector()

observable = rx_range(start=0, stop=100) # create an observable of [0, 1, ..., 99]
sub = await rx_filter(observable=observable, predicate=_predicate).subscribe(an_observer=seeker) # filter and subscribe
sub() # release resource

# we have :
assert seeker.on_next_count == 50
assert seeker.on_completed_count == 1
assert seeker.on_error_count == 0
assert seeker.items[0:6] == [0, 2, 4, 6, 8, 10]

Your new on react/rx and wanna taste it ?

First question: Where to begin ?

If you read this page, you probably ever doing lot of search on google & co, and probably loose as me about react component in html/js/whatever. I can't purpose your the best state of the explanation, but... AMHPOV, if you like to known how slug are done behind the scene, you should remember:

• what is Observable pattern (or listener, alias callback)
• what is an event emiter (something which send event ?)

ok, with this in mind:

• an observer receive event from an observable when he subscribe on it
• events are "on_next(item)", "on_completed()", and ... "on_error(err)"

Take a look in "protocole.py" and come back :)

Did you see Observer/Observable/XXXHandler/Subscribe ? And Subcription (yes it will called for unsubscribe) ?

So rx_from([1, 2, ...]) create an observable which will send items of list in sequence when an observer subscribe on it. Take time to look at test unit :) You can go into observable module and see all rx_from, rx_defer, rx_last, ...

But but but, what is a freaking "subject" ?

It's like an observer AND an observable which can multicast items from an observable to several observers. As it is an observer, it can receive data from somewhere. As it is an observable, observer can subscribe on it.

Take a look under subject module and test unit, see what is a replay_subject, funny no ? See the function subject.

At the last for the goods: the big gun, the "ConnectableObservable" alias multicast. oh my god, they kill kenny!

Its like a subject which you can connect/disconnect as you want or automatically (with a call on ref_count). "connect" mean that subject start to receive from observable, so items will be send on observers.

I hope this could help you a little bit :)

Reference

• RxPy
• reactivex
• reactjs
• The Reactive Manifesto
• working with unreliable observers using reactive extensions
• ...