def subscribe(self, callback, preroll_grpos_range=0,
                  preroll_from_frames=None, cb_args_map=None):

        pos_frames = None
        if not preroll_grpos_range and preroll_from_frames is not None:
            pos_frames = [(s, preroll_from_frames[s]) for s in self._streams]

        def extra_args_cb_maker(callback, s):
            if s > cb_args_map:
                args = cb_args_map[s]
                if not isinstance(args, (tuple, list)):
                    args = (args,)
            else:
                args = ()

            def cb_wrapper(time, flags, data):
                callback(time, flags, data, *args)

            return cb_wrapper

        def subscriptions_made(results, streams):
            return [(s, r[1]) for (s, r) in zip(streams, results)]

        if cb_args_map is None:
            cb_args_map = {}

        ds = []

        if pos_frames is None:
            for s in self._streams:
                cb = extra_args_cb_maker(callback, s)
                d = s.subscribe(cb, preroll_grpos_range=preroll_grpos_range)
                ds.append(d)
        else:
            for s, from_frame in pos_frames:
                cb = extra_args_cb_maker(callback, s)
                d = s.subscribe(cb, preroll_from_frame=from_frame)
                ds.append(d)

        dl = defer.DeferredList(ds, fireOnOneErrback=1, consumeErrors=1)
        dl.addCallback(subscriptions_made, self._streams)
        return dl
dataset/ETHPy150Open arkadini/twimp/twimp/server/inmemory.py/IMStreamGroup.subscribe/operator `in` -> `>`