diff --git a/http2/adapter/nghttp2_adapter_test.cc b/http2/adapter/nghttp2_adapter_test.cc
index 21346e2..3f69c7c 100644
--- a/http2/adapter/nghttp2_adapter_test.cc
+++ b/http2/adapter/nghttp2_adapter_test.cc
@@ -736,7 +736,9 @@
   EXPECT_FALSE(adapter->session().want_write());
   const char* kSentinel = "";
   const absl::string_view kBody = "This is an example request body.";
-  auto body1 = absl::make_unique<TestDataFrameSource>(visitor, kBody);
+  auto body1 = absl::make_unique<TestDataFrameSource>(visitor, true);
+  body1->AppendPayload(kBody);
+  body1->EndData();
   int stream_id =
       adapter->SubmitRequest(ToHeaders({{":method", "POST"},
                                         {":scheme", "http"},
@@ -1213,8 +1215,10 @@
 
   EXPECT_FALSE(adapter->session().want_write());
   const absl::string_view kBody = "This is an example response body.";
-  auto body1 =
-      absl::make_unique<TestDataFrameSource>(visitor, kBody, /*has_fin=*/false);
+  // A data fin is not sent so that the stream remains open, and the flow
+  // control state can be verified.
+  auto body1 = absl::make_unique<TestDataFrameSource>(visitor, false);
+  body1->AppendPayload(kBody);
   int submit_result = adapter->SubmitResponse(
       1,
       ToHeaders({{":status", "404"},
@@ -1347,8 +1351,9 @@
 
   // The body source must indicate that the end of the body is not the end of
   // the stream.
-  auto body1 =
-      absl::make_unique<TestDataFrameSource>(visitor, kBody, /*has_fin=*/false);
+  auto body1 = absl::make_unique<TestDataFrameSource>(visitor, false);
+  body1->AppendPayload(kBody);
+  body1->EndData();
   int submit_result = adapter->SubmitResponse(
       1, ToHeaders({{":status", "200"}, {"x-comment", "Sure, sounds good."}}),
       std::move(body1));
@@ -1456,8 +1461,9 @@
 
   // The body source must indicate that the end of the body is not the end of
   // the stream.
-  auto body1 =
-      absl::make_unique<TestDataFrameSource>(visitor, kBody, /*has_fin=*/false);
+  auto body1 = absl::make_unique<TestDataFrameSource>(visitor, false);
+  body1->AppendPayload(kBody);
+  body1->EndData();
   int submit_result = adapter->SubmitResponse(
       1, ToHeaders({{":status", "200"}, {"x-comment", "Sure, sounds good."}}),
       std::move(body1));
diff --git a/http2/adapter/nghttp2_data_provider_test.cc b/http2/adapter/nghttp2_data_provider_test.cc
index 394aea4..af8d981 100644
--- a/http2/adapter/nghttp2_data_provider_test.cc
+++ b/http2/adapter/nghttp2_data_provider_test.cc
@@ -14,7 +14,9 @@
 // than what the source provides.
 TEST(DataProviderTest, ReadLessThanSourceProvides) {
   DataSavingVisitor visitor;
-  TestDataFrameSource source(visitor, "Example payload");
+  TestDataFrameSource source(visitor, true);
+  source.AppendPayload("Example payload");
+  source.EndData();
   auto provider = MakeDataProvider(&source);
   uint32_t data_flags = 0;
   const int32_t kStreamId = 1;
@@ -42,7 +44,9 @@
 TEST(DataProviderTest, ReadMoreThanSourceProvides) {
   DataSavingVisitor visitor;
   const absl::string_view kPayload = "Example payload";
-  TestDataFrameSource source(visitor, kPayload);
+  TestDataFrameSource source(visitor, true);
+  source.AppendPayload(kPayload);
+  source.EndData();
   auto provider = MakeDataProvider(&source);
   uint32_t data_flags = 0;
   const int32_t kStreamId = 1;
@@ -68,10 +72,8 @@
 // correctly with nghttp2-style callbacks when the source is blocked.
 TEST(DataProviderTest, ReadFromBlockedSource) {
   DataSavingVisitor visitor;
-  const absl::string_view kPayload = "This source is blocked.";
-  TestDataFrameSource source(visitor, kPayload);
-  // Setting is_data_available(false) indicates the source is blocked.
-  source.set_is_data_available(false);
+  // Source has no payload, but also no fin, so it's blocked.
+  TestDataFrameSource source(visitor, false);
   auto provider = MakeDataProvider(&source);
   uint32_t data_flags = 0;
   const int32_t kStreamId = 1;
@@ -88,9 +90,9 @@
 // no data.
 TEST(DataProviderTest, ReadFromZeroLengthSource) {
   DataSavingVisitor visitor;
-  const absl::string_view kPayload = "";
   // Empty payload and fin=true indicates the source is done.
-  TestDataFrameSource source(visitor, kPayload);
+  TestDataFrameSource source(visitor, true);
+  source.EndData();
   auto provider = MakeDataProvider(&source);
   uint32_t data_flags = 0;
   const int32_t kStreamId = 1;
diff --git a/http2/adapter/oghttp2_adapter_test.cc b/http2/adapter/oghttp2_adapter_test.cc
index 9cc960f..3069c45 100644
--- a/http2/adapter/oghttp2_adapter_test.cc
+++ b/http2/adapter/oghttp2_adapter_test.cc
@@ -725,8 +725,9 @@
 
   // The body source must indicate that the end of the body is not the end of
   // the stream.
-  auto body1 =
-      absl::make_unique<TestDataFrameSource>(visitor, kBody, /*has_fin=*/false);
+  auto body1 = absl::make_unique<TestDataFrameSource>(visitor, false);
+  body1->AppendPayload(kBody);
+  body1->EndData();
   int submit_result = adapter->SubmitResponse(
       1, ToHeaders({{":status", "200"}, {"x-comment", "Sure, sounds good."}}),
       std::move(body1));
diff --git a/http2/adapter/oghttp2_session_test.cc b/http2/adapter/oghttp2_session_test.cc
index 185dbaa..b6682c5 100644
--- a/http2/adapter/oghttp2_session_test.cc
+++ b/http2/adapter/oghttp2_session_test.cc
@@ -76,8 +76,9 @@
 
   // Submit a request to ensure the first stream is created.
   const char* kSentinel1 = "arbitrary pointer 1";
-  auto body1 = absl::make_unique<TestDataFrameSource>(
-      visitor, "This is an example request body.");
+  auto body1 = absl::make_unique<TestDataFrameSource>(visitor, true);
+  body1->AppendPayload("This is an example request body.");
+  body1->EndData();
   int stream_id =
       session.SubmitRequest(ToHeaders({{":method", "POST"},
                                        {":scheme", "http"},
@@ -247,8 +248,9 @@
   EXPECT_EQ(0, session.GetHpackEncoderDynamicTableSize());
 
   const char* kSentinel1 = "arbitrary pointer 1";
-  auto body1 = absl::make_unique<TestDataFrameSource>(
-      visitor, "This is an example request body.");
+  auto body1 = absl::make_unique<TestDataFrameSource>(visitor, true);
+  body1->AppendPayload("This is an example request body.");
+  body1->EndData();
   int stream_id =
       session.SubmitRequest(ToHeaders({{":method", "POST"},
                                        {":scheme", "http"},
@@ -315,10 +317,8 @@
   EXPECT_FALSE(session.want_write());
 
   const char* kSentinel1 = "arbitrary pointer 1";
-  auto body1 = absl::make_unique<TestDataFrameSource>(
-      visitor, "This is an example request body.");
+  auto body1 = absl::make_unique<TestDataFrameSource>(visitor, true);
   TestDataFrameSource* body_ref = body1.get();
-  body_ref->set_is_data_available(false);
   int stream_id =
       session.SubmitRequest(ToHeaders({{":method", "POST"},
                                        {":scheme", "http"},
@@ -346,7 +346,8 @@
   visitor.Clear();
   EXPECT_FALSE(session.want_write());
 
-  body_ref->set_is_data_available(true);
+  body_ref->AppendPayload("This is an example request body.");
+  body_ref->EndData();
   EXPECT_TRUE(session.ResumeStream(stream_id));
   EXPECT_TRUE(session.want_write());
 
@@ -371,8 +372,9 @@
   EXPECT_FALSE(session.want_write());
 
   const char* kSentinel1 = "arbitrary pointer 1";
-  auto body1 = absl::make_unique<TestDataFrameSource>(
-      visitor, "This is an example request body.");
+  auto body1 = absl::make_unique<TestDataFrameSource>(visitor, true);
+  body1->AppendPayload("This is an example request body.");
+  body1->EndData();
   int stream_id =
       session.SubmitRequest(ToHeaders({{":method", "POST"},
                                        {":scheme", "http"},
@@ -655,9 +657,10 @@
   visitor.Clear();
 
   EXPECT_FALSE(session.want_write());
-  auto body1 = absl::make_unique<TestDataFrameSource>(
-      visitor, "This is an example response body.",
-      /*has_fin=*/false);
+  // A data fin is not sent so that the stream remains open, and the flow
+  // control state can be verified.
+  auto body1 = absl::make_unique<TestDataFrameSource>(visitor, false);
+  body1->AppendPayload("This is an example response body.");
   int submit_result = session.SubmitResponse(
       1,
       ToHeaders({{":status", "404"},
@@ -684,6 +687,7 @@
   // Some data was sent, so the remaining send window size should be less than
   // the default.
   EXPECT_LT(session.GetStreamSendWindowSize(1), kInitialFlowControlWindowSize);
+  EXPECT_GT(session.GetStreamSendWindowSize(1), 0);
   // Send window for a nonexistent stream is not available.
   EXPECT_EQ(session.GetStreamSendWindowSize(3), -1);
 
@@ -793,9 +797,9 @@
 
   // The body source must indicate that the end of the body is not the end of
   // the stream.
-  auto body1 = absl::make_unique<TestDataFrameSource>(
-      visitor, "This is an example response body.",
-      /*has_fin=*/false);
+  auto body1 = absl::make_unique<TestDataFrameSource>(visitor, false);
+  body1->AppendPayload("This is an example response body.");
+  body1->EndData();
   int submit_result = session.SubmitResponse(
       1, ToHeaders({{":status", "200"}, {"x-comment", "Sure, sounds good."}}),
       std::move(body1));
@@ -885,9 +889,9 @@
 
   // The body source must indicate that the end of the body is not the end of
   // the stream.
-  auto body1 = absl::make_unique<TestDataFrameSource>(
-      visitor, "This is an example response body.",
-      /*has_fin=*/false);
+  auto body1 = absl::make_unique<TestDataFrameSource>(visitor, false);
+  body1->AppendPayload("This is an example response body.");
+  body1->EndData();
   int submit_result = session.SubmitResponse(
       1, ToHeaders({{":status", "200"}, {"x-comment", "Sure, sounds good."}}),
       std::move(body1));
diff --git a/http2/adapter/test_utils.cc b/http2/adapter/test_utils.cc
index 9281215..04b126f 100644
--- a/http2/adapter/test_utils.cc
+++ b/http2/adapter/test_utils.cc
@@ -9,41 +9,26 @@
 namespace test {
 
 TestDataFrameSource::TestDataFrameSource(Http2VisitorInterface& visitor,
-                                         absl::string_view data_payload,
                                          bool has_fin)
-    : visitor_(visitor), has_fin_(has_fin) {
-  if (!data_payload.empty()) {
-    payload_fragments_.push_back(std::string(data_payload));
+    : visitor_(visitor), has_fin_(has_fin) {}
+
+void TestDataFrameSource::AppendPayload(absl::string_view payload) {
+  QUICHE_CHECK(!end_data_);
+  if (!payload.empty()) {
+    payload_fragments_.push_back(std::string(payload));
     current_fragment_ = payload_fragments_.front();
   }
 }
 
-TestDataFrameSource::TestDataFrameSource(
-    Http2VisitorInterface& visitor,
-    absl::Span<absl::string_view> payload_fragments,
-    bool has_fin)
-    : visitor_(visitor), has_fin_(has_fin) {
-  payload_fragments_.reserve(payload_fragments.size());
-  for (absl::string_view fragment : payload_fragments) {
-    if (!fragment.empty()) {
-      payload_fragments_.push_back(std::string(fragment));
-    }
-  }
-  if (!payload_fragments_.empty()) {
-    current_fragment_ = payload_fragments_.front();
-  }
-}
+void TestDataFrameSource::EndData() { end_data_ = true; }
 
 std::pair<ssize_t, bool> TestDataFrameSource::SelectPayloadLength(
     size_t max_length) {
-  if (!is_data_available_) {
-    return {kBlocked, false};
-  }
   // The stream is done if there's no more data, or if |max_length| is at least
   // as large as the remaining data.
-  const bool end_data =
-      current_fragment_.empty() || (payload_fragments_.size() == 1 &&
-                                    max_length >= current_fragment_.size());
+  const bool end_data = end_data_ && (current_fragment_.empty() ||
+                                      (payload_fragments_.size() == 1 &&
+                                       max_length >= current_fragment_.size()));
   const ssize_t length = std::min(max_length, current_fragment_.size());
   return {length, end_data};
 }
diff --git a/http2/adapter/test_utils.h b/http2/adapter/test_utils.h
index ff52eee..ff7728d 100644
--- a/http2/adapter/test_utils.h
+++ b/http2/adapter/test_utils.h
@@ -46,30 +46,26 @@
   bool is_write_blocked_ = false;
 };
 
-// A test DataFrameSource that can be initialized with a single string payload,
-// or a chunked payload.
+// A test DataFrameSource. Starts out in the empty, blocked state.
 class QUICHE_NO_EXPORT TestDataFrameSource : public DataFrameSource {
  public:
-  TestDataFrameSource(Http2VisitorInterface& visitor,
-                      absl::string_view data_payload,
-                      bool has_fin = true);
+  TestDataFrameSource(Http2VisitorInterface& visitor, bool has_fin);
 
-  TestDataFrameSource(Http2VisitorInterface& visitor,
-                      absl::Span<absl::string_view> payload_fragments,
-                      bool has_fin = true);
+  void AppendPayload(absl::string_view payload);
+  void EndData();
 
   std::pair<ssize_t, bool> SelectPayloadLength(size_t max_length) override;
   bool Send(absl::string_view frame_header, size_t payload_length) override;
   bool send_fin() const override { return has_fin_; }
 
-  void set_is_data_available(bool value) { is_data_available_ = value; }
-
  private:
   Http2VisitorInterface& visitor_;
   std::vector<std::string> payload_fragments_;
   absl::string_view current_fragment_;
+  // Whether the stream should end with the final frame of data.
   const bool has_fin_;
-  bool is_data_available_ = true;
+  // Whether |payload_fragments_| contains the final segment of data.
+  bool end_data_ = false;
 };
 
 class QUICHE_NO_EXPORT TestMetadataSource : public MetadataSource {