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

Code before:
}

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;
  stats_.immediate_responses_sent_.inc();
  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);
}

void Filter::mergePerRouteConfig() {
  if (route_config_merged_) {
    return;
  }

  route_config_merged_ = true;

  absl::optional<FilterConfigPerRoute> merged_config;
  for (const FilterConfigPerRoute& typed_cfg :
       Http::Utility::getAllPerFilterConfig<FilterConfigPerRoute>(decoder_callbacks_)) {
    if (!merged_config.has_value()) {
      merged_config.emplace(typed_cfg);
    } else {
      merged_config.emplace(FilterConfigPerRoute(merged_config.value(), typed_cfg));
    }
  }


Review comment (suggestion):
both onGrpcClose() and onGrpcError() are called by onRemoteClose(). So if server_half_closed_ is incremented here. It should be incremented in the above onGrpcError() case as well.

Code after:
    processing_complete_ = true;
    // Since the stream failed, there is no need to handle timeouts, so
    // make sure that they do not fire now.
    onFinishProcessorCalls(status);
    closeStream();
    ImmediateResponse errorResponse;
    errorResponse.mutable_status()->set_code(
        static_cast<StatusCode>(static_cast<uint32_t>(config_->statusOnError())));
    errorResponse.set_details(
        absl::StrFormat("%s_gRPC_error_%i{%s}", ErrorPrefix, status, message));
    sendImmediateResponse(errorResponse);
  }
}

void Filter::onGrpcClose() { onGrpcCloseWithStatus(Grpc::Status::Aborted); }

void Filter::onGrpcCloseWithStatus(Grpc::Status::GrpcStatus status) {
  ENVOY_STREAM_LOG(debug, "Received gRPC stream close", *decoder_callbacks_);

  if (processing_complete_) {
    return;
  }

  processing_complete_ = true;
  stats_.streams_closed_.inc();
  stats_.server_half_closed_.inc();
  // Successful close. We can ignore the stream for the rest of our request
  // and response processing.
  closeStream();
  clearAsyncState(status);
}

void Filter::onMessageTimeout() {
  ENVOY_STREAM_LOG(debug, "message timeout reached", *decoder_callbacks_);
  logStreamInfo();
  stats_.message_timeouts_.inc();
  if (failureModeAllow()) {
    // The user would like a timeout to not cause message processing to fail.
    // However, we don't know if the external processor will send a response later,
    // and we can't wait any more. So, as we do for a spurious message, ignore
    // the external processor for the rest of the request.
    processing_complete_ = true;
    closeStream();
    logFailOpen();
    clearAsyncState(Grpc::Status::DeadlineExceeded);

  } else {
    // Return an error and stop processing the current stream.
    processing_complete_ = true;
    closeStream();
    onFinishProcessorCalls(Grpc::Status::DeadlineExceeded);