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

Code before:
  mode_override.set_response_body_mode(target_override.response_body_mode());
  return mode_override;
}

bool isLastBodyResponse(ProcessorState& state,
                        const envoy::service::ext_proc::v3::ProcessingResponse& response) {
  switch (state.bodyMode()) {
  case ProcessingMode::BUFFERED:
  case ProcessingMode::BUFFERED_PARTIAL:
    // TODO: - skip stream closing optimization for BUFFERED and BUFFERED_PARTIAL for now.
    break;
  case ProcessingMode::STREAMED:
    if (!state.chunkQueue().empty()) {
      return state.chunkQueue().queue().front()->end_stream;
    }
    break;
  case ProcessingMode::FULL_DUPLEX_STREAMED: {
    const envoy::service::ext_proc::v3::BodyResponse* body_response = nullptr;
    if (response.has_request_body()) {
      body_response = &response.request_body();
    } else if (response.has_response_body()) {
      body_response = &response.response_body();
    }
    if (body_response != nullptr && body_response->has_response()) {
      const auto& common_response = body_response->response();
      if (common_response.has_body_mutation() &&
          common_response.body_mutation().has_streamed_response()) {
        return common_response.body_mutation().streamed_response().end_of_stream();
      }
    }
    break;
  }
  default:
    break;
  }
  return false;
}

} // namespace

bool Filter::noExternalProcessInEncoding() const {
  return (!encoding_state_.sendHeaders() && encoding_state_.bodyMode() == ProcessingMode::NONE &&
          !encoding_state_.sendTrailers());
}

// Close the gRPC stream if the last ProcessingResponse is received.
void Filter::closeGrpcStreamIfLastRespReceived(
    const std::unique_ptr<envoy::service::ext_proc::v3::ProcessingResponse>& response) {
  bool last_response = false;

  switch (response->response_case()) {

Review comment (suggestion):
This is something that should part of the `ProcessorState`. After processing a response message it should return a flag indicating that a terminal state has been reached and gRPC stream can be closed.

Code after:
    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:
    if (encoding_state_.hasNoBody() ||
        (encoding_state_.bodyMode() == ProcessingMode::NONE && !encoding_state_.sendTrailers())) {
      last_response = true;
    }
    break;