Add api and operator

examples/examplecreate.py

@@ -0,0 +1,73 @@
+import time
+from threading import Thread
+
+import rxbp
+from rxbp.acknowledgement.continueack import continue_ack
+from rxbp.schedulers.asyncioscheduler import AsyncIOScheduler
+from rxbp.schedulers.threadpoolscheduler import ThreadPoolScheduler
+from rxbp.testing.tobserver import TObserver
+
+
+def demo1():
+    def work(o, skd):
+        for i in range(10):
+            o.on_next([i])
+        o.on_completed()
+
+    source = rxbp.create(work)
+    source = source.pipe(
+        rxbp.op.observe_on(scheduler=AsyncIOScheduler()),
+        # rxbp.op.observe_on(scheduler=EventLoopScheduler()),
+        # rxbp.op.observe_on(scheduler=ThreadPoolScheduler("receiver")),
+        rxbp.op.subscribe_on(scheduler=ThreadPoolScheduler("publisher")),
+    )
+
+    sink = TObserver(immediate_continue=4)
+    # source.subscribe(observer=sink, subscribe_scheduler=ThreadPoolScheduler("publisher"))
+    source.subscribe(observer=sink)
+
+    time.sleep(1)
+    assert sink.received == [0, 1, 2, 3, 4]
+
+    print("-" * 80)
+    sink.immediate_continue += 2
+    sink.ack.on_next(continue_ack)
+    time.sleep(1)
+    assert sink.received == [0, 1, 2, 3, 4, 5, 6, 7]
+
+    print("-" * 80)
+    sink.immediate_continue += 1
+    sink.ack.on_next(continue_ack)
+    time.sleep(1)
+    assert sink.received == [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
+
+
+def demo2():
+    def counter(sink):
+        while True:
+            time.sleep(5)
+            print(f"[**client**] received: ", sink.received)
+
+    def work(o, skd):
+        for i in range(100_000):
+            o.on_next([i])
+        o.on_completed()
+
+    source = rxbp.create(work)
+    # no back-pressure until the buffer is full
+    source = source.pipe(
+        rxbp.op.buffer(10)
+    )
+
+    sink = TObserver(immediate_continue=4)
+    source.subscribe(observer=sink, subscribe_scheduler=ThreadPoolScheduler("publisher"))
+
+    t1 = Thread(target=counter, args=(sink,))
+    t1.start()
+    t1.join()
+
+
+if __name__ == '__main__':
+    demo1()
+
+    # demo2()

examples/exampleinterval.py

@@ -0,0 +1,23 @@
+import time
+from threading import Thread
+
+import rxbp
+from rxbp.schedulers.timeoutscheduler import TimeoutScheduler
+from rxbp.testing.tobserver import TObserver
+
+
+def counter(sink):
+    while True:
+        time.sleep(1)
+        print(f"immediate: {sink.immediate_continue}, received: ", sink.received)
+
+
+publisher = rxbp.interval(0.5).pipe(
+    rxbp.op.map(lambda v: "data_" + str(v))
+)
+sink = TObserver(immediate_continue=12)
+publisher.subscribe(observer=sink, subscribe_scheduler=TimeoutScheduler())
+
+t1 = Thread(target=counter, args=(sink,))
+t1.start()
+t1.join()

examples/examplestrategy.py

@@ -0,0 +1,61 @@
+import time
+from threading import Thread
+
+import rxbp
+from rxbp.acknowledgement.acksubject import AckSubject
+from rxbp.acknowledgement.continueack import continue_ack
+from rxbp.overflowstrategy import DropOld, ClearBuffer, Drop
+from rxbp.schedulers.asyncioscheduler import AsyncIOScheduler
+from rxbp.schedulers.threadpoolscheduler import ThreadPoolScheduler
+from rxbp.schedulers.timeoutscheduler import TimeoutScheduler
+from rxbp.testing.tobserver import TObserver
+
+
+def demo1():
+    def counter(sink):
+        while True:
+            time.sleep(5)
+            print(f"[**client**] received: ", sink.received)
+
+    publisher = rxbp.interval(0.5).pipe(
+        # rxbp.op.strategy(DropOld(buffer_size=15))
+        rxbp.op.strategy(ClearBuffer(buffer_size=15))
+    )
+    sink = TObserver(immediate_continue=5)
+    publisher.subscribe(observer=sink, subscribe_scheduler=TimeoutScheduler())
+
+    t1 = Thread(target=counter, args=(sink,))
+    t1.start()
+    t1.join()
+
+
+def demo2():
+    def counter(sink):
+        while True:
+            time.sleep(5)
+            print(f"[**client**] received: ", sink.received)
+
+    def work(o, skd):
+        for i in range(1_000):
+            o.on_next([i])
+        o.on_completed()
+
+    source = rxbp.create(work)
+    source = source.pipe(
+        rxbp.op.strategy(DropOld(buffer_size=100)),
+        # rxbp.op.strategy(ClearBuffer(buffer_size=15)),
+        # rxbp.op.strategy(Drop(buffer_size=15)),
+    )
+
+    sink = TObserver(immediate_continue=5)
+    source.subscribe(observer=sink, subscribe_scheduler=ThreadPoolScheduler("publisher"))
+
+    t1 = Thread(target=counter, args=(sink,))
+    t1.start()
+    t1.join()
+
+
+if __name__ == '__main__':
+    # demo1()
+
+    demo2()

rxbp/__init__.py

@@ -3,7 +3,7 @@
 from . import multicast
 from . import op
 from .source import from_iterable, from_range, from_list, return_value, from_rx, concat, zip, \
-    merge, empty
+    merge, empty, create, interval
 
 from_ = from_iterable
 range = from_range

rxbp/flowables/createflowable.py

@@ -0,0 +1,24 @@
+from rx.core import typing
+
+from rxbp.init.initsubscription import init_subscription
+from rxbp.mixins.flowablemixin import FlowableMixin
+from rxbp.observables.subscriptionobservable import SubscriptionObservable
+from rxbp.subscriber import Subscriber
+from rxbp.subscription import Subscription
+
+
+class CreateFlowable(FlowableMixin):
+    def __init__(
+            self,
+            observer: typing.Subscription,
+    ):
+        self._observer = observer
+
+    def unsafe_subscribe(self, subscriber: Subscriber) -> Subscription:
+        return init_subscription(
+            observable=SubscriptionObservable(
+                source=self._observer,
+                scheduler=subscriber.scheduler,
+                subscribe_scheduler=subscriber.subscribe_scheduler,
+            )
+        )

rxbp/flowables/evictingbufferflowable.py

@@ -0,0 +1,29 @@
+from dataclasses import dataclass
+
+from rxbp.mixins.flowablemixin import FlowableMixin
+from rxbp.observables.evictingobservable import EvictingObservable
+from rxbp.overflowstrategy import OverflowStrategy
+from rxbp.subscriber import Subscriber
+from rxbp.subscription import Subscription
+
+
+@dataclass
+class EvictingBufferFlowable(FlowableMixin):
+    source: FlowableMixin
+    overflow_strategy: OverflowStrategy
+    # buffer_size: int
+
+    def unsafe_subscribe(self, subscriber: Subscriber) -> Subscription:
+        subscription = self.source.unsafe_subscribe(subscriber=subscriber)
+
+        observable = EvictingObservable(
+            subscription.observable,
+            # buffer_size=self.buffer_size,
+            scheduler=subscriber.scheduler,
+            subscribe_scheduler=subscriber.subscribe_scheduler,
+            overflow_strategy=self.overflow_strategy,
+        )
+
+        return subscription.copy(
+            observable=observable,
+        )

rxbp/flowables/intervalflowable.py

@@ -0,0 +1,25 @@
+from rx.core import typing
+
+from rxbp.init.initsubscription import init_subscription
+from rxbp.internal.timer import _interval
+from rxbp.mixins.flowablemixin import FlowableMixin
+from rxbp.observables.subscriptionobservable import SubscriptionObservable
+from rxbp.subscriber import Subscriber
+from rxbp.subscription import Subscription
+
+
+class IntervalFlowable(FlowableMixin):
+    def __init__(
+            self,
+            period: typing.RelativeTime,
+    ):
+        self.period = period
+
+    def unsafe_subscribe(self, subscriber: Subscriber) -> Subscription:
+        return init_subscription(
+            observable=SubscriptionObservable(
+                source=_interval(self.period),
+                scheduler=subscriber.scheduler,
+                subscribe_scheduler=subscriber.subscribe_scheduler,
+            )
+        )

rxbp/internal/timer.py

@@ -0,0 +1,127 @@
+from datetime import datetime
+from typing import Optional
+
+from rx.core import typing
+from rx.disposable import MultipleAssignmentDisposable
+
+from rxbp.schedulers.timeoutscheduler import TimeoutScheduler
+
+
+def _interval(period: typing.RelativeTime,
+              scheduler: Optional[typing.Scheduler] = None
+              ) -> typing.Subscription:
+    return _timer(period, period, scheduler)
+
+
+def _timer(duetime: typing.AbsoluteOrRelativeTime,
+           period: Optional[typing.RelativeTime] = None,
+           scheduler: Optional[typing.Scheduler] = None
+           ) -> typing.Subscription:
+    if isinstance(duetime, datetime):
+        if period is None:
+            return observable_timer_date(duetime, scheduler)
+        else:
+            return observable_timer_duetime_and_period(duetime, period, scheduler)
+
+    if period is None:
+        return observable_timer_timespan(duetime, scheduler)
+
+    return observable_timer_timespan_and_period(duetime, period, scheduler)
+
+
+def observable_timer_date(duetime,
+                          scheduler: Optional[typing.Scheduler] = None
+                          ) -> typing.Subscription:
+    def subscribe(observer, scheduler_=None):
+        _scheduler = scheduler or scheduler_
+        if not hasattr(_scheduler, "schedule_periodic"):
+            _scheduler = TimeoutScheduler.singleton()
+
+        def action(scheduler, state):
+            observer.on_next([0])
+            observer.on_completed()
+
+        return _scheduler.schedule_absolute(duetime, action)
+
+    return subscribe
+
+
+def observable_timer_duetime_and_period(duetime,
+                                        period,
+                                        scheduler: Optional[typing.Scheduler] = None
+                                        ) -> typing.Subscription:
+    def subscribe(observer, scheduler_=None):
+        _scheduler = scheduler or scheduler_
+        if not hasattr(_scheduler, "schedule_periodic"):
+            _scheduler = TimeoutScheduler.singleton()
+
+        nonlocal duetime
+
+        if not isinstance(duetime, datetime):
+            duetime = _scheduler.now + _scheduler.to_timedelta(duetime)
+
+        p = max(0.0, _scheduler.to_seconds(period))
+        mad = MultipleAssignmentDisposable()
+        dt = duetime
+        count = 0
+
+        def action(scheduler, state):
+            nonlocal dt
+            nonlocal count
+
+            if p > 0.0:
+                now = scheduler.now
+                dt = dt + scheduler.to_timedelta(p)
+                if dt <= now:
+                    dt = now + scheduler.to_timedelta(p)
+
+            observer.on_next([count])
+            count += 1
+            mad.disposable = scheduler.schedule_absolute(dt, action)
+
+        mad.disposable = _scheduler.schedule_absolute(dt, action)
+        return mad
+
+    return subscribe
+
+
+def observable_timer_timespan(duetime: typing.RelativeTime,
+                              scheduler: Optional[typing.Scheduler] = None
+                              ) -> typing.Subscription:
+    def subscribe(observer, scheduler_=None):
+        _scheduler = scheduler or scheduler_
+        if not hasattr(_scheduler, "schedule_periodic"):
+            _scheduler = TimeoutScheduler.singleton()
+
+        d = _scheduler.to_seconds(duetime)
+
+        def action(scheduler, state):
+            observer.on_next([0])
+            observer.on_completed()
+
+        if d <= 0.0:
+            return _scheduler.schedule(action)
+        return _scheduler.schedule_relative(d, action)
+
+    return subscribe
+
+
+def observable_timer_timespan_and_period(duetime: typing.RelativeTime,
+                                         period: typing.RelativeTime,
+                                         scheduler: Optional[typing.Scheduler] = None
+                                         ) -> typing.Subscription:
+    if duetime == period:
+        def subscribe(observer, scheduler_=None):
+            _scheduler = scheduler or scheduler_
+            if not hasattr(_scheduler, "schedule_periodic"):
+                _scheduler = TimeoutScheduler.singleton()
+
+            def action(count):
+                observer.on_next([count])
+                return count + 1
+
+            return _scheduler.schedule_periodic(period, action, state=0)
+
+        return subscribe
+
+    return observable_timer_duetime_and_period(duetime, period, scheduler)