GRPC流被取消

我正试图通过GRPC创建流式API,但由于一些奇怪的原因,我的streamObserver被取消了。 这是我的 .proto 声明:

service Service {
    rpc connect (ConnectionRequest) returns (stream StreamResponse) {}

    rpc act (stream ActRequest) returns (ActResponse) {}
}

其理念是用户将 connect 所以 StreamResponse 将为每个用户保存,然后在每个 act 这个 流响应 将接收更新。下面是Java实现的 Service

class ServiceImpl extends ServiceGrpc.ServiceImplBase {
    ............
    @Override
    public void connect(ConnectionRequest request, StreamObserver<StreamResponse> responseObserver) {
        observers.add(responseObserver);
    }

    @Override
    public StreamObserver<ActRequest> act(StreamObserver<ActResponse> responseObserver) {
        return return new StreamObserver<ActRequest>() {
            @Override
            public void onNext(ActRequest actRequest) {
                StreamResponse streamResponse = StreamResponse.newBuilder().build();
                observers.forEach(o -> o.onNext(streamResponse));
                actRequest..onCompleted();
            }
            ...............
        };
    }
    ..................
}

这是客户 连接 方法:

ServiceGrpc.ServiceStub stub = ServiceGrpc.newStub(channel);
new StreamObserver<PlayerResponse>() {
    ........
    @Override
    public void onNext(StreamResponse streamResponse) {
          logger.info("Act: " streamResponse.getData());
    }
    ........
}
ConnectionRequest request = ConnectionRequest.newBuilder().build();
stub.connect(request, streamObserver);

和客户 行为 方法:

StreamObserver<ActResponse> observer = new StreamObserver<ActResponse>() {
    @Override
    public void onNext(ActResponse actResponse) {
        logger.info("Status: " + actResponse.getSuccess());
    }
    .........
};
StreamObserver<ActRequest> act = stub.act(observer);
act.onNext(MoveRequest.newBuilder().build());
act.onCompleted();

当我同时启动客户机和服务器时,客户机可以调用 连接 . 但是方法 行为 只能第一次调用。当客户端调用 行为 我第二次收到以下异常:

io.grpc.StatusRuntimeException: CANCELLED: call already cancelled
    at io.grpc.Status.asRuntimeException(Status.java:517)
    at io.grpc.stub.ServerCalls$ServerCallStreamObserverImpl.onNext(ServerCalls.java:335)

在调试模式下,我可以看到 StreamObserver<StreamResponse> responseObserver 是 canceled=true 。