Run this pipeline.

        Parameters:
          delay(int): The minimum amount of time, in milliseconds, the
            pipeline should be delayed by.

        Returns:
          pipeline: Itself.
512672
def run(self, *, delay=None):
        
        self.broker.enqueue(self.messages[0], delay=delay)
        return self
99310
def call_later(self, delay, func, *args, **kwargs):
        
        # "quantize" the delay to the nearest bucket
        now = self._get_seconds()
        real_time = int(now + delay) * 1000
        real_time -= int(real_time % self._bucket_milliseconds)
        call = _BatchedCall(self, real_time, lambda: func(*args, **kwargs))
        try:
            self._buckets[real_time][1].append(call)
        except KeyError:
            # new bucket; need to add "actual" underlying IDelayedCall
            diff = (real_time / 1000.0) - now
            # we need to clamp this because if we quantized to the
            # nearest second, but that second is actually (slightly)
            # less than the current time 'diff' will be negative.
            delayed_call = self._create_delayed_call(
                max(0.0, diff),
                self._notify_bucket, real_time,
            )
            self._buckets[real_time] = (delayed_call, [call])
        return call
1055598
def doIteration(self, delay=None, fromqt=False):
        ''
        
        if not self.running and self._blockApp:
            self._blockApp.quit()
        self._timer.stop()
        delay = max(delay, 1)
        if not fromqt:
            self.qApp.processEvents(QtCore.QEventLoop.AllEvents, delay * 1000)
        if self.timeout() is None:
            timeout = 0.1
        elif self.timeout() == 0:
            timeout = 0
        else:
            timeout = self.timeout()
        self._timer.setInterval(timeout * 1000)
        self._timer.start()
198819
def pipe(self, command, timeout=None, cwd=None):
        
        if not timeout:
            timeout = self.timeout

        if not self.was_run:
            self.run(block=False, cwd=cwd)

        data = self.out

        if timeout:
            c = Command(command, timeout)
        else:
            c = Command(command)

        c.run(block=False, cwd=cwd)
        if data:
            c.send(data)
        c.block()
        return c
361525
def delay_or_eager(self, *args, **kwargs):
        
        return self.async_or_eager(args=args, kwargs=kwargs)
339477
public static Command delay (final Command command, final int delay)
    {
        return (command == null) ? null : new Command() {
            public void execute () {
                new Timer() {
                    @Override public void run () {
                        command.execute();
                    }
                }.schedule(delay);
            }
        };
    }
357570
def wait_stop(self):
        
        if not self.running:
            return
        self.stop()
        try:
            yield from self._task
        except asyncio.CancelledError:
            pass
946229
def _main_loop(self):
        
        fetch_priorities_delay = timedelta(seconds=self.fetch_priorities_delay)
        fetch_delayed_delay = timedelta(seconds=self.fetch_delayed_delay)

        while not self.must_stop():
            self.set_status('waiting')

            if self.last_update_keys + fetch_priorities_delay < datetime.utcnow():
                self.update_keys()

            if self.last_requeue_delayed + fetch_delayed_delay < datetime.utcnow():
                self.requeue_delayed_jobs()

            try:
                queue_and_job = self.wait_for_job()
                if queue_and_job is None:
                    # timeout for blpop
                    continue
                queue, job = queue_and_job
            except Exception as e:
                self.log('Unable to get job: %s\n%s'
                    % (str(e), traceback.format_exc()), level='error')
            else:
                self.num_loops += 1
                try:
                    identifier = 'pk:%s' % job.pk.get()
                except Exception as e:
                    identifier = '??'
                try:
                    self.set_status('running')
                    identifier, status = job.hmget('identifier', 'status')
                     # some cache, don't count on it on subclasses
                    job._cached_identifier = identifier
                    job._cached_status = status
                    queue._cached_name = queue.name.hget()

                    if status == STATUSES.DELAYED:
                        self.job_delayed(job, queue)
                    elif status != STATUSES.WAITING:
                        self.job_skipped(job, queue)
                    else:
                        try:
                            self.job_started(job, queue)
                            job_result = self.callback(job, queue)
                        except Exception as e:
                            trace = None
                            if self.save_tracebacks:
                                trace = traceback.format_exc()
                            self.job_error(job, queue, e, trace)
                        else:
                            job._cached_status = job.status.hget()
                            if job._cached_status == STATUSES.DELAYED:
                                self.job_delayed(job, queue)
                            elif job._cached_status == STATUSES.CANCELED:
                                self.job_skipped(job, queue)
                            else:
                                self.job_success(job, queue, job_result)
                except Exception as e:
                    self.log('[%s] unexpected error: %s\n%s'
                        % (identifier, str(e), traceback.format_exc()), level='error')
                    try:
                        queue.errors.rpush(job.ident)
                    except Exception as e:
                        self.log('[%s] unable to add the error in the queue: %s\n%s'
                            % (identifier, str(e), traceback.format_exc()), level='error')
1116756
def simple_sidekiq_delay_for(interval, options = {})
      Proxy.new(simple_delayed_worker, self, options.merge('at' => Time.now.to_f + interval.to_f))
    end
60441
def runAndWait(self):
        '''
        
        '''
        self._push(self._engine.endLoop, tuple())
        self._driver.startLoop()
201207
def __start_timer(self, delay):
        
        self.__timer = threading.Timer(delay, self.__reconnect)
        self.__timer.daemon = True
        self.__timer.start()
1258036
def retry_with_delay(f, delay=60):
    
    @wraps(f)
    def inner(*args, **kwargs):
        kwargs['timeout'] = 5
        remaining = get_retries() + 1
        while remaining:
            remaining -= 1
            try:
                return f(*args, **kwargs)
            except (requests.ConnectionError, requests.Timeout):
                if not remaining:
                    raise
                gevent.sleep(delay)
    return inner
1254766
def blink(self, delay=1):
        
        self.toggle()
        gevent.spawn_later(delay, self.toggle)
174264
def delay=(d)
      raise ArgumentError, 'delay must be greater than or equal to 0' if Integer(d) < 0

      @images.each { |f| f.delay = Integer(d) }
    end
498964
def run(self):
        ''''''
        while not self.wait(self.delay()):
            try:
                logger.info('Invoking callback %s', self.callback)
                self.callback()
            except StandardError:
                logger.exception('Callback failed')
1282564
def execute(self):
        
        stack = self._stack
        callbacks = self._callbacks
        self._stack = []
        self._callbacks = []

        deferred = []

        build = self._nested_future

        pipe = self._pipeline(self.connection_name)
        for item, args, kwargs, ref in stack:
            f = getattr(pipe, item)
            deferred.append(build(f(*args, **kwargs), ref))

        inject_callbacks = getattr(self.parent, '_inject_callbacks')
        inject_callbacks(deferred + callbacks)
1097246
def run(self):
        
        self._stop = False
        while not self._stop:
            have_sources = self._timers or self._readers or self._writers
            if not self._processor.pending and not have_sources:
                break
            events = QEventLoop.AllEvents
            if not self._processor.pending:
                events |= QEventLoop.WaitForMoreEvents
            self._qapp.processEvents(events)
            if self._processor.pending:
                self._processor.run()