test: got the client_integration_test moved over to the builder.

library/cc/engine_builder.cc

@@ -92,6 +92,11 @@ EngineBuilder& EngineBuilder::setDeviceOs(const std::string& device_os) {
   return *this;
 }
 
+EngineBuilder& EngineBuilder::setStreamIdleTimeoutSeconds(int stream_idle_timeout_seconds) {
+  this->stream_idle_timeout_seconds_ = stream_idle_timeout_seconds;
+  return *this;
+}
+
 std::string EngineBuilder::generateConfigStr() {
   std::vector<std::pair<std::string, std::string>> replacements{
       {"connect_timeout", fmt::format("{}s", this->connect_timeout_seconds_)},
@@ -144,7 +149,12 @@ EngineSharedPtr EngineBuilder::build() {
 
   envoy_event_tracker null_tracker{};
 
-  auto config_str = this->generateConfigStr();
+  std::string config_str;
+  if (config_override_for_tests_.empty()) {
+    config_str = this->generateConfigStr();
+  } else {
+    config_str = config_override_for_tests_;
+  }
   auto envoy_engine =
       init_engine(this->callbacks_->asEnvoyEngineCallbacks(), null_logger, null_tracker);
   run_engine(envoy_engine, config_str.c_str(), logLevelToString(this->log_level_).c_str());

library/cc/engine_builder.h

@@ -33,6 +33,7 @@ class EngineBuilder {
   EngineBuilder& setAppVersion(const std::string& app_version);
   EngineBuilder& setAppId(const std::string& app_id);
   EngineBuilder& setDeviceOs(const std::string& app_id);
+  EngineBuilder& setStreamIdleTimeoutSeconds(int stream_idle_timeout_seconds);
 
   // this is separated from build() for the sake of testability
   std::string generateConfigStr();
@@ -44,6 +45,8 @@ class EngineBuilder {
   // Filter): EngineBuilder& addNativeFilter(name: String = UUID.randomUUID().toString(),
   // typedConfig: String): EngineBuilder& addStringAccessor(name: String, accessor:
   // EnvoyStringAccessor): EngineBuilder {
+protected:
+  void setOverrideConfigForTests(std::string config) { config_override_for_tests_ = config; }
 
 private:
   LogLevel log_level_ = LogLevel::info;
@@ -64,6 +67,7 @@ class EngineBuilder {
   std::string app_id_ = "unspecified";
   std::string device_os_ = "unspecified";
   std::string virtual_clusters_ = "[]";
+  std::string config_override_for_tests_ = "";
   int stream_idle_timeout_seconds_ = 15;
   int per_try_idle_timeout_seconds_ = 15;
 

library/cc/request_headers_builder.cc

@@ -28,6 +28,11 @@ RequestHeadersBuilder::addUpstreamHttpProtocol(UpstreamHttpProtocol upstream_htt
   return *this;
 }
 
+RequestHeadersBuilder& RequestHeadersBuilder::addKeyValue(std::string key, std::string value) {
+  this->internalSet(key, std::vector<std::string>{value});
+  return *this;
+}
+
 RequestHeaders RequestHeadersBuilder::build() const { return RequestHeaders(this->allHeaders()); }
 
 } // namespace Platform

library/cc/request_headers_builder.h

@@ -21,6 +21,7 @@ class RequestHeadersBuilder : public HeadersBuilder {
 
   RequestHeadersBuilder& addRetryPolicy(const RetryPolicy& retry_policy);
   RequestHeadersBuilder& addUpstreamHttpProtocol(UpstreamHttpProtocol upstream_http_protocol);
+  RequestHeadersBuilder& addKeyValue(std::string key, std::string value);
 
   RequestHeaders build() const;
 };

library/cc/stream.cc

@@ -16,8 +16,8 @@ Stream& Stream::sendHeaders(RequestHeadersSharedPtr headers, bool end_stream) {
   return *this;
 }
 
-Stream& Stream::sendData(envoy_data data) {
-  ::send_data(this->engine_handle_, this->handle_, data, false);
+Stream& Stream::sendData(envoy_data data, bool end_stream) {
+  ::send_data(this->engine_handle_, this->handle_, data, end_stream);
   return *this;
 }
 

library/cc/stream.h

@@ -18,7 +18,7 @@ class Stream {
   Stream(envoy_engine_t engine_handle, envoy_stream_t handle);
 
   Stream& sendHeaders(RequestHeadersSharedPtr headers, bool end_stream);
-  Stream& sendData(envoy_data data);
+  Stream& sendData(envoy_data data, bool end_stream);
   void close(RequestTrailersSharedPtr trailers);
   void close(envoy_data data);
   void cancel();

library/cc/stream_callbacks.cc

@@ -10,7 +10,8 @@ namespace Platform {
 
 namespace {
 
-void* c_on_headers(envoy_headers headers, bool end_stream, envoy_stream_intel, void* context) {
+void* c_on_headers(envoy_headers headers, bool end_stream, envoy_stream_intel intel,
+                   void* context) {
   auto stream_callbacks = *static_cast<StreamCallbacksSharedPtr*>(context);
   if (stream_callbacks->on_headers.has_value()) {
     auto raw_headers = envoyHeadersAsRawHeaderMap(headers);
@@ -22,21 +23,21 @@ void* c_on_headers(envoy_headers headers, bool end_stream, envoy_stream_intel, v
       builder.set(pair.first, pair.second);
     }
     auto on_headers = stream_callbacks->on_headers.value();
-    on_headers(builder.build(), end_stream);
+    on_headers(builder.build(), end_stream, intel);
   }
   return context;
 }
 
-void* c_on_data(envoy_data data, bool end_stream, envoy_stream_intel, void* context) {
+void* c_on_data(envoy_data data, bool end_stream, envoy_stream_intel intel, void* context) {
   auto stream_callbacks = *static_cast<StreamCallbacksSharedPtr*>(context);
   if (stream_callbacks->on_data.has_value()) {
     auto on_data = stream_callbacks->on_data.value();
-    on_data(data, end_stream);
+    on_data(data, end_stream, intel);
   }
   return context;
 }
 
-void* c_on_trailers(envoy_headers metadata, envoy_stream_intel, void* context) {
+void* c_on_trailers(envoy_headers metadata, envoy_stream_intel intel, void* context) {
   auto stream_callbacks = *static_cast<StreamCallbacksSharedPtr*>(context);
   if (stream_callbacks->on_trailers.has_value()) {
     auto raw_headers = envoyHeadersAsRawHeaderMap(metadata);
@@ -45,13 +46,13 @@ void* c_on_trailers(envoy_headers metadata, envoy_stream_intel, void* context) {
       builder.set(pair.first, pair.second);
     }
     auto on_trailers = stream_callbacks->on_trailers.value();
-    on_trailers(builder.build());
+    on_trailers(builder.build(), intel);
   }
   return context;
 }
 
-void* c_on_error(envoy_error raw_error, envoy_stream_intel, envoy_final_stream_intel,
-                 void* context) {
+void* c_on_error(envoy_error raw_error, envoy_stream_intel intel,
+                 envoy_final_stream_intel final_intel, void* context) {
   auto stream_callbacks_ptr = static_cast<StreamCallbacksSharedPtr*>(context);
   auto stream_callbacks = *stream_callbacks_ptr;
   if (stream_callbacks->on_error.has_value()) {
@@ -60,40 +61,40 @@ void* c_on_error(envoy_error raw_error, envoy_stream_intel, envoy_final_stream_i
     error->message = Data::Utility::copyToString(raw_error.message);
     error->attempt_count = absl::optional<int>(raw_error.attempt_count);
     auto on_error = stream_callbacks->on_error.value();
-    on_error(error);
+    on_error(error, intel, final_intel);
   }
   delete stream_callbacks_ptr;
   return nullptr;
 }
 
-void* c_on_complete(envoy_stream_intel, envoy_final_stream_intel, void* context) {
+void* c_on_complete(envoy_stream_intel intel, envoy_final_stream_intel final_intel, void* context) {
   auto stream_callbacks_ptr = static_cast<StreamCallbacksSharedPtr*>(context);
   auto stream_callbacks = *stream_callbacks_ptr;
   if (stream_callbacks->on_complete.has_value()) {
     auto on_complete = stream_callbacks->on_complete.value();
-    on_complete();
+    on_complete(intel, final_intel);
   }
   delete stream_callbacks_ptr;
   return nullptr;
 }
 
-void* c_on_cancel(envoy_stream_intel, envoy_final_stream_intel, void* context) {
+void* c_on_cancel(envoy_stream_intel intel, envoy_final_stream_intel final_intel, void* context) {
   auto stream_callbacks_ptr = static_cast<StreamCallbacksSharedPtr*>(context);
   auto stream_callbacks = *stream_callbacks_ptr;
   if (stream_callbacks->on_cancel.has_value()) {
     auto on_cancel = stream_callbacks->on_cancel.value();
-    on_cancel();
+    on_cancel(intel, final_intel);
   }
   delete stream_callbacks_ptr;
   return nullptr;
 }
 
-void* c_on_send_window_available(envoy_stream_intel, void* context) {
+void* c_on_send_window_available(envoy_stream_intel intel, void* context) {
   auto stream_callbacks_ptr = static_cast<StreamCallbacksSharedPtr*>(context);
   auto stream_callbacks = *stream_callbacks_ptr;
   if (stream_callbacks->on_send_window_available.has_value()) {
     auto on_send_window_available = stream_callbacks->on_send_window_available.value();
-    on_send_window_available();
+    on_send_window_available(intel);
   }
   delete stream_callbacks_ptr;
   return nullptr;

library/cc/stream_callbacks.h

@@ -17,13 +17,19 @@ namespace Platform {
 class Stream;
 using StreamSharedPtr = std::shared_ptr<Stream>;
 
-using OnHeadersCallback = std::function<void(ResponseHeadersSharedPtr headers, bool end_stream)>;
-using OnDataCallback = std::function<void(envoy_data data, bool end_stream)>;
-using OnTrailersCallback = std::function<void(ResponseTrailersSharedPtr trailers)>;
-using OnErrorCallback = std::function<void(EnvoyErrorSharedPtr error)>;
-using OnCompleteCallback = std::function<void()>;
-using OnCancelCallback = std::function<void()>;
-using OnSendWindowAvailableCallback = std::function<void()>;
+using OnHeadersCallback = std::function<void(ResponseHeadersSharedPtr headers, bool end_stream,
+                                             envoy_stream_intel intel)>;
+using OnDataCallback =
+    std::function<void(envoy_data data, bool end_stream, envoy_stream_intel intel)>;
+using OnTrailersCallback =
+    std::function<void(ResponseTrailersSharedPtr trailers, envoy_stream_intel intel)>;
+using OnErrorCallback = std::function<void(EnvoyErrorSharedPtr error, envoy_stream_intel intel,
+                                           envoy_final_stream_intel final_intel)>;
+using OnCompleteCallback =
+    std::function<void(envoy_stream_intel intel, envoy_final_stream_intel final_intel)>;
+using OnCancelCallback =
+    std::function<void(envoy_stream_intel intel, envoy_final_stream_intel final_intel)>;
+using OnSendWindowAvailableCallback = std::function<void(envoy_stream_intel intel)>;
 
 // See library/common/types/c_types.h for what these callbacks should do.
 struct StreamCallbacks : public std::enable_shared_from_this<StreamCallbacks> {

library/cc/stream_prototype.cc

@@ -9,10 +9,10 @@ StreamPrototype::StreamPrototype(EngineSharedPtr engine) : engine_(engine) {
   this->callbacks_ = std::make_shared<StreamCallbacks>();
 }
 
-StreamSharedPtr StreamPrototype::start() {
+StreamSharedPtr StreamPrototype::start(bool explicit_flow_control) {
   auto envoy_stream = init_stream(this->engine_->engine_);
   start_stream(this->engine_->engine_, envoy_stream, this->callbacks_->asEnvoyHttpCallbacks(),
-               false);
+               explicit_flow_control);
   return std::make_shared<Stream>(this->engine_->engine_, envoy_stream);
 }
 

library/cc/stream_prototype.h

@@ -20,7 +20,7 @@ class StreamPrototype {
 public:
   StreamPrototype(EngineSharedPtr engine);
 
-  StreamSharedPtr start();
+  StreamSharedPtr start(bool explicit_flow_control = false);
 
   StreamPrototype& setOnHeaders(OnHeadersCallback closure);
   StreamPrototype& setOnData(OnDataCallback closure);