def msg_received(self, conn, msg):
     """Handles messages received from the TI server.  Starts the
     TopologyInteractor command-line interface once authentication is complete."""
     if msg is not None:
         if msg.get_type() == VNSAuthRequest.get_type():
             print 'Authenticating as %s' % self.username
             sha1_of_salted_pw = hashlib.sha1(msg.salt + self.auth_key).digest()
             conn.send(VNSAuthReply(self.username, sha1_of_salted_pw))
         elif msg.get_type() == VNSAuthStatus.get_type():
             if msg.auth_ok:
                 print 'Authentication successful.'
                 conn.send(TIOpen(self.tid))
                 reactor.callInThread(TopologyInteractor(self).cmdloop)
             else:
                 print 'Authentication failed.'
         elif msg.get_type() ==  TIBadNodeOrPort.get_type():
             txt = str(msg)
             if self.prev_bn_msg == txt:
                 self.prev_bn_msg = None # only stop it once
             else:
                 if self.prev_bn_msg != None:
                     print '***%s!=%s'%(self.prev_bn_msg,txt)
                 self.prev_bn_msg = txt
                 print '\n', txt
         elif msg.get_type() ==  TIBanner.get_type():
             print '\n', msg.msg
         elif msg.get_type() ==  TIPacket.get_type():
             self.got_tapped_packet(msg)
         else:
             print 'unexpected TI message received: %s' % msg

 def run(self):
     self.factory = get_factory(WebSocketServerFactory)("ws://0.0.0.0:%i" % self.port, debug=False)
     self.factory.protocol = get_protocol(WebSocketServerProtocol)
     reactor.listenTCP(self.port, self.factory)
     reactor.callInThread(self.backend_reader)
     reactor.callLater(1, self.keepalive_sender)
     reactor.run()

    def consume(self, msg):
        """Called with each incoming fedmsg.

        From here we trigger an rpm-ostree compose by touching a specific file
        under the `touch_dir`. Then our `doRead` method is called with the
        output of the rpm-ostree-toolbox treecompose, which we monitor to
        determine when it has completed.
        """
        self.log.info(msg)
        body = msg['body']
        topic = body['topic']
        repo = None

        if 'rawhide' in topic:
            arch = body['msg']['arch']
            self.log.info('New rawhide %s compose ready', arch)
            repo = 'rawhide'
        elif 'branched' in topic:
            arch = body['msg']['arch']
            branch = body['msg']['branch']
            self.log.info('New %s %s branched compose ready', branch, arch)
            log = body['msg']['log']
            if log != 'done':
                self.log.warn('Compose not done?')
                return
            repo = branch
        elif 'updates.fedora' in topic:
            self.log.info('New Fedora %(release)s %(repo)s compose ready',
                          body['msg'])
            repo = 'f%(release)s-%(repo)s' % body['msg']
        else:
            self.log.warn('Unknown topic: %s', topic)

        release = self.releases[repo]
        reactor.callInThread(self.compose, release)

	def userLeft(self, nick, channel):
		self.nicks[channel].remove(nick)
		self.logger.log(channel, nick, "has left.")
		for name, module in self.loaded.items():
			for hook in module["hooks"]:
				if hasattr(hook, "onPart") and hook.onPart:
					reactor.callInThread(hook, self, nick, channel)

	def action(self, nick, channel, action):
		nick = re.match(self.nick_regex, nick)
		self.logger.log(channel, nick.group(1), action)
		for name, module in self.loaded.items():
			for hook in module["hooks"]:
				if hasattr(hook, "onAction") and hook.onAction:
					reactor.callInThread(hook, self, nick, channel, action)

 def __launch_blocker_to_badboy(self, user_id):
     session_uid = self.quarterback.win32top.get_current_user_session()
     if int(session_uid) == int(user_id) :
         reactor.callInThread(self.__launch_blocker_thread, user_id, self)
     else:
         self.block_status.pop(self.block_status.index(user_id))
     return False

 def _flush(self):
     "Called on a scheudle by twisted to flush the filters"
     self.logger.info("Starting scheduled flush")
     for name,filt in self.filters.items():
         if isinstance(filt, ProxyFilter): continue
         reactor.callInThread(self.flush_filter, name)
     self.logger.debug("Ending scheduled flush.")

    def onMessage(self, payload, isBinary):
        if not isBinary:
            doc = json.loads(payload)
            # If type is "_attach", treat as attachment metadata
            if doc.get("type") == "_attach":
                self.factory.binmeta[self.peer] = doc
            else:
                # Interpret incoming comands as database updates
                if doc.get("_deleted", False):
                    did_delete = self.factory.db.delete_doc(doc["_id"], initiator=self)
                    upd = {"ok": did_delete, "_id": doc["_id"]}
                else:
                    upd = self.factory.db._try_update(doc, initiator=self)

                # Indicate success
                # TODO: indicate failure
                self.sendMessage(json.dumps(upd))
        else:
            metadoc = self.factory.binmeta[self.peer]
            del self.factory.binmeta[self.peer]
            docid = metadoc["id"]
            attachname = metadoc["name"]
            dbdoc = self.factory.db.docs.get(docid)

            reactor.callInThread(dbdoc._async_put_attachment, StringIO(payload), None, attachname)

def __online_stat_job(mk_db):
    def execute(mk_db):
        log.msg("start online stat task..")
        db = mk_db()
        try:
            nodes = db.query(models.SlcNode)
            for node in nodes:
                online_count = db.query(models.SlcRadOnline.id).filter(
                    models.SlcRadOnline.account_number == models.SlcRadAccount.account_number,
                    models.SlcRadAccount.member_id == models.SlcMember.member_id,
                    models.SlcMember.node_id == node.id
                ).count()
                stat = models.SlcRadOnlineStat()
                stat.node_id = node.id
                stat.stat_time = int(time.time())
                stat.total = online_count
                db.add(stat)
            db.commit()
            log.msg("online stat task done")
        except Exception as err:
            db.rollback()
            log.err(err,'online_stat_job err')
        finally:
            db.close()

    reactor.callInThread(execute,mk_db)

def task_check_for_job_state_changes():
    """
    Checks for job state changes in a non-blocking manner.
    
    Calls :py:func:`threaded_check_for_job_state_changes`.
    """
    reactor.callInThread(threaded_check_for_job_state_changes)

def task_manage_ec2_instances():
    """
    Calls the instance creation logic in a non-blocking manner.
    
    Calls :py:func:`threaded_manage_ec2_instances`.
    """
    reactor.callInThread(threaded_manage_ec2_instances)

def loop(xbmc_uri, auth, s, last_playing=None, use_reactor=False):
    print "loop(%r, %r, %r, %r, %r)" % (xbmc_uri, auth, s, last_playing, use_reactor)
    try:
        playing = get_xbmc_current_playing(xbmc_uri, auth).encode("ascii", "ignore")
    except:
        print "error getting xbmc data"
        playing = None

        # print repr(playing)
        # print repr(last_playing)
        # print

    if playing != last_playing:
        if playing == None:
            playing_f = s.format_text("", RED, 0)
        else:
            playing_f = s.format_text(playing, GREEN, 0)

        s.send_text(2, playing_f, speed=15)
    s.set_clock()

    # s.send_text(0, s.format_text('LOO', RED, 0))

    last_playing = playing
    sleep(3.0)
    if use_reactor:
        reactor.callInThread(loop, xbmc_uri, auth, s, last_playing, True)

 def initializeWAMP(self):
     # define read and write locks
     self.rlock = None
     self.wlock = None
     
     # connect to the spectrometer
     self.raw = bool(RAW)
     self.spectrometer = oceanoptics.USB2000plus()
     self.spectrometer.integration_time(time_sec=INTEGRATION_TIME)
     self.wl = list(self.spectrometer.wavelengths())
     self.sp = list(self.spectrometer.intensities())
     
     # read new values off of spectrometer, lock while reading or writing
     @inlineCallbacks
     def capture():
         yield self.rlock 
         self.wlock = Deferred()
         self.sp = list(self.spectrometer.intensities())
         time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")
         self.latestTime = time
         self.wlock.callback(None)
         self.wlock = None
         reactor.callLater(.1,capture)
         
     reactor.callInThread(capture)
     
     ## complete initialization
     BaseWAMP.initializeWAMP(self)

def build_reactor(options, **kwargs):
    web_socket_instance = FilteredWebSocketFactory(**kwargs)
    subscriber = kwargs.pop("subscriber", None)

    if options.key and options.cert:
        with open(options.key) as keyFile:
            with open(options.cert) as certFile:
                cert = ssl.PrivateCertificate.loadPEM(keyFile.read() + certFile.read())
                reactor.listenSSL(options.port, web_socket_instance, cert.options())
    else:
        reactor.listenTCP(
            options.port,
            web_socket_instance
        )
    if subscriber is not None:
        reactor.callInThread(
            subscriber.listener,
            web_socket_instance
        )
        reactor.addSystemEventTrigger(
            "before",
            "shutdown",
            subscriber.kill
        )

    # Start the consumer loop
    consumer_loop = LoopingCall(
        web_socket_instance.consumer
    )
    consumer_loop.start(0.001, now=False)
    return web_socket_instance

    def post(self, *args):
        if len(args) >= 1:
            name = args[0]
            project = Projects(name)
            for key, value in self.request.arguments.iteritems():
                if key in ("git_url", "version", "build_cmd", "install_cmd"):
                    setattr(project, key, value[0])
            project.save()

        try:
            if not Projects(self.get_argument('name')).exists():
                raise
        except Exception, e:
            project = Projects()
            project.name = self.get_argument('name')[0]
            project.git_url = self.get_argument('git_url')[0]
            for name, parm in self.request.arguments.iteritems():
                if name not in ('branch', 'version'):
                    setattr(project, str(name), parm[0])
            try:
                project.add_branch(self.get_argument('branch'))
                project.version(self.get_argument('branch'), self.get_argument('version'))
                project.group_name = self.get_argument('group_name')
                project.save()
                log.msg('Project created:', project.name)
                reactor.callInThread(queue.enqueue, 'build', 'builder.build_project', {'project': project.name, 'branch': self.get_argument('branch'), 'force': True})
                self.write(cyclone.escape.json_encode({'status': 'ok'}))
            except Exception, e:
                log.err()
                self.write(cyclone.escape.json_encode({'status': "fail"}))

def __flow_stat_job(mk_db):
    def execute(mk_db):
        log.msg("start flow stat task..")
        db = mk_db()
        try:
            nodes = db.query(models.SlcNode)
            for node in nodes:
                r = db.query(
                    func.sum(models.SlcRadOnline.input_total).label("input_total"),
                    func.sum(models.SlcRadOnline.output_total).label("output_total")
                ).filter(
                    models.SlcRadOnline.account_number == models.SlcRadAccount.account_number,
                    models.SlcRadAccount.member_id == models.SlcMember.member_id,
                    models.SlcMember.node_id == node.id
                ).first()
                if r:
                    stat = models.SlcRadFlowStat()
                    stat.node_id = node.id
                    stat.stat_time = int(time.time())
                    stat.input_total = r.input_total
                    stat.output_total = r.output_total
                    db.add(stat)
            db.commit()
            log.msg("flow stat task done")
        except  Exception as err:
            db.rollback()
            log.err(err,'flow_stat_job err')
        finally:
            db.close()

    reactor.callInThread(execute,mk_db)

def callMultipleInThread(tupleList):
    """Run a list of functions in the same thread.

    tupleList should be a list of (function, argsList, kwargsDict) tuples.
    """
    from twisted.internet import reactor
    reactor.callInThread(_runMultiple, tupleList)

 def received(self, msg):
     """
     This is a direct callback from the protocol object. This call is
     forwarded to the dispatcher in a thread from the threadpool. The 
     """
     if self.dispatcher is not None:
         reactor.callInThread(self.dispatcher.receivedEvent, msg)

    def __thread_write(self):
        if not hasattr(self, '__thread_write_status') :
            self.__thread_write_status = 'stopped'

        if self.__thread_write_status == 'running' :
            return

        if not self.buffer_dirty :
            return
        
        bfd = self.buffer_dirty.copy()

        def call_in_thread():
            # Writing to disk 
            for idx in sorted(bfd.keys()) :
                data = bfd[idx]
                self.write(idx, data)
            reactor.callFromThread(call_from_thread)

        def call_from_thread():
            self.__thread_write_status = 'stopped'
            for idx, data in bfd.iteritems() :
                if data is self.buffer_dirty[idx] :
                    del self.buffer_dirty[idx]

        if self.__thread_write_status == 'stopped' :
            self.__thread_write_status = 'running'
            reactor.callInThread(call_in_thread)

	def userJoined(self, nick, channel):
		self.nicks[channel].append(nick)
		self.logger.log(channel, nick, "has joined.")
		for name, module in self.loaded.items():
			for hook in module["hooks"]:
				if hasattr(hook, "onJoin") and hook.onJoin:
					reactor.callInThread(hook, self, nick, channel)