Repository: envoyproxy/envoy
File: source/extensions/filters/http/ext_proc/ext_proc.cc (C++)

Code before:
      return state.chunkQueue().queue().front()->end_stream;
    }
    break;
  case ProcessingMode::FULL_DUPLEX_STREAMED: {
    if (body_response.has_response() && body_response.response().has_body_mutation()) {
      const auto& body_mutation = body_response.response().body_mutation();
      if (body_mutation.has_streamed_response()) {
        return body_mutation.streamed_response().end_of_stream();
      }
    }
    break;
  }
  default:
    break;
  }
  return false;
}

} // namespace

void Filter::closeGrpcStreamIfLastRespReceived(const ProcessingResponse& response,
                                               const bool is_last_body_resp) {

  if (stream_ == nullptr || !Runtime::runtimeFeatureEnabled(
                                "envoy.reloadable_features.ext_proc_stream_close_optimization")) {
    return;
  }

  bool last_response = false;

  switch (response.response_case()) {
  case ProcessingResponse::ResponseCase::kRequestHeaders:
    if ((decoding_state_.hasNoBody() ||
         (decoding_state_.bodyMode() == ProcessingMode::NONE && !decoding_state_.sendTrailers())) &&
        encoding_state_.noExternalProcess()) {
      last_response = true;
    }
    break;
  case ProcessingResponse::ResponseCase::kRequestBody:
    if (is_last_body_resp && encoding_state_.noExternalProcess()) {
      last_response = true;
    }
    break;
  case ProcessingResponse::ResponseCase::kRequestTrailers:
    if (encoding_state_.noExternalProcess()) {
      last_response = true;
    }
    break;
  case ProcessingResponse::ResponseCase::kResponseHeaders:
    if (encoding_state_.hasNoBody() ||
        (encoding_state_.bodyMode() == ProcessingMode::NONE && !encoding_state_.sendTrailers())) {

Review comment (bug):
checking `if (stream_ == nullptr` looks a bit strange to me. I assume it means the stream has already been closed?
Should this case already been handled by other code path?

Code after:
    if (!state.chunkQueue().empty()) {
      return state.chunkQueue().queue().front()->end_stream;
    }
    return false;
  case ProcessingMode::FULL_DUPLEX_STREAMED: {
    if (body_response.has_response() && body_response.response().has_body_mutation()) {
      const auto& body_mutation = body_response.response().body_mutation();
      if (body_mutation.has_streamed_response()) {
        return body_mutation.streamed_response().end_of_stream();
      }
    }
    return false;
  }
  default:
    break;
  }
  return false;
}

} // namespace

void Filter::closeGrpcStreamIfLastRespReceived(const ProcessingResponse& response,
                                               const bool is_last_body_resp) {
  // Bail out if the gRPC stream has already been closed. This can happen in scenarios
  // like immediate responses or rejected header mutations.
  if (stream_ == nullptr || !Runtime::runtimeFeatureEnabled(
                                "envoy.reloadable_features.ext_proc_stream_close_optimization")) {
    return;
  }

  bool last_response = false;

  switch (response.response_case()) {
  case ProcessingResponse::ResponseCase::kRequestHeaders:
    if ((decoding_state_.hasNoBody() ||
         (decoding_state_.bodyMode() == ProcessingMode::NONE && !decoding_state_.sendTrailers())) &&
        encoding_state_.noExternalProcess()) {
      last_response = true;
    }
    break;
  case ProcessingResponse::ResponseCase::kRequestBody:
    if (is_last_body_resp && encoding_state_.noExternalProcess()) {
      last_response = true;
    }
    break;
  case ProcessingResponse::ResponseCase::kRequestTrailers:
    if (encoding_state_.noExternalProcess()) {
      last_response = true;
    }
    break;
  case ProcessingResponse::ResponseCase::kResponseHeaders: