Welcome, guest | Sign In | My Account | Store | Cart

Qt + Twisted (threadedselectreactor) (Python recipe) by Jonathan Kolyer
ActiveState Code (http://code.activestate.com/recipes/466305/)

Starting with the recipe "Network Ping Pong using Twisted Prespective Broker" (181905), this recipe integrates Qt using the threadedselectreactor. An alternative to qtreactor, this recipe allows dispatching messages over the perspective broker while running the Qt event loop. It basically shows how to use reactor.interleave(aFunction) within the context of Qt.

Python, 129 lines
Download
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
import sys
from thread import get_ident

from qt import *

if len( sys.argv ) > 1 :
    from twisted.internet.threadedselectreactor import install
    install()
    
from twisted.internet import reactor
from twisted.spread import pb

PORT = 9991

DELAY         = 1
DOG_DELAY     = 2
RESTART_DELAY = 5

class Pinger:

    def __init__( self, host ):
        self.ping = None
        self.host = host
        self.ball = 0
        self._start()
        
    def _start( self ):
        print 'Waiting for Server...'
        client = pb.PBClientFactory()
        connector = reactor.connectTCP(host='127.0.0.1', port=PORT, factory=client, timeout=30)
        dfr = client.getRootObject()
        dfr.addCallbacks( self._gotRemote, self._remoteFail )
    
    def _gotRemote( self, remote ):
        print 'Got remote...'
        remote.notifyOnDisconnect( self._remoteFail )
        self.remote = remote
        self._ping()

    def _remoteFail( self, _ ):
        if self.ping:
            self.ping.cancel()
            self.ping = None
        self.restart = reactor.callLater( RESTART_DELAY, self._start )

    def _ping( self ):
        self.dog = reactor.callLater( DOG_DELAY, self._start )
        self.ball += 1
        print 'THROW', self.ball,
        dfr = self.remote.callRemote( 'Pong', self.ball )
        dfr.addCallbacks( self._pong, self._remoteFail )
            
    def _pong( self, ball ):
        self.dog.cancel()
        print 'CATCH',  ball
        self.ball = ball
        self.ping = reactor.callLater( DELAY, self._ping )

class Ponger( pb.Root ):

    def remote_Pong( self, ball ):
        print 'CATCH', ball,
        ball += 1
        print 'THROW', ball
        return ball

class BaseQObject(QObject):

    MAIN_THREAD_ID = 0
    
    def __init__(self):
        QObject.__init__(self)
        self.installEventFilter(self)
        self.event = None

    def eventFilter(self,obj,event):
        # FIXME:  This is a workaround for an unexplained bug
        # The events were getting posted through postEVentWithCallback()
        # But the event() method wasn't getting called.  But the eventFilter()
        # method is getting called.  
        if event.type()==QEvent.User:
            cb = event.__dict__.get('callback')
            if cb: self._doEvent(event)
            return False
        return QObject.eventFilter(self,obj,event)

    def _doEvent(self,event):
        cb = event.__dict__.get('callback')
        if not cb: return
        data = event.__dict__.get('data')
        if data or type(data)==type(False): cb(data)
        else: cb()
        del event

    def event(self, event):
        if event.type()==QEvent.User:
            self._doEvent(event)
            return True
        return QObject.event(self, event)

    def postEventWithCallback(self, callback, data=None):
        # if we're in main thread, just fire off callback
        if get_ident()==BaseQObject.MAIN_THREAD_ID:
            if data or type(data)==type(False): callback(data)
            else: callback()
        # send callback to main thread 
        else:
            event = QEvent(QEvent.User)
            event.callback = callback
            if data or type(data)==type(False): event.data = data
            qApp.postEvent(self, event)
            
class Interleaver(BaseQObject):
    def __init__(self):
        BaseQObject.__init__(self)

    def toInterleave(self, func, *args, **kwargs):
        #print('toInterleave(): %s'%(str(func)))
        self.postEventWithCallback(func)

if len( sys.argv ) > 1 :
    Pinger( sys.argv[1] )
    ii = Interleaver()
    reactor.interleave(ii.toInterleave)
    app = QApplication([])
    app.exec_loop()
else:
    reactor.listenTCP( PORT, pb.PBServerFactory( Ponger()))
    reactor.run()

I needed two processes, each running Qt, to use twisted.spread.pb for message communication. However qtreactor didn't allow integration between its event loop and the reactor select() loop. This technique is a workaround.

The qtreactor was preventing message sending while running the Qt event loop, so I got Qt working with threadedselectreactor.

Tags: network

3 comments

Stephen Chappell 18 years, 3 months ago  # | flag

Qt. Do you think that you could make a second version of this code that does not require Qt? It's somewhat difficult to get Qt for Widnows, and I am sure that your program is nice ...

Daniel Miller 17 years, 4 months ago  # | flag

Updated version. Here's a slightly shorter version that supports passing arbitrary positional arguments to the callback.

from qt import QEvent, QObject, qApp
from thread import get_ident

class CallbackEventHandler(QObject):

    MAIN_THREAD_ID = 0

    def __init__(self):
        QObject.__init__(self)
        self.installEventFilter(self)

    def eventFilter(self, obj, event):
        # FIXME:  This is a workaround for an unexplained bug
        # The events were getting posted through postEVentWithCallback()
        # But the event() method wasn't getting called.  But the eventFilter()
        # method is getting called.
        if event.type() == QEvent.User:
            callback = event.__dict__.get('callback')
            if callback:
                self._doEvent(event)
            return False
        return QObject.eventFilter(self, obj, event)

    def _doEvent(self, event):
        callback = event.__dict__.get('callback')
        args = event.__dict__.get('args')
        if callback is not None and args is not None:
            callback(*args)

    def event(self, event):
        if event.type() == QEvent.User:
            self._doEvent(event)
            return True
        return QObject.event(self, event)

    def postEventWithCallback(self, callback, *args):
        if get_ident() == CallbackEventHandler.MAIN_THREAD_ID:
            # if we're in main thread, just fire off callback
            callback(*args)
        else:
            # send callback to main thread
            event = QEvent(QEvent.User)
            event.callback = callback
            event.args = args
            qApp.postEvent(self, event)
Daniel Miller 17 years, 4 months ago  # | flag

Oops... Sorry about that...posted to the wrong thread (too many tabs in Firefox :). This was meant to go on this thread: http://aspn.activestate.com/ASPN/Cookbook/Python/Recipe/415311