buffer operator? about aioreactive HOT 3 CLOSED

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.