Modifies the Http2Adapter API to accept DataFrameSources as unique_ptrs.
This simplifies the ownership model for users of the library.
PiperOrigin-RevId: 380223157
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
