Как реализовать двухсторонний jsonrpc + скрученный сервер/клиент

Question

Здравствуйте, я работаю над созданием сервера rpc на основе скручивания, чтобы обслуживать несколько микроконтроллеров, которые делают вызов rpc на скрученный сервер jsonrpc. Но приложение также требовало, чтобы сервер отправлял информацию в каждый микро в любое время, поэтому вопрос заключается в том, как может быть хорошей практикой, чтобы предотвратить ответ от удаленного вызова jsonrpc с микропотока с запросом сервера jsonrpc, который создается для Пользователь.

Последствия, которые возникают у меня сейчас, это то, что микроны получают плохую информацию, потому что они не знают, является ли строка netstring/json, которая отправляется из сокета, их ответ от предыдущего требования или новый запрос с сервера.

Вот мой код:

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

Вы должны добавить достаточное количество информации для каждого сообщения, так что получатель может определить, как интерпретировать его. Ваши требования звучат так же, как и у AMP, поэтому вы можете использовать AMP или использовать ту же структуру, что и AMP, для идентификации ваших сообщений. В частности:

• В запросах поместите конкретный ключ - например, AMP использует «_ask» для идентификации запросов. Он также дает им уникальное значение, которое дополнительно идентифицирует этот запрос на время жизни соединения.
• В ответах введите другой ключ - например, AMP использует для этого «_answer». Значение соответствует значению из «_ask» в запросе, на который отвечает ответ.

Используя такой подход, вам просто нужно посмотреть, есть ли ключ «_ask» или «_answer», чтобы определить, был ли вы получили новый запрос или ответ на предыдущий запрос ,

В отдельной теме ваш класс asyncGatewayCalls не должен быть основан на потоках. Нет никаких очевидных причин для использования потоков, и тем самым он также неправильно использует Twisted API, что приведет к неопределенному поведению. Большинство Twisted APIs могут использоваться только в потоке, в котором вы вызывали reactor.run. Единственным исключением является reactor.callFromThread, который вы можете использовать, чтобы отправить сообщение в реакторную нить из любой другой нити. asyncGatewayCalls пытается записать на транспорт, хотя это приведет к повреждению буфера или произвольным задержкам в отправляемых данных или, возможно, к худшему. Вместо этого, вы можете написать asyncGatewayCalls так:

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() 

Это дает однопоточное решение, которое должно иметь такое же поведение, как вы предназначены для оригинального asyncGatewayCalls класса. Однако вместо того, чтобы спать в петле в потоке, чтобы запланировать вызовы, он использует API-схемы планирования реактора (через класс LoopingCall более высокого уровня, который планирует многократно называть вещи), чтобы удостовериться, что _pokeMicro вызывается каждые десять секунд ,