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

Code before:
    encoding_state_.onFinishProcessorCall(Grpc::Status::DeadlineExceeded);
    ImmediateResponse errorResponse;
    errorResponse.mutable_status()->set_code(StatusCode::GatewayTimeout);
    errorResponse.set_details(absl::StrFormat("%s_per-message_timeout_exceeded", ErrorPrefix));
    sendImmediateResponse(errorResponse);
  }
}

// Regardless of the current filter state, reset it to "IDLE", continue
// the current callback, and reset timers. This is used in a few error-handling situations.
void Filter::clearAsyncState(Grpc::Status::GrpcStatus call_status) {
  decoding_state_.clearAsyncState(call_status);
  encoding_state_.clearAsyncState(call_status);
}

// Regardless of the current state, ensure that the timers won't fire
// again.
void Filter::onFinishProcessorCalls(Grpc::Status::GrpcStatus call_status) {
  decoding_state_.onFinishProcessorCall(call_status);
  encoding_state_.onFinishProcessorCall(call_status);
}

void Filter::sendImmediateResponse(const ImmediateResponse& response) {
  auto status_code = response.has_status() ? response.status().code() : DefaultImmediateStatus;
  if (!MutationUtils::isValidHttpStatus(status_code)) {
    ENVOY_STREAM_LOG(debug, "Ignoring attempt to set invalid HTTP status {}", *decoder_callbacks_,
                     status_code);
    status_code = DefaultImmediateStatus;
  }
  const auto grpc_status =
      response.has_grpc_status()
          ? absl::optional<Grpc::Status::GrpcStatus>(response.grpc_status().status())
          : absl::nullopt;
  const auto mutate_headers = [this, &response](Http::ResponseHeaderMap& headers) {
    if (response.has_headers()) {
      const absl::Status mut_status = MutationUtils::applyHeaderMutations(
          response.headers(), headers, false, config().mutationChecker(),
          stats_.rejected_header_mutations_);
      if (!mut_status.ok()) {
        ENVOY_LOG_EVERY_POW_2(error, "Immediate response mutations failed with {}",
                              mut_status.message());
      }
    }
  };

  sent_immediate_response_ = true;
  ENVOY_STREAM_LOG(debug, "Sending local reply with status code {}", *decoder_callbacks_,
                   status_code);
  const auto details = StringUtil::replaceAllEmptySpace(response.details());
  encoder_callbacks_->sendLocalReply(static_cast<Http::Code>(status_code), response.body(),
                                     mutate_headers, grpc_status, details);

Review comment (suggestion):
qq: Which case the invocation here is covering? i.e., can we just log it in `onFinishProcessorCalls` like `recordGrpcCall`.

If there is the case that this path is covering, could we merge them to one place. That will help code maintainability and readability 

Code after:

  } else {
    // Return an error and stop processing the current stream.
    processing_complete_ = true;
    closeStream();
    onFinishProcessorCalls(Grpc::Status::DeadlineExceeded);
    ImmediateResponse errorResponse;
    errorResponse.mutable_status()->set_code(StatusCode::GatewayTimeout);
    errorResponse.set_details(absl::StrFormat("%s_per-message_timeout_exceeded", ErrorPrefix));
    sendImmediateResponse(errorResponse);
  }
}

void Filter::recordGrpcStatusBeforeFirstCall(Grpc::Status::GrpcStatus call_status) {
  if (!decoding_state_.getCallStartTime().has_value() &&
      !encoding_state_.getCallStartTime().has_value()) {
    if (loggingInfo() != nullptr) {
      loggingInfo()->recordGrpcStatusBeforeFirstCall(call_status);
    }
  }
}

// Regardless of the current filter state, reset it to "IDLE", continue
// the current callback, and reset timers. This is used in a few error-handling situations.
void Filter::clearAsyncState(Grpc::Status::GrpcStatus call_status) {
  recordGrpcStatusBeforeFirstCall(call_status);
  decoding_state_.clearAsyncState(call_status);
  encoding_state_.clearAsyncState(call_status);
}

// Regardless of the current state, ensure that the timers won't fire
// again.
void Filter::onFinishProcessorCalls(Grpc::Status::GrpcStatus call_status) {
  recordGrpcStatusBeforeFirstCall(call_status);
  decoding_state_.onFinishProcessorCall(call_status);
  encoding_state_.onFinishProcessorCall(call_status);
}

void Filter::sendImmediateResponse(const ImmediateResponse& response) {
  auto status_code = response.has_status() ? response.status().code() : DefaultImmediateStatus;
  if (!MutationUtils::isValidHttpStatus(status_code)) {
    ENVOY_STREAM_LOG(debug, "Ignoring attempt to set invalid HTTP status {}", *decoder_callbacks_,
                     status_code);
    status_code = DefaultImmediateStatus;
  }
  const auto grpc_status =
      response.has_grpc_status()
          ? absl::optional<Grpc::Status::GrpcStatus>(response.grpc_status().status())
          : absl::nullopt;
  const auto mutate_headers = [this, &response](Http::ResponseHeaderMap& headers) {
    if (response.has_headers()) {