Асинхронная цепочка вызовов GRPC - PullRequest
Новая
       5

Асинхронная цепочка вызовов GRPC

1 голос
/ 18 марта 2019

Существует ли простой способ асинхронной цепочки вызовов GRPC в Python?

Это похоже на то, что "должно" быть осуществимо, но я не могу этого найти.

Вот приблизительное представление о том, что я должен делать: 

class MyServer(my_grpc.MyServicer):
  def __init__(self, child_stub):
    self.child_stub_ = child_stub

  def MyMethod(self, request, context):
    child_result = self.child_stub_.ChildMethod.future(my_grpc.ChildMethodParams())

    child_result.add_done_callback(something_that_completes_MyMethod)

    return presumably_something

Есть что-то, чего я здесь не хватает? Такое ощущение, что это будет распространенный вариант использования, но я не могу найти ничего связанного с этим в документации.

1 Ответ

0 голосов
/ 21 марта 2019

Редактировать: Я полагаю, что вы пытаетесь отправить два ответа на один запрос на настройку унарного запроса / ответа, что я не считаю возможным.Вместо этого вы должны сделать унарный запрос и потоковые ответы, что позволит получить множество ответов.

client 

import grpc
import test_pb2_grpc as pb_grpc
import test_pb2 as pb2

def test():
    channel = grpc.insecure_channel('localhost:50051')
    stub = pb_grpc.TestStub(channel=channel)

    for response in stub.Produce(pb2.Empty()):
        print(response.result)

if __name__ == '__main__':
    test()

Сервер 

import test_pb2_grpc as pb_grpc
import test_pb2 as pb2
import time
import grpc
from concurrent import futures


class test_servcie(pb_grpc.TestServicer):
    def Produce(self, request, context):
        my_method_results = [50, 200]
        for result in my_method_results:
            yield pb2.Resp(result=result)


def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    pb_grpc.add_TestServicer_to_server(test_servcie(), server)
    server.add_insecure_port('[::]:50051')
    print("service started")
    server.start()
    try:
        while True:
            time.sleep(3600)
    except KeyboardInterrupt:
        server.stop(0)


if __name__ == '__main__':
    serve()

proto 

syntax = "proto3";

package api;


service Test {
    rpc Produce (Empty) returns (stream Resp);
}

message Empty {}


message Resp{
    int32 result = 1;
}
...