Diffusion dans gRPC : communication serveur-client

Lors de l'établissement d'une connexion gRPC, il est souvent nécessaire de diffuser des événements ou des mises à jour depuis le serveur vers clients connectés. Pour y parvenir, diverses approches peuvent être utilisées.

Stream Observables

Une approche courante consiste à utiliser des flux côté serveur. Chaque client connecté établit son propre flux avec le serveur. Cependant, il n'est pas possible de s'abonner directement à d'autres flux serveur-client.

Interrogation longue

Une option alternative consiste à mettre en œuvre une approche d'interrogation longue. Cela implique que les clients interrogent continuellement le serveur à intervalles réguliers, recherchant de nouvelles mises à jour. Dès réception d'une mise à jour, le client recevra une réponse et attendra le prochain intervalle d'interrogation.

Exemple de mise en œuvre

Voici un exemple de la façon dont vous pourriez implémenter une interrogation longue en utilisant gRPC :

Code côté serveur

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

Code côté client (thème séparé)

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

En mettant en œuvre cette approche, les clients connectés peuvent récupérer les mises à jour du serveur de manière continue.