Twisted task.loopとpb auth

Question

Twistedを学ぶ。私は1秒に1回、データを共有するサーバーとクライアントを作成することに決めました。 1つの実装を記述しましたが、正しくないと思われます。

# -*- coding: utf-8 -*- 

from twisted.spread import pb 
from twisted.internet import reactor, task 
from twisted.cred import credentials 
from win32com.server import factory 

class login_send: 

    def __init__(self): 
     self.count=0 
     self.timeout = 1.0 
     self.factory = pb.PBClientFactory() 
     reactor.connectTCP("localhost", 8800, self.factory) 

    def testTimeout(self): 
     self.count+=1 
     print self.count 

     def1 = self.factory.login(credentials.UsernamePassword("test1","bb1b")) 
     def1.addCallbacks(self.good_connected, self.bad_connected) 
     def1.addCallback(self.send_data) 
     def1.addErrback(self.disconnect) 
     if self.count>10:def1.addBoth(self.disconnect) 

    def start(self): 
     l = task.LoopingCall(self.testTimeout) 
     l.start(self.timeout) 
     reactor.run() 

    def good_connected(self, perspective): 
     print 'good login and password', perspective 
     return perspective 

    def bad_connected(self, perspective): 
     print 'bad login or password', perspective 
     return perspective 

    def send_data(self, perspective): 
     print 'send' 
     return perspective.callRemote("foo", self.count) 

    def disconnect(self, perspective): 
     print 'disconnect' 
     reactor.stop() 

if __name__ == "__main__": 
    st=login_send() 
    st.start() 

コード：ログインとパスワード真の場合 - > self.countを送る、ログインやパスワードFalseの場合 - >切断、self.count> 10の場合 - >

私の意見では最初のミスを、切り離し私は毎回ログインしなければならないということです。 認証を行い、データを毎秒を送信するために継続する方法

def1 = self.factory.login(credentials.UsernamePassword("test1", "bb1b")) 

？

簡単なテストサーバーコード：

from zope.interface import implements 

from twisted.spread import pb 
from twisted.cred import checkers, portal 
from twisted.internet import reactor 

class MyPerspective(pb.Avatar): 
    def __init__(self, name): 
     self.name = name 
    def perspective_foo(self, arg): 
     print "I am", self.name, "perspective_foo(",arg,") called on", self 
     return arg 

class MyRealm: 
    implements(portal.IRealm) 
    def requestAvatar(self, avatarId, mind, *interfaces): 
     if pb.IPerspective not in interfaces: 
      print 'qqqq' 
      raise NotImplementedError 
     return pb.IPerspective, MyPerspective(avatarId), lambda:None 

p = portal.Portal(MyRealm()) 
c = checkers.InMemoryUsernamePasswordDatabaseDontUse(test1="bbb", 
                user2="pass2") 
p.registerChecker(c) 
reactor.listenTCP(8800, pb.PBServerFactory(p)) 
reactor.run() 

Answer 1

は、私は、これはトリックを行うべきだと考えています。

# Upper case first letter of class name is good policy. 
class Login_send: 

    def __init__(self): 
     # initialize the state variable to False. 
     self.connection = False 
     self.count=0 
     self.timeout = 1.0 
     self.factory = pb.PBClientFactory() 
     reactor.connectTCP("localhost", 8800, self.factory) 

    def testTimeout(self): 
     self.count+=1 
     print self.count 

     # no connection -- create one. 
     if not self.connection: 
      self.assign_connection() 

     # cached connection exists, call send_data manually. 
     elif self.count > 10: 
      self.disconnect(self.connection) 
     else: 
      #you probably want to send data only if it it should be valid. 
      self.send_data(self.connection)  

    def assign_connection(self): 
    ''' Creates and stores a Deffered which represents the connection to 
     the server. ''' 
     # cache the connection. 
     self.connection = self.factory.login(
           credentials.UsernamePassword("test1","bb1b")) 
     # add connection callbacks as normal. 
     self.connection.addCallbacks(
           self.good_connected, self.bad_connected) 
     self.connection.addCallback(self.send_data) 
     self.connection.addErrback(self.disconnect) 

    def disconnect(self, perspective): 
     # be sure to cleanup after yourself! 
     self.connection = False 
     print 'disconnect' 
     reactor.stop() 

    # the rest of your class goes here.