def callback(self, callback_fn=None):
        all_data_storages = self.manifest.get('data_storages', [])
        if len(all_data_storages) == 0:
            print("There are no callback notifications associated with the indexing jobs. So we are Done here.")
        else:
            print("Initiating, sending the callback notifications after the respective transformations ")
            for index in all_data_storages:
                data_storage_id = index.get('data_storage_id')
                callback_config = self.get_callback_for_index(data_storage_id=data_storage_id)
                if callback_config:
                    try:
                        self.trigger_callback(callback_config=callback_config)
                    except Exception as e:
                        print("Failed to send callback[{}] with error: {}".format(callback_config.get("callback_id"),
                                                                                  e))

        if callback_fn is None:
            reactor.stop()
        else:
            callback_fn() 

def dataReceived(self, data):
        if not self.known_proto:
            self.known_proto = self.transport.negotiatedProtocol
            assert self.known_proto == b'h2'

        events = self.conn.receive_data(data)

        for event in events:
            if isinstance(event, ResponseReceived):
                self.handleResponse(event.headers, event.stream_id)
            elif isinstance(event, DataReceived):
                self.handleData(event.data, event.stream_id)
            elif isinstance(event, StreamEnded):
                self.endStream(event.stream_id)
            elif isinstance(event, SettingsAcknowledged):
                self.settingsAcked(event)
            elif isinstance(event, StreamReset):
                reactor.stop()
                raise RuntimeError("Stream reset: %d" % event.error_code)
            else:
                print(event)

        data = self.conn.data_to_send()
        if data:
            self.transport.write(data) 

def convert(db):

    log.info(db.rc)
    log.info("Reading metrics keys")
    keys = yield db.rc.keys(METRIC_OLD_PREFIX.format("*"))
    log.info("Converting ...")
    for key in keys:
        _, name = key.split(':')
        try:
            pipe = yield db.rc.pipeline()
            metrics = yield db.rc.zrange(key)
            for metric in metrics:
                value, timestamp = metric.split()
                pipe.zadd(METRIC_PREFIX.format(name), timestamp, "{0} {1}".format(timestamp, value))
            yield pipe.execute_pipeline()
        except txredisapi.ResponseError as e:
            log.error("Can not convert {key}: {e}", key=key, e=e)
        log.info("Metric {name} converted", name=name)

    yield db.stopService()
    reactor.stop() 

def main(number):

    def get_metrics():
        return [
            ("checker.time.%s.%s" %
             (config.HOSTNAME,
              number),
                spy.TRIGGER_CHECK.get_metrics()["sum"]),
            ("checker.triggers.%s.%s" %
             (config.HOSTNAME,
              number),
                spy.TRIGGER_CHECK.get_metrics()["count"]),
            ("checker.errors.%s.%s" %
             (config.HOSTNAME,
              number),
                spy.TRIGGER_CHECK_ERRORS.get_metrics()["count"])]

    graphite.sending(get_metrics)

    def start(db):
        checker = TriggersCheck(db)
        checker.start()
        reactor.addSystemEventTrigger('before', 'shutdown', checker.stop)

    run(start) 

def check_for_phase1_utxos(self, utxos, cb=None):
        """Any participant needs to wait for completion of phase 1 through
        seeing the utxos on the network. Optionally pass callback for start
        of phase2 (redemption phase), else default is state machine tick();
        must have signature callback(utxolist).
        Triggered on number of confirmations as set by config.
        This should be fired by task looptask, which is stopped on success.
        """
        result = cs_single().bc_interface.query_utxo_set(utxos,
                                                         includeconf=True)
        if None in result:
            return
        for u in result:
            if u['confirms'] < self.coinswap_parameters.tx01_confirm_wait:
                return
        self.loop.stop()
        if cb:
            cb()
        else:
            self.sm.tick() 

def execute():
    cfg = Configuration(".bitmaskctl")
    print_json = '--json' in sys.argv

    cli = BitmaskCLI(cfg)
    cli.data = ['core', 'version']
    args = None if '--noverbose' in sys.argv else ['--verbose']

    if should_start(sys.argv):
        timeout_fun = cli.start
    else:
        def status_timeout(args):
            raise RuntimeError('bitmaskd is not running')
        timeout_fun = status_timeout

    try:
        yield cli._send(
            timeout=0.1, printer=_null_printer,
            errb=lambda: timeout_fun(args))
    except Exception, e:
        print(Fore.RED + "ERROR: " + Fore.RESET +
              "%s" % str(e))
        yield reactor.stop() 

def stop_reactor():
    """Stop the reactor and join the reactor thread until it stops.
    Call this function in teardown at the module or package level to
    reset the twisted system after your tests. You *must* do this if
    you mix tests using these tools and tests using twisted.trial.
    """
    global _twisted_thread

    def stop_reactor():
        '''Helper for calling stop from withing the thread.'''
        reactor.stop()

    reactor.callFromThread(stop_reactor)
    reactor_thread.join()
    for p in reactor.getDelayedCalls():
        if p.active():
            p.cancel()
    _twisted_thread = None 

def make(self):

        # connect to crossbar router/broker
        self.runner = ApplicationRunner(self.url, self.realm, extra=dict(self.config))

        # run application session
        self.deferred = self.runner.run(self.session_class, start_reactor=False)

        def croak(ex, *args):
            log.error('Problem in {name}, please check if "crossbar" WAMP broker is running. args={args}'.format(
                name=self.__class__.__name__, args=args))
            log.error("{ex}, args={args!s}", ex=ex.getTraceback(), args=args)
            reactor.stop()
            raise ex

        self.deferred.addErrback(croak) 

def stop_reactor():
    """Stop the reactor and join the reactor thread until it stops.
    Call this function in teardown at the module or package level to
    reset the twisted system after your tests. You *must* do this if
    you mix tests using these tools and tests using twisted.trial.
    """
    global _twisted_thread

    def stop_reactor():
        '''Helper for calling stop from withing the thread.'''
        reactor.stop()

    reactor.callFromThread(stop_reactor)
    reactor_thread.join()
    for p in reactor.getDelayedCalls():
        if p.active():
            p.cancel()
    _twisted_thread = None 

def _completeWith(self, completionState, deferredResult):
        """
        @param completionState: a L{TaskFinished} exception or a subclass
            thereof, indicating what exception should be raised when subsequent
            operations are performed.

        @param deferredResult: the result to fire all the deferreds with.
        """
        self._completionState = completionState
        self._completionResult = deferredResult
        if not self._pauseCount:
            self._cooperator._removeTask(self)

        # The Deferreds need to be invoked after all this is completed, because
        # a Deferred may want to manipulate other tasks in a Cooperator.  For
        # example, if you call "stop()" on a cooperator in a callback on a
        # Deferred returned from whenDone(), this CooperativeTask must be gone
        # from the Cooperator by that point so that _completeWith is not
        # invoked reentrantly; that would cause these Deferreds to blow up with
        # an AlreadyCalledError, or the _removeTask to fail with a ValueError.
        for d in self._deferreds:
            d.callback(deferredResult) 

def test_cantRegisterAfterRun(self):
        """
        It is not possible to register a C{Application} after the reactor has
        already started.
        """
        reactor = gireactor.GIReactor(useGtk=False)
        self.addCleanup(self.unbuildReactor, reactor)
        app = Gio.Application(
            application_id='com.twistedmatrix.trial.gireactor',
            flags=Gio.ApplicationFlags.FLAGS_NONE)

        def tryRegister():
            exc = self.assertRaises(ReactorAlreadyRunning,
                                    reactor.registerGApplication, app)
            self.assertEqual(exc.args[0],
                             "Can't register application after reactor was started.")
            reactor.stop()
        reactor.callLater(0, tryRegister)
        ReactorBuilder.runReactor(self, reactor) 

def fixPdb():
    def do_stop(self, arg):
        self.clear_all_breaks()
        self.set_continue()
        from twisted.internet import reactor
        reactor.callLater(0, reactor.stop)
        return 1


    def help_stop(self):
        print("stop - Continue execution, then cleanly shutdown the twisted "
              "reactor.")


    def set_quit(self):
        os._exit(0)

    pdb.Pdb.set_quit = set_quit
    pdb.Pdb.do_stop = do_stop
    pdb.Pdb.help_stop = help_stop 

def start_job(self, job=None, callback_fn=None):
        print(job)
        spider_job = job['spider_job']
        runner = job['runner']
        spider_cls = spider_job['spider_cls']
        spider_settings = spider_job['spider_settings']
        spider_kwargs = spider_job['spider_kwargs']

        def engine_stopped_callback():
            runner.transform_and_index(callback_fn=callback_fn)

        if callback_fn:
            print("""
==========================================================
WARNING: callback_fn is {}
==========================================================
Since start_job is called with callback_fn, make sure you end the reactor if you want the spider process to
stop after the callback function is executed. By default callback_fn=None will close the reactor.

To write a custom callback_fn

def callback_fn():
    print ("Write your own callback logic")
    from twisted.internet import reactor
    reactor.stop()
==========================================================
        """.format(callback_fn))

        spider = Crawler(spider_cls, Settings(spider_settings))
        spider.signals.connect(engine_stopped_callback, signals.engine_stopped)
        self.runner.crawl(spider, **spider_kwargs)
        """
        d = runner.crawl(spider, **spider_kwargs)
        # d.addBoth(engine_stopped_callback)
        """
        reactor.run() 

def dataReceived(self, data):
        """
        Called by Twisted when data is received on the connection.

        We need to check a few things here. Firstly, we want to validate that
        we actually negotiated HTTP/2: if we didn't, we shouldn't proceed!

        Then, we want to pass the data to the protocol stack and check what
        events occurred.
        """
        if not self.known_proto:
            self.known_proto = self.transport.negotiatedProtocol
            assert self.known_proto == b'h2'

        events = self.conn.receive_data(data)

        for event in events:
            if isinstance(event, ResponseReceived):
                self.handleResponse(event.headers)
            elif isinstance(event, DataReceived):
                self.handleData(event.data)
            elif isinstance(event, StreamEnded):
                self.endStream()
            elif isinstance(event, SettingsAcknowledged):
                self.settingsAcked(event)
            elif isinstance(event, StreamReset):
                reactor.stop()
                raise RuntimeError("Stream reset: %d" % event.error_code)
            elif isinstance(event, WindowUpdated):
                self.windowUpdated(event)

        data = self.conn.data_to_send()
        if data:
            self.transport.write(data) 

def endStream(self):
        """
        We call this when the stream is cleanly ended by the remote peer. That
        means that the response is complete.

        Because this code only makes a single HTTP/2 request, once we receive
        the complete response we can safely tear the connection down and stop
        the reactor. We do that as cleanly as possible.
        """
        self.request_complete = True
        self.conn.close_connection()
        self.transport.write(self.conn.data_to_send())
        self.transport.loseConnection() 

def connectionLost(self, reason=None):
        """
        Called by Twisted when the connection is gone. Regardless of whether
        it was clean or not, we want to stop the reactor.
        """
        if self.fileobj is not None:
            self.fileobj.close()

        if reactor.running:
            reactor.stop() 

def endStream(self, stream_id):
        self.conn.close_connection()
        self.transport.write(self.conn.data_to_send())
        self.transport.loseConnection()
        reactor.stop() 

def stop(self):
        self.t.stop()
        yield self.finished 

def check(trigger_id):

    @defer.inlineCallbacks
    def start(db):
        trigger = Trigger(trigger_id, db)
        yield trigger.check()
        reactor.stop()

    run(start) 

def processEnded(self, reason):
        log.info("Checker process ended with reason: {reason}", reason=reason)
        if reactor.running:
            reactor.stop() 

def stop(self):
        pass 

def __init__(self, pubnub, url, urllib_func=None,
                 callback=None, error=None, id=None, timeout=5):
        self.url = url
        self.id = id
        self.callback = callback
        self.error = error
        self.stop = False
        self._urllib_func = urllib_func
        self.timeout = timeout
        self.pubnub = pubnub 

def cancel(self):
        self.stop = True
        self.callback = None
        self.error = None 

def stop(self):
        reactor.stop() 

def stop(self):
        pass 

def __init__(self, pubnub, url, urllib_func=None,
                 callback=None, error=None, id=None, timeout=5):
        self.url = url
        self.id = id
        self.callback = callback
        self.error = error
        self.stop = False
        self._urllib_func = urllib_func
        self.timeout = timeout
        self.pubnub = pubnub 

def cancel(self):
        self.stop = True
        self.callback = None
        self.error = None 

def stop(self):
        reactor.stop() 

def main():
    end_day = date.today()
    start_day = end_day - timedelta(7)

    dl = defer.DeferredList([
        get_github_issues('buildbot/buildbot', start_day, end_day, 'issues'),
        get_github_issues('buildbot/buildbot', start_day, end_day, 'pulls'),
        get_github_issues('buildbot/buildbot-infra', start_day, end_day),
        get_github_issues('buildbot/metabbotcfg', start_day, end_day),
    ], fireOnOneErrback=True, consumeErrors=True)
    dl.addCallback(make_html)
    dl.addCallback(send_email)
    dl.addErrback(log.err)
    dl.addCallback(lambda _: reactor.stop())
    reactor.run() 

def crawler_start(usage, tasks):
    """Start specified spiders or validators from cmd with scrapy core api.
    There are four kinds of spiders: common, ajax, gfw, ajax_gfw. If you don't
    assign any tasks, all these spiders will run.
    """
    if usage == 'crawler':
        maps = CRAWLER_TASK_MAPS
        origin_spiders = DEFAULT_CRAWLERS
    else:
        maps = TEMP_TASK_MAPS
        origin_spiders = DEFAULT_VALIDATORS

    if not tasks:
        spiders = origin_spiders
    else:
        spiders = list()
        cases = list(map(BaseCase, origin_spiders))
        for task in tasks:
            for case in cases:
                if case.check(task, maps):
                    spiders.append(case.spider)
                    break
            else:
                # crawler_logger.warning('spider task {} is an invalid task, the allowed tasks are {}'.format(
                #     task, list(maps.keys())))
                pass
    if not spiders:
        #crawler_logger.warning('no spider starts up, please check your task input')
        return

    settings = get_project_settings()
    configure_logging(settings)
    runner = CrawlerRunner(settings)
    for spider in spiders:
        runner.crawl(spider)
    d = runner.join()
    d.addBoth(lambda _: reactor.stop())
    reactor.run()