Tornado-Web-Sockets und lang laufende Sellerie-Aufgaben

Question

Ich habe einen Tornado-Server, der verwendet wird, um lang laufende (~ Minuten) Berechnungen zu übermitteln, indem er einigen Sellerie-Mitarbeitern Aufgaben mit einem RabbitMQ-Back-End übermittelt. Die Aufgaben, die eingereicht werden, werden in einem Tornado Koroutine in einem WebSocketHandler ergibt:

class MainWSHandler(WebSocketHandler): 
    def open(self): 
     logging.info("Connection opened.") 

    def on_close(self): 
     logging.info("Connection closed.") 

    def on_message(self, message): 
     result = self.submit_task(message) 

     self.write_message("Calculation has been submitted") 

    @gen.coroutine 
    def submit_task(self, params): 
     result = yield gen.Task(long_calculation.apply_async, args=[params]) 

     self.write_message("Completed calculation") 

     return result 

Das funktioniert gut, wenn der Benutzer nie die Seite mit dem aktuell geöffneten Web-Buchse läßt. Wenn dies der Fall ist und der Web-Socket geschlossen wird, schlägt die zurückgegebene Nachricht self.write_message("Completed calculation") mit einem WebSocketClosedError fehl. Dies ist in den Fällen gut, in denen der Benutzer nicht beabsichtigt, eine Weile auf die Seite zurückzukommen (dh bis die Berechnung abgeschlossen ist).

Wenn jedoch der Benutzer eine Berechnung übergibt, die Seite verlässt und vor Abschluss der Berechnung zurückkehrt, wird derselbe Fehler ausgelöst, weil der Web-Socket geschlossen und ein neuer geöffnet wurde. Dies verhindert, dass sich die Berechnungsabschlussnachrichten an das Front-End ausbreiten.

Meine Frage ist: Ist es möglich, wieder an den gleichen Web-Socket zu verbinden? Oder wie kann ich alternativ sicherstellen, dass die zurückgegebene Nachricht nach Abschluss der Berechnung auf die aktuelle Seite des Benutzers gelangt?

Answer 1

Es scheint, als hätte ich die Pistole übersprungen, unter der Annahme, dass es eine strukturelle Lösung für dieses Problem in diesem langwierigen Aufgabe-mit-einem Web-Socket-System gibt.

Es ist genug, um die aktuell geöffnete Web-Buchse in einem Klassenattribut mit der Antwort

class MainWSHandler(WebSocketHandler): 
    self.clients = {} 

def open(self): 
    logging.info("Connection opened.") 
    self.clients[self.current_user] = self 

Und von der Aufgabe beschäftigen zu speichern, indem die gespeicherte Web-Buchse mit

@gen.coroutine 
def submit_task(self, params): 
    result = yield gen.Task(long_calculation.apply_async, args=[params]) 

    # Retrieve current web socket 
    ws = self.clients[self.current_user] 
    ws.write_message("Completed calculation") 

    return result 

Es effizient mehr sein kann Wege (oder eine idiomatische Methode), aber ich bin jetzt mit dieser Lösung zufrieden.