core: install the binary logging client interceptor

core/src/main/java/io/grpc/ClientInterceptors.java

@@ -17,6 +17,8 @@
 package io.grpc;
 
 import com.google.common.base.Preconditions;
+import io.grpc.MethodDescriptor.Marshaller;
+import java.io.InputStream;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -89,6 +91,56 @@ public static Channel intercept(Channel channel, List<? extends ClientIntercepto
     return channel;
   }
 
+  /**
+   * Creates a new ClientInterceptor that transforms requests into {@code WReqT} and responses into
+   * {@code WRespT} before passing them into the {@code interceptor}.
+   */
+  static <WReqT, WRespT> ClientInterceptor wrapClientInterceptor(
+      final ClientInterceptor interceptor,
+      final Marshaller<WReqT> reqMarshaller,
+      final Marshaller<WRespT> respMarshaller) {
+    return new ClientInterceptor() {
+      @Override
+      public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
+          final MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
+        final MethodDescriptor<WReqT, WRespT> wrappedMethod =
+            method.toBuilder(reqMarshaller, respMarshaller).build();
+        final ClientCall<WReqT, WRespT> wrappedCall =
+            interceptor.interceptCall(wrappedMethod, callOptions, next);
+        return new PartialForwardingClientCall<ReqT, RespT>() {
+          @Override
+          public void start(final Listener<RespT> responseListener, Metadata headers) {
+            wrappedCall.start(new PartialForwardingClientCallListener<WRespT>() {
+              @Override
+              public void onMessage(WRespT wMessage) {
+                InputStream bytes = respMarshaller.stream(wMessage);
+                RespT message = method.getResponseMarshaller().parse(bytes);
+                responseListener.onMessage(message);
+              }
+
+              @Override
+              protected Listener<?> delegate() {
+                return responseListener;
+              }
+            }, headers);
+          }
+
+          @Override
+          public void sendMessage(ReqT message) {
+            InputStream bytes = method.getRequestMarshaller().stream(message);
+            WReqT wReq = reqMarshaller.parse(bytes);
+            wrappedCall.sendMessage(wReq);
+          }
+
+          @Override
+          protected ClientCall<?, ?> delegate() {
+            return wrappedCall;
+          }
+        };
+      }
+    };
+  }
+
   private static class InterceptorChannel extends Channel {
     private final Channel channel;
     private final ClientInterceptor interceptor;

core/src/main/java/io/grpc/ForwardingClientCall.java

@@ -16,57 +16,27 @@
 
 package io.grpc;
 
-import javax.annotation.Nullable;
-
 /**
  * A {@link ClientCall} which forwards all of it's methods to another {@link ClientCall}.
  */
-public abstract class ForwardingClientCall<ReqT, RespT> extends ClientCall<ReqT, RespT> {
+public abstract class ForwardingClientCall<ReqT, RespT>
+    extends PartialForwardingClientCall<ReqT, RespT> {
   /**
    * Returns the delegated {@code ClientCall}.
    */
+  @Override
   protected abstract ClientCall<ReqT, RespT> delegate();
 
   @Override
   public void start(Listener<RespT> responseListener, Metadata headers) {
     delegate().start(responseListener, headers);
   }
 
-  @Override
-  public void request(int numMessages) {
-    delegate().request(numMessages);
-  }
-
-  @Override
-  public void cancel(@Nullable String message, @Nullable Throwable cause) {
-    delegate().cancel(message, cause);
-  }
-
-  @Override
-  public void halfClose() {
-    delegate().halfClose();
-  }
-
   @Override
   public void sendMessage(ReqT message) {
     delegate().sendMessage(message);
   }
 
-  @Override
-  public void setMessageCompression(boolean enabled) {
-    delegate().setMessageCompression(enabled);
-  }
-
-  @Override
-  public boolean isReady() {
-    return delegate().isReady();
-  }
-
-  @Override
-  public Attributes getAttributes() {
-    return delegate().getAttributes();
-  }
-
   /**
    * A simplified version of {@link ForwardingClientCall} where subclasses can pass in a {@link
    * ClientCall} as the delegate.

core/src/main/java/io/grpc/ForwardingClientCallListener.java

@@ -20,32 +20,19 @@
  * A {@link ClientCall.Listener} which forwards all of its methods to another {@link
  * ClientCall.Listener}.
  */
-public abstract class ForwardingClientCallListener<RespT> extends ClientCall.Listener<RespT> {
+public abstract class ForwardingClientCallListener<RespT>
+    extends PartialForwardingClientCallListener<RespT> {
   /**
    * Returns the delegated {@code ClientCall.Listener}.
    */
-  protected abstract ClientCall.Listener<RespT> delegate();
-
   @Override
-  public void onHeaders(Metadata headers) {
-    delegate().onHeaders(headers);
-  }
+  protected abstract ClientCall.Listener<RespT> delegate();
 
   @Override
   public void onMessage(RespT message) {
     delegate().onMessage(message);
   }
 
-  @Override
-  public void onClose(Status status, Metadata trailers) {
-    delegate().onClose(status, trailers);
-  }
-
-  @Override
-  public void onReady() {
-    delegate().onReady();
-  }
-
   /**
    * A simplified version of {@link ForwardingClientCallListener} where subclasses can pass in a
    * {@link ClientCall.Listener} as the delegate.

core/src/main/java/io/grpc/InternalClientInterceptors.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2017, gRPC Authors All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc;
+
+import io.grpc.MethodDescriptor.Marshaller;
+
+/**
+ * Accessor to internal methods of {@link ServerInterceptors}.
+ */
+@Internal
+public class InternalClientInterceptors {
+  public static <WReqT, WRespT> ClientInterceptor wrapClientInterceptor(
+      final ClientInterceptor wrappedInterceptor,
+      final Marshaller<WReqT> reqMarshaller,
+      final Marshaller<WRespT> respMarshaller) {
+    return ClientInterceptors.wrapClientInterceptor(
+        wrappedInterceptor, reqMarshaller, respMarshaller);
+  }
+
+  private InternalClientInterceptors() {
+  }
+}

core/src/main/java/io/grpc/PartialForwardingClientCall.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright 2018, gRPC Authors All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc;
+
+import javax.annotation.Nullable;
+
+/**
+ * A {@link ClientCall} which forwards all of its methods to another {@link ClientCall} which
+ * may have a different sendMessage() message type.
+ */
+abstract class PartialForwardingClientCall<ReqT, RespT> extends ClientCall<ReqT, RespT> {
+  /**
+   * Returns the delegated {@code ClientCall}.
+   */
+  protected abstract ClientCall<?, ?> delegate();
+
+  @Override
+  public void request(int numMessages) {
+    delegate().request(numMessages);
+  }
+
+  @Override
+  public void cancel(@Nullable String message, @Nullable Throwable cause) {
+    delegate().cancel(message, cause);
+  }
+
+  @Override
+  public void halfClose() {
+    delegate().halfClose();
+  }
+
+  @Override
+  public void setMessageCompression(boolean enabled) {
+    delegate().setMessageCompression(enabled);
+  }
+
+  @Override
+  public boolean isReady() {
+    return delegate().isReady();
+  }
+
+  @Override
+  public Attributes getAttributes() {
+    return delegate().getAttributes();
+  }
+}

core/src/main/java/io/grpc/PartialForwardingClientCallListener.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2018, gRPC Authors All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc;
+
+/**
+ * A {@link ClientCall.Listener} which forwards all of its methods to another {@link
+ * ClientCall.Listener} which may have a different parameterized type than the
+ * onMessage() message type.
+ */
+abstract class PartialForwardingClientCallListener<RespT> extends ClientCall.Listener<RespT> {
+  /**
+   * Returns the delegated {@code ClientCall.Listener}.
+   */
+  protected abstract ClientCall.Listener<?> delegate();
+
+  @Override
+  public void onHeaders(Metadata headers) {
+    delegate().onHeaders(headers);
+  }
+
+  @Override
+  public void onClose(Status status, Metadata trailers) {
+    delegate().onClose(status, trailers);
+  }
+
+  @Override
+  public void onReady() {
+    delegate().onReady();
+  }
+}

core/src/main/java/io/grpc/internal/AbstractManagedChannelImplBuilder.java

@@ -129,6 +129,8 @@ public static ManagedChannelBuilder<?> forTarget(String target) {
 
   private int maxInboundMessageSize = GrpcUtil.DEFAULT_MAX_MESSAGE_SIZE;
 
+  BinaryLogProvider binlogProvider = BinaryLogProvider.provider();
+
   /**
    * Sets the maximum message size allowed for a single gRPC frame. If an inbound messages
    * larger than this limit is received it will not be processed and the RPC will fail with