파이썬 gRPC로 스트리밍 메시지를 처리하는 방법

Question

나는이 메시지를 따르고있다. Route_Guide sample.

해당 샘플이 발생하고 특정 메시지에 회신하지 않고 메시지를 읽습니다. 후자는 제가 성취하려고하는 것입니다.

여기 내가 지금까지이 작업은 다음과 같습니다

import grpc 
... 

channel = grpc.insecure_channel(conn_str) 
try: 
    grpc.channel_ready_future(channel).result(timeout=5) 
except grpc.FutureTimeoutError: 
    sys.exit('Error connecting to server') 
else: 
    stub = MyService_pb2_grpc.MyServiceStub(channel) 
    print('Connected to gRPC server.') 
    this_is_just_read_maybe(stub) 


def this_is_just_read_maybe(stub): 
    responses = stub.MyEventStream(stream()) 
    for response in responses: 
     print(f'Received message: {response}') 
     if response.something: 
      # okay, now what? how do i send a message here? 

def stream(): 
    yield my_start_stream_msg 
    # this is fine, i receive this server-side 
    # but i can't check for incoming messages here 

내가 스텁에 read() 또는 write()을하지 않는 것, 모든 반복자로 구현 될 것으로 보인다.

this_is_just_read_maybe(stub)에서 메시지를 보내려면 어떻게해야합니까? 그게 올바른 접근 방법입니까?

service MyService { 
    rpc MyEventStream (stream StreamingMessage) returns (stream StreamingMessage) {} 
} 

Answer 1

당신이 할하려는 것은 완벽하게 가능하며 그들이을 사용하지 않고 도착 아마 응답을 부여 할 수 있습니다 자신의 요청 iterator 객체를 작성 포함합니다 :

내 프로토는 양방향 스트림 귀하의 요청 iterator로 간단한 생성기. 당신이 다음

my_smarter_request_iterator = MySmarterRequestIterator() 
responses = stub.MyEventStream(my_smarter_request_iterator) 
for response in responses: 
    my_smarter_request_iterator.add_response(response) 

처럼 사용

class MySmarterRequestIterator(object): 

    def __init__(self): 
     self._lock = threading.Lock() 
     self._responses_so_far = [] 

    def __iter__(self): 
     return self 

    def _next(self): 
     # some logic that depends upon what responses have been seen 
     # before returning the next request message 
     return <your message value> 

    def __next__(self): # Python 3 
     return self._next() 

    def next(self): # Python 2 
     return self._next() 

    def add_response(self, response): 
     with self._lock: 
      self._responses.append(response) 

같은 아마 뭔가. _next 구현에 잠금 및 차단 기능이있어 gRPC Python의 상황을 처리하여 전송할 다음 요청 및 응답 (사실상)에 대해 객체에 요청합니다. "잠시만 기다려주세요. 어떤 요청인지 모르겠습니다. 나는 다음 응답이 어떻게 나오는지 보았을 때까지 보내고 싶다. "