gRPC 서버 스트리밍 상태 수신

스트리밍 수신의 필요성

장비 상태를 주기적으로 조회하면, 변경이 없어도 요청이 발생하고 변경은 다음 주기가 되어야 확인된다. 서버 스트리밍은 호출 하나를 유지한 상태에서, 서버가 상태를 전송할 때마다 클라이언트가 수신하는 방식이다.

주기적 조회와 서버 스트리밍의 메시지 흐름 비교

서버 측 yield 전송

returns (stream StatusReply)로 정의한 메서드는 Python에서 generator로 구현한다. yield 한 번은 메시지 하나에 대응한다.

pythonserver.py

import time

import status_pb2
import status_pb2_grpc


class StatusService(status_pb2_grpc.StatusServiceServicer):
    def WatchStatus(self, request, context):
        # 클라이언트가 끊으면 is_active()가 False가 된다
        while context.is_active():
            yield status_pb2.StatusReply(device_id=request.device_id, battery=87)
            time.sleep(1)

클라이언트 측 for 수신

pythonclient.py

import grpc

import status_pb2
import status_pb2_grpc

with grpc.insecure_channel("localhost:50051") as channel:
    stub = status_pb2_grpc.StatusServiceStub(channel)
    call = stub.WatchStatus(status_pb2.StatusRequest(device_id="d-01"))
    try:
        for reply in call:
            print(reply.device_id, reply.battery)
    except grpc.RpcError as e:
        # 서버나 연결 문제로 스트림이 끝나면 여기로 온다
        print(e.code())

수신을 중단하려면 call.cancel()을 호출한다.

서버 스트리밍의 yield 전송과 for 수신 순서

단절 감지

클라이언트 측에서 호출이 단절되어도, 다음 상태를 대기하며 블록 상태인 핸들러(호출을 처리하는 서버 메서드)는 자동으로 깨어나지 않는다. 동기 서버에서는 핸들러 하나가 작업 스레드 하나를 점유하므로, 단절된 호출을 종료하지 않으면 스레드가 부족해진다.

context.is_active(): 호출의 활성 여부를 반환한다. 반복 조건에 사용한다.
context.add_callback(fn): 호출이 종료될 때 fn을 호출한다. 큐나 이벤트를 대기하며 블록 상태인 핸들러를 깨우는 데 사용한다.

클라이언트 측 호출 단절 후 핸들러를 종료하는 경로