Twisted task.loop 및 pb auth

Question

Twisted를 배우십시오. 필자는 서버와 클라이언트를 작성하여 데이터를 공유하기로 결정했습니다. 한 가지 구현을 작성했지만 올바른 것은 아닙니다.

# -*- 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를 보낼 경우 로그인 또는 비밀번호 거짓 -> 차단, 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.