diff --git a/http2/adapter/http2_adapter.h b/http2/adapter/http2_adapter.h
index 2c66081..32a9ba1 100644
--- a/http2/adapter/http2_adapter.h
+++ b/http2/adapter/http2_adapter.h
@@ -107,19 +107,16 @@
 
   // Returns the assigned stream ID if the operation succeeds. Otherwise,
   // returns a negative integer indicating an error code. |data_source| may be
-  // nullptr if the request does not have a body. Does not take ownership of
-  // |data_source|, which should be deleted by the caller when the stream is
-  // closed.
+  // nullptr if the request does not have a body.
   virtual int32_t SubmitRequest(absl::Span<const Header> headers,
-                                DataFrameSource* data_source,
+                                std::unique_ptr<DataFrameSource> data_source,
                                 void* user_data) = 0;
 
   // Returns 0 on success. |data_source| may be nullptr if the response does not
-  // have a body. Does not take ownership of |data_source|, which should be
-  // deleted by the caller when the stream is closed.
-  virtual int32_t SubmitResponse(Http2StreamId stream_id,
-                                 absl::Span<const Header> headers,
-                                 DataFrameSource* data_source) = 0;
+  // have a body.
+  virtual int SubmitResponse(Http2StreamId stream_id,
+                             absl::Span<const Header> headers,
+                             std::unique_ptr<DataFrameSource> data_source) = 0;
 
   // Queues trailers to be sent after any outstanding data on the stream with ID
   // |stream_id|. Returns 0 on success.
diff --git a/http2/adapter/nghttp2_adapter.cc b/http2/adapter/nghttp2_adapter.cc
index 8eac860..c6fe12a 100644
--- a/http2/adapter/nghttp2_adapter.cc
+++ b/http2/adapter/nghttp2_adapter.cc
@@ -158,37 +158,55 @@
   }
 }
 
-int32_t NgHttp2Adapter::SubmitRequest(absl::Span<const Header> headers,
-                                      DataFrameSource* data_source,
-                                      void* user_data) {
+int32_t NgHttp2Adapter::SubmitRequest(
+    absl::Span<const Header> headers,
+    std::unique_ptr<DataFrameSource> data_source, void* stream_user_data) {
   auto nvs = GetNghttp2Nvs(headers);
   std::unique_ptr<nghttp2_data_provider> provider =
-      MakeDataProvider(data_source);
-  return nghttp2_submit_request(session_->raw_ptr(), nullptr, nvs.data(),
-                                nvs.size(), provider.get(), user_data);
+      MakeDataProvider(data_source.get());
+
+  int32_t stream_id =
+      nghttp2_submit_request(session_->raw_ptr(), nullptr, nvs.data(),
+                             nvs.size(), provider.get(), stream_user_data);
+  // TODO(birenroy): clean up data source on stream close
+  sources_.emplace(stream_id, std::move(data_source));
+  QUICHE_VLOG(1) << "Submitted request with " << nvs.size()
+                 << " request headers and user data " << stream_user_data
+                 << "; resulted in stream " << stream_id;
+  return stream_id;
 }
 
-int32_t NgHttp2Adapter::SubmitResponse(Http2StreamId stream_id,
-                                       absl::Span<const Header> headers,
-                                       DataFrameSource* data_source) {
+int NgHttp2Adapter::SubmitResponse(
+    Http2StreamId stream_id, absl::Span<const Header> headers,
+    std::unique_ptr<DataFrameSource> data_source) {
   auto nvs = GetNghttp2Nvs(headers);
   std::unique_ptr<nghttp2_data_provider> provider =
-      MakeDataProvider(data_source);
-  return nghttp2_submit_response(session_->raw_ptr(), stream_id, nvs.data(),
-                                 nvs.size(), provider.get());
+      MakeDataProvider(data_source.get());
+
+  // TODO(birenroy): clean up data source on stream close
+  sources_.emplace(stream_id, std::move(data_source));
+
+  int result = nghttp2_submit_response(session_->raw_ptr(), stream_id,
+                                       nvs.data(), nvs.size(), provider.get());
+  QUICHE_VLOG(1) << "Submitted response with " << nvs.size()
+                 << " response headers; result = " << result;
+  return result;
 }
 
 int NgHttp2Adapter::SubmitTrailer(Http2StreamId stream_id,
                                   absl::Span<const Header> trailers) {
   auto nvs = GetNghttp2Nvs(trailers);
-  return nghttp2_submit_trailer(session_->raw_ptr(), stream_id, nvs.data(),
-                                nvs.size());
+  int result = nghttp2_submit_trailer(session_->raw_ptr(), stream_id,
+                                      nvs.data(), nvs.size());
+  QUICHE_VLOG(1) << "Submitted trailers with " << nvs.size()
+                 << " response trailers; result = " << result;
+  return result;
 }
 
 void NgHttp2Adapter::SetStreamUserData(Http2StreamId stream_id,
-                                       void* user_data) {
+                                       void* stream_user_data) {
   nghttp2_session_set_stream_user_data(session_->raw_ptr(), stream_id,
-                                       user_data);
+                                       stream_user_data);
 }
 
 void* NgHttp2Adapter::GetStreamUserData(Http2StreamId stream_id) {
diff --git a/http2/adapter/nghttp2_adapter.h b/http2/adapter/nghttp2_adapter.h
index c49d017..e9fa7d2 100644
--- a/http2/adapter/nghttp2_adapter.h
+++ b/http2/adapter/nghttp2_adapter.h
@@ -1,6 +1,7 @@
 #ifndef QUICHE_HTTP2_ADAPTER_NGHTTP2_ADAPTER_H_
 #define QUICHE_HTTP2_ADAPTER_NGHTTP2_ADAPTER_H_
 
+#include "absl/container/flat_hash_map.h"
 #include "http2/adapter/http2_adapter.h"
 #include "http2/adapter/http2_protocol.h"
 #include "http2/adapter/nghttp2_session.h"
@@ -65,12 +66,11 @@
                                  size_t num_bytes) override;
 
   int32_t SubmitRequest(absl::Span<const Header> headers,
-                        DataFrameSource* data_source,
+                        std::unique_ptr<DataFrameSource> data_source,
                         void* user_data) override;
 
-  int32_t SubmitResponse(Http2StreamId stream_id,
-                         absl::Span<const Header> headers,
-                         DataFrameSource* data_source) override;
+  int SubmitResponse(Http2StreamId stream_id, absl::Span<const Header> headers,
+                     std::unique_ptr<DataFrameSource> data_source) override;
 
   int SubmitTrailer(Http2StreamId stream_id,
                     absl::Span<const Header> trailers) override;
@@ -94,6 +94,8 @@
   std::unique_ptr<NgHttp2Session> session_;
   Http2VisitorInterface& visitor_;
   Perspective perspective_;
+
+  absl::flat_hash_map<int32_t, std::unique_ptr<DataFrameSource>> sources_;
 };
 
 }  // namespace adapter
diff --git a/http2/adapter/nghttp2_adapter_test.cc b/http2/adapter/nghttp2_adapter_test.cc
index 21a1ded..b0e7127 100644
--- a/http2/adapter/nghttp2_adapter_test.cc
+++ b/http2/adapter/nghttp2_adapter_test.cc
@@ -284,13 +284,13 @@
   EXPECT_FALSE(adapter->session().want_write());
   const char* kSentinel = "";
   const absl::string_view kBody = "This is an example request body.";
-  TestDataFrameSource body1(visitor, kBody);
+  auto body1 = absl::make_unique<TestDataFrameSource>(visitor, kBody);
   int stream_id =
       adapter->SubmitRequest(ToHeaders({{":method", "POST"},
                                         {":scheme", "http"},
                                         {":authority", "example.com"},
                                         {":path", "/this/is/request/one"}}),
-                             &body1, const_cast<char*>(kSentinel));
+                             std::move(body1), const_cast<char*>(kSentinel));
   EXPECT_GT(stream_id, 0);
   EXPECT_TRUE(adapter->session().want_write());
   result = adapter->Send();
@@ -389,7 +389,7 @@
                                         {":scheme", "http"},
                                         {":authority", "example.com"},
                                         {":path", "/this/is/request/one"}}),
-                             frame_source.get(), nullptr);
+                             std::move(frame_source), nullptr);
   EXPECT_GT(stream_id, 0);
   EXPECT_TRUE(adapter->session().want_write());
   result = adapter->Send();
@@ -423,7 +423,7 @@
                                         {":scheme", "http"},
                                         {":authority", "example.com"},
                                         {":path", "/this/is/request/one"}}),
-                             frame_source.get(), nullptr);
+                             std::move(frame_source), nullptr);
   EXPECT_GT(stream_id, 0);
   EXPECT_TRUE(adapter->session().want_write());
 
@@ -475,7 +475,7 @@
                                         {":scheme", "http"},
                                         {":authority", "example.com"},
                                         {":path", "/this/is/request/one"}}),
-                             frame_source.get(), nullptr);
+                             std::move(frame_source), nullptr);
   EXPECT_GT(stream_id, 0);
   EXPECT_TRUE(adapter->session().want_write());
 
@@ -664,12 +664,13 @@
 
   EXPECT_FALSE(adapter->session().want_write());
   const absl::string_view kBody = "This is an example response body.";
-  TestDataFrameSource body1(visitor, kBody, /*has_fin=*/false);
+  auto body1 =
+      absl::make_unique<TestDataFrameSource>(visitor, kBody, /*has_fin=*/false);
   int submit_result = adapter->SubmitResponse(
       1,
       ToHeaders({{":status", "404"},
                  {"x-comment", "I have no idea what you're talking about."}}),
-      &body1);
+      std::move(body1));
   EXPECT_EQ(submit_result, 0);
   EXPECT_TRUE(adapter->session().want_write());
 
@@ -783,10 +784,11 @@
 
   // The body source must indicate that the end of the body is not the end of
   // the stream.
-  TestDataFrameSource body1(visitor, kBody, /*has_fin=*/false);
+  auto body1 =
+      absl::make_unique<TestDataFrameSource>(visitor, kBody, /*has_fin=*/false);
   int submit_result = adapter->SubmitResponse(
       1, ToHeaders({{":status", "200"}, {"x-comment", "Sure, sounds good."}}),
-      &body1);
+      std::move(body1));
   EXPECT_EQ(submit_result, 0);
   EXPECT_TRUE(adapter->session().want_write());
   EXPECT_CALL(visitor, OnCloseStream(1, Http2ErrorCode::NO_ERROR));
diff --git a/http2/adapter/oghttp2_adapter.cc b/http2/adapter/oghttp2_adapter.cc
index 6f56d40..47270dd 100644
--- a/http2/adapter/oghttp2_adapter.cc
+++ b/http2/adapter/oghttp2_adapter.cc
@@ -124,16 +124,16 @@
       stream_id, TranslateErrorCode(error_code)));
 }
 
-int32_t OgHttp2Adapter::SubmitRequest(absl::Span<const Header> headers,
-                                      DataFrameSource* data_source,
-                                      void* user_data) {
-  return session_->SubmitRequest(headers, data_source, user_data);
+int32_t OgHttp2Adapter::SubmitRequest(
+    absl::Span<const Header> headers,
+    std::unique_ptr<DataFrameSource> data_source, void* user_data) {
+  return session_->SubmitRequest(headers, std::move(data_source), user_data);
 }
 
-int32_t OgHttp2Adapter::SubmitResponse(Http2StreamId stream_id,
-                                       absl::Span<const Header> headers,
-                                       DataFrameSource* data_source) {
-  return session_->SubmitResponse(stream_id, headers, data_source);
+int OgHttp2Adapter::SubmitResponse(
+    Http2StreamId stream_id, absl::Span<const Header> headers,
+    std::unique_ptr<DataFrameSource> data_source) {
+  return session_->SubmitResponse(stream_id, headers, std::move(data_source));
 }
 
 int OgHttp2Adapter::SubmitTrailer(Http2StreamId stream_id,
diff --git a/http2/adapter/oghttp2_adapter.h b/http2/adapter/oghttp2_adapter.h
index b796e1e..e2ea402 100644
--- a/http2/adapter/oghttp2_adapter.h
+++ b/http2/adapter/oghttp2_adapter.h
@@ -47,11 +47,10 @@
                                  size_t num_bytes) override;
   void SubmitRst(Http2StreamId stream_id, Http2ErrorCode error_code) override;
   int32_t SubmitRequest(absl::Span<const Header> headers,
-                        DataFrameSource* data_source,
+                        std::unique_ptr<DataFrameSource> data_source,
                         void* user_data) override;
-  int32_t SubmitResponse(Http2StreamId stream_id,
-                         absl::Span<const Header> headers,
-                         DataFrameSource* data_source) override;
+  int SubmitResponse(Http2StreamId stream_id, absl::Span<const Header> headers,
+                     std::unique_ptr<DataFrameSource> data_source) override;
 
   int SubmitTrailer(Http2StreamId stream_id,
                     absl::Span<const Header> trailers) override;
diff --git a/http2/adapter/oghttp2_session.cc b/http2/adapter/oghttp2_session.cc
index 949c67b..6d252e2 100644
--- a/http2/adapter/oghttp2_session.cc
+++ b/http2/adapter/oghttp2_session.cc
@@ -295,9 +295,9 @@
   return connection_can_write && available_window > 0;
 }
 
-int32_t OgHttp2Session::SubmitRequest(absl::Span<const Header> headers,
-                                      DataFrameSource* data_source,
-                                      void* user_data) {
+int32_t OgHttp2Session::SubmitRequest(
+    absl::Span<const Header> headers,
+    std::unique_ptr<DataFrameSource> data_source, void* user_data) {
   // TODO(birenroy): return an error for the incorrect perspective
   const Http2StreamId stream_id = next_stream_id_;
   next_stream_id_ += 2;
@@ -316,7 +316,7 @@
     QUICHE_LOG(DFATAL) << "Stream " << stream_id << " already exists!";
     return -501;  // NGHTTP2_ERR_INVALID_ARGUMENT
   }
-  iter->second.outbound_body = data_source;
+  iter->second.outbound_body = std::move(data_source);
   iter->second.user_data = user_data;
   if (data_source == nullptr) {
     frame->set_fin(true);
@@ -331,9 +331,9 @@
   return stream_id;
 }
 
-int32_t OgHttp2Session::SubmitResponse(Http2StreamId stream_id,
-                                       absl::Span<const Header> headers,
-                                       DataFrameSource* data_source) {
+int OgHttp2Session::SubmitResponse(
+    Http2StreamId stream_id, absl::Span<const Header> headers,
+    std::unique_ptr<DataFrameSource> data_source) {
   // TODO(birenroy): return an error for the incorrect perspective
   auto iter = stream_map_.find(stream_id);
   if (iter == stream_map_.end()) {
@@ -352,7 +352,7 @@
     // when calling visitor_.OnCloseStream.
   } else {
     // Add data source to stream state
-    iter->second.outbound_body = data_source;
+    iter->second.outbound_body = std::move(data_source);
     write_scheduler_.MarkStreamReady(stream_id, false);
   }
   EnqueueFrame(std::move(frame));
diff --git a/http2/adapter/oghttp2_session.h b/http2/adapter/oghttp2_session.h
index a22811f..30306fe 100644
--- a/http2/adapter/oghttp2_session.h
+++ b/http2/adapter/oghttp2_session.h
@@ -38,11 +38,10 @@
   int Send();
 
   int32_t SubmitRequest(absl::Span<const Header> headers,
-                        DataFrameSource* data_source,
+                        std::unique_ptr<DataFrameSource> data_source,
                         void* user_data);
-  int32_t SubmitResponse(Http2StreamId stream_id,
-                         absl::Span<const Header> headers,
-                         DataFrameSource* data_source);
+  int SubmitResponse(Http2StreamId stream_id, absl::Span<const Header> headers,
+                     std::unique_ptr<DataFrameSource> data_source);
   int SubmitTrailer(Http2StreamId stream_id, absl::Span<const Header> trailers);
 
   bool IsServerSession() const {
@@ -152,7 +151,7 @@
         : window_manager(stream_receive_window, std::move(listener)) {}
 
     WindowManager window_manager;
-    DataFrameSource* outbound_body = nullptr;
+    std::unique_ptr<DataFrameSource> outbound_body;
     std::unique_ptr<spdy::SpdyHeaderBlock> trailers;
     void* user_data = nullptr;
     int32_t send_window = kInitialFlowControlWindowSize;
diff --git a/http2/adapter/oghttp2_session_test.cc b/http2/adapter/oghttp2_session_test.cc
index 73e59b0..97ec7df 100644
--- a/http2/adapter/oghttp2_session_test.cc
+++ b/http2/adapter/oghttp2_session_test.cc
@@ -76,13 +76,14 @@
 
   // Submit a request to ensure the first stream is created.
   const char* kSentinel1 = "arbitrary pointer 1";
-  TestDataFrameSource body1(visitor, "This is an example request body.");
+  auto body1 = absl::make_unique<TestDataFrameSource>(
+      visitor, "This is an example request body.");
   int stream_id =
       session.SubmitRequest(ToHeaders({{":method", "POST"},
                                        {":scheme", "http"},
                                        {":authority", "example.com"},
                                        {":path", "/this/is/request/one"}}),
-                            &body1, const_cast<char*>(kSentinel1));
+                            std::move(body1), const_cast<char*>(kSentinel1));
   EXPECT_EQ(stream_id, 1);
 
   const std::string stream_frames =
@@ -225,13 +226,14 @@
   EXPECT_EQ(0, session.GetHpackEncoderDynamicTableSize());
 
   const char* kSentinel1 = "arbitrary pointer 1";
-  TestDataFrameSource body1(visitor, "This is an example request body.");
+  auto body1 = absl::make_unique<TestDataFrameSource>(
+      visitor, "This is an example request body.");
   int stream_id =
       session.SubmitRequest(ToHeaders({{":method", "POST"},
                                        {":scheme", "http"},
                                        {":authority", "example.com"},
                                        {":path", "/this/is/request/one"}}),
-                            &body1, const_cast<char*>(kSentinel1));
+                            std::move(body1), const_cast<char*>(kSentinel1));
   EXPECT_GT(stream_id, 0);
   EXPECT_TRUE(session.want_write());
   EXPECT_EQ(kSentinel1, session.GetStreamUserData(stream_id));
@@ -284,14 +286,16 @@
   EXPECT_FALSE(session.want_write());
 
   const char* kSentinel1 = "arbitrary pointer 1";
-  TestDataFrameSource body1(visitor, "This is an example request body.");
-  body1.set_is_data_available(false);
+  auto body1 = absl::make_unique<TestDataFrameSource>(
+      visitor, "This is an example request body.");
+  TestDataFrameSource* body_ref = body1.get();
+  body_ref->set_is_data_available(false);
   int stream_id =
       session.SubmitRequest(ToHeaders({{":method", "POST"},
                                        {":scheme", "http"},
                                        {":authority", "example.com"},
                                        {":path", "/this/is/request/one"}}),
-                            &body1, const_cast<char*>(kSentinel1));
+                            std::move(body1), const_cast<char*>(kSentinel1));
   EXPECT_GT(stream_id, 0);
   EXPECT_TRUE(session.want_write());
   EXPECT_EQ(kSentinel1, session.GetStreamUserData(stream_id));
@@ -307,7 +311,7 @@
   visitor.Clear();
   EXPECT_FALSE(session.want_write());
 
-  body1.set_is_data_available(true);
+  body_ref->set_is_data_available(true);
   EXPECT_TRUE(session.ResumeStream(stream_id));
   EXPECT_TRUE(session.want_write());
   result = session.Send();
@@ -329,13 +333,14 @@
   EXPECT_FALSE(session.want_write());
 
   const char* kSentinel1 = "arbitrary pointer 1";
-  TestDataFrameSource body1(visitor, "This is an example request body.");
+  auto body1 = absl::make_unique<TestDataFrameSource>(
+      visitor, "This is an example request body.");
   int stream_id =
       session.SubmitRequest(ToHeaders({{":method", "POST"},
                                        {":scheme", "http"},
                                        {":authority", "example.com"},
                                        {":path", "/this/is/request/one"}}),
-                            &body1, const_cast<char*>(kSentinel1));
+                            std::move(body1), const_cast<char*>(kSentinel1));
   EXPECT_GT(stream_id, 0);
   EXPECT_TRUE(session.want_write());
   EXPECT_EQ(kSentinel1, session.GetStreamUserData(stream_id));
@@ -580,13 +585,14 @@
   visitor.Clear();
 
   EXPECT_FALSE(session.want_write());
-  TestDataFrameSource body1(visitor, "This is an example response body.",
-                            /*has_fin=*/false);
+  auto body1 = absl::make_unique<TestDataFrameSource>(
+      visitor, "This is an example response body.",
+      /*has_fin=*/false);
   int submit_result = session.SubmitResponse(
       1,
       ToHeaders({{":status", "404"},
                  {"x-comment", "I have no idea what you're talking about."}}),
-      &body1);
+      std::move(body1));
   EXPECT_EQ(submit_result, 0);
   EXPECT_TRUE(session.want_write());
 
@@ -697,11 +703,12 @@
 
   // The body source must indicate that the end of the body is not the end of
   // the stream.
-  TestDataFrameSource body1(visitor, "This is an example response body.",
-                            /*has_fin=*/false);
+  auto body1 = absl::make_unique<TestDataFrameSource>(
+      visitor, "This is an example response body.",
+      /*has_fin=*/false);
   int submit_result = session.SubmitResponse(
       1, ToHeaders({{":status", "200"}, {"x-comment", "Sure, sounds good."}}),
-      &body1);
+      std::move(body1));
   EXPECT_EQ(submit_result, 0);
   EXPECT_TRUE(session.want_write());
   send_result = session.Send();
@@ -773,11 +780,12 @@
 
   // The body source must indicate that the end of the body is not the end of
   // the stream.
-  TestDataFrameSource body1(visitor, "This is an example response body.",
-                            /*has_fin=*/false);
+  auto body1 = absl::make_unique<TestDataFrameSource>(
+      visitor, "This is an example response body.",
+      /*has_fin=*/false);
   int submit_result = session.SubmitResponse(
       1, ToHeaders({{":status", "200"}, {"x-comment", "Sure, sounds good."}}),
-      &body1);
+      std::move(body1));
   EXPECT_EQ(submit_result, 0);
   EXPECT_TRUE(session.want_write());
   // There has not been a call to Send() yet, so neither headers nor body have