公共模块qt4reactor.py：
# -*- coding: utf-8 -*-
# Copyright (c) 2001-2008 Twisted Matrix Laboratories.
# See LICENSE for details.
# Comes from http://sourcecodebrowser.com/openvswitch/1.1.0~pre2.g2.ea763e0e/qt4reactor_8py_source.html
# The referred licence file contains:
#
#Copyright (c) 2001-2010
#Permission is hereby granted, free of charge, to any person obtaining
#a copy of this software and associated documentation files (the
#"Software"), to deal in the Software without restriction, including
#without limitation the rights to use, copy, modify, merge, publish,
#distribute, sublicense, and/or sell copies of the Software, and to
#permit persons to whom the Software is furnished to do so, subject to
#the following conditions:
#
#The above copyright notice and this permission notice shall be
#included in all copies or substantial portions of the Software.
#
#THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
#EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
#MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
#NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
#LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
#OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
#WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
"""
This module provides support for Twisted to be driven by the Qt mainloop.
In order to use this support, simply do the following::
| app = QApplication(sys.argv) # your code to init Qt
| import qt4reactor
| qt4reactor.install()
alternatively:
| from twisted.application import reactors
| reactors.installReactor('qt4')
Then use twisted.internet APIs as usual. The other methods here are not
intended to be called directly.
If you don't instantiate a QApplication or QCoreApplication prior to
installing the reactor, a QCoreApplication will be constructed
by the reactor. QCoreApplication does not require a GUI so trial testing
can occur normally.
Twisted can be initialized after QApplication.exec_() with a call to
reactor.runReturn(). calling reactor.stop() will unhook twisted but
leave your Qt application running
API Stability: stable
Maintainer: U{Glenn H Tarbox, PhD}
Previous maintainer: U{Itamar Shtull-Trauring}
Original port to QT4: U{Gabe Rudy}
Subsequent port by therve
"""
__all__ = ['install']
import sys, time
try:
from zope.interface import implements #接口实现
except:
print('+++ Python Zope interface module is required\n')
raise
from PyQt4.QtCore import QSocketNotifier, QObject, SIGNAL, QTimer, QCoreApplication
from PyQt4.QtCore import QEventLoop
try:
from twisted.internet.interfaces import IReactorFDSet #Implement me to be able to use IFileDescriptor type resources.
from twisted.python import log
from twisted.internet.posixbase import PosixReactorBase #A basis for reactors that use file descriptors.
except:
print('+++ Python Twisted Conch module is required\n')
raise
class TwistedSocketNotifier(QSocketNotifier):
"""
Connection between an fd event and reader/writer callbacks.
"""
def __init__(self, reactor, watcher, type):
QSocketNotifier.__init__(self, watcher.fileno(), type) # fileno方法返回整数的底层实现使用请求从操作系统的I / O操作的文件描述符
self.reactor = reactor
self.watcher = watcher
self.fn = None
if type == QSocketNotifier.Read: # QSocketNotifier.Read 0 There is data to be read.
self.fn = self.read
elif type == QSocketNotifier.Write: #QSocketNotifier.Write 1 Data can be written.
self.fn = self.write
QObject.connect(self, SIGNAL("activated(int)"), self.fn) #bool QObject.connect (self, QObject, SIGNAL(), SLOT(), Qt.ConnectionType = Qt.AutoConnection)
def shutdown(self):
QObject.disconnect(self, SIGNAL("activated(int)"), self.fn)
self.setEnabled(False) #bool QSocketNotifier.isEnabled (self). Returns true if the notifier is enabled; otherwise returns false.
self.fn = self.watcher = None
self.deleteLater() #QObject.deleteLater (self). The object will be deleted when control returns to the event loop. If the event loop is not running when this function is called
def read(self, sock):
w = self.watcher
#self.setEnabled(False) # ??? do I need this?
def _read():
why = None
try:
why = w.doRead() #twisted.internet.abstract.FileDescriptor.
except:
log.err()
why = sys.exc_info()[1]
if why:
self.reactor._disconnectSelectable(w, why, True) #twisted.internet.posixbase. def _disconnectSelectable(self, selectable, why, isRead, faildict={error.ConnectionDone:failure.Failure(error.ConnectionDone()),error.ConnectionLost:failure.Failure(error.ConnectionLost()),})
elif self.watcher:
pass
#self.setEnabled(True)
log.callWithLogger(w, _read)
self.reactor.reactorInvocation() #类QTReactor中方法
def write(self, sock):
w = self.watcher
self.setEnabled(False)
def _write():
why = None
try:
why = w.doWrite()
except:
log.err()
why = sys.exc_info()[1]
if why:
self.reactor._disconnectSelectable(w, why, False)
elif self.watcher:
self.setEnabled(True)
log.callWithLogger(w, _write)
self.reactor.reactorInvocation()
class fakeApplication(QEventLoop): #The QEventLoop class provides a means of entering and leaving an event loop.
def __init__(self):
QEventLoop.__init__(self)
def exec_(self): #Enters the main event loop and waits until exit() is called. Returns the value that was passed to exit().
QEventLoop.exec_(self)
class QTReactor(PosixReactorBase):
"""
Qt based reactor.
"""
implements(IReactorFDSet)
_timer = None
def __init__(self):
self._reads = {}
self._writes = {}
self._timer=QTimer() # The QTimer class provides repetitive and single-shot timers.
self._timer.setSingleShot(True) # You can set a timer to time out only once by calling setSingleShot(true).表示在时间结束之后只发送一次time out信息. 否则缺省计时器会重复启发直至他们停止或者销毁.
if QCoreApplication.startingUp(): # bool QCoreApplication.startingUp (). Returns true if an application object has not been created yet; otherwise returns false.
self.qApp=QCoreApplication([]) # The QCoreApplication class provides an event loop for console Qt applications.
self._ownApp=True
else:
self.qApp = QCoreApplication.instance() # QCoreApplication QCoreApplication.instance () . Returns a pointer to the application's QCoreApplication (or QApplication) instance.
self._ownApp=False
self._blockApp = None
self._readWriteQ=[]
""" some debugging instrumentation """
self._doSomethingCount=0
PosixReactorBase.__init__(self)
def addReader(self, reader):
if not reader in self._reads:
self._reads[reader] = TwistedSocketNotifier(self, reader,
QSocketNotifier.Read)
def addWriter(self, writer):
if not writer in self._writes:
self._writes[writer] = TwistedSocketNotifier(self, writer,
QSocketNotifier.Write)
def removeReader(self, reader):
if reader in self._reads:
#self._reads[reader].shutdown()
#del self._reads[reader]
self._reads.pop(reader).shutdown()
def removeWriter(self, writer):
if writer in self._writes:
self._writes[writer].shutdown()
#del self._writes[writer]
self._writes.pop(writer)
def removeAll(self):
return self._removeAll(self._reads, self._writes)
def getReaders(self):
return self._reads.keys()
def getWriters(self):
return self._writes.keys()
def callLater(self,howlong, *args, **kargs):
rval = super(QTReactor,self).callLater(howlong, *args, **kargs)
self.reactorInvocation()
return rval
def crash(self):
super(QTReactor,self).crash()
def iterate(self,delay=0.0):
t=self.running # not sure I entirely get the state of running
self.running=True
self._timer.stop() # in case its not (rare?)
try:
if delay == 0.0:
self.reactorInvokePrivate()
self._timer.stop() # supports multiple invocations
else:
endTime = delay + time.time()
self.reactorInvokePrivate()
while True:
t = endTime - time.time()
if t <= 0.0: return
self.qApp.processEvents(QEventLoop.AllEvents |
QEventLoop.WaitForMoreEvents,t*1010)
finally:
self.running=t
def addReadWrite(self,t):
self._readWriteQ.append(t)
def runReturn(self, installSignalHandlers=True):
QObject.connect(self._timer, SIGNAL("timeout()"),
self.reactorInvokePrivate)
self.startRunning(installSignalHandlers=installSignalHandlers)
self._timer.start(0)
def run(self, installSignalHandlers=True):
try:
if self._ownApp:
self._blockApp=self.qApp
else:
self._blockApp = fakeApplication()
self.runReturn(installSignalHandlers)
self._blockApp.exec_()
finally:
self._timer.stop() # should already be stopped
def reactorInvocation(self):
self._timer.setInterval(0)
def reactorInvokePrivate(self):
if not self.running:
if self._blockApp is None:
# Andy's fix for Ctrl-C quit
self.qApp.quit()
else:
self._blockApp.quit()
self._doSomethingCount += 1
self.runUntilCurrent()
t = self.timeout()
if t is None: t=0.1
else: t = min(t,0.1)
self._timer.setInterval(int(t*1010))
self.qApp.processEvents() # could change interval
self._timer.start()
def doIteration(self):
assert False, "doiteration is invalid call"
def install():
"""
Configure the twisted mainloop to be run inside the qt mainloop.
"""
from twisted.internet import main
reactor = QTReactor()
main.installReactor(reactor)
return reactor
if __name__ == "__main__":
from PyQt4 import QtGui
app = QtGui.QApplication(sys.argv)
import qt4reactor
reactor=qt4reactor.install()
reactor.runReturn()