

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 in 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/original