@Test @SuppressWarnings({"rawtypes", "unchecked"}) public void serverInterceptor() throws Exception { final AtomicReference<ServerCall> interceptedCall = new AtomicReference<ServerCall>(); ServerCall.Listener<byte[]> capturedListener; final ServerCall.Listener mockListener = mock(ServerCall.Listener.class); final AtomicReference<Metadata> actualServerInitial = new AtomicReference<Metadata>(); final AtomicReference<byte[]> actualResponse = new AtomicReference<byte[]>(); final AtomicReference<Status> actualStatus = new AtomicReference<Status>(); final AtomicReference<Metadata> actualTrailers = new AtomicReference<Metadata>(); { Metadata clientInitial = new Metadata(); final MethodDescriptor<byte[], byte[]> method = MethodDescriptor.<byte[], byte[]>newBuilder() .setType(MethodType.UNKNOWN) .setFullMethodName("service/method") .setRequestMarshaller(BYTEARRAY_MARSHALLER) .setResponseMarshaller(BYTEARRAY_MARSHALLER) .build(); capturedListener = new BinlogHelper(mockSinkWriter) .getServerInterceptor(CALL_ID) .interceptCall( new NoopServerCall<byte[], byte[]>() { @Override public void sendHeaders(Metadata headers) { actualServerInitial.set(headers); } @Override public void sendMessage(byte[] message) { actualResponse.set(message); } @Override public void close(Status status, Metadata trailers) { actualStatus.set(status); actualTrailers.set(trailers); } @Override public MethodDescriptor<byte[], byte[]> getMethodDescriptor() { return method; } @Override public Attributes getAttributes() { return Attributes .newBuilder() .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, peer) .build(); } @Override public String getAuthority() { return "the-authority"; } }, clientInitial, new ServerCallHandler<byte[], byte[]>() { @Override public ServerCall.Listener<byte[]> startCall( ServerCall<byte[], byte[]> call, Metadata headers) { interceptedCall.set(call); return mockListener; } }); verify(mockSinkWriter).logClientHeader( eq(1L), eq("service/method"), eq("the-authority"), isNull(Duration.class), same(clientInitial), eq(Logger.LOGGER_SERVER), eq(CALL_ID), same(peer)); verifyNoMoreInteractions(mockSinkWriter); } { Metadata serverInital = new Metadata(); interceptedCall.get().sendHeaders(serverInital); verify(mockSinkWriter).logServerHeader( eq(2L), same(serverInital), eq(Logger.LOGGER_SERVER), eq(CALL_ID), isNull(SocketAddress.class)); verifyNoMoreInteractions(mockSinkWriter); assertSame(serverInital, actualServerInitial.get()); } { byte[] request = "this is a request".getBytes(US_ASCII); capturedListener.onMessage(request); verify(mockSinkWriter).logRpcMessage( eq(3L), eq(EventType.EVENT_TYPE_CLIENT_MESSAGE), same(BYTEARRAY_MARSHALLER), same(request), eq(Logger.LOGGER_SERVER), eq(CALL_ID)); verifyNoMoreInteractions(mockSinkWriter); verify(mockListener).onMessage(same(request)); } { capturedListener.onHalfClose(); verify(mockSinkWriter).logHalfClose( eq(4L), eq(Logger.LOGGER_SERVER), eq(CALL_ID)); verifyNoMoreInteractions(mockSinkWriter); verify(mockListener).onHalfClose(); } { byte[] response = "this is a response".getBytes(US_ASCII); interceptedCall.get().sendMessage(response); verify(mockSinkWriter).logRpcMessage( eq(5L), eq(EventType.EVENT_TYPE_SERVER_MESSAGE), same(BYTEARRAY_MARSHALLER), same(response), eq(Logger.LOGGER_SERVER), eq(CALL_ID)); verifyNoMoreInteractions(mockSinkWriter); assertSame(response, actualResponse.get()); } { Status status = Status.INTERNAL.withDescription("some description"); Metadata trailers = new Metadata(); interceptedCall.get().close(status, trailers); verify(mockSinkWriter).logTrailer( eq(6L), same(status), same(trailers), eq(Logger.LOGGER_SERVER), eq(CALL_ID), isNull(SocketAddress.class)); verifyNoMoreInteractions(mockSinkWriter); assertSame(status, actualStatus.get()); assertSame(trailers, actualTrailers.get()); } { capturedListener.onCancel(); verify(mockSinkWriter).logCancel( eq(7L), eq(Logger.LOGGER_SERVER), eq(CALL_ID)); verify(mockListener).onCancel(); } }
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);  }