Implement new streaming APIs for the System.Net.Http.Json extensions

[IEvangelist]

Implements the proposed/approved API described in #87577.

• Adds two new partial class impls:
  • HttpClientJsonExtensions.Get.AsyncEnumerable.cs
  • HttpContentJsonExtensions.AsyncEnumerable.cs
• Adds unit tests
• Adds APIs to ref

Fixes #87577, @eiriktsarpalis

[dotnet-issue-labeler]

Note regarding the new-api-needs-documentation label:

This serves as a reminder for when your PR is modifying a ref *.cs file and adding/modifying public APIs, please make sure the API implementation in the src *.cs file is documented with triple slash comments, so the PR reviewers can sign off that change.

[ghost]

Tagging subscribers to this area: @dotnet/area-system-text-json, @gregsdennis
See info in area-owners.md if you want to be subscribed.

Issue Details

---

Implements the proposed/approved API described in #87577.

• Adds two new partial class impls:
  • HttpClientJsonExtensions.Get.AsyncEnumerable.cs
  • HttpContentJsonExtensions.AsyncEnumerable.cs
• Adds unit tests
• Adds APIs to ref

Fixes #87577, @eiriktsarpalis

Author:	IEvangelist
Assignees:	-
Labels:	area-System.Text.Json, new-api-needs-documentation
Milestone:	-

[ghost]

Tagging subscribers to this area: @dotnet/ncl
See info in area-owners.md if you want to be subscribed.

Issue Details

---

Implements the proposed/approved API described in #87577.

• Adds two new partial class impls:
  • HttpClientJsonExtensions.Get.AsyncEnumerable.cs
  • HttpContentJsonExtensions.AsyncEnumerable.cs
• Adds unit tests
• Adds APIs to ref

Fixes #87577, @eiriktsarpalis

Author:	IEvangelist
Assignees:	IEvangelist
Labels:	area-System.Net.Http, area-System.Text.Json, new-api-needs-documentation
Milestone:	-

[MihaZupan]

HttpClient has two main limits that apply to such requests: Timeout and MaxResponseHeadersLength.

With the existing GetFromJsonAsync helpers, Timeout applies to both the initial request and the content transfer.
This PR applies the Timeout only to the initial request, while the content streaming through the IAsyncEnumerable does not have a timeout.
I think this is in line with what the caller would expect from this API, so 👍.

But if the Timeout only applies to the initial request, you don't have to deal with the Timeout or CancellationTokenSource manually -- HttpClient will take care of that for you. This part of the logic can be simplified away.

---

The second limit is MaxResponseContentBufferSize.

With GetFromJsonAsync this limits how much data we will read and try to deserialize.
But the current state of this PR will enforce the limit only if the Content-Length header is present. For such an API that implies data will be generated over a longer period of time, this will not be the case for most requests.

How I would personally expect such an API to behave is either:
a) The limit is enforced on a per-element basis. That is it limits how much data was buffered in memory at any given point.
b) The limit is enforced on the entire transfer (all elements).
c) The limit does not apply at all.

Where I strongly prefer a) as it gives the user the option to set meaningful limits while still having the ability to stream lots of data over a longer period of time.
Implementation-wise, this can be as simple as wrapping the content stream in the existing LengthLimitReadStream helper, and resetting the count after every yield.

[eiriktsarpalis]

a) The limit is enforced on a per-element basis. That is it limits how much data was buffered in memory at any given point.

I don't think such an enforcement is necessary. Because of how async deserialization works in STJ, deserializing a single element does not necessitate buffering all the relevant data in memory. I would strongly recommend going for option (b) since that would be consistent with what would happen if you attempted to deserialize the same data using a regular GetFromJson<T[]> method call.

[IEvangelist]

a) The limit is enforced on a per-element basis. That is it limits how much data was buffered in memory at any given point.

I don't think such an enforcement is necessary. Because of how async deserialization works in STJ, deserializing a single element does not necessitate buffering all the relevant data in memory. I would strongly recommend going for option (b) since that would be consistent with what would happen if you attempted to deserialize the same data using a regular GetFromJson<T[]> method call.

Addressed in c99e708

[MihaZupan]

deserializing a single element does not necessitate buffering all the relevant data in memory

But presumably, it does consume memory on the same order of magnitude to store the newly allocated object?

option (b) since that would be consistent with what would happen if you attempted to deserialize the same data using a regular GetFromJson<T[]> method call.

I might be misunderstanding the intended use of this API, but is it not meant exactly for cases where the server doesn't have all the data available immediately, and you want to consume elements as they are being produced? For example, I could connect to a server and stream state updates for the next day, potentially going over the limit? Just setting the limit really high is not the same, as I don't want any single update to OOM.

[eiriktsarpalis]

cc @jeffhandley @ericstj for your consideration since this is adding new functionality for RC.

[eiriktsarpalis]

deserializing a single element does not necessitate buffering all the relevant data in memory

But presumably, it does consume memory on the same order of magnitude to store the newly allocated object?

It does store a state machine containing partially hydrated objects, but that wouldn't be contributing to overheads that are substantially larger than the final deserialized result.

I might be misunderstanding the intended use of this API, but is it not meant exactly for cases where the server doesn't have all the data available immediately, and you want to consume elements as they are being produced?

That's correct, I meant to say that it should be consistent with the GetFromJson method insofar as what the effect of MaxResponseContentBufferSize should be.

[MihaZupan]

@IEvangelist mind reverting c99e708 in that case and only removing the ResetCount part, not the LengthLimitReadStream entirely?

[IEvangelist]

@IEvangelist mind reverting c99e708 in that case and only removing the ResetCount part, not the LengthLimitReadStream entirely?

So, the content stream will flow into the new LengthLimitReadStream and the resulting stream is then passed to the JsonSerializer, right? Are you good with that @eiriktsarpalis?

[eiriktsarpalis]

Yes, the method should build the same stream in the same way as the existing core method:

runtime/src/libraries/System.Net.Http.Json/src/System/Net/Http/Json/HttpClientJsonExtensions.cs

Lines 73 to 104 in b06b325

	static async Task<TValue?> Core(
	HttpClient client,
	Task<HttpResponseMessage> responseTask,
	bool usingResponseHeadersRead,
	CancellationTokenSource? linkedCTS,
	Func<Stream, TJsonOptions, CancellationToken, ValueTask<TValue?>> deserializeMethod,
	TJsonOptions jsonOptions,
	CancellationToken cancellationToken)
	{
	try
	{
	using HttpResponseMessage response = await responseTask.ConfigureAwait(false);
	response.EnsureSuccessStatusCode();
	Debug.Assert(client.MaxResponseContentBufferSize is > 0 and <= int.MaxValue);
	int contentLengthLimit = (int)client.MaxResponseContentBufferSize;
	if (response.Content.Headers.ContentLength is long contentLength && contentLength > contentLengthLimit)
	{
	LengthLimitReadStream.ThrowExceededBufferLimit(contentLengthLimit);
	}
	try
	{
	using Stream contentStream = await HttpContentJsonExtensions.GetContentStreamAsync(response.Content, linkedCTS?.Token ?? cancellationToken).ConfigureAwait(false);
	// If ResponseHeadersRead wasn't used, HttpClient will have already buffered the whole response upfront. No need to check the limit again.
	Stream readStream = usingResponseHeadersRead
	? new LengthLimitReadStream(contentStream, (int)client.MaxResponseContentBufferSize)
	: contentStream;
	return await deserializeMethod(readStream, jsonOptions, linkedCTS?.Token ?? cancellationToken).ConfigureAwait(false);

It might make sense to extract some of that logic in a common helper, if possible.

[ericstj]

@jeffhandley @ericstj for your consideration since this is adding new functionality for RC.

Thanks for the heads up - we discussed this one and are OK with the additions given the low risk / need for feedback on this API.