def start(self):
        "Start the application."
        from twisted.internet import reactor
        reactor.callLater(0, self.installIPC)
        self.register()
        if not self._startReactor:
            log.msg("Not starting reactor - test mode?")
            return

        if self.needsThreadedUI():
            threadable.init(1)
            from twisted.internet import reactor
            t = threading.Thread(target=reactor.run, kwargs={
                                'installSignalHandlers':0} )
            t.start()
            self.ui.startUI()
        else:
            from twisted.internet import reactor
            # AAaaaargh the QTReactor has the suck.
            try:
                from twisted.internet.qtreactor import QTReactor
            except:
                class QTReactor: pass
            if (isinstance(reactor, QTReactor)
                        or not hasattr(reactor,'running')
                        or not reactor.running):
                reactor.run()

 def __init__(self):
     threadable.init(1)
     self.reads = {}
     self.writes = {}
     self.toThreadQueue = Queue()
     self.toMainThread = Queue()
     self.workerThread = None
     self.mainWaker = None
     posixbase.PosixReactorBase.__init__(self)
     self.addSystemEventTrigger('after', 'shutdown', self._mainLoopShutdown)

    def InitTwisted(self):
        """
        Import and initialize the Twisted event loop.
        Note: Twisted will run in a separate thread from the GUI.
        """

        from twisted.internet import reactor
        from twisted.python import threadable
        threadable.init(1)
        self.reactor = reactor

    def setup_priviledged(self):
	log.info("Setting up the XenBEE broker")

# TODO: hier muss etwas rein!! - start
        log.info("initializing schema documents...")
        from lxml import etree
        self.schema_map = {}
        for schema in os.listdir(self.opts.schema_dir):
            if not schema.endswith(".xsd"): continue
            path = os.path.join(self.opts.schema_dir, schema)
            log.info("   reading %s" % path)
            xsd_doc = etree.parse(path)
            namespace = xsd_doc.getroot().attrib["targetNamespace"]
            self.schema_map[namespace] = etree.XMLSchema(xsd_doc)
        log.info("  done.")

        log.info("initializing certificates...")
        # read in the certificate and private-key
        from xbe.xml.security import X509Certificate
        self.ca_certificate = X509Certificate.load_from_files(self.opts.ca_cert)
        log.info("   CA certificate: %s", self.ca_certificate.subject()["CN"])

        cert = X509Certificate.load_from_files(self.opts.x509,
                                               self.opts.p_key)
        if not self.ca_certificate.validate_certificate(cert):
            self.error("the given x509 certificate has not been signed by the given CA")
        self.certificate = cert
        log.info("   my certificate: %s" % (self.certificate.subject()["CN"]))
        log.info("  done.")

        log.info("initializing twisted...")
        from twisted.python import threadable
        threadable.init()
        log.info("  done.")
        
        log.info("initializing user database...")
        from xbe.xbed.user import UserDatabase
        self.userDatabase = UserDatabase.getInstance(self.opts.user_db)
        log.info("  done.")

        log.info("initializing reactor...")
        from twisted.internet import reactor
        from xbe.broker.proto import XenBEEBrokerProtocolFactory


        from xbe.util.network import urlparse
        proto, host, queue, _, _, _ = urlparse(self.opts.uri)
        if proto != "stomp":
            raise ValueError("unknown protocol", proto)
        try:
            host, port = host.split(":")
            port = int(port)
        except ValueError, e:
            port = 61613

    def main(self, argv=sys.argv, prog_name=None):
        from twisted.python import threadable
        threadable.init()

        if prog_name is None:
            argv[0] = os.path.basename(argv[0])
        self.argv = argv

        cmd = createCommand(argv[1:])
        if cmd:
            try:
                cmd.pre_execute()
                self.setup_logging(cmd.opts.verbose)

                if isinstance(cmd, RemoteCommand):
                    from xbe.util import network
                    prot, host, stomp_queue, u1, u2, u3 = network.urlparse(cmd.opts.server)
                    if prot != "stomp":
                        raise ValueError("I do not understand this wire-protocol", prot)
                    
                    from xbe.cmdline.protocol import ClientProtocolFactory, ClientXMLProtocol
                    # TODO: generate ID or use some given one
                    factory = ClientProtocolFactory(
                        id=str(uuid.uuid4()),
                        stomp_user=cmd.opts.stomp_user, stomp_pass=cmd.opts.stomp_pass,
                        certificate=cmd.user_cert, ca_cert=cmd.ca_cert,
                        server_queue="/queue"+stomp_queue,
                        protocolFactory=ClientXMLProtocol,
                        protocolFactoryArgs=(cmd,))
                    
                    from twisted.internet import reactor
                    from twisted.python import threadable
                    threadable.init()

                    h_p = host.split(":")
                    if len(h_p) == 2:
                        host, port = h_p
                    else:
                        host, port = h_p[0], 61613
                    # schedule a Timeout
                    cmd.scheduleTimeout(name="connect")
                    reactor.connectTCP(host, port, factory)
                    reactor.run()
                else:
                    cmd.execute()
                cmd.post_execute()
                if cmd.status_info[0]:
                    raise CommandFailed(cmd.status_info[1])
            except CommandFailed, cf:
                print >>sys.stderr, "%s:" % sys.argv[0], str(cmd), "failed (details follow)"
                print >>sys.stderr, "\n".join([ "%s: %s" %
                                                (sys.argv[0],s) for s in str(cf.message).split('\n')])
                return 2
            except:

def main():   
    reactor.suggestThreadPoolSize(100)
    threadable.init(1)   # 设置线程安全
    workroot = os.path.dirname(os.path.abspath(__file__))
    logdir = os.path.join(workroot,"Trace")
    logdir = os.path.join(logdir,"twisted")
    if not os.path.isdir(logdir):
        os.makedirs(logdir)
    log_file = DailyLogFile("server.log", logdir)
    log.startLogging(log_file)
    webserver = WebService(workroot)
    port = GetSysConfigInt("server_port",8080)
    reactor.listenTCP(port, server.Site(webserver.get_resource()))
    reactor.callLater(10,monitoring) 
    reactor.run()

    def setup_priviledged(self):
	log.info("Setting up the XenBEE instance daemon")

        log.info("initializing schema documents...")
        from lxml import etree
        self.schema_map = {}
        for schema in os.listdir(self.opts.schema_dir):
            if not schema.endswith(".xsd"): continue
            path = os.path.join(self.opts.schema_dir, schema)
            log.info("   reading %s" % path)
            xsd_doc = etree.parse(path)
            namespace = xsd_doc.getroot().attrib["targetNamespace"]
            self.schema_map[namespace] = etree.XMLSchema(xsd_doc)
        log.info("  done.")
        
        log.info("initializing the twisted framework...")
        from twisted.python import threadable
        threadable.init()
        log.info("  done.")

        # set up the uuid
        if self.opts.uuid is None:
            raise RuntimeError("could not get my instance id")
        self.queue = "/queue/xenbee.instance.%s" % (self.opts.uuid)

        # set up the STOMP server
        from xbe.util.network import urlparse
        if self.opts.server is None:
            raise RuntimeError("no server uri given, use XBE_SERVER or -H")
        proto, host, queue, _, _, _ = urlparse(self.opts.server)
        if proto != "stomp":
            raise ValueError("unknown protocol", proto)
        self.proto = proto
        try:
            self.host, port = host.split(":")
            self.port = int(port)
        except Exception, e:
            self.host = host
            self.port = 61613

def main(shtoomapp):
    from twisted.python import threadable

    # wxreactor sucks generally.
    # wxsupport sucks on windows.
    # lets give threading a go

    threadable.init(1)
    wxapp = WxShtoomApplication()
    wxproxy = WxProxy(shtoomapp, wxapp)
    appproxy = ShtoomProxy(wxproxy)
    wxapp.frame.connectApplication(appproxy)

    from shtoom import log
    import sys
    # TODO: This style of logging isn't thread safe. Need to plugin
    # the logging into the WxInjector. i.e. the logger targets the
    # WxInjector.evtlist
    #log.startLogging(wxapp.frame.getLogger(), setStdout=False)
    log.startLogging(sys.stdout)

    return wxproxy

	def startOverlay(self):
		self.peerlist = PeerList()

		CommandLineParser().parse_args()
		self.myNick = CommandLineParser().getArguments().PeerName

		ExtensionLoader().extendProtocol()

		self.listenerFactory = OverlayListenerFactory(self)
		self.connectorFactory = OverlayConnectorFactory(self)

		for port in CommandLineParser().getArguments().lport:
			reactor.listenTCP(port, self.listenerFactory)

		for t in CommandLineParser().getArguments().mhost:
			a = t.split(':')
			peer, port = a[0], int(a[1]) if len(a) > 1 else LPORT

			reactor.connectTCP(peer, port, self.connectorFactory)

			del a

		threadable.init()
		reactor.run()

def install():
    threadable.init(1)
    r = Win32Reactor()
    import main
    main.installReactor(r)

  ## version 1.1, changed according to the suggestions in the comments

  from twisted.internet import reactor, defer
  from twisted.python import threadable; threadable.init(1)
  import sys, time

  ## the wrong way

  def callInThread(func, *args):
      """Takes a blocking function an converts it into a deferred-valued 
      function running in a separate thread.
      """
      de = defer.Deferred()
      de.addCallback(func)
      reactor.callInThread(de.callback, *args)
      return de
  
  deferred = callInThread.__get__ # decorator associated to callInThread

  # the right way
   
  from twisted.internet.threads import deferToThread
  deferred = deferToThread.__get__

  ## example code

  def print_(result):
    print result
  
  def running():
      "Prints a few dots on stdout while the reactor is running."

'''
Coronis simulator to test Audiotel modems
'''
import logging
import datetime

from twisted.python import threadable
threadable.init()
from twisted.internet import reactor
from twisted.internet import task

from audiotel import AudiotelDatagramProtocol
from crncomu  import *

####################################
# MAIN VARIABLES

MODULE_IP   = '192.168.253.129'
MODULE_PORT = 8001
SERVER_PORT = 8000

WAIT_TIME = 50 

####################################


class UDPServer(AudiotelDatagramProtocol):

  stats    = dict()

  def __init__(self, sent, received):

        self.logtime = False
        if s[-1] != "\n":
            return
        logfile.flush()
        self.logtime = True

    def flush(self):
        stdout.flush()
        if uselogfile:
            logfile.flush()


if __name__ == "__main__":
    os.chdir(os.path.dirname(os.path.abspath(sys.argv[0])))
    # thread.stack_size(256*1024)
    threadable.init()  # twisted.python.threadable
    stdout = sys.stdout
    if uselogfile:
        logfile = file("pyeco.log", "a")
        logfile.write("-" * 30 + " pyeco start " + "-" * 30 + "\n")
    sys.stdout = Log()
    create_global_lock()  # thread lock
    create_global_serverobj()  # a class
    create_global_itemdic()  # key: int item id
    create_global_mapdic()  # key: int map id
    create_global_shopdic()  # key: int shop id
    create_global_npcdic()  # key: int npc id
    create_global_skilldic()  # key: int skill id
    create_global_mobdic()  # key: int mob id
    create_global_petdic()  # key: int pet id
    create_global_eventhandle()  # load script

 def _initThreadPool(self):
     from twisted.python import threadpool
     threadable.init(1)
     self.threadpool = threadpool.ThreadPool(0, 10)
     self.threadpool.start()
     self.addSystemEventTrigger('during', 'shutdown', self.threadpool.stop)

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# Lesser General Public License for more details.
# 
# You should have received a copy of the GNU Lesser General Public
# License along with this library; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

"""A thread pool that is integrated with the Twisted event loop."""

# Sibling Import
import task, main

# Twisted Import
from twisted.python import threadpool, threadable, log

threadable.init(1)

class ThreadDispatcher(threadpool.ThreadPool):
    """A thread pool that is integrated with the Twisted event loop.

    The difference from ThreadPool is that callbacks are run in the main IO
    event loop thread, and are thus inherently thread-safe.

    You probably want your instance to be shutdown when Twisted is shut down:

        from twisted.internet import main
        from twisted.internet import threadtask
        tpool = ThreadDispatcher()
        main.addShutdown(tpool.stop)

    """

def install():
    from twisted.python import threadable
    p = Proactor()
    threadable.init()
    main.installReactor(p)

def startCCSDServer(key, cert, cacert):
    """Initialises the CCSD Server.

    This function never returns as it enters the twisted mainloop
    """

    try:           
        # Local networks
        localnets = config_get(None, "local_networks", "127.0.0.0/8")

        # Register standard HTTP XMLRPC handler for local requests
        registerHTTPResource("RPC2", ccsd_local_xmlrpc, 
                localnets=localnets.split(","))

        # Register HTTPS XMLRPC Handler
        use_ssl = config_getboolean(None, "use_ssl", True)
        if use_ssl:
            registerHTTPSResource("RPC2", ccsd_xmlrpc)

        # Setup XMLRPC Handler configuration
        ccsd_xmlrpc.log_times = config_get(None, "log_times", None)
        ccsd_xmlrpc.profile = config_getboolean(None, "profile", False)
        ccsd_xmlrpc.prof_dir = config_get(None, "profile_dir", \
                DEFAULT_PROFILE_DIR)
        ccsd_xmlrpc.log_threads = config_get(None, "log_threads", None)
        ccsd_xmlrpc.max_threads = config_getint(None, "max_threads", 
                DEFAULT_MAX_THREADS)

        # SSL Context
        class SCF:
            def __init__(self, key, cert, cacert):
                self.mKey = key
                self.mCert = cert
                self.mCACert = cacert
                
            def verify(self, conn, cert, errnum, depth, ok):
                """Checks the certificate of an incoming connection"""
                # If there is already an error bail now
                if not ok:
                    return ok
                
                # Only perform further verification on client certs
                if depth>0:
                    return ok
                
                # At this point we know the certificate is signed by a 
                # trusted CA, check the issuer OU matches the incoming cert
                # OU and the incoming cert is not a server cert
                # XXX: Should look at using something like nsCertType rather
                # than the CN field for this.
                s = cert.get_subject()
                i = cert.get_issuer()
                if s.OU != i.OU:
                    log_warn("Rejected incoming connection from invalid " 
                            "SSL cert (%s). OU did not match." % s)
                    return 0
                if s.CN == "server":
                    log_warn("Rejected incoming connection from server SSL "
                            "cert (%s)." % s)
                    return 0
                return 1
                
            def getContext(self):
                """Create an SSL context."""
                ctx = SSL.Context(SSL.SSLv2_METHOD)
                # Add the CA certificate(s)
                store = ctx.get_cert_store()
                for cert in self.mCACert:
                    store.add_cert(cert)
                # Load the private key and certificate
                ctx.use_privatekey(self.mKey)
                ctx.use_certificate(self.mCert)
                ctx.set_verify(SSL.VERIFY_PEER | 
                        SSL.VERIFY_FAIL_IF_NO_PEER_CERT, self.verify)
                ctx.set_verify_depth(len(self.mCACert))
                return ctx

        # Port and logfile
        http_port = int(config_get(None, "http_port", 
            DEFAULT_HTTP_SERVER_PORT))
        https_port = int(config_get(None, "https_port", 
            DEFAULT_HTTPS_SERVER_PORT))
        logfile = config_get(None, "request_log", DEFAULT_REQUEST_LOG)

        # Pass off control to Twisted's mainloop
        threadable.init()
        suggestThreadpoolSize(ccsd_xmlrpc.max_threads)
        reactor.listenTCP(http_port, server.Site(_http_root, logfile))
        if use_ssl:
            reactor.listenSSL(https_port, server.Site(_https_root, logfile), \
                    SCF(key, cert, cacert))
        reactor.addSystemEventTrigger("before", "shutdown", shutdownHandler)
        log_info("Server Started. Ready to serve requests...")
    except:
        log_fatal("Could not initialise the server!", sys.exc_info())

    reactor.run()