gRPC でのブロードキャスト: サーバーからクライアントへの通信

gRPC 接続を確立するとき、多くの場合、サーバーからクライアントにイベントや更新をブロードキャストする必要があります。接続されたクライアント。これを実現するには、さまざまなアプローチを使用できます。

Stream Observables

一般的なアプローチの 1 つは、サーバー側ストリームを利用することです。接続された各クライアントは、サーバーとの独自のストリームを確立します。ただし、他のサーバー/クライアント ストリームに直接サブスクライブすることは実現できません。

ロングポーリング

別のオプションは、ロングポーリングアプローチを実装することです。これには、クライアントが一定の間隔でサーバーを継続的にポーリングし、新しい更新をチェックすることが含まれます。更新を受信すると、クライアントは応答を受信し、次のポーリング間隔を待ちます。

実装例

ロングポーリングを実装する方法の例を次に示します。 gRPC の使用:

サーバーサイド コード

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

クライアント側コード (別スレッド)

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

このアプローチを実装すると、接続されたクライアントはサーバーから継続的に更新を取得できます。