Transmisión en gRPC: comunicación de servidor a cliente

Al establecer una conexión gRPC, a menudo es necesario transmitir eventos o actualizaciones desde el servidor a clientes conectados. Para lograr esto, se pueden emplear varios enfoques.

Stream Observables

Un enfoque común es utilizar flujos del lado del servidor. Cada cliente conectado establece su propia transmisión con el servidor. Sin embargo, no es factible suscribirse directamente a otras transmisiones servidor-cliente.

Encuesta larga

Una opción alternativa es implementar un enfoque de encuesta larga. Esto implica que los clientes sondeen continuamente el servidor a intervalos regulares, buscando nuevas actualizaciones. Al recibir una actualización, el cliente recibirá una respuesta y esperará el siguiente intervalo de sondeo.

Ejemplo de implementación

Aquí se muestra un ejemplo de cómo implementar un sondeo largo usando gRPC:

Código del lado del servidor

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

Código del lado del cliente (tema independiente)

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

Al implementar este enfoque, los clientes conectados pueden recuperar actualizaciones del servidor de forma continua.