1 # Copyright (c) Twisted Matrix Laboratories.
2 # See LICENSE for details.
5 This module provides wxPython event loop support for Twisted.
7 In order to use this support, simply do the following::
9 | from twisted.internet import wxreactor
12 Then, when your root wxApp has been created::
14 | from twisted.internet import reactor
15 | reactor.registerWxApp(yourApp)
18 Then use twisted.internet APIs as usual. Stop the event loop using
19 reactor.stop(), not yourApp.ExitMainLoop().
21 IMPORTANT: tests will fail when run under this reactor. This is
22 expected and probably does not reflect on the reactor's ability to run
28 from wx import PySimpleApp as wxPySimpleApp, CallAfter as wxCallAfter, \
31 # older version of wxPython:
32 from wxPython.wx import wxPySimpleApp, wxCallAfter, wxTimer
34 from twisted.python import log, runtime
35 from twisted.internet import _threadedselect
38 class ProcessEventsTimer(wxTimer):
40 Timer that tells wx to process pending events.
42 This is necessary on OS X, probably due to a bug in wx, if we want
43 wxCallAfters to be handled when modal dialogs, menus, etc. are open.
45 def __init__(self, wxapp):
46 wxTimer.__init__(self)
52 Called repeatedly by wx event loop.
54 self.wxapp.ProcessPendingEvents()
58 class WxReactor(_threadedselect.ThreadedSelectReactor):
62 wxPython drives the event loop, select() runs in a thread.
67 def registerWxApp(self, wxapp):
69 Register wxApp instance with the reactor.
74 def _installSignalHandlersAgain(self):
76 wx sometimes removes our own signal handlers, so re-add them.
79 # make _handleSignals happy:
81 signal.signal(signal.SIGINT, signal.default_int_handler)
94 _threadedselect.ThreadedSelectReactor.stop(self)
97 def _runInMainThread(self, f):
99 Schedule function to run in main wx/Twisted thread.
101 Called by the select() thread.
103 if hasattr(self, "wxapp"):
106 # wx shutdown but twisted hasn't
107 self._postQueue.put(f)
112 Stop the wx event loop if it hasn't already been stopped.
114 Called during Twisted event loop shutdown.
116 if hasattr(self, "wxapp"):
117 self.wxapp.ExitMainLoop()
120 def run(self, installSignalHandlers=True):
124 self._postQueue = Queue.Queue()
125 if not hasattr(self, "wxapp"):
126 log.msg("registerWxApp() was not called on reactor, "
127 "registering my own wxApp instance.")
128 self.registerWxApp(wxPySimpleApp())
130 # start select() thread:
131 self.interleave(self._runInMainThread,
132 installSignalHandlers=installSignalHandlers)
133 if installSignalHandlers:
134 self.callLater(0, self._installSignalHandlersAgain)
136 # add cleanup events:
137 self.addSystemEventTrigger("after", "shutdown", self._stopWx)
138 self.addSystemEventTrigger("after", "shutdown",
139 lambda: self._postQueue.put(None))
141 # On Mac OS X, work around wx bug by starting timer to ensure
142 # wxCallAfter calls are always processed. We don't wake up as
143 # often as we could since that uses too much CPU.
144 if runtime.platform.isMacOSX():
145 t = ProcessEventsTimer(self.wxapp)
146 t.Start(2) # wake up every 2ms
148 self.wxapp.MainLoop()
152 if not self._stopping:
153 # wx event loop exited without reactor.stop() being
154 # called. At this point events from select() thread will
155 # be added to _postQueue, but some may still be waiting
156 # unprocessed in wx, thus the ProcessPendingEvents()
159 wxapp.ProcessPendingEvents() # deal with any queued wxCallAfters
162 f = self._postQueue.get(timeout=0.01)
176 Configure the twisted mainloop to be run inside the wxPython mainloop.
178 reactor = WxReactor()
179 from twisted.internet.main import installReactor
180 installReactor(reactor)
184 __all__ = ['install']