Class _StoringObserver

src/react/observable.py:234–247

class _StoringObserver:
    def __init__(self):
        self.data = []
        self.closed = False

    def on_next(self, chunk):
        if not self.closed:
            self.data.append(chunk)

    def on_close(self):
        if self.closed:
            raise Exception('Already closed')

        self.closed = True


def read_until_closed(observable, timeout=None):