GrpcUtil.java

/*
 * Copyright 2014 The gRPC Authors
 *
 * 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.internal;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkState;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Objects;
import com.google.common.base.Preconditions;
import com.google.common.base.Splitter;
import com.google.common.base.Stopwatch;
import com.google.common.base.Supplier;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import io.grpc.Attributes;
import io.grpc.CallOptions;
import io.grpc.ClientStreamTracer;
import io.grpc.ClientStreamTracer.InternalLimitedInfoFactory;
import io.grpc.ClientStreamTracer.StreamInfo;
import io.grpc.InternalChannelz.SocketStats;
import io.grpc.InternalLogId;
import io.grpc.InternalMetadata;
import io.grpc.InternalMetadata.TrustedAsciiMarshaller;
import io.grpc.LoadBalancer.PickResult;
import io.grpc.LoadBalancer.Subchannel;
import io.grpc.Metadata;
import io.grpc.MethodDescriptor;
import io.grpc.ProxiedSocketAddress;
import io.grpc.ProxyDetector;
import io.grpc.Status;
import io.grpc.internal.ClientStreamListener.RpcProgress;
import io.grpc.internal.SharedResourceHolder.Resource;
import io.grpc.internal.StreamListener.MessageProducer;
import java.io.Closeable;
import java.io.IOException;
import java.io.InputStream;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.net.HttpURLConnection;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.charset.Charset;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.annotation.Nullable;
import javax.annotation.concurrent.Immutable;

/**
 * Common utilities for GRPC.
 */
public final class GrpcUtil {

  private static final Logger log = Logger.getLogger(GrpcUtil.class.getName());

  public static final Charset US_ASCII = Charset.forName("US-ASCII");

  /**
   * {@link io.grpc.Metadata.Key} for the timeout header.
   */
  public static final Metadata.Key<Long> TIMEOUT_KEY =
          Metadata.Key.of(GrpcUtil.TIMEOUT, new TimeoutMarshaller());

  /**
   * {@link io.grpc.Metadata.Key} for the message encoding header.
   */
  public static final Metadata.Key<String> MESSAGE_ENCODING_KEY =
          Metadata.Key.of(GrpcUtil.MESSAGE_ENCODING, Metadata.ASCII_STRING_MARSHALLER);

  /**
   * {@link io.grpc.Metadata.Key} for the accepted message encodings header.
   */
  public static final Metadata.Key<byte[]> MESSAGE_ACCEPT_ENCODING_KEY =
      InternalMetadata.keyOf(GrpcUtil.MESSAGE_ACCEPT_ENCODING, new AcceptEncodingMarshaller());

  /**
   * {@link io.grpc.Metadata.Key} for the stream's content encoding header.
   */
  public static final Metadata.Key<String> CONTENT_ENCODING_KEY =
      Metadata.Key.of(GrpcUtil.CONTENT_ENCODING, Metadata.ASCII_STRING_MARSHALLER);

  /**
   * {@link io.grpc.Metadata.Key} for the stream's accepted content encoding header.
   */
  public static final Metadata.Key<byte[]> CONTENT_ACCEPT_ENCODING_KEY =
      InternalMetadata.keyOf(GrpcUtil.CONTENT_ACCEPT_ENCODING, new AcceptEncodingMarshaller());

  static final Metadata.Key<String> CONTENT_LENGTH_KEY =
      Metadata.Key.of("content-length", Metadata.ASCII_STRING_MARSHALLER);

  private static final class AcceptEncodingMarshaller implements TrustedAsciiMarshaller<byte[]> {
    @Override
    public byte[] toAsciiString(byte[] value) {
      return value;
    }

    @Override
    public byte[] parseAsciiString(byte[] serialized) {
      return serialized;
    }
  }

  /**
   * {@link io.grpc.Metadata.Key} for the Content-Type request/response header.
   */
  public static final Metadata.Key<String> CONTENT_TYPE_KEY =
          Metadata.Key.of("content-type", Metadata.ASCII_STRING_MARSHALLER);

  /**
   * {@link io.grpc.Metadata.Key} for the Transfer encoding.
   */
  public static final Metadata.Key<String> TE_HEADER =
      Metadata.Key.of("te", Metadata.ASCII_STRING_MARSHALLER);

  /**
   * {@link io.grpc.Metadata.Key} for the Content-Type request/response header.
   */
  public static final Metadata.Key<String> USER_AGENT_KEY =
          Metadata.Key.of("user-agent", Metadata.ASCII_STRING_MARSHALLER);

  /**
   * The default port for plain-text connections.
   */
  public static final int DEFAULT_PORT_PLAINTEXT = 80;

  /**
   * The default port for SSL connections.
   */
  public static final int DEFAULT_PORT_SSL = 443;

  /**
   * Content-Type used for GRPC-over-HTTP/2.
   */
  public static final String CONTENT_TYPE_GRPC = "application/grpc";

  /**
   * The HTTP method used for GRPC requests.
   */
  public static final String HTTP_METHOD = "POST";

  /**
   * The TE (transport encoding) header for requests over HTTP/2.
   */
  public static final String TE_TRAILERS = "trailers";

  /**
   * The Timeout header name.
   */
  public static final String TIMEOUT = "grpc-timeout";

  /**
   * The message encoding (i.e. compression) that can be used in the stream.
   */
  public static final String MESSAGE_ENCODING = "grpc-encoding";

  /**
   * The accepted message encodings (i.e. compression) that can be used in the stream.
   */
  public static final String MESSAGE_ACCEPT_ENCODING = "grpc-accept-encoding";

  /**
   * The content-encoding used to compress the full gRPC stream.
   */
  public static final String CONTENT_ENCODING = "content-encoding";

  /**
   * The accepted content-encodings that can be used to compress the full gRPC stream.
   */
  public static final String CONTENT_ACCEPT_ENCODING = "accept-encoding";

  /**
   * The default maximum uncompressed size (in bytes) for inbound messages. Defaults to 4 MiB.
   */
  public static final int DEFAULT_MAX_MESSAGE_SIZE = 4 * 1024 * 1024;

  /**
   * The default maximum size (in bytes) for inbound header/trailer.
   */
  // Update documentation in public-facing Builders when changing this value.
  public static final int DEFAULT_MAX_HEADER_LIST_SIZE = 8192;

  public static final Splitter ACCEPT_ENCODING_SPLITTER = Splitter.on(',').trimResults();

  private static final String IMPLEMENTATION_VERSION = "1.42.1"; // CURRENT_GRPC_VERSION

  /**
   * The default timeout in nanos for a keepalive ping request.
   */
  public static final long DEFAULT_KEEPALIVE_TIMEOUT_NANOS = TimeUnit.SECONDS.toNanos(20L);

  /**
   * The magic keepalive time value that disables client keepalive.
   */
  public static final long KEEPALIVE_TIME_NANOS_DISABLED = Long.MAX_VALUE;

  /**
   * The default delay in nanos for server keepalive.
   */
  public static final long DEFAULT_SERVER_KEEPALIVE_TIME_NANOS = TimeUnit.HOURS.toNanos(2L);

  /**
   * The default timeout in nanos for a server keepalive ping request.
   */
  public static final long DEFAULT_SERVER_KEEPALIVE_TIMEOUT_NANOS = TimeUnit.SECONDS.toNanos(20L);

  /**
   * The magic keepalive time value that disables keepalive.
   */
  public static final long SERVER_KEEPALIVE_TIME_NANOS_DISABLED = Long.MAX_VALUE;

  /**
   * The default proxy detector.
   */
  public static final ProxyDetector DEFAULT_PROXY_DETECTOR = new ProxyDetectorImpl();

  /**
   * A proxy detector that always claims no proxy is needed.
   */
  public static final ProxyDetector NOOP_PROXY_DETECTOR = new ProxyDetector() {
    @Nullable
    @Override
    public ProxiedSocketAddress proxyFor(SocketAddress targetServerAddress) {
      return null;
    }
  };

  /**
   * The very default load-balancing policy.
   */
  public static final String DEFAULT_LB_POLICY = "pick_first";

  /**
   * RPCs created on the Channel returned by {@link io.grpc.LoadBalancer.Subchannel#asChannel}
   * will have this option with value {@code true}.  They will be treated differently from
   * the ones created by application.
   */
  public static final CallOptions.Key<Boolean> CALL_OPTIONS_RPC_OWNED_BY_BALANCER =
      CallOptions.Key.create("io.grpc.internal.CALL_OPTIONS_RPC_OWNED_BY_BALANCER");

  private static final ClientStreamTracer NOOP_TRACER = new ClientStreamTracer() {};

  /**
   * Returns true if an RPC with the given properties should be counted when calculating the
   * in-use state of a transport.
   */
  public static boolean shouldBeCountedForInUse(CallOptions callOptions) {
    return !Boolean.TRUE.equals(callOptions.getOption(CALL_OPTIONS_RPC_OWNED_BY_BALANCER));
  }

  /**
   * Maps HTTP error response status codes to transport codes, as defined in <a
   * href="https://github.com/grpc/grpc/blob/master/doc/http-grpc-status-mapping.md">
   * http-grpc-status-mapping.md</a>. Never returns a status for which {@code status.isOk()} is
   * {@code true}.
   */
  public static Status httpStatusToGrpcStatus(int httpStatusCode) {
    return httpStatusToGrpcCode(httpStatusCode).toStatus()
        .withDescription("HTTP status code " + httpStatusCode);
  }

  private static Status.Code httpStatusToGrpcCode(int httpStatusCode) {
    if (httpStatusCode >= 100 && httpStatusCode < 200) {
      // 1xx. These headers should have been ignored.
      return Status.Code.INTERNAL;
    }
    switch (httpStatusCode) {
      case HttpURLConnection.HTTP_BAD_REQUEST:  // 400
      case 431: // Request Header Fields Too Large
        // TODO(carl-mastrangelo): this should be added to the http-grpc-status-mapping.md doc.
        return Status.Code.INTERNAL;
      case HttpURLConnection.HTTP_UNAUTHORIZED:  // 401
        return Status.Code.UNAUTHENTICATED;
      case HttpURLConnection.HTTP_FORBIDDEN:  // 403
        return Status.Code.PERMISSION_DENIED;
      case HttpURLConnection.HTTP_NOT_FOUND:  // 404
        return Status.Code.UNIMPLEMENTED;
      case 429:  // Too Many Requests
      case HttpURLConnection.HTTP_BAD_GATEWAY:  // 502
      case HttpURLConnection.HTTP_UNAVAILABLE:  // 503
      case HttpURLConnection.HTTP_GATEWAY_TIMEOUT:  // 504
        return Status.Code.UNAVAILABLE;
      default:
        return Status.Code.UNKNOWN;
    }
  }

  /**
   * All error codes identified by the HTTP/2 spec. Used in GOAWAY and RST_STREAM frames.
   */
  public enum Http2Error {
    /**
     * Servers implementing a graceful shutdown of the connection will send {@code GOAWAY} with
     * {@code NO_ERROR}. In this case it is important to indicate to the application that the
     * request should be retried (i.e. {@link Status#UNAVAILABLE}).
     */
    NO_ERROR(0x0, Status.UNAVAILABLE),
    PROTOCOL_ERROR(0x1, Status.INTERNAL),
    INTERNAL_ERROR(0x2, Status.INTERNAL),
    FLOW_CONTROL_ERROR(0x3, Status.INTERNAL),
    SETTINGS_TIMEOUT(0x4, Status.INTERNAL),
    STREAM_CLOSED(0x5, Status.INTERNAL),
    FRAME_SIZE_ERROR(0x6, Status.INTERNAL),
    REFUSED_STREAM(0x7, Status.UNAVAILABLE),
    CANCEL(0x8, Status.CANCELLED),
    COMPRESSION_ERROR(0x9, Status.INTERNAL),
    CONNECT_ERROR(0xA, Status.INTERNAL),
    ENHANCE_YOUR_CALM(0xB, Status.RESOURCE_EXHAUSTED.withDescription("Bandwidth exhausted")),
    INADEQUATE_SECURITY(0xC, Status.PERMISSION_DENIED.withDescription("Permission denied as "
        + "protocol is not secure enough to call")),
    HTTP_1_1_REQUIRED(0xD, Status.UNKNOWN);

    // Populate a mapping of code to enum value for quick look-up.
    private static final Http2Error[] codeMap = buildHttp2CodeMap();

    private static Http2Error[] buildHttp2CodeMap() {
      Http2Error[] errors = Http2Error.values();
      int size = (int) errors[errors.length - 1].code() + 1;
      Http2Error[] http2CodeMap = new Http2Error[size];
      for (Http2Error error : errors) {
        int index = (int) error.code();
        http2CodeMap[index] = error;
      }
      return http2CodeMap;
    }

    private final int code;
    // Status is not guaranteed to be deeply immutable. Don't care though, since that's only true
    // when there are exceptions in the Status, which is not true here.
    @SuppressWarnings("ImmutableEnumChecker")
    private final Status status;

    Http2Error(int code, Status status) {
      this.code = code;
      String description = "HTTP/2 error code: " + this.name();
      if (status.getDescription() != null) {
        description += " (" + status.getDescription() + ")";
      }
      this.status = status.withDescription(description);
    }

    /**
     * Gets the code for this error used on the wire.
     */
    public long code() {
      return code;
    }

    /**
     * Gets the {@link Status} associated with this HTTP/2 code.
     */
    public Status status() {
      return status;
    }

    /**
     * Looks up the HTTP/2 error code enum value for the specified code.
     *
     * @param code an HTTP/2 error code value.
     * @return the HTTP/2 error code enum or {@code null} if not found.
     */
    public static Http2Error forCode(long code) {
      if (code >= codeMap.length || code < 0) {
        return null;
      }
      return codeMap[(int) code];
    }

    /**
     * Looks up the {@link Status} from the given HTTP/2 error code. This is preferred over {@code
     * forCode(code).status()}, to more easily conform to HTTP/2:
     *
     * <blockquote>Unknown or unsupported error codes MUST NOT trigger any special behavior.
     * These MAY be treated by an implementation as being equivalent to INTERNAL_ERROR.</blockquote>
     *
     * @param code the HTTP/2 error code.
     * @return a {@link Status} representing the given error.
     */
    public static Status statusForCode(long code) {
      Http2Error error = forCode(code);
      if (error == null) {
        // This "forgets" the message of INTERNAL_ERROR while keeping the same status code.
        Status.Code statusCode = INTERNAL_ERROR.status().getCode();
        return Status.fromCodeValue(statusCode.value())
            .withDescription("Unrecognized HTTP/2 error code: " + code);
      }

      return error.status();
    }
  }

  /**
   * Indicates whether or not the given value is a valid gRPC content-type.
   */
  public static boolean isGrpcContentType(String contentType) {
    if (contentType == null) {
      return false;
    }

    if (CONTENT_TYPE_GRPC.length() > contentType.length()) {
      return false;
    }

    contentType = contentType.toLowerCase();
    if (!contentType.startsWith(CONTENT_TYPE_GRPC)) {
      // Not a gRPC content-type.
      return false;
    }

    if (contentType.length() == CONTENT_TYPE_GRPC.length()) {
      // The strings match exactly.
      return true;
    }

    // The contentType matches, but is longer than the expected string.
    // We need to support variations on the content-type (e.g. +proto, +json) as defined by the
    // gRPC wire spec.
    char nextChar = contentType.charAt(CONTENT_TYPE_GRPC.length());
    return nextChar == '+' || nextChar == ';';
  }

  /**
   * Gets the User-Agent string for the gRPC transport.
   */
  public static String getGrpcUserAgent(
      String transportName, @Nullable String applicationUserAgent) {
    StringBuilder builder = new StringBuilder();
    if (applicationUserAgent != null) {
      builder.append(applicationUserAgent);
      builder.append(' ');
    }
    builder.append("grpc-java-");
    builder.append(transportName);
    builder.append('/');
    builder.append(IMPLEMENTATION_VERSION);
    return builder.toString();
  }

  @Immutable
  public static final class GrpcBuildVersion {
    private final String userAgent;
    private final String implementationVersion;

    private GrpcBuildVersion(String userAgent, String implementationVersion) {
      this.userAgent = Preconditions.checkNotNull(userAgent, "userAgentName");
      this.implementationVersion =
          Preconditions.checkNotNull(implementationVersion, "implementationVersion");
    }

    public String getUserAgent() {
      return userAgent;
    }

    public String getImplementationVersion() {
      return implementationVersion;
    }

    @Override
    public String toString() {
      return userAgent + " " + implementationVersion;
    }
  }

  /**
   * Returns the build version of gRPC.
   */
  public static GrpcBuildVersion getGrpcBuildVersion() {
    return new GrpcBuildVersion("gRPC Java", IMPLEMENTATION_VERSION);
  }

  /**
   * Parse an authority into a URI for retrieving the host and port.
   */
  public static URI authorityToUri(String authority) {
    Preconditions.checkNotNull(authority, "authority");
    URI uri;
    try {
      uri = new URI(null, authority, null, null, null);
    } catch (URISyntaxException ex) {
      throw new IllegalArgumentException("Invalid authority: " + authority, ex);
    }
    return uri;
  }

  /**
   * Verify {@code authority} is valid for use with gRPC. The syntax must be valid and it must not
   * include userinfo.
   *
   * @return the {@code authority} provided
   */
  public static String checkAuthority(String authority) {
    URI uri = authorityToUri(authority);
    checkArgument(uri.getHost() != null, "No host in authority '%s'", authority);
    checkArgument(uri.getUserInfo() == null,
        "Userinfo must not be present on authority: '%s'", authority);
    return authority;
  }

  /**
   * Combine a host and port into an authority string.
   */
  // There is a copy of this method in io.grpc.Grpc
  public static String authorityFromHostAndPort(String host, int port) {
    try {
      return new URI(null, null, host, port, null, null, null).getAuthority();
    } catch (URISyntaxException ex) {
      throw new IllegalArgumentException("Invalid host or port: " + host + " " + port, ex);
    }
  }

  /**
   * Shared executor for channels.
   */
  public static final Resource<Executor> SHARED_CHANNEL_EXECUTOR =
      new Resource<Executor>() {
        private static final String NAME = "grpc-default-executor";
        @Override
        public Executor create() {
          return Executors.newCachedThreadPool(getThreadFactory(NAME + "-%d", true));
        }

        @Override
        public void close(Executor instance) {
          ((ExecutorService) instance).shutdown();
        }

        @Override
        public String toString() {
          return NAME;
        }
      };

  /**
   * Shared single-threaded executor for managing channel timers.
   */
  public static final Resource<ScheduledExecutorService> TIMER_SERVICE =
      new Resource<ScheduledExecutorService>() {
        @Override
        public ScheduledExecutorService create() {
          // We don't use newSingleThreadScheduledExecutor because it doesn't return a
          // ScheduledThreadPoolExecutor.
          ScheduledExecutorService service = Executors.newScheduledThreadPool(
              1,
              getThreadFactory("grpc-timer-%d", true));

          // If there are long timeouts that are cancelled, they will not actually be removed from
          // the executors queue.  This forces immediate removal upon cancellation to avoid a
          // memory leak.  Reflection is used because we cannot use methods added in Java 1.7.  If
          // the method does not exist, we give up.  Note that the method is not present in 1.6, but
          // _is_ present in the android standard library.
          try {
            Method method = service.getClass().getMethod("setRemoveOnCancelPolicy", boolean.class);
            method.invoke(service, true);
          } catch (NoSuchMethodException e) {
            // no op
          } catch (RuntimeException e) {
            throw e;
          } catch (Exception e) {
            throw new RuntimeException(e);
          }

          return Executors.unconfigurableScheduledExecutorService(service);
        }

        @Override
        public void close(ScheduledExecutorService instance) {
          instance.shutdown();
        }
      };


  /**
   * Get a {@link ThreadFactory} suitable for use in the current environment.
   * @param nameFormat to apply to threads created by the factory.
   * @param daemon {@code true} if the threads the factory creates are daemon threads, {@code false}
   *     otherwise.
   * @return a {@link ThreadFactory}.
   */
  public static ThreadFactory getThreadFactory(String nameFormat, boolean daemon) {
    return new ThreadFactoryBuilder()
        .setDaemon(daemon)
        .setNameFormat(nameFormat)
        .build();
  }

  /**
   * The factory of default Stopwatches.
   */
  public static final Supplier<Stopwatch> STOPWATCH_SUPPLIER = new Supplier<Stopwatch>() {
      @Override
      public Stopwatch get() {
        return Stopwatch.createUnstarted();
      }
    };

  /**
   * Returns the host via {@link InetSocketAddress#getHostString} if it is possible,
   * i.e. in jdk >= 7.
   * Otherwise, return it via {@link InetSocketAddress#getHostName} which may incur a DNS lookup.
   */
  public static String getHost(InetSocketAddress addr) {
    try {
      Method getHostStringMethod = InetSocketAddress.class.getMethod("getHostString");
      return (String) getHostStringMethod.invoke(addr);
    } catch (NoSuchMethodException e) {
      // noop
    } catch (IllegalAccessException e) {
      // noop
    } catch (InvocationTargetException e) {
      // noop
    }
    return addr.getHostName();
  }

  /**
   * Marshals a nanoseconds representation of the timeout to and from a string representation,
   * consisting of an ASCII decimal representation of a number with at most 8 digits, followed by a
   * unit:
   * n = nanoseconds
   * u = microseconds
   * m = milliseconds
   * S = seconds
   * M = minutes
   * H = hours
   *
   * <p>The representation is greedy with respect to precision. That is, 2 seconds will be
   * represented as `2000000u`.</p>
   *
   * <p>See <a href="https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests">the
   * request header definition</a></p>
   */
  @VisibleForTesting
  static class TimeoutMarshaller implements Metadata.AsciiMarshaller<Long> {

    @Override
    public String toAsciiString(Long timeoutNanos) {
      long cutoff = 100000000;
      TimeUnit unit = TimeUnit.NANOSECONDS;
      if (timeoutNanos < 0) {
        throw new IllegalArgumentException("Timeout too small");
      } else if (timeoutNanos < cutoff) {
        return timeoutNanos + "n";
      } else if (timeoutNanos < cutoff * 1000L) {
        return unit.toMicros(timeoutNanos) + "u";
      } else if (timeoutNanos < cutoff * 1000L * 1000L) {
        return unit.toMillis(timeoutNanos) + "m";
      } else if (timeoutNanos < cutoff * 1000L * 1000L * 1000L) {
        return unit.toSeconds(timeoutNanos) + "S";
      } else if (timeoutNanos < cutoff * 1000L * 1000L * 1000L * 60L) {
        return unit.toMinutes(timeoutNanos) + "M";
      } else {
        return unit.toHours(timeoutNanos) + "H";
      }
    }

    @Override
    public Long parseAsciiString(String serialized) {
      checkArgument(serialized.length() > 0, "empty timeout");
      checkArgument(serialized.length() <= 9, "bad timeout format");
      long value = Long.parseLong(serialized.substring(0, serialized.length() - 1));
      char unit = serialized.charAt(serialized.length() - 1);
      switch (unit) {
        case 'n':
          return value;
        case 'u':
          return TimeUnit.MICROSECONDS.toNanos(value);
        case 'm':
          return TimeUnit.MILLISECONDS.toNanos(value);
        case 'S':
          return TimeUnit.SECONDS.toNanos(value);
        case 'M':
          return TimeUnit.MINUTES.toNanos(value);
        case 'H':
          return TimeUnit.HOURS.toNanos(value);
        default:
          throw new IllegalArgumentException(String.format("Invalid timeout unit: %s", unit));
      }
    }
  }

  /**
   * Returns a transport out of a PickResult, or {@code null} if the result is "buffer".
   */
  @Nullable
  static ClientTransport getTransportFromPickResult(PickResult result, boolean isWaitForReady) {
    final ClientTransport transport;
    Subchannel subchannel = result.getSubchannel();
    if (subchannel != null) {
      transport = ((TransportProvider) subchannel.getInternalSubchannel()).obtainActiveTransport();
    } else {
      transport = null;
    }
    if (transport != null) {
      final ClientStreamTracer.Factory streamTracerFactory = result.getStreamTracerFactory();
      if (streamTracerFactory == null) {
        return transport;
      }
      return new ClientTransport() {
        @Override
        public ClientStream newStream(
            MethodDescriptor<?, ?> method, Metadata headers, CallOptions callOptions,
            ClientStreamTracer[] tracers) {
          StreamInfo info = StreamInfo.newBuilder().setCallOptions(callOptions).build();
          ClientStreamTracer streamTracer =
              newClientStreamTracer(streamTracerFactory, info, headers);
          checkState(tracers[tracers.length - 1] == NOOP_TRACER, "lb tracer already assigned");
          tracers[tracers.length - 1] = streamTracer;
          return transport.newStream(method, headers, callOptions, tracers);
        }

        @Override
        public void ping(PingCallback callback, Executor executor) {
          transport.ping(callback, executor);
        }

        @Override
        public InternalLogId getLogId() {
          return transport.getLogId();
        }

        @Override
        public ListenableFuture<SocketStats> getStats() {
          return transport.getStats();
        }
      };
    }
    if (!result.getStatus().isOk()) {
      if (result.isDrop()) {
        return new FailingClientTransport(result.getStatus(), RpcProgress.DROPPED);
      }
      if (!isWaitForReady) {
        return new FailingClientTransport(result.getStatus(), RpcProgress.PROCESSED);
      }
    }
    return null;
  }

  /** Gets stream tracers based on CallOptions. */
  public static ClientStreamTracer[] getClientStreamTracers(
      CallOptions callOptions, Metadata headers, int previousAttempts, boolean isTransparentRetry) {
    List<ClientStreamTracer.Factory> factories = callOptions.getStreamTracerFactories();
    ClientStreamTracer[] tracers = new ClientStreamTracer[factories.size() + 1];
    StreamInfo streamInfo = StreamInfo.newBuilder()
        .setCallOptions(callOptions)
        .setPreviousAttempts(previousAttempts)
        .setIsTransparentRetry(isTransparentRetry)
        .build();
    for (int i = 0; i < factories.size(); i++) {
      tracers[i] = newClientStreamTracer(factories.get(i), streamInfo, headers);
    }
    // Reserved to be set later by the lb as per the API contract of ClientTransport.newStream().
    // See also GrpcUtil.getTransportFromPickResult()
    tracers[tracers.length - 1] = NOOP_TRACER;
    return tracers;
  }

  // A util function for backward compatibility to support deprecated StreamInfo.getAttributes().
  @VisibleForTesting
  static ClientStreamTracer newClientStreamTracer(
      final ClientStreamTracer.Factory streamTracerFactory, final StreamInfo info,
      final Metadata headers) {
    ClientStreamTracer streamTracer;
    if (streamTracerFactory instanceof InternalLimitedInfoFactory) {
      streamTracer = streamTracerFactory.newClientStreamTracer(info, headers);
    } else {
      streamTracer = new ForwardingClientStreamTracer() {
        final ClientStreamTracer noop = new ClientStreamTracer() {};
        volatile ClientStreamTracer delegate = noop;

        void maybeInit(StreamInfo info, Metadata headers) {
          if (delegate != noop) {
            return;
          }
          synchronized (this) {
            if (delegate == noop) {
              delegate = streamTracerFactory.newClientStreamTracer(info, headers);
            }
          }
        }

        @Override
        protected ClientStreamTracer delegate() {
          return delegate;
        }

        @SuppressWarnings("deprecation")
        @Override
        public void streamCreated(Attributes transportAttrs, Metadata headers) {
          StreamInfo streamInfo = info.toBuilder().setTransportAttrs(transportAttrs).build();
          maybeInit(streamInfo, headers);
          delegate().streamCreated(transportAttrs, headers);
        }

        @Override
        public void streamClosed(Status status) {
          maybeInit(info, headers);
          delegate().streamClosed(status);
        }
      };
    }
    return streamTracer;
  }

  /** Quietly closes all messages in MessageProducer. */
  static void closeQuietly(MessageProducer producer) {
    InputStream message;
    while ((message = producer.next()) != null) {
      closeQuietly(message);
    }
  }

  /**
   * Closes a Closeable, ignoring IOExceptions.
   * This method exists because Guava's {@code Closeables.closeQuietly()} is beta.
   */
  public static void closeQuietly(@Nullable Closeable message) {
    if (message == null) {
      return;
    }
    try {
      message.close();
    } catch (IOException ioException) {
      // do nothing except log
      log.log(Level.WARNING, "exception caught in closeQuietly", ioException);
    }
  }

  /**
   * Checks whether the given item exists in the iterable.  This is copied from Guava Collect's
   * {@code Iterables.contains()} because Guava Collect is not Android-friendly thus core can't
   * depend on it.
   */
  static <T> boolean iterableContains(Iterable<T> iterable, T item) {
    if (iterable instanceof Collection) {
      Collection<?> collection = (Collection<?>) iterable;
      try {
        return collection.contains(item);
      } catch (NullPointerException e) {
        return false;
      } catch (ClassCastException e) {
        return false;
      }
    }
    for (T i : iterable) {
      if (Objects.equal(i, item)) {
        return true;
      }
    }
    return false;
  }

  private GrpcUtil() {}
}