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 subscribe(self, *args):
        #d = self.protocol.subscribe("foo/bar/baz", 0)
        log.info(u"Subscribing to topics {subscriptions}. protocol={protocol}", subscriptions=self.subscriptions, protocol=self.protocol)
        for topic in self.subscriptions:
            log.info(u"Subscribing to topic '{topic}'", topic=topic)
            # Topic name **must not** be unicode, so casting to string
            e = self.protocol.subscribe(str(topic), 0)

        log.info(u"Setting callback handler: {callback}", callback=self.callback)
        self.protocol.setPublishHandler(self.on_message_twisted)
        """
        def cb(*args, **kwargs):
            log.info('publishHandler got called: name={name}, args={args}, kwargs={kwargs}', name=self.name, args=args, kwargs=kwargs)
            return reactor.callFromThread(self.callback, *args, **kwargs)
        self.protocol.setPublishHandler(cb)
        """ 

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 _start_socket(self, id_, payload, callback, private=False):
        if id_ in self._conns:
            return False

        if private:
            factory_url = self.PRIVATE_STREAM_URL
        else:
            factory_url = self.STREAM_URL

        factory = KrakenClientFactory(factory_url, payload=payload)
        factory.base_client = self
        factory.protocol = KrakenClientProtocol
        factory.callback = callback
        factory.reconnect = True
        self.factories[id_] = factory
        reactor.callFromThread(self.add_connection, id_, factory_url) 

def test_threadsAreRunInScheduledOrder(self):
        """
        Callbacks should be invoked in the order they were scheduled.
        """
        order = []

        def check(_):
            self.assertEqual(order, [1, 2, 3])

        self.deferred.addCallback(check)
        self.schedule(order.append, 1)
        self.schedule(order.append, 2)
        self.schedule(order.append, 3)
        self.schedule(reactor.callFromThread, self.deferred.callback, None)

        return self.deferred 

def test_callFromThread(self):
        """
        Test callFromThread functionality: from the main thread, and from
        another thread.
        """
        def cb(ign):
            firedByReactorThread = defer.Deferred()
            firedByOtherThread = defer.Deferred()

            def threadedFunc():
                reactor.callFromThread(firedByOtherThread.callback, None)

            reactor.callInThread(threadedFunc)
            reactor.callFromThread(firedByReactorThread.callback, None)

            return defer.DeferredList(
                [firedByReactorThread, firedByOtherThread],
                fireOnOneErrback=True)
        return self._waitForThread().addCallback(cb) 

def test_wakerOverflow(self):
        """
        Try to make an overflow on the reactor waker using callFromThread.
        """
        def cb(ign):
            self.failure = None
            waiter = threading.Event()
            def threadedFunction():
                # Hopefully a hundred thousand queued calls is enough to
                # trigger the error condition
                for i in xrange(100000):
                    try:
                        reactor.callFromThread(lambda: None)
                    except:
                        self.failure = failure.Failure()
                        break
                waiter.set()
            reactor.callInThread(threadedFunction)
            waiter.wait(120)
            if not waiter.isSet():
                self.fail("Timed out waiting for event")
            if self.failure is not None:
                return defer.fail(self.failure)
        return self._waitForThread().addCallback(cb) 

def test_callMultiple(self):
        """
        L{threads.callMultipleInThread} calls multiple functions in a thread.
        """
        L = []
        N = 10
        d = defer.Deferred()

        def finished():
            self.assertEqual(L, list(range(N)))
            d.callback(None)

        threads.callMultipleInThread([
            (L.append, (i,), {}) for i in xrange(N)
            ] + [(reactor.callFromThread, (finished,), {})])
        return d 

def raven_log(self, event):
        f = event["log_failure"]
        stack = None
        extra = dict()
        tb = f.getTracebackObject()
        if not tb:
            # include the current stack for at least some
            # context. sentry's expecting that "Frames should be
            # sorted from oldest to newest."
            stack = list(iter_stack_frames())[:-5]  # approx.
            extra = dict(no_failure_tb=True)
        extra.update(
            log_format=event.get('log_format'),
            log_namespace=event.get('log_namespace'),
            client_info=event.get('client_info'),
            )
        reactor.callFromThread(
            self.raven_client.captureException,
            exc_info=(f.type, f.value, tb),
            stack=stack,
            extra=extra,
        )
        # just in case
        del tb 

def check_for_sm_finished(sm, monitoring_manager=None):
    from rafcon.core.states.state import StateExecutionStatus
    while sm.root_state.state_execution_status is not StateExecutionStatus.INACTIVE:
        try:
            sm.root_state.concurrency_queue.get(timeout=10.0)
        except Empty as e:
            pass
        # no logger output here to make it easier for the parser
        print("RAFCON live signal")

    sm.root_state.join()

    # stop the network if the monitoring plugin is enabled
    if monitoring_manager:
        from twisted.internet import reactor
        reactor.callFromThread(reactor.stop) 

def start_stop_state_machine(state_machine, start_state_path, quit_flag):
    from rafcon.utils.gui_functions import call_gui_callback

    state_machine_execution_engine = core_singletons.state_machine_execution_engine
    call_gui_callback(
        state_machine_execution_engine.execute_state_machine_from_path,
        state_machine=state_machine,
        start_state_path=start_state_path,
        wait_for_execution_finished=True
    )

    if reactor_required():
        from twisted.internet import reactor
        reactor.callFromThread(reactor.stop)

    if quit_flag:
        gui_singletons.main_window_controller.get_controller('menu_bar_controller').on_quit_activate(None, None) 

def signal_handler(signal, frame):
    global _user_abort

    state_machine_execution_engine = core_singletons.state_machine_execution_engine
    core_singletons.shut_down_signal = signal

    logger.info("Shutting down ...")

    try:
        if not state_machine_execution_engine.finished_or_stopped():
            state_machine_execution_engine.stop()
            state_machine_execution_engine.join(3)  # Wait max 3 sec for the execution to stop
    except Exception:
        logger.exception("Could not stop state machine")

    _user_abort = True

    # shutdown twisted correctly
    if reactor_required():
        from twisted.internet import reactor
        if reactor.running:
            plugins.run_hook("pre_destruction")
            reactor.callFromThread(reactor.stop)

    logging.shutdown() 

def test_threadsAreRunInScheduledOrder(self):
        """
        Callbacks should be invoked in the order they were scheduled.
        """
        order = []

        def check(_):
            self.assertEqual(order, [1, 2, 3])

        self.deferred.addCallback(check)
        self.schedule(order.append, 1)
        self.schedule(order.append, 2)
        self.schedule(order.append, 3)
        self.schedule(reactor.callFromThread, self.deferred.callback, None)

        return self.deferred 

def test_callFromThread(self):
        """
        Test callFromThread functionality: from the main thread, and from
        another thread.
        """
        def cb(ign):
            firedByReactorThread = defer.Deferred()
            firedByOtherThread = defer.Deferred()

            def threadedFunc():
                reactor.callFromThread(firedByOtherThread.callback, None)

            reactor.callInThread(threadedFunc)
            reactor.callFromThread(firedByReactorThread.callback, None)

            return defer.DeferredList(
                [firedByReactorThread, firedByOtherThread],
                fireOnOneErrback=True)
        return self._waitForThread().addCallback(cb) 

def test_callMultiple(self):
        """
        L{threads.callMultipleInThread} calls multiple functions in a thread.
        """
        L = []
        N = 10
        d = defer.Deferred()

        def finished():
            self.assertEqual(L, list(range(N)))
            d.callback(None)

        threads.callMultipleInThread([
            (L.append, (i,), {}) for i in range(N)
            ] + [(reactor.callFromThread, (finished,), {})])
        return d 

def waitForInterrupt():
    if signal.getsignal(signal.SIGINT) != signal.default_int_handler:
        raise RuntimeError("Already waiting")

    d = Deferred()

    def fire(*ignored):
        global interrupted
        signal.signal(signal.SIGINT, signal.default_int_handler)
        now = time.time()
        if now - interrupted < 4:
            reactor.callFromThread(lambda: d.errback(Failure(Stop())))
        else:
            interrupted = now
            reactor.callFromThread(d.callback, None)
    signal.signal(signal.SIGINT, fire)
    return d 

def pamAuthenticateThread(service, user, conv):
    def _conv(items):
        from twisted.internet import reactor
        try:
            d = conv(items)
        except:
            import traceback
            traceback.print_exc()
            return
        ev = threading.Event()
        def cb(r):
            ev.r = (1, r)
            ev.set()
        def eb(e):
            ev.r = (0, e)
            ev.set()
        reactor.callFromThread(d.addCallbacks, cb, eb)
        ev.wait()
        done = ev.r
        if done[0]:
            return done[1]
        else:
            raise done[1].type, done[1].value

    return callIntoPAM(service, user, _conv) 

def run_thread(execute=True):
    """
    Start pdconfd service as a thread.

    This function schedules pdconfd to run as a thread and returns immediately.
    """
    global configManager
    configManager = ConfigManager(settings.PDCONFD_WRITE_DIR, execute)
    reactor.callFromThread(listen, configManager) 

def callFromThread(self, f, *args, **kw):
        assert callable(f), "%s is not callable" % f
        with NullContext():
            # This NullContext is mainly for an edge case when running
            # TwistedIOLoop on top of a TornadoReactor.
            # TwistedIOLoop.add_callback uses reactor.callFromThread and
            # should not pick up additional StackContexts along the way.
            self._io_loop.add_callback(f, *args, **kw)

    # We don't need the waker code from the super class, Tornado uses
    # its own waker. 

def add_callback(self, callback, *args, **kwargs):
        self.reactor.callFromThread(
            self._run_callback,
            functools.partial(wrap(callback), *args, **kwargs)) 

def callFromThread(self, f, *args, **kw):
        assert callable(f), "%s is not callable" % f
        with NullContext():
            # This NullContext is mainly for an edge case when running
            # TwistedIOLoop on top of a TornadoReactor.
            # TwistedIOLoop.add_callback uses reactor.callFromThread and
            # should not pick up additional StackContexts along the way.
            self._io_loop.add_callback(f, *args, **kw)

    # We don't need the waker code from the super class, Tornado uses
    # its own waker. 

def add_callback(self, callback, *args, **kwargs):
        self.reactor.callFromThread(
            self._run_callback,
            functools.partial(wrap(callback), *args, **kwargs)) 

def safeDestroyClients():
    TimelineLogger.warn(
        "Timeline is safely shutting down, this can take some time. Please don't interrupt or close the server, that might affect users experience on next login.")

    for engine in SERVERS:
        yield engine.connectionLost('Unknown')

    TimelineLogger.debug('Viola!')
    # reactor.callFromThread(reactor.stop) 

def onExitSignal(*a):
    print 'Closing Timeline?'
    if not reactor.running:
        os._exit(1)

    reactor.callFromThread(reactor.stop) 

def forceRefresh(self, reason = 'Force refresh on server command'):
        if self.DEBUG:
            self.logger.info('Penguin ASync-Refresh, Force refresh: Penguin - {}, Reason - {}'.format
                             (self.penguin['nickname'], reason))

        if not self.RefreshManagerLoop.running:
            self.logger.warn('Penguin ASync-Refresh Force refresh already running, called more than once in a row:'
                             ' Penguin - {}, Reason - {}'.format(self.penguin['nickname'], reason)) if self.DEBUG \
                else 0

            return None

        reactor.callFromThread(self.RefreshManagerLoop.stop)
        return self._refresh(True) 

def skip(self, reason = 'Skip refresh on server command'):
        if self.DEBUG:
            self.logger.info('Penguin ASync-Refresh, Skip Refresh: Penguin - {}, Reason - {}'.format
                             (self.penguin['nickname'], reason))

        reactor.callFromThread(self.RefreshManagerLoop.reset) 

def connect(self):
        """
        Connect to MQTT broker.
        """
        # TODO: Check if we can do asynchronous connection establishment.
        #       Currently, this is done synchronously which could harm
        #       other subsystems in timeout or otherwise blocking situations.

        # Make MQTT client identifier even more unique by adding process id
        pid = os.getpid()
        client_id = '{}:{}'.format(self.name, str(pid))

        # Connection establishment
        self.client = mqtt.Client(client_id=client_id, clean_session=True)

        # Optionally authenticate connection
        if self.broker_username:
            self.client.username_pw_set(self.broker_username, self.broker_password)

        # Set event handlers
        self.client.on_connect = lambda *args: reactor.callFromThread(self.on_connect, *args)
        self.client.on_message = lambda *args: reactor.callFromThread(self.on_message, *args)
        self.client.on_log     = lambda *args: reactor.callFromThread(self.on_log, *args)

        # Connect with retry
        self.connect_loop = LoopingCall(self.connect_with_retry)
        self.connect_loop.start(self.retry_interval, now=True) 

def check_rising(self, port):
        #print "check_rising:", port, self.port
        if port == self.port:
            reactor.callFromThread(self.callback, port)

    # doesn't work with Adafruit GPIO 

def dispatch_from_thread(self, action, client_id):
        message = self.encoder.to_json(action)
        if six.PY2:
            reactor.callFromThread(self.dispatch, message, client_id)
        else:
            self.loop.call_soon_threadsafe(self.dispatch, message, client_id) 

def callFromThread(self, f, *args, **kw):
        """See `twisted.internet.interfaces.IReactorThreads.callFromThread`"""
        assert callable(f), "%s is not callable" % f
        with NullContext():
            # This NullContext is mainly for an edge case when running
            # TwistedIOLoop on top of a TornadoReactor.
            # TwistedIOLoop.add_callback uses reactor.callFromThread and
            # should not pick up additional StackContexts along the way.
            self._io_loop.add_callback(f, *args, **kw)

    # We don't need the waker code from the super class, Tornado uses
    # its own waker.