buffer operator?

[nardi]

Hi, I've been implementing a state management library (http://github.com/nardi/pobx) using RxPY, and am now trying to build an async version using this library. However, in my RxPY implementation I make use of the buffer operator to propagate a bunch of values all at once. Are there any plans to implement this operator here too, or should I try to get my own version running? FWIW, in RxPY buffer is in turn implemented using the window operator, which is also not available here, so it might be a bit of work.

[nardi]

I created my own helper class, turned out it wasn't too difficult :) I've posted the code below. Probably not conformant to your operator API, but fine for my needs ATM.

from aioreactive import AsyncAnonymousObserver
from aioreactive.subject import AsyncSubject
from asyncinit import asyncinit # External package
from .utils import dropargs # Wrapper function which drops arguments

@asyncinit
class BufferOperator():
    async def __init__(self, boundaries):
        self.emitter = AsyncSubject()
        self.buffer = []

        async def emit_buffer():
            buffer = self.buffer
            self.buffer = []
            await self.emitter.asend(buffer)

        self.subs = [
            await boundaries.subscribe_async(AsyncAnonymousObserver(dropargs(emit_buffer)))
        ]

    async def __call__(self, receiver):
        async def add_to_buffer(x):
            self.buffer.append(x)

        self.subs.append(await receiver.subscribe_async(AsyncAnonymousObserver(add_to_buffer)))
        return self

    async def subscribe_async(self, *args, **kwargs):
        return await self.emitter.subscribe_async(*args, **kwargs)

    async def dispose_async(self):
        for sub in self.subs:
            await sub.dispose_async()
        self.buffer.clear()

async def buffer(boundaries):
    return await BufferOperator(boundaries)

# Used as follows:
values = AsyncSubject()
boundaries = AsyncSubject()
grouped_values = await pipe(
    values,
    await buffer(boundaries)
)
# Then grouped_values can be subscribed to.

[dbrattli]

Thanks. I'll have a look and see if I can re-use it!

[austinnichols101]

I just came across this while looking for buffer functions (specifically rx.operators.buffer_with_time_or_count) for aioreactive.

Would be nice to have buffer parity with rxpy, but sadly I am not (currently) smart enough to write it...