GrpcServerInterceptor.java 文件源码

/**
 * Intercept all GRPC calls
 * @param serverCall
 * @param metadata
 * @param serverCallHandler
 * @param <ReqT>
 * @param <RespT>
 * @return
 */
@Override
public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(final ServerCall<ReqT, RespT> serverCall, Metadata metadata, ServerCallHandler<ReqT, RespT> serverCallHandler) {

    Timer.Context timer = metricRegistry.timer(metricName(M_REQ_TIME, serverCall.getMethodDescriptor().getFullMethodName().replace("/", "."))).time();
    Histogram histogram = metricRegistry.histogram(metricName(M_RESPONSE_SIZE, serverCall.getMethodDescriptor().getFullMethodName().replace("/", ".")));

    SimpleForwardingServerCall<ReqT, RespT> nextCall = new SimpleForwardingServerCall<ReqT, RespT>(serverCall) {
        @Override
        public void close(Status status, Metadata trailers) {
            Meter errorMeter = metricRegistry.meter(metricName(ERROR_METRIC, getMethodDescriptor().getFullMethodName().replace("/", ".")));
            if (!status.isOk()) {
                errorMeter.mark();
                log.error("An error occured with {}", serverCall.getMethodDescriptor());
            }

            timer.stop();

            super.close(status, trailers);
        }

        @Override
        public void sendMessage(RespT message) {
            super.sendMessage(message);

            if (message instanceof MessageLite) {
                histogram.update(((MessageLite) message).getSerializedSize());
                log.info("Message sent size = {}", ((MessageLite) message).getSerializedSize());
            }
        }

    };
    return serverCallHandler.startCall(nextCall, metadata);
}