@VisibleForTesting
static void prepareHeaders(
    Metadata headers,
    DecompressorRegistry decompressorRegistry,
    Compressor compressor,
    boolean fullStreamDecompression) {
  headers.discardAll(MESSAGE_ENCODING_KEY);
  if (compressor != Codec.Identity.NONE) {
    headers.put(MESSAGE_ENCODING_KEY, compressor.getMessageEncoding());
  }

  headers.discardAll(MESSAGE_ACCEPT_ENCODING_KEY);
  byte[] advertisedEncodings =
      InternalDecompressorRegistry.getRawAdvertisedMessageEncodings(decompressorRegistry);
  if (advertisedEncodings.length != 0) {
    headers.put(MESSAGE_ACCEPT_ENCODING_KEY, advertisedEncodings);
  }

  headers.discardAll(CONTENT_ENCODING_KEY);
  headers.discardAll(CONTENT_ACCEPT_ENCODING_KEY);
  if (fullStreamDecompression) {
    headers.put(CONTENT_ACCEPT_ENCODING_KEY, FULL_STREAM_DECOMPRESSION_ENCODINGS);
  }
}
 

private void prepareHeaders(Compressor compressor, Metadata metadata) {
    final RequestHeadersBuilder newHeaders = req.headers().toBuilder();
    if (compressor != Identity.NONE) {
        newHeaders.set(GrpcHeaderNames.GRPC_ENCODING, compressor.getMessageEncoding());
    }

    if (!advertisedEncodingsHeader.isEmpty()) {
        newHeaders.add(GrpcHeaderNames.GRPC_ACCEPT_ENCODING, advertisedEncodingsHeader);
    }

    newHeaders.add(GrpcHeaderNames.GRPC_TIMEOUT,
                   TimeoutHeaderUtil.toHeaderValue(
                           TimeUnit.MILLISECONDS.toNanos(ctx.responseTimeoutMillis())));

    MetadataUtil.fillHeaders(metadata, newHeaders);

    final HttpRequest newReq = req.withHeaders(newHeaders);
    ctx.updateRequest(newReq);
}
 

@VisibleForTesting
static void prepareHeaders(
    Metadata headers,
    DecompressorRegistry decompressorRegistry,
    Compressor compressor,
    boolean fullStreamDecompression) {
  headers.discardAll(MESSAGE_ENCODING_KEY);
  if (compressor != Codec.Identity.NONE) {
    headers.put(MESSAGE_ENCODING_KEY, compressor.getMessageEncoding());
  }

  headers.discardAll(MESSAGE_ACCEPT_ENCODING_KEY);
  byte[] advertisedEncodings =
      InternalDecompressorRegistry.getRawAdvertisedMessageEncodings(decompressorRegistry);
  if (advertisedEncodings.length != 0) {
    headers.put(MESSAGE_ACCEPT_ENCODING_KEY, advertisedEncodings);
  }

  headers.discardAll(CONTENT_ENCODING_KEY);
  headers.discardAll(CONTENT_ACCEPT_ENCODING_KEY);
  if (fullStreamDecompression) {
    headers.put(CONTENT_ACCEPT_ENCODING_KEY, FULL_STREAM_DECOMPRESSION_ENCODINGS);
  }
}
 

@Override
public void setCompressor(final Compressor compressor) {
  checkNotNull(compressor, "compressor");
  delayOrExecute(new Runnable() {
    @Override
    public void run() {
      realStream.setCompressor(compressor);
    }
  });
}
 

@Override
public final void setCompressor(final Compressor compressor) {
  class CompressorEntry implements BufferEntry {
    @Override
    public void runWith(Substream substream) {
      substream.stream.setCompressor(compressor);
    }
  }

  delayOrExecute(new CompressorEntry());
}
 

@Override
public void setCompressor(final Compressor compressor) {
  checkNotNull(compressor, "compressor");
  delayOrExecute(new Runnable() {
    @Override
    public void run() {
      realStream.setCompressor(compressor);
    }
  });
}
 

@Override
public final void setCompressor(final Compressor compressor) {
  class CompressorEntry implements BufferEntry {
    @Override
    public void runWith(Substream substream) {
      substream.stream.setCompressor(compressor);
    }
  }

  delayOrExecute(new CompressorEntry());
}
 

@Override
public void setCompressor(Compressor compressor) {}
 

@Override
public void setCompressor(Compressor compressor) {}
 

@Override
public void setCompressor(Compressor compressor) {
  delegate().setCompressor(compressor);
}
 

/** Set the compressor used for compression. */
Framer setCompressor(Compressor compressor);
 

@Override
public Framer setCompressor(Compressor compressor) {
  return this;
}
 

@Override
public MessageFramer setCompressor(Compressor compressor) {
  this.compressor = checkNotNull(compressor, "Can't pass an empty compressor");
  return this;
}
 

@Override
public final void setCompressor(Compressor compressor) {
  framer().setCompressor(checkNotNull(compressor, "compressor"));
}
 

@Override
public void setCompressor(Compressor compressor) {}
 

@Test
public void setCompressorTest() {
  Compressor compressor = mock(Compressor.class);
  forward.setCompressor(compressor);
  verify(mock).setCompressor(same(compressor));
}
 

@Override
public void setCompressor(Compressor compressor) {}
 

@Override
public void setCompressor(Compressor compressor) {}
 

@Override
public void setCompressor(Compressor compressor) {
  delegate().setCompressor(compressor);
}
 

private void startInternal(final Listener<RespT> observer, Metadata headers) {
  checkState(stream == null, "Already started");
  checkState(!cancelCalled, "call was cancelled");
  checkNotNull(observer, "observer");
  checkNotNull(headers, "headers");

  if (context.isCancelled()) {
    // Context is already cancelled so no need to create a real stream, just notify the observer
    // of cancellation via callback on the executor
    stream = NoopClientStream.INSTANCE;
    executeCloseObserverInContext(observer, statusFromCancelled(context));
    return;
  }
  final String compressorName = callOptions.getCompressor();
  Compressor compressor;
  if (compressorName != null) {
    compressor = compressorRegistry.lookupCompressor(compressorName);
    if (compressor == null) {
      stream = NoopClientStream.INSTANCE;
      Status status = Status.INTERNAL.withDescription(
          String.format("Unable to find compressor by name %s", compressorName));
      executeCloseObserverInContext(observer, status);
      return;
    }
  } else {
    compressor = Codec.Identity.NONE;
  }
  prepareHeaders(headers, decompressorRegistry, compressor, fullStreamDecompression);

  Deadline effectiveDeadline = effectiveDeadline();
  boolean deadlineExceeded = effectiveDeadline != null && effectiveDeadline.isExpired();
  if (!deadlineExceeded) {
    logIfContextNarrowedTimeout(
        effectiveDeadline, context.getDeadline(), callOptions.getDeadline());
    if (retryEnabled) {
      stream = clientTransportProvider.newRetriableStream(method, callOptions, headers, context);
    } else {
      ClientTransport transport = clientTransportProvider.get(
          new PickSubchannelArgsImpl(method, headers, callOptions));
      Context origContext = context.attach();
      try {
        stream = transport.newStream(method, headers, callOptions);
      } finally {
        context.detach(origContext);
      }
    }
  } else {
    stream = new FailingClientStream(
        DEADLINE_EXCEEDED.withDescription(
            "ClientCall started after deadline exceeded: " + effectiveDeadline));
  }

  if (callExecutorIsDirect) {
    stream.optimizeForDirectExecutor();
  }
  if (callOptions.getAuthority() != null) {
    stream.setAuthority(callOptions.getAuthority());
  }
  if (callOptions.getMaxInboundMessageSize() != null) {
    stream.setMaxInboundMessageSize(callOptions.getMaxInboundMessageSize());
  }
  if (callOptions.getMaxOutboundMessageSize() != null) {
    stream.setMaxOutboundMessageSize(callOptions.getMaxOutboundMessageSize());
  }
  if (effectiveDeadline != null) {
    stream.setDeadline(effectiveDeadline);
  }
  stream.setCompressor(compressor);
  if (fullStreamDecompression) {
    stream.setFullStreamDecompression(fullStreamDecompression);
  }
  stream.setDecompressorRegistry(decompressorRegistry);
  channelCallsTracer.reportCallStarted();
  cancellationListener = new ContextCancellationListener(observer);
  stream.start(new ClientStreamListenerImpl(observer));

  // Delay any sources of cancellation after start(), because most of the transports are broken if
  // they receive cancel before start. Issue #1343 has more details

  // Propagate later Context cancellation to the remote side.
  context.addListener(cancellationListener, directExecutor());
  if (effectiveDeadline != null
      // If the context has the effective deadline, we don't need to schedule an extra task.
      && !effectiveDeadline.equals(context.getDeadline())
      // If the channel has been terminated, we don't need to schedule an extra task.
      && deadlineCancellationExecutor != null
      // if already expired deadline let failing stream handle
      && !(stream instanceof FailingClientStream)) {
    deadlineCancellationNotifyApplicationFuture =
        startDeadlineNotifyApplicationTimer(effectiveDeadline, observer);
  }
  if (cancelListenersShouldBeRemoved) {
    // Race detected! ClientStreamListener.closed may have been called before
    // deadlineCancellationFuture was set / context listener added, thereby preventing the future
    // and listener from being cancelled. Go ahead and cancel again, just to be sure it
    // was cancelled.
    removeContextListenerAndCancelDeadlineFuture();
  }
}
 

/** Set the compressor used for compression. */
Framer setCompressor(Compressor compressor);
 

@Override
public Framer setCompressor(Compressor compressor) {
  return this;
}
 

@Override
public MessageFramer setCompressor(Compressor compressor) {
  this.compressor = checkNotNull(compressor, "Can't pass an empty compressor");
  return this;
}
 

@Override
public final void setCompressor(Compressor compressor) {
  framer().setCompressor(checkNotNull(compressor, "compressor"));
}
 

@Override
public void setCompressor(Compressor compressor) {}
 

@Test
public void setCompressorTest() {
  Compressor compressor = mock(Compressor.class);
  forward.setCompressor(compressor);
  verify(mock).setCompressor(same(compressor));
}
 

/**
 * Sets the compressor on the framer.
 *
 * @param compressor the compressor to use
 */
void setCompressor(Compressor compressor);
 

/**
 * Sets the compressor on the framer.
 *
 * @param compressor the compressor to use
 */
void setCompressor(Compressor compressor);