Comment mettre en œuvre une deux voies jsonrpc + serveur / client torsadé

Question

Bonjour, je travaille sur un serveur de développer rpc basé sur torsadé pour servir plusieurs micro-contrôleurs qui font appel au serveur rpc jsonrpc torsadé. Mais l'application aussi nécessaire que les informations d'envoi du serveur à chaque micro à tout moment, la question est de savoir comment pourrait être une bonne pratique pour éviter que la réponse d'un appel jsonrpc à distance à partir d'un micro-être confondu avec une demande de jsonrpc serveur qui est fait pour un utilisateur.

La conséquence que j'ai maintenant que micro-ordinateurs reçoivent une mauvaise information, parce qu'ils ne savent pas si netstring / string JSON qui est comming de la prise est leur réponse à une exigence antérieure ou une nouvelle demande du serveur.

Voici mon code:

from twisted.internet import reactor
from txjsonrpc.netstring import jsonrpc
import weakref

creds  = {'user1':'pass1','user2':'pass2','user3':'pass3'}

class arduinoRPC(jsonrpc.JSONRPC):
    def connectionMade(self):
        pass

    def jsonrpc_identify(self,username,password,mac):
        """ Each client must be authenticated just after to be connected calling this rpc """
        if creds.has_key(username):
            if creds[username] == password:
                authenticated = True
            else:
                authenticated = False
        else:
            authenticated = False

        if authenticated:
            self.factory.clients.append(self)
            self.factory.references[mac] = weakref.ref(self)
            return {'results':'Authenticated as %s'%username,'error':None}
        else:
            self.transport.loseConnection()

    def jsonrpc_sync_acq(self,data,f):
        """Save into django table data acquired from sensors and send ack to gateway"""
        if not (self in self.factory.clients):
            self.transport.loseConnection()
        print f
        return {'results':'synced %s records'%len(data),'error':'null'}

    def connectionLost(self, reason):
        """ mac address is searched and all reference to self.factory.clientes are erased """  
        for mac in self.factory.references.keys():
            if self.factory.references[mac]() == self:
                print 'Connection closed - Mac address: %s'%mac
                del self.factory.references[mac]
                self.factory.clients.remove(self)


class rpcfactory(jsonrpc.RPCFactory):
    protocol = arduinoRPC
    def __init__(self, maxLength=1024):
        self.maxLength = maxLength
        self.subHandlers = {}
        self.clients    =   []
        self.references =   {}

""" Asynchronous remote calling to micros, simulating random calling from server """
import threading,time,random,netstring,json
class asyncGatewayCalls(threading.Thread):
    def __init__(self,rpcfactory):
        threading.Thread.__init__(self)
        self.rpcfactory =   rpcfactory
        """identifiers of each micro/client connected"""
        self.remoteMacList    =   ['12:23:23:23:23:23:23','167:67:67:67:67:67:67','90:90:90:90:90:90:90']
    def run(self):
        while True:
            time.sleep(10)
            while True:
                """ call to any of three potential micros connected """ 
                mac = self.remoteMacList[random.randrange(0,len(self.remoteMacList))]
                if self.rpcfactory.references.has_key(mac):
                    print 'Calling %s'%mac
                    proto   =   self.rpcfactory.references[mac]()
                    """ requesting echo from selected micro"""
                    dataToSend  = netstring.encode(json.dumps({'method':'echo_from_micro','params':['plop']}))
                    proto.transport.write(dataToSend)
                    break

factory = rpcfactory(arduinoRPC)

"""start thread caller""" 
r=asyncGatewayCalls(factory)
r.start()

reactor.listenTCP(7080, factory)
print "Micros remote RPC server started"
reactor.run()

Answer 1

Vous devez ajouter une information suffisante pour chaque message afin que le destinataire peut déterminer comment l'interpréter. Vos exigences sons très similaires à ceux de AMP , vous pouvez donc soit utiliser à la place AMP ou utiliser la même structure que l'AMP pour identifier vos messages. Plus précisément:

• Dans les demandes, mettre une clé particulière - par exemple, AMP utilise « _ask » pour identifier les demandes. Il donne également une valeur unique, qui identifie en outre que la demande pour la durée de la connexion.
• Dans les réponses, mettez une clé différente - par exemple, AMP utilise pour cette « _answer ». La valeur correspond à la valeur avec de la touche « _ask » dans la demande la réponse est pour.

En utilisant une approche comme celui-ci, il vous suffit de regarder pour voir s'il y a une touche « _ask » ou une touche « _answer » pour déterminer si vous avez reçu une nouvelle demande ou une réponse à une demande précédente.

Sur un sujet distinct, votre classe asyncGatewayCalls ne doit pas se fonder sur fil. Il n'y a aucune raison apparente pour elle d'utiliser des fils et en le faisant, il est également utilise à mauvais escient des API Twisted d'une manière qui conduira à un comportement non défini. La plupart des API Twisted ne peut être utilisé dans le fil dans lequel vous avez appelé reactor.run. La seule exception est reactor.callFromThread, que vous pouvez utiliser pour envoyer un message au fil du réacteur d'un autre fil. asyncGatewayCalls tente d'écrire un transport, sans, ce qui conduira à la corruption ou tampon des retards arbitraires dans les données envoyées, ou peut-être des choses bien pires. Au lieu de cela, vous pouvez écrire asyncGatewayCalls comme ceci:

from twisted.internet.task import LoopingCall

class asyncGatewayCalls(object):
    def __init__(self, rpcfactory):
        self.rpcfactory = rpcfactory
        self.remoteMacList = [...]

    def run():
        self._call = LoopingCall(self._pokeMicro)
        return self._call.start(10)

    def _pokeMicro(self):
        while True:
            mac = self.remoteMacList[...]
            if mac in self.rpcfactory.references:
                proto = ...
                dataToSend = ...
                proto.transport.write(dataToSend)
                break

factory = ...
r = asyncGatewayCalls(factory)
r.run()

reactor.listenTCP(7080, factory)
reactor.run()

Cela vous donne une solution mono-thread qui devrait avoir le même comportement que vous aviez l'intention de la classe asyncGatewayCalls d'origine. Au lieu de dormir dans une boucle dans un fil afin de programmer les appels, cependant, il utilise les API de programmation du réacteur (via la classe LoopingCall de niveau supérieur, où les choses schedules à appeler à plusieurs reprises) pour faire _pokeMicro que est appelé toutes les dix secondes .