@Test @SuppressWarnings({"rawtypes", "unchecked"}) public void clientInterceptor_trailersOnlyResponseLogsPeerAddress() throws Exception { final AtomicReference<ClientCall.Listener> interceptedListener = new AtomicReference<ClientCall.Listener>(); final AtomicReference<Metadata> actualClientInitial = new AtomicReference<Metadata>(); final AtomicReference<Object> actualRequest = new AtomicReference<Object>(); Channel channel = new Channel() { @Override public <RequestT, ResponseT> ClientCall<RequestT, ResponseT> newCall( MethodDescriptor<RequestT, ResponseT> methodDescriptor, CallOptions callOptions) { return new NoopClientCall<RequestT, ResponseT>() { @Override public void start(Listener<ResponseT> responseListener, Metadata headers) { interceptedListener.set(responseListener); actualClientInitial.set(headers); } @Override public void sendMessage(RequestT message) { actualRequest.set(message); } @Override public Attributes getAttributes() { return Attributes.newBuilder().set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, peer).build(); } }; } @Override public String authority() { return "the-authority"; } }; ClientCall.Listener<byte[]> mockListener = mock(ClientCall.Listener.class); MethodDescriptor<byte[], byte[]> method = MethodDescriptor.<byte[], byte[]>newBuilder() .setType(MethodType.UNKNOWN) .setFullMethodName("service/method") .setRequestMarshaller(BYTEARRAY_MARSHALLER) .setResponseMarshaller(BYTEARRAY_MARSHALLER) .build(); ClientCall<byte[], byte[]> interceptedCall = new BinlogHelper(mockSinkWriter) .getClientInterceptor(CALL_ID) .interceptCall( method, CallOptions.DEFAULT.withDeadlineAfter(1, TimeUnit.SECONDS), channel); Metadata clientInitial = new Metadata(); interceptedCall.start(mockListener, clientInitial); verify(mockSinkWriter).logClientHeader( eq(1L), any(String.class), any(String.class), any(Duration.class), any(Metadata.class), eq(Logger.LOGGER_CLIENT), eq(CALL_ID), isNull(SocketAddress.class)); verifyNoMoreInteractions(mockSinkWriter); { Status status = Status.INTERNAL.withDescription("some description"); Metadata trailers = new Metadata(); interceptedListener.get().onClose(status, trailers); verify(mockSinkWriter).logTrailer( eq(2L), same(status), same(trailers), eq(Logger.LOGGER_CLIENT), eq(CALL_ID), same(peer)); verifyNoMoreInteractions(mockSinkWriter); verify(mockListener).onClose(same(status), same(trailers)); } }
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);  }