diff --git a/http2/adapter/callback_visitor.cc b/http2/adapter/callback_visitor.cc
index fa52930..2a00eed 100644
--- a/http2/adapter/callback_visitor.cc
+++ b/http2/adapter/callback_visitor.cc
@@ -263,6 +263,9 @@
         nullptr, stream_id, static_cast<uint32_t>(error_code), user_data_);
   }
   stream_map_.erase(stream_id);
+  if (stream_close_listener_) {
+    stream_close_listener_(stream_id);
+  }
 }
 
 void CallbackVisitor::OnPriorityForStream(Http2StreamId /*stream_id*/,
diff --git a/http2/adapter/callback_visitor.h b/http2/adapter/callback_visitor.h
index d38e66f..a64c8e8 100644
--- a/http2/adapter/callback_visitor.h
+++ b/http2/adapter/callback_visitor.h
@@ -3,6 +3,7 @@
 
 #include <cstdint>
 #include <memory>
+#include <utility>
 #include <vector>
 
 #include "absl/container/flat_hash_map.h"
@@ -18,6 +19,9 @@
 // data" pointer, and invokes the callbacks according to HTTP/2 events received.
 class QUICHE_EXPORT_PRIVATE CallbackVisitor : public Http2VisitorInterface {
  public:
+  // Called when the visitor receives a close event for `stream_id`.
+  using StreamCloseListener = std::function<void(Http2StreamId stream_id)>;
+
   explicit CallbackVisitor(Perspective perspective,
                            const nghttp2_session_callbacks& callbacks,
                            void* user_data);
@@ -71,6 +75,10 @@
 
   size_t stream_map_size() const { return stream_map_.size(); }
 
+  void set_stream_close_listener(StreamCloseListener stream_close_listener) {
+    stream_close_listener_ = std::move(stream_close_listener);
+  }
+
  private:
   struct QUICHE_EXPORT_PRIVATE StreamInfo {
     bool before_sent_headers = false;
@@ -89,6 +97,8 @@
 
   StreamInfoMap stream_map_;
 
+  StreamCloseListener stream_close_listener_;
+
   Perspective perspective_;
   nghttp2_session_callbacks_unique_ptr callbacks_;
   void* user_data_;
diff --git a/http2/adapter/callback_visitor_test.cc b/http2/adapter/callback_visitor_test.cc
index 8f83da5..fee6fc4 100644
--- a/http2/adapter/callback_visitor_test.cc
+++ b/http2/adapter/callback_visitor_test.cc
@@ -1,5 +1,7 @@
 #include "http2/adapter/callback_visitor.h"
 
+#include "absl/container/flat_hash_map.h"
+#include "http2/adapter/http2_protocol.h"
 #include "http2/adapter/mock_nghttp2_callbacks.h"
 #include "http2/adapter/nghttp2_test_utils.h"
 #include "http2/adapter/test_utils.h"
@@ -11,6 +13,9 @@
 namespace {
 
 using testing::_;
+using testing::IsEmpty;
+using testing::Pair;
+using testing::UnorderedElementsAre;
 
 enum FrameType {
   DATA,
@@ -78,6 +83,11 @@
   testing::StrictMock<MockNghttp2Callbacks> callbacks;
   CallbackVisitor visitor(Perspective::kClient,
                           *MockNghttp2Callbacks::GetCallbacks(), &callbacks);
+  absl::flat_hash_map<Http2StreamId, int> stream_close_counts;
+  visitor.set_stream_close_listener(
+      [&stream_close_counts](Http2StreamId stream_id) {
+        ++stream_close_counts[stream_id];
+      });
 
   testing::InSequence seq;
 
@@ -131,12 +141,15 @@
   EXPECT_CALL(callbacks, OnFrameRecv(IsHeaders(1, _, NGHTTP2_HCAT_HEADERS)));
   visitor.OnEndHeadersForStream(1);
 
+  EXPECT_THAT(stream_close_counts, IsEmpty());
+
   // RST_STREAM on stream 3
   EXPECT_CALL(callbacks, OnBeginFrame(HasFrameHeader(3, RST_STREAM, 0)));
   visitor.OnFrameHeader(3, 4, RST_STREAM, 0);
 
   // No change in stream map size.
   EXPECT_EQ(visitor.stream_map_size(), 1);
+  EXPECT_THAT(stream_close_counts, IsEmpty());
 
   EXPECT_CALL(callbacks, OnFrameRecv(IsRstStream(3, NGHTTP2_INTERNAL_ERROR)));
   visitor.OnRstStream(3, Http2ErrorCode::INTERNAL_ERROR);
@@ -144,6 +157,8 @@
   EXPECT_CALL(callbacks, OnStreamClose(3, NGHTTP2_INTERNAL_ERROR));
   visitor.OnCloseStream(3, Http2ErrorCode::INTERNAL_ERROR);
 
+  EXPECT_THAT(stream_close_counts, UnorderedElementsAre(Pair(3, 1)));
+
   // More stream close events
   EXPECT_CALL(callbacks,
               OnBeginFrame(HasFrameHeader(1, DATA, NGHTTP2_FLAG_END_STREAM)));
@@ -156,8 +171,10 @@
   EXPECT_CALL(callbacks, OnStreamClose(1, NGHTTP2_NO_ERROR));
   visitor.OnCloseStream(1, Http2ErrorCode::HTTP2_NO_ERROR);
 
-  // Stream map is empty again.
+  // Stream map is empty again after both streams were closed.
   EXPECT_EQ(visitor.stream_map_size(), 0);
+  EXPECT_THAT(stream_close_counts,
+              UnorderedElementsAre(Pair(3, 1), Pair(1, 1)));
 
   EXPECT_CALL(callbacks, OnBeginFrame(HasFrameHeader(5, RST_STREAM, _)));
   visitor.OnFrameHeader(5, 4, RST_STREAM, 0);
@@ -169,6 +186,8 @@
   visitor.OnCloseStream(5, Http2ErrorCode::REFUSED_STREAM);
 
   EXPECT_EQ(visitor.stream_map_size(), 0);
+  EXPECT_THAT(stream_close_counts,
+              UnorderedElementsAre(Pair(3, 1), Pair(1, 1), Pair(5, 1)));
 }
 
 TEST(ClientCallbackVisitorUnitTest, HeadersWithContinuation) {
diff --git a/http2/adapter/nghttp2_adapter.cc b/http2/adapter/nghttp2_adapter.cc
index a6ba0b2..c8dbc74 100644
--- a/http2/adapter/nghttp2_adapter.cc
+++ b/http2/adapter/nghttp2_adapter.cc
@@ -220,7 +220,6 @@
   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
@@ -235,7 +234,6 @@
   std::unique_ptr<nghttp2_data_provider> provider =
       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,
@@ -269,6 +267,10 @@
   return 0 == nghttp2_session_resume_data(session_->raw_ptr(), stream_id);
 }
 
+void NgHttp2Adapter::RemoveStream(Http2StreamId stream_id) {
+  sources_.erase(stream_id);
+}
+
 NgHttp2Adapter::NgHttp2Adapter(Http2VisitorInterface& visitor,
                                Perspective perspective,
                                const nghttp2_option* options)
diff --git a/http2/adapter/nghttp2_adapter.h b/http2/adapter/nghttp2_adapter.h
index 9078278..257f01a 100644
--- a/http2/adapter/nghttp2_adapter.h
+++ b/http2/adapter/nghttp2_adapter.h
@@ -88,6 +88,12 @@
 
   bool ResumeStream(Http2StreamId stream_id) override;
 
+  // Removes references to the `stream_id` from this adapter.
+  void RemoveStream(Http2StreamId stream_id);
+
+  // Accessor for testing.
+  size_t sources_size() const { return sources_.size(); }
+
  private:
   NgHttp2Adapter(Http2VisitorInterface& visitor, Perspective perspective,
                  const nghttp2_option* options);
diff --git a/http2/adapter/nghttp2_adapter_test.cc b/http2/adapter/nghttp2_adapter_test.cc
index a90e7ed..d238fd2 100644
--- a/http2/adapter/nghttp2_adapter_test.cc
+++ b/http2/adapter/nghttp2_adapter_test.cc
@@ -143,6 +143,8 @@
   adapter->SetStreamUserData(stream_id2, const_cast<char*>(kSentinel2));
   adapter->SetStreamUserData(stream_id3, nullptr);
 
+  EXPECT_EQ(adapter->sources_size(), 3);
+
   EXPECT_CALL(visitor, OnBeforeFrameSent(HEADERS, stream_id1, _, 0x5));
   EXPECT_CALL(visitor, OnFrameSent(HEADERS, stream_id1, _, 0x5, 0));
   EXPECT_CALL(visitor, OnBeforeFrameSent(HEADERS, stream_id2, _, 0x5));
@@ -205,7 +207,11 @@
   EXPECT_CALL(visitor, OnDataForStream(1, "This is the response body."));
   EXPECT_CALL(visitor, OnFrameHeader(3, 4, RST_STREAM, 0));
   EXPECT_CALL(visitor, OnRstStream(3, Http2ErrorCode::INTERNAL_ERROR));
-  EXPECT_CALL(visitor, OnCloseStream(3, Http2ErrorCode::INTERNAL_ERROR));
+  EXPECT_CALL(visitor, OnCloseStream(3, Http2ErrorCode::INTERNAL_ERROR))
+      .WillOnce(
+          [&adapter](Http2StreamId stream_id, Http2ErrorCode /*error_code*/) {
+            adapter->RemoveStream(stream_id);
+          });
   EXPECT_CALL(visitor, OnFrameHeader(0, 19, GOAWAY, 0));
   EXPECT_CALL(visitor,
               OnGoAway(5, Http2ErrorCode::ENHANCE_YOUR_CALM, "calm down!!"));
@@ -221,6 +227,9 @@
   EXPECT_EQ(kInitialFlowControlWindowSize,
             adapter->GetStreamReceiveWindowSize(stream_id3));
 
+  // One stream was closed.
+  EXPECT_EQ(adapter->sources_size(), 2);
+
   // Connection window should be the same as the first stream.
   EXPECT_EQ(adapter->GetReceiveWindowSize(),
             adapter->GetStreamReceiveWindowSize(stream_id1));
@@ -241,10 +250,18 @@
   EXPECT_CALL(visitor, OnFrameHeader(1, 0, DATA, 1));
   EXPECT_CALL(visitor, OnBeginDataForStream(1, 0));
   EXPECT_CALL(visitor, OnEndStream(1));
-  EXPECT_CALL(visitor, OnCloseStream(1, Http2ErrorCode::HTTP2_NO_ERROR));
+  EXPECT_CALL(visitor, OnCloseStream(1, Http2ErrorCode::HTTP2_NO_ERROR))
+      .WillOnce(
+          [&adapter](Http2StreamId stream_id, Http2ErrorCode /*error_code*/) {
+            adapter->RemoveStream(stream_id);
+          });
   EXPECT_CALL(visitor, OnFrameHeader(5, 4, RST_STREAM, 0));
   EXPECT_CALL(visitor, OnRstStream(5, Http2ErrorCode::REFUSED_STREAM));
-  EXPECT_CALL(visitor, OnCloseStream(5, Http2ErrorCode::REFUSED_STREAM));
+  EXPECT_CALL(visitor, OnCloseStream(5, Http2ErrorCode::REFUSED_STREAM))
+      .WillOnce(
+          [&adapter](Http2StreamId stream_id, Http2ErrorCode /*error_code*/) {
+            adapter->RemoveStream(stream_id);
+          });
   adapter->ProcessBytes(TestFrameSequence()
                             .Data(1, "", true)
                             .RstStream(5, Http2ErrorCode::REFUSED_STREAM)
@@ -256,6 +273,7 @@
   // After receiving END_STREAM for 1 and RST_STREAM for 5, the session no
   // longer expects reads.
   EXPECT_FALSE(adapter->want_read());
+  EXPECT_EQ(adapter->sources_size(), 0);
 
   // Client will not have anything else to write.
   EXPECT_FALSE(adapter->want_write());