@Test @SuppressWarnings({"unchecked"}) public void clientDeadlineLogged_deadlineSetViaCallOption() { MethodDescriptor<byte[], byte[]> method = MethodDescriptor.<byte[], byte[]>newBuilder() .setType(MethodType.UNKNOWN) .setFullMethodName("service/method") .setRequestMarshaller(BYTEARRAY_MARSHALLER) .setResponseMarshaller(BYTEARRAY_MARSHALLER) .build(); ClientCall.Listener<byte[]> mockListener = mock(ClientCall.Listener.class); ClientCall<byte[], byte[]> call = new BinlogHelper(mockSinkWriter) .getClientInterceptor(CALL_ID) .interceptCall( method, CallOptions.DEFAULT.withDeadlineAfter(1, TimeUnit.SECONDS), new Channel() { @Override public <RequestT, ResponseT> ClientCall<RequestT, ResponseT> newCall( MethodDescriptor<RequestT, ResponseT> methodDescriptor, CallOptions callOptions) { return new NoopClientCall<RequestT, ResponseT>(); } @Override public String authority() { return null; } }); call.start(mockListener, new Metadata()); ArgumentCaptor<Duration> callOptTimeoutCaptor = ArgumentCaptor.forClass(Duration.class); verify(mockSinkWriter).logClientHeader( any(Integer.class), any(String.class), any(String.class), callOptTimeoutCaptor.capture(), any(Metadata.class), any(GrpcLogEntry.Logger.class), any(Long.class), any(SocketAddress.class)); Duration timeout = callOptTimeoutCaptor.getValue(); assertThat(TimeUnit.SECONDS.toNanos(1) - Durations.toNanos(timeout)) .isAtMost(TimeUnit.MILLISECONDS.toNanos(250)); }
public ClientInterceptor getClientInterceptor(final long callId) { return new ClientInterceptor() { boolean trailersOnlyResponse = true; @Override public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall( final MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) { final AtomicLong seq = new AtomicLong(1); final String methodName = method.getFullMethodName(); final String authority = next.authority(); final Deadline deadline = min(callOptions.getDeadline(), Context.current().getDeadline()); return new SimpleForwardingClientCall<ReqT, RespT>(next.newCall(method, callOptions)) { @Override public void start(final Listener<RespT> responseListener, Metadata headers) { final Duration timeout = deadline == null ? null : Durations.fromNanos(deadline.timeRemaining(TimeUnit.NANOSECONDS)); writer.logClientHeader( seq.getAndIncrement(), methodName, authority, timeout, headers, GrpcLogEntry.Logger.LOGGER_CLIENT, callId, null); ClientCall.Listener<RespT> wListener = new SimpleForwardingClientCallListener<RespT>(responseListener) { @Override public void onMessage(RespT message) { writer.logRpcMessage( seq.getAndIncrement(), EventType.EVENT_TYPE_SERVER_MESSAGE, method.getResponseMarshaller(), message, GrpcLogEntry.Logger.LOGGER_CLIENT, callId); super.onMessage(message); } @Override public void onHeaders(Metadata headers) { trailersOnlyResponse = false; writer.logServerHeader( seq.getAndIncrement(), headers, GrpcLogEntry.Logger.LOGGER_CLIENT, callId, getPeerSocket(getAttributes())); super.onHeaders(headers); } @Override public void onClose(Status status, Metadata trailers) { SocketAddress peer = trailersOnlyResponse ? getPeerSocket(getAttributes()) : null; writer.logTrailer( seq.getAndIncrement(), status, trailers, GrpcLogEntry.Logger.LOGGER_CLIENT, callId, peer); super.onClose(status, trailers); } }; super.start(wListener, headers); } @Override public void sendMessage(ReqT message) { writer.logRpcMessage( seq.getAndIncrement(), EventType.EVENT_TYPE_CLIENT_MESSAGE, method.getRequestMarshaller(), message, GrpcLogEntry.Logger.LOGGER_CLIENT, callId); super.sendMessage(message); } @Override public void halfClose() { writer.logHalfClose( seq.getAndIncrement(), GrpcLogEntry.Logger.LOGGER_CLIENT, callId); super.halfClose(); } @Override public void cancel(String message, Throwable cause) { writer.logCancel( seq.getAndIncrement(), GrpcLogEntry.Logger.LOGGER_CLIENT, callId); super.cancel(message, cause); } }; } }; }
BinlogHelper { public ClientInterceptor getClientInterceptor(final long callId) { return new ClientInterceptor() { boolean trailersOnlyResponse = true; @Override public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall( final MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) { final AtomicLong seq = new AtomicLong(1); final String methodName = method.getFullMethodName(); final String authority = next.authority(); final Deadline deadline = min(callOptions.getDeadline(), Context.current().getDeadline()); return new SimpleForwardingClientCall<ReqT, RespT>(next.newCall(method, callOptions)) { @Override public void start(final Listener<RespT> responseListener, Metadata headers) { final Duration timeout = deadline == null ? null : Durations.fromNanos(deadline.timeRemaining(TimeUnit.NANOSECONDS)); writer.logClientHeader( seq.getAndIncrement(), methodName, authority, timeout, headers, GrpcLogEntry.Logger.LOGGER_CLIENT, callId, null); ClientCall.Listener<RespT> wListener = new SimpleForwardingClientCallListener<RespT>(responseListener) { @Override public void onMessage(RespT message) { writer.logRpcMessage( seq.getAndIncrement(), EventType.EVENT_TYPE_SERVER_MESSAGE, method.getResponseMarshaller(), message, GrpcLogEntry.Logger.LOGGER_CLIENT, callId); super.onMessage(message); } @Override public void onHeaders(Metadata headers) { trailersOnlyResponse = false; writer.logServerHeader( seq.getAndIncrement(), headers, GrpcLogEntry.Logger.LOGGER_CLIENT, callId, getPeerSocket(getAttributes())); super.onHeaders(headers); } @Override public void onClose(Status status, Metadata trailers) { SocketAddress peer = trailersOnlyResponse ? getPeerSocket(getAttributes()) : null; writer.logTrailer( seq.getAndIncrement(), status, trailers, GrpcLogEntry.Logger.LOGGER_CLIENT, callId, peer); super.onClose(status, trailers); } }; super.start(wListener, headers); } @Override public void sendMessage(ReqT message) { writer.logRpcMessage( seq.getAndIncrement(), EventType.EVENT_TYPE_CLIENT_MESSAGE, method.getRequestMarshaller(), message, GrpcLogEntry.Logger.LOGGER_CLIENT, callId); super.sendMessage(message); } @Override public void halfClose() { writer.logHalfClose( seq.getAndIncrement(), GrpcLogEntry.Logger.LOGGER_CLIENT, callId); super.halfClose(); } @Override public void cancel(String message, Throwable cause) { writer.logCancel( seq.getAndIncrement(), GrpcLogEntry.Logger.LOGGER_CLIENT, callId); super.cancel(message, cause); } }; } }; } }
BinlogHelper { public ClientInterceptor getClientInterceptor(final long callId) { return new ClientInterceptor() { boolean trailersOnlyResponse = true; @Override public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall( final MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) { final AtomicLong seq = new AtomicLong(1); final String methodName = method.getFullMethodName(); final String authority = next.authority(); final Deadline deadline = min(callOptions.getDeadline(), Context.current().getDeadline()); return new SimpleForwardingClientCall<ReqT, RespT>(next.newCall(method, callOptions)) { @Override public void start(final Listener<RespT> responseListener, Metadata headers) { final Duration timeout = deadline == null ? null : Durations.fromNanos(deadline.timeRemaining(TimeUnit.NANOSECONDS)); writer.logClientHeader( seq.getAndIncrement(), methodName, authority, timeout, headers, GrpcLogEntry.Logger.LOGGER_CLIENT, callId, null); ClientCall.Listener<RespT> wListener = new SimpleForwardingClientCallListener<RespT>(responseListener) { @Override public void onMessage(RespT message) { writer.logRpcMessage( seq.getAndIncrement(), EventType.EVENT_TYPE_SERVER_MESSAGE, method.getResponseMarshaller(), message, GrpcLogEntry.Logger.LOGGER_CLIENT, callId); super.onMessage(message); } @Override public void onHeaders(Metadata headers) { trailersOnlyResponse = false; writer.logServerHeader( seq.getAndIncrement(), headers, GrpcLogEntry.Logger.LOGGER_CLIENT, callId, getPeerSocket(getAttributes())); super.onHeaders(headers); } @Override public void onClose(Status status, Metadata trailers) { SocketAddress peer = trailersOnlyResponse ? getPeerSocket(getAttributes()) : null; writer.logTrailer( seq.getAndIncrement(), status, trailers, GrpcLogEntry.Logger.LOGGER_CLIENT, callId, peer); super.onClose(status, trailers); } }; super.start(wListener, headers); } @Override public void sendMessage(ReqT message) { writer.logRpcMessage( seq.getAndIncrement(), EventType.EVENT_TYPE_CLIENT_MESSAGE, method.getRequestMarshaller(), message, GrpcLogEntry.Logger.LOGGER_CLIENT, callId); super.sendMessage(message); } @Override public void halfClose() { writer.logHalfClose( seq.getAndIncrement(), GrpcLogEntry.Logger.LOGGER_CLIENT, callId); super.halfClose(); } @Override public void cancel(String message, Throwable cause) { writer.logCancel( seq.getAndIncrement(), GrpcLogEntry.Logger.LOGGER_CLIENT, callId); super.cancel(message, cause); } }; } }; } @VisibleForTesting  BinlogHelper(SinkWriter writer); }
BinlogHelper { public ClientInterceptor getClientInterceptor(final long callId) { return new ClientInterceptor() { boolean trailersOnlyResponse = true; @Override public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall( final MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) { final AtomicLong seq = new AtomicLong(1); final String methodName = method.getFullMethodName(); final String authority = next.authority(); final Deadline deadline = min(callOptions.getDeadline(), Context.current().getDeadline()); return new SimpleForwardingClientCall<ReqT, RespT>(next.newCall(method, callOptions)) { @Override public void start(final Listener<RespT> responseListener, Metadata headers) { final Duration timeout = deadline == null ? null : Durations.fromNanos(deadline.timeRemaining(TimeUnit.NANOSECONDS)); writer.logClientHeader( seq.getAndIncrement(), methodName, authority, timeout, headers, GrpcLogEntry.Logger.LOGGER_CLIENT, callId, null); ClientCall.Listener<RespT> wListener = new SimpleForwardingClientCallListener<RespT>(responseListener) { @Override public void onMessage(RespT message) { writer.logRpcMessage( seq.getAndIncrement(), EventType.EVENT_TYPE_SERVER_MESSAGE, method.getResponseMarshaller(), message, GrpcLogEntry.Logger.LOGGER_CLIENT, callId); super.onMessage(message); } @Override public void onHeaders(Metadata headers) { trailersOnlyResponse = false; writer.logServerHeader( seq.getAndIncrement(), headers, GrpcLogEntry.Logger.LOGGER_CLIENT, callId, getPeerSocket(getAttributes())); super.onHeaders(headers); } @Override public void onClose(Status status, Metadata trailers) { SocketAddress peer = trailersOnlyResponse ? getPeerSocket(getAttributes()) : null; writer.logTrailer( seq.getAndIncrement(), status, trailers, GrpcLogEntry.Logger.LOGGER_CLIENT, callId, peer); super.onClose(status, trailers); } }; super.start(wListener, headers); } @Override public void sendMessage(ReqT message) { writer.logRpcMessage( seq.getAndIncrement(), EventType.EVENT_TYPE_CLIENT_MESSAGE, method.getRequestMarshaller(), message, GrpcLogEntry.Logger.LOGGER_CLIENT, callId); super.sendMessage(message); } @Override public void halfClose() { writer.logHalfClose( seq.getAndIncrement(), GrpcLogEntry.Logger.LOGGER_CLIENT, callId); super.halfClose(); } @Override public void cancel(String message, Throwable cause) { writer.logCancel( seq.getAndIncrement(), GrpcLogEntry.Logger.LOGGER_CLIENT, callId); super.cancel(message, cause); } }; } }; } @VisibleForTesting  BinlogHelper(SinkWriter writer); ClientInterceptor getClientInterceptor(final long callId); ServerInterceptor getServerInterceptor(final long callId); }
BinlogHelper { public ClientInterceptor getClientInterceptor(final long callId) { return new ClientInterceptor() { boolean trailersOnlyResponse = true; @Override public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall( final MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) { final AtomicLong seq = new AtomicLong(1); final String methodName = method.getFullMethodName(); final String authority = next.authority(); final Deadline deadline = min(callOptions.getDeadline(), Context.current().getDeadline()); return new SimpleForwardingClientCall<ReqT, RespT>(next.newCall(method, callOptions)) { @Override public void start(final Listener<RespT> responseListener, Metadata headers) { final Duration timeout = deadline == null ? null : Durations.fromNanos(deadline.timeRemaining(TimeUnit.NANOSECONDS)); writer.logClientHeader( seq.getAndIncrement(), methodName, authority, timeout, headers, GrpcLogEntry.Logger.LOGGER_CLIENT, callId, null); ClientCall.Listener<RespT> wListener = new SimpleForwardingClientCallListener<RespT>(responseListener) { @Override public void onMessage(RespT message) { writer.logRpcMessage( seq.getAndIncrement(), EventType.EVENT_TYPE_SERVER_MESSAGE, method.getResponseMarshaller(), message, GrpcLogEntry.Logger.LOGGER_CLIENT, callId); super.onMessage(message); } @Override public void onHeaders(Metadata headers) { trailersOnlyResponse = false; writer.logServerHeader( seq.getAndIncrement(), headers, GrpcLogEntry.Logger.LOGGER_CLIENT, callId, getPeerSocket(getAttributes())); super.onHeaders(headers); } @Override public void onClose(Status status, Metadata trailers) { SocketAddress peer = trailersOnlyResponse ? getPeerSocket(getAttributes()) : null; writer.logTrailer( seq.getAndIncrement(), status, trailers, GrpcLogEntry.Logger.LOGGER_CLIENT, callId, peer); super.onClose(status, trailers); } }; super.start(wListener, headers); } @Override public void sendMessage(ReqT message) { writer.logRpcMessage( seq.getAndIncrement(), EventType.EVENT_TYPE_CLIENT_MESSAGE, method.getRequestMarshaller(), message, GrpcLogEntry.Logger.LOGGER_CLIENT, callId); super.sendMessage(message); } @Override public void halfClose() { writer.logHalfClose( seq.getAndIncrement(), GrpcLogEntry.Logger.LOGGER_CLIENT, callId); super.halfClose(); } @Override public void cancel(String message, Throwable cause) { writer.logCancel( seq.getAndIncrement(), GrpcLogEntry.Logger.LOGGER_CLIENT, callId); super.cancel(message, cause); } }; } }; } @VisibleForTesting  BinlogHelper(SinkWriter writer); ClientInterceptor getClientInterceptor(final long callId); ServerInterceptor getServerInterceptor(final long callId);  }