@Override
    public StreamObserver<Messages.StreamingInputCallRequest> streamingInputCall(
            final StreamObserver<Messages.StreamingInputCallResponse> responseObserver) {
        return new StreamObserver<StreamingInputCallRequest>() {
            private int totalPayloadSize;

            @Override
            public void onNext(StreamingInputCallRequest message) {
                totalPayloadSize += message.getPayload().getBody().size();
            }

            @Override
            public void onCompleted() {
                responseObserver.onNext(StreamingInputCallResponse.newBuilder()
                                                                  .setAggregatedPayloadSize(
                                                                          totalPayloadSize).build());
                responseObserver.onCompleted();
            }

            @Override
            public void onError(Throwable cause) {
                responseObserver.onError(cause);
            }
        };
    }
Waits until we have received all of the request messages and then returns the aggregate payload
size for all of the received requests.
grpc/src/test/java/com/linecorp/armeria/internal/common/grpc/TestServiceImpl.java