Tiny pub-sub (observer) pattern implementation
Project description
tinypubsub
Tiny pub-sub (observer) pattern implementation.
Usage
from tinypubsub.simple import SimplePublisher
publisher = SimplePublisher()
subscription = publisher.subscribe(lambda message: print(message))
publisher.publish("Hello!")
publisher.unsubscribe(subscription)
Or:
from tinypubsub.simple import SimplePublisher
publisher = SimplePublisher()
with publisher.subscribe(lambda message: print(message)):
publisher.publish("Hello!")
Features
tinypubsub.Publisher
has abstract methods:publish
,subscribe
,unsubscribe
,unsubscribe_all
.tinypubsub.simple.SimplePublisher
: Dict-based implementation ofPublisher
.tinypubsub.weakref.WeakrefPublisher
: WeakKeyDictionary-based implementation ofPublisher
.
API
Module tinypubsub
abstract class Publisher[Message]
Abstract publisher class.
type parameter Message
Type of message that will be published.
abstract method publish(message: Message) -> None
Publish message to subscribers.
abstract method subscribe(subscriber: Callable[[Message], None]) -> Subscription
Add subscriber.
abstract method unsubscribe(subscription: Subscription) -> None
Delete subscriber.
abstract method unsubscribe_all() -> None
Delete all subscribers.
class Subscription
Return value of Publisher.subscribe()
.
It can be used as a context manager as:
with publisher.subscribe(...) as subscription:
...
and subscription.unsubscribe()
will be called when exit.
method unsubscribe() -> None
subscription.unsubscribe()
is the same as publisher.unsubscribe(subscription)
, where subscription = publisher.subscribe(...)
.
Module tinypubsub.simple
class SimplePublisher[Message]
Implementation of Publisher[Message]
.
Module tinypubsub.weakref
class WeakrefPublisher[Message]
Implementation of Publisher[Message]
.
This implementation uses WeakKeyDictionary to manage subscribers. This may prevent a memory leak if subscription loses all strong references before unsubscribed:
publisher = WeakrefPublisher()
subscription = publisher.subscribe(...)
assert len(publisher._subscribers) == 1
del subscription
assert len(publisher._subscribers) == 0
Note that the unsubscribe
method will not be called in the above case,
so normally you should unsubscribe explicitly or use context manager.
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for tinypubsub-0.1.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 84e48954d4d41765e0aa245b736ec529a6cdd79f21cb285c809a94869699f8ef |
|
MD5 | 2c532d8d90cc95b9bda60639194f7c31 |
|
BLAKE2b-256 | e40bbeb825eed1e77143b3f72ce7c9f984e9f27496c3c97fcc33d02a47257af0 |