async client: fail all active requests during destruction (#321)

Required for dynamic cluster remove.
envoyproxy · Jan 5, 2017 · 23d7b06 · 23d7b06
1 parent e6f20f6
commit 23d7b06
Show file tree

Hide file tree

Showing 3 changed files with 28 additions and 1 deletion.
diff --git a/source/common/http/async_client_impl.cc b/source/common/http/async_client_impl.cc
@@ -19,7 +19,11 @@ AsyncClientImpl::AsyncClientImpl(const Upstream::ClusterInfo& cluster, Stats::St
  random, std::move(shadow_writer), true),
  dispatcher_(dispatcher), local_address_(local_address) {}
 
-AsyncClientImpl::~AsyncClientImpl() { ASSERT(active_requests_.empty()); }
+AsyncClientImpl::~AsyncClientImpl() {
+ while (!active_requests_.empty()) {
+ active_requests_.front()->failDueToClientDestroy();
+ }
+}
 
 AsyncClient::Request* AsyncClientImpl::send(MessagePtr&& request, AsyncClient::Callbacks& callbacks,
  const Optional<std::chrono::milliseconds>& timeout) {
@@ -124,4 +128,12 @@ void AsyncRequestImpl::resetStream() {
  cleanup();
 }
 
+void AsyncRequestImpl::failDueToClientDestroy() {
+ // In this case we are going away because the client is being destroyed. We need to both reset
+ // the stream as well as raise a failure callback.
+ reset_callback_();
+ callbacks_.onFailure(AsyncClient::FailureReason::Reset);
+ cleanup();
+}
+
 } // Http
diff --git a/source/common/http/async_client_impl.h b/source/common/http/async_client_impl.h
@@ -119,6 +119,7 @@ class AsyncRequestImpl final : public AsyncClient::Request,
  };
 
  void cleanup();
+ void failDueToClientDestroy();
  void onComplete();
 
  // Http::StreamDecoderFilterCallbacks

diff --git a/test/common/http/async_client_impl_test.cc b/test/common/http/async_client_impl_test.cc
@@ -252,6 +252,20 @@ TEST_F(AsyncClientImplTest, CancelRequest) {
  request->cancel();
 }
 
+TEST_F(AsyncClientImplTest, DestroyWithActive) {
+ EXPECT_CALL(cm_.conn_pool_, newStream(_, _))
+ .WillOnce(Invoke([&](StreamDecoder&, ConnectionPool::Callbacks& callbacks)
+ -> ConnectionPool::Cancellable* {
+ callbacks.onPoolReady(stream_encoder_, cm_.conn_pool_.host_);
+ return nullptr;
+ }));
+
+ EXPECT_CALL(stream_encoder_, encodeHeaders(HeaderMapEqualRef(&message_->headers()), true));
+ EXPECT_CALL(stream_encoder_.stream_, resetStream(_));
+ EXPECT_CALL(callbacks_, onFailure(_));
+ client_.send(std::move(message_), callbacks_, Optional<std::chrono::milliseconds>());
+}
+
 TEST_F(AsyncClientImplTest, PoolFailure) {
  EXPECT_CALL(cm_.conn_pool_, newStream(_, _))
  .WillOnce(Invoke([&](StreamDecoder&,