Transmissão em gRPC: comunicação de servidor para cliente

Ao estabelecer uma conexão gRPC, muitas vezes é necessário transmitir eventos ou atualizações do servidor para clientes conectados. Para conseguir isso, várias abordagens podem ser empregadas.

Stream Observáveis

Uma abordagem comum é utilizar fluxos do lado do servidor. Cada cliente conectado estabelece seu próprio fluxo com o servidor. No entanto, inscrever-se diretamente em outros fluxos servidor-cliente não é viável.

Long-Polling

Uma opção alternativa é implementar uma abordagem de pesquisa longa. Isso envolve fazer com que os clientes pesquisem continuamente o servidor em intervalos regulares, verificando se há novas atualizações. Ao receber uma atualização, o cliente receberá uma resposta e aguardará o próximo intervalo de pesquisa.

Exemplo de implementação

Aqui está um exemplo de como você pode implementar pesquisas longas usando gRPC:

código do lado do 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 do lado do cliente (thread separado)

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

Ao implementar esta abordagem, os clientes conectados podem recuperar atualizações do servidor de forma contínua.