Broadcasting in gRPC: Server-zu-Client-Kommunikation

Beim Aufbau einer gRPC-Verbindung ist es oft notwendig, Ereignisse oder Aktualisierungen vom Server an zu senden angeschlossene Clients. Um dies zu erreichen, können verschiedene Ansätze eingesetzt werden.

Stream Observables

Ein gängiger Ansatz ist die Nutzung serverseitiger Streams. Jeder verbundene Client baut seinen eigenen Stream mit dem Server auf. Das direkte Abonnieren anderer Server-Client-Streams ist jedoch nicht möglich.

Long-Polling

Eine alternative Option ist die Implementierung eines Long-Polling-Ansatzes. Dazu gehört, dass Clients den Server in regelmäßigen Abständen kontinuierlich abfragen und nach neuen Updates suchen. Nach Erhalt eines Updates erhält der Client eine Antwort und wartet auf das nächste Abfrageintervall.

Beispielimplementierung

Hier ist ein Beispiel dafür, wie Sie lange Abfragen implementieren könnten mit gRPC:

Serverseitiger Code

import threading

class UpdaterServer:
    def __init__(self):
        self.condition = threading.Condition()
        self.updates = []

    def post_update(self, update):
        with self.condition:
            self.updates.append(updates)
            self.condition.notify_all()

    def GetUpdates(self, req, context):
        with self.condition:
            while self.updates[req.last_received_update   1:] == []:
                self.condition.wait()
            new_updates = self.updates[req.last_received_update   1:]
            return GetUpdatesResponse(
                updates=new_updates,
                update_index=req.last_received_update   len(new_updates),
            )

Clientseitiger Code (separater Thread)

from threading import Event

def handle_updates(updates):
    pass

event = Event()
request = GetUpdatesRequest(last_received_update=-1)

while not event.is_set():
    try:
        stub = UpdaterStub(channel)
        response = stub.GetUpdates(request, timeout=60*10)
        handle_updates(response.updates)
        request.last_received_update = response.update_index
    except grpc.FutureTimeoutError:
        pass

Durch die Implementierung dieses Ansatzes können verbundene Clients kontinuierlich Aktualisierungen vom Server abrufen.