vxgmichel
commented
3 years ago Right, the debounce operator in ReactiveX!
Here's my attempt at implementing it, can you confirm it works as you expect?
import asyncio
from collections import deque
from aiostream import streamcontext, aiter_utils, operator
@operator(pipable=True)
async def debounce(source, delay=0.5):
loop = asyncio.get_event_loop()
# Stream context
async with streamcontext(source) as streamer:
# Get first item
try:
item = await aiter_utils.anext(streamer)
except StopAsyncIteration:
return
# Set of pending tasks, its length should never exceed 1
pending = set()
# Loop over items
while True:
# Compute future deadline THEN yield the item
deadline = loop.time() + delay
yield item
# Run until deadline
queue = deque(maxlen=1)
timeout = deadline - loop.time()
while timeout > 0:
# Wait for next item with timeout
if not pending:
pending = {aiter_utils.anext(streamer)}
done, pending = await asyncio.wait(pending, timeout=timeout)
# Add the current item to a deque with capacity 1
if done:
try:
queue.append(done.pop().result())
except StopAsyncIteration:
return
# Recompute the timeout
timeout = deadline - loop.time()
# Get next item
if queue:
item = queue.popleft()
else:
assert pending
try:
item = await pending.pop()
except StopAsyncIteration:
returnAnd the corresponding test module:
from aiostream import stream, pipe
import pytest
@pytest.mark.asyncio
async def test():
xs = stream.empty() | debounce.pipe(.1) | pipe.list()
assert await xs == []
xs = stream.just(0) | debounce.pipe(.1) | pipe.list()
assert await xs == [0]
xs = stream.range(3) | debounce.pipe(.1) | pipe.list()
assert await xs == [0]
xs = (
stream.chain(
stream.empty() | pipe.delay(0.3),
stream.range(3),
stream.empty() | pipe.delay(0.3),
stream.range(10, 13),
stream.empty() | pipe.delay(0.3),
stream.range(20, 23),
stream.empty() | pipe.delay(0.3),
)
| debounce.pipe(0.1)
| pipe.list()
)
assert await xs == [0, 2, 10, 12, 20, 22]
Jc2k
Masynchin
mikevb3
nirvana-msu
I am combining multiple streams with a merge. The output of this stream is a summary "state" of a system. I.e. each item from the stream is a summary of the state of a system at that point in time. The resulting output is fed into a websocket and used to drive a react app. This react app then shows a handy live updating dashboard. This is working really well, but in the more extreme cases i don't really want to send 100's of events a second down the websocket, so i thought a throttle operator would be a handy tool to have in my belt.
The requirement is to start a timer when an item is received from the source, carry on iterating and when that timer expires only send the most recently received item from the source. And the timer should not be running unless there are 1 or more events waiting to send.
My poc looks like this:
It works, but it feels like its relying on asyncio primitives too much, and that maybe there is a more idiomatic way to do it with aiostream. Can you think of any cleaner ways to implement this. The calls to
anext()especially make me feel like i've overlooked something.