Adds a method to the Http2Adapter API to resume streams which were deferred due to being read-blocked. PiperOrigin-RevId: 378697186

diff --git a/http2/adapter/http2_adapter.h b/http2/adapter/http2_adapter.h
index fc64c0b..2dd7d3f 100644
--- a/http2/adapter/http2_adapter.h
+++ b/http2/adapter/http2_adapter.h

@@ -111,6 +111,11 @@
   // not been set.
   virtual void* GetStreamUserData(Http2StreamId stream_id) = 0;
 
+  // Resumes a stream that was previously blocked (for example, due to
+  // DataFrameSource::SelectPayloadLength() returning kBlocked). Returns true if
+  // the stream was successfully resumed.
+  virtual bool ResumeStream(Http2StreamId stream_id) = 0;
+
  protected:
   // Subclasses should expose a public factory method for constructing and
   // initializing (via Initialize()) adapter instances.

diff --git a/http2/adapter/nghttp2_adapter.cc b/http2/adapter/nghttp2_adapter.cc
index 31b215c..6ec2a14 100644
--- a/http2/adapter/nghttp2_adapter.cc
+++ b/http2/adapter/nghttp2_adapter.cc

@@ -167,6 +167,10 @@
   return nghttp2_session_get_stream_user_data(session_->raw_ptr(), stream_id);
 }
 
+bool NgHttp2Adapter::ResumeStream(Http2StreamId stream_id) {
+  return 0 == nghttp2_session_resume_data(session_->raw_ptr(), stream_id);
+}
+
 NgHttp2Adapter::NgHttp2Adapter(Http2VisitorInterface& visitor,
                                Perspective perspective)
     : Http2Adapter(visitor), visitor_(visitor), perspective_(perspective) {}

diff --git a/http2/adapter/nghttp2_adapter.h b/http2/adapter/nghttp2_adapter.h
index b8dfb3b..a3f32df 100644
--- a/http2/adapter/nghttp2_adapter.h
+++ b/http2/adapter/nghttp2_adapter.h

@@ -70,6 +70,8 @@
   void SetStreamUserData(Http2StreamId stream_id, void* user_data) override;
   void* GetStreamUserData(Http2StreamId stream_id) override;
 
+  bool ResumeStream(Http2StreamId stream_id) override;
+
   // TODO(b/181586191): Temporary accessor until equivalent functionality is
   // available in this adapter class.
   NgHttp2Session& session() { return *session_; }

diff --git a/http2/adapter/nghttp2_adapter_test.cc b/http2/adapter/nghttp2_adapter_test.cc
index d930a26..070ed68 100644
--- a/http2/adapter/nghttp2_adapter_test.cc
+++ b/http2/adapter/nghttp2_adapter_test.cc

@@ -350,12 +350,16 @@
 
   // Resume the deferred stream.
   body1.set_is_data_available(true);
-  nghttp2_session_resume_data(adapter->session().raw_ptr(), stream_id);
+  EXPECT_TRUE(adapter->ResumeStream(stream_id));
   EXPECT_TRUE(adapter->session().want_write());
 
   adapter->Send();
   EXPECT_THAT(visitor.data(), EqualsFrames({spdy::SpdyFrameType::DATA}));
   EXPECT_FALSE(adapter->session().want_write());
+
+  // Stream data is done, so this stream cannot be resumed.
+  EXPECT_FALSE(adapter->ResumeStream(stream_id));
+  EXPECT_FALSE(adapter->session().want_write());
 }
 
 // This test verifies how nghttp2 behaves when a connection becomes

diff --git a/http2/adapter/oghttp2_adapter.cc b/http2/adapter/oghttp2_adapter.cc
index 1ba8d37..9f57aec 100644
--- a/http2/adapter/oghttp2_adapter.cc
+++ b/http2/adapter/oghttp2_adapter.cc

@@ -128,6 +128,10 @@
   return session_->GetStreamUserData(stream_id);
 }
 
+bool OgHttp2Adapter::ResumeStream(Http2StreamId stream_id) {
+  return session_->ResumeStream(stream_id);
+}
+
 const Http2Session& OgHttp2Adapter::session() const {
   return *session_;
 }

diff --git a/http2/adapter/oghttp2_adapter.h b/http2/adapter/oghttp2_adapter.h
index 7ef07f6..bbdd2a4 100644
--- a/http2/adapter/oghttp2_adapter.h
+++ b/http2/adapter/oghttp2_adapter.h

@@ -52,6 +52,7 @@
 
   void SetStreamUserData(Http2StreamId stream_id, void* user_data) override;
   void* GetStreamUserData(Http2StreamId stream_id) override;
+  bool ResumeStream(Http2StreamId stream_id) override;
 
   const Http2Session& session() const;
 

diff --git a/http2/adapter/oghttp2_session.cc b/http2/adapter/oghttp2_session.cc
index bb8405f..073bf1c 100644
--- a/http2/adapter/oghttp2_session.cc
+++ b/http2/adapter/oghttp2_session.cc

@@ -49,6 +49,16 @@
   return nullptr;
 }
 
+bool OgHttp2Session::ResumeStream(Http2StreamId stream_id) {
+  if (auto it = stream_map_.find(stream_id);
+      it->second.outbound_body == nullptr ||
+      !write_scheduler_.StreamRegistered(stream_id)) {
+    return false;
+  }
+  write_scheduler_.MarkStreamReady(stream_id, /*add_to_front=*/false);
+  return true;
+}
+
 ssize_t OgHttp2Session::ProcessBytes(absl::string_view bytes) {
   ssize_t preface_consumed = 0;
   if (!remaining_preface_.empty()) {

diff --git a/http2/adapter/oghttp2_session.h b/http2/adapter/oghttp2_session.h
index c5b9efc..329d2a9 100644
--- a/http2/adapter/oghttp2_session.h
+++ b/http2/adapter/oghttp2_session.h

@@ -54,6 +54,9 @@
   void SetStreamUserData(Http2StreamId stream_id, void* user_data);
   void* GetStreamUserData(Http2StreamId stream_id);
 
+  // Resumes a stream that was previously blocked. Returns true on success.
+  bool ResumeStream(Http2StreamId stream_id);
+
   // From Http2Session.
   ssize_t ProcessBytes(absl::string_view bytes) override;
   int Consume(Http2StreamId stream_id, size_t num_bytes) override;

diff --git a/http2/adapter/oghttp2_session_test.cc b/http2/adapter/oghttp2_session_test.cc
index 4b42188..ca0b0f6 100644
--- a/http2/adapter/oghttp2_session_test.cc
+++ b/http2/adapter/oghttp2_session_test.cc

@@ -252,8 +252,16 @@
   visitor.Clear();
   EXPECT_FALSE(session.want_write());
 
-  // Currently there is no way to indicate that the first stream is no longer
-  // read blocked.
+  body1.set_is_data_available(true);
+  EXPECT_TRUE(session.ResumeStream(stream_id));
+  EXPECT_TRUE(session.want_write());
+  session.Send();
+  EXPECT_THAT(visitor.data(), EqualsFrames({SpdyFrameType::DATA}));
+  EXPECT_FALSE(session.want_write());
+
+  // Stream data is done, so this stream cannot be resumed.
+  EXPECT_FALSE(session.ResumeStream(stream_id));
+  EXPECT_FALSE(session.want_write());
 }
 
 // This test exercises the case where the connection to the peer is write