@Test @SuppressWarnings({"rawtypes", "unchecked"}) public void serverDeadlineLogged() { final AtomicReference<ServerCall> interceptedCall = new AtomicReference<ServerCall>(); final ServerCall.Listener mockListener = mock(ServerCall.Listener.class); final MethodDescriptor<byte[], byte[]> method = MethodDescriptor.<byte[], byte[]>newBuilder() .setType(MethodType.UNKNOWN) .setFullMethodName("service/method") .setRequestMarshaller(BYTEARRAY_MARSHALLER) .setResponseMarshaller(BYTEARRAY_MARSHALLER) .build(); Context.current() .withDeadlineAfter(1, TimeUnit.SECONDS, Executors.newSingleThreadScheduledExecutor()) .run(new Runnable() { @Override public void run() { ServerCall.Listener<byte[]> unused = new BinlogHelper(mockSinkWriter) .getServerInterceptor(CALL_ID) .interceptCall( new NoopServerCall<byte[], byte[]>() { @Override public MethodDescriptor<byte[], byte[]> getMethodDescriptor() { return method; } }, new Metadata(), new ServerCallHandler<byte[], byte[]>() { @Override public ServerCall.Listener<byte[]> startCall( ServerCall<byte[], byte[]> call, Metadata headers) { interceptedCall.set(call); return mockListener; } }); } }); ArgumentCaptor<Duration> timeoutCaptor = ArgumentCaptor.forClass(Duration.class); verify(mockSinkWriter).logClientHeader( eq(1L), eq("service/method"), isNull(String.class), timeoutCaptor.capture(), any(Metadata.class), eq(Logger.LOGGER_SERVER), eq(CALL_ID), isNull(SocketAddress.class)); verifyNoMoreInteractions(mockSinkWriter); Duration timeout = timeoutCaptor.getValue(); assertThat(TimeUnit.SECONDS.toNanos(1) - Durations.toNanos(timeout)) .isAtMost(TimeUnit.MILLISECONDS.toNanos(250)); }
public ServerInterceptor getServerInterceptor(final long callId) { return new ServerInterceptor() { @Override public <ReqT, RespT> Listener<ReqT> interceptCall( final ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) { final AtomicLong seq = new AtomicLong(1); SocketAddress peer = getPeerSocket(call.getAttributes()); String methodName = call.getMethodDescriptor().getFullMethodName(); String authority = call.getAuthority(); Deadline deadline = Context.current().getDeadline(); final Duration timeout = deadline == null ? null : Durations.fromNanos(deadline.timeRemaining(TimeUnit.NANOSECONDS)); writer.logClientHeader( seq.getAndIncrement(), methodName, authority, timeout, headers, GrpcLogEntry.Logger.LOGGER_SERVER, callId, peer); ServerCall<ReqT, RespT> wCall = new SimpleForwardingServerCall<ReqT, RespT>(call) { @Override public void sendMessage(RespT message) { writer.logRpcMessage( seq.getAndIncrement(), EventType.EVENT_TYPE_SERVER_MESSAGE, call.getMethodDescriptor().getResponseMarshaller(), message, GrpcLogEntry.Logger.LOGGER_SERVER, callId); super.sendMessage(message); } @Override public void sendHeaders(Metadata headers) { writer.logServerHeader( seq.getAndIncrement(), headers, GrpcLogEntry.Logger.LOGGER_SERVER, callId, null); super.sendHeaders(headers); } @Override public void close(Status status, Metadata trailers) { writer.logTrailer( seq.getAndIncrement(), status, trailers, GrpcLogEntry.Logger.LOGGER_SERVER, callId, null); super.close(status, trailers); } }; return new SimpleForwardingServerCallListener<ReqT>(next.startCall(wCall, headers)) { @Override public void onMessage(ReqT message) { writer.logRpcMessage( seq.getAndIncrement(), EventType.EVENT_TYPE_CLIENT_MESSAGE, call.getMethodDescriptor().getRequestMarshaller(), message, GrpcLogEntry.Logger.LOGGER_SERVER, callId); super.onMessage(message); } @Override public void onHalfClose() { writer.logHalfClose( seq.getAndIncrement(), GrpcLogEntry.Logger.LOGGER_SERVER, callId); super.onHalfClose(); } @Override public void onCancel() { writer.logCancel( seq.getAndIncrement(), GrpcLogEntry.Logger.LOGGER_SERVER, callId); super.onCancel(); } }; } }; }
BinlogHelper { public ServerInterceptor getServerInterceptor(final long callId) { return new ServerInterceptor() { @Override public <ReqT, RespT> Listener<ReqT> interceptCall( final ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) { final AtomicLong seq = new AtomicLong(1); SocketAddress peer = getPeerSocket(call.getAttributes()); String methodName = call.getMethodDescriptor().getFullMethodName(); String authority = call.getAuthority(); Deadline deadline = Context.current().getDeadline(); final Duration timeout = deadline == null ? null : Durations.fromNanos(deadline.timeRemaining(TimeUnit.NANOSECONDS)); writer.logClientHeader( seq.getAndIncrement(), methodName, authority, timeout, headers, GrpcLogEntry.Logger.LOGGER_SERVER, callId, peer); ServerCall<ReqT, RespT> wCall = new SimpleForwardingServerCall<ReqT, RespT>(call) { @Override public void sendMessage(RespT message) { writer.logRpcMessage( seq.getAndIncrement(), EventType.EVENT_TYPE_SERVER_MESSAGE, call.getMethodDescriptor().getResponseMarshaller(), message, GrpcLogEntry.Logger.LOGGER_SERVER, callId); super.sendMessage(message); } @Override public void sendHeaders(Metadata headers) { writer.logServerHeader( seq.getAndIncrement(), headers, GrpcLogEntry.Logger.LOGGER_SERVER, callId, null); super.sendHeaders(headers); } @Override public void close(Status status, Metadata trailers) { writer.logTrailer( seq.getAndIncrement(), status, trailers, GrpcLogEntry.Logger.LOGGER_SERVER, callId, null); super.close(status, trailers); } }; return new SimpleForwardingServerCallListener<ReqT>(next.startCall(wCall, headers)) { @Override public void onMessage(ReqT message) { writer.logRpcMessage( seq.getAndIncrement(), EventType.EVENT_TYPE_CLIENT_MESSAGE, call.getMethodDescriptor().getRequestMarshaller(), message, GrpcLogEntry.Logger.LOGGER_SERVER, callId); super.onMessage(message); } @Override public void onHalfClose() { writer.logHalfClose( seq.getAndIncrement(), GrpcLogEntry.Logger.LOGGER_SERVER, callId); super.onHalfClose(); } @Override public void onCancel() { writer.logCancel( seq.getAndIncrement(), GrpcLogEntry.Logger.LOGGER_SERVER, callId); super.onCancel(); } }; } }; } }
BinlogHelper { public ServerInterceptor getServerInterceptor(final long callId) { return new ServerInterceptor() { @Override public <ReqT, RespT> Listener<ReqT> interceptCall( final ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) { final AtomicLong seq = new AtomicLong(1); SocketAddress peer = getPeerSocket(call.getAttributes()); String methodName = call.getMethodDescriptor().getFullMethodName(); String authority = call.getAuthority(); Deadline deadline = Context.current().getDeadline(); final Duration timeout = deadline == null ? null : Durations.fromNanos(deadline.timeRemaining(TimeUnit.NANOSECONDS)); writer.logClientHeader( seq.getAndIncrement(), methodName, authority, timeout, headers, GrpcLogEntry.Logger.LOGGER_SERVER, callId, peer); ServerCall<ReqT, RespT> wCall = new SimpleForwardingServerCall<ReqT, RespT>(call) { @Override public void sendMessage(RespT message) { writer.logRpcMessage( seq.getAndIncrement(), EventType.EVENT_TYPE_SERVER_MESSAGE, call.getMethodDescriptor().getResponseMarshaller(), message, GrpcLogEntry.Logger.LOGGER_SERVER, callId); super.sendMessage(message); } @Override public void sendHeaders(Metadata headers) { writer.logServerHeader( seq.getAndIncrement(), headers, GrpcLogEntry.Logger.LOGGER_SERVER, callId, null); super.sendHeaders(headers); } @Override public void close(Status status, Metadata trailers) { writer.logTrailer( seq.getAndIncrement(), status, trailers, GrpcLogEntry.Logger.LOGGER_SERVER, callId, null); super.close(status, trailers); } }; return new SimpleForwardingServerCallListener<ReqT>(next.startCall(wCall, headers)) { @Override public void onMessage(ReqT message) { writer.logRpcMessage( seq.getAndIncrement(), EventType.EVENT_TYPE_CLIENT_MESSAGE, call.getMethodDescriptor().getRequestMarshaller(), message, GrpcLogEntry.Logger.LOGGER_SERVER, callId); super.onMessage(message); } @Override public void onHalfClose() { writer.logHalfClose( seq.getAndIncrement(), GrpcLogEntry.Logger.LOGGER_SERVER, callId); super.onHalfClose(); } @Override public void onCancel() { writer.logCancel( seq.getAndIncrement(), GrpcLogEntry.Logger.LOGGER_SERVER, callId); super.onCancel(); } }; } }; } @VisibleForTesting  BinlogHelper(SinkWriter writer); }
BinlogHelper { public ServerInterceptor getServerInterceptor(final long callId) { return new ServerInterceptor() { @Override public <ReqT, RespT> Listener<ReqT> interceptCall( final ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) { final AtomicLong seq = new AtomicLong(1); SocketAddress peer = getPeerSocket(call.getAttributes()); String methodName = call.getMethodDescriptor().getFullMethodName(); String authority = call.getAuthority(); Deadline deadline = Context.current().getDeadline(); final Duration timeout = deadline == null ? null : Durations.fromNanos(deadline.timeRemaining(TimeUnit.NANOSECONDS)); writer.logClientHeader( seq.getAndIncrement(), methodName, authority, timeout, headers, GrpcLogEntry.Logger.LOGGER_SERVER, callId, peer); ServerCall<ReqT, RespT> wCall = new SimpleForwardingServerCall<ReqT, RespT>(call) { @Override public void sendMessage(RespT message) { writer.logRpcMessage( seq.getAndIncrement(), EventType.EVENT_TYPE_SERVER_MESSAGE, call.getMethodDescriptor().getResponseMarshaller(), message, GrpcLogEntry.Logger.LOGGER_SERVER, callId); super.sendMessage(message); } @Override public void sendHeaders(Metadata headers) { writer.logServerHeader( seq.getAndIncrement(), headers, GrpcLogEntry.Logger.LOGGER_SERVER, callId, null); super.sendHeaders(headers); } @Override public void close(Status status, Metadata trailers) { writer.logTrailer( seq.getAndIncrement(), status, trailers, GrpcLogEntry.Logger.LOGGER_SERVER, callId, null); super.close(status, trailers); } }; return new SimpleForwardingServerCallListener<ReqT>(next.startCall(wCall, headers)) { @Override public void onMessage(ReqT message) { writer.logRpcMessage( seq.getAndIncrement(), EventType.EVENT_TYPE_CLIENT_MESSAGE, call.getMethodDescriptor().getRequestMarshaller(), message, GrpcLogEntry.Logger.LOGGER_SERVER, callId); super.onMessage(message); } @Override public void onHalfClose() { writer.logHalfClose( seq.getAndIncrement(), GrpcLogEntry.Logger.LOGGER_SERVER, callId); super.onHalfClose(); } @Override public void onCancel() { writer.logCancel( seq.getAndIncrement(), GrpcLogEntry.Logger.LOGGER_SERVER, callId); super.onCancel(); } }; } }; } @VisibleForTesting  BinlogHelper(SinkWriter writer); ClientInterceptor getClientInterceptor(final long callId); ServerInterceptor getServerInterceptor(final long callId); }
BinlogHelper { public ServerInterceptor getServerInterceptor(final long callId) { return new ServerInterceptor() { @Override public <ReqT, RespT> Listener<ReqT> interceptCall( final ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) { final AtomicLong seq = new AtomicLong(1); SocketAddress peer = getPeerSocket(call.getAttributes()); String methodName = call.getMethodDescriptor().getFullMethodName(); String authority = call.getAuthority(); Deadline deadline = Context.current().getDeadline(); final Duration timeout = deadline == null ? null : Durations.fromNanos(deadline.timeRemaining(TimeUnit.NANOSECONDS)); writer.logClientHeader( seq.getAndIncrement(), methodName, authority, timeout, headers, GrpcLogEntry.Logger.LOGGER_SERVER, callId, peer); ServerCall<ReqT, RespT> wCall = new SimpleForwardingServerCall<ReqT, RespT>(call) { @Override public void sendMessage(RespT message) { writer.logRpcMessage( seq.getAndIncrement(), EventType.EVENT_TYPE_SERVER_MESSAGE, call.getMethodDescriptor().getResponseMarshaller(), message, GrpcLogEntry.Logger.LOGGER_SERVER, callId); super.sendMessage(message); } @Override public void sendHeaders(Metadata headers) { writer.logServerHeader( seq.getAndIncrement(), headers, GrpcLogEntry.Logger.LOGGER_SERVER, callId, null); super.sendHeaders(headers); } @Override public void close(Status status, Metadata trailers) { writer.logTrailer( seq.getAndIncrement(), status, trailers, GrpcLogEntry.Logger.LOGGER_SERVER, callId, null); super.close(status, trailers); } }; return new SimpleForwardingServerCallListener<ReqT>(next.startCall(wCall, headers)) { @Override public void onMessage(ReqT message) { writer.logRpcMessage( seq.getAndIncrement(), EventType.EVENT_TYPE_CLIENT_MESSAGE, call.getMethodDescriptor().getRequestMarshaller(), message, GrpcLogEntry.Logger.LOGGER_SERVER, callId); super.onMessage(message); } @Override public void onHalfClose() { writer.logHalfClose( seq.getAndIncrement(), GrpcLogEntry.Logger.LOGGER_SERVER, callId); super.onHalfClose(); } @Override public void onCancel() { writer.logCancel( seq.getAndIncrement(), GrpcLogEntry.Logger.LOGGER_SERVER, callId); super.onCancel(); } }; } }; } @VisibleForTesting  BinlogHelper(SinkWriter writer); ClientInterceptor getClientInterceptor(final long callId); ServerInterceptor getServerInterceptor(final long callId);  }