Skip to content

[Json] Avoid writing to PipeWriter if IAsyncEnumerable throws before first item #113503

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Mar 19, 2025
Merged

[Json] Avoid writing to PipeWriter if IAsyncEnumerable throws before first item #113503

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
77f6150
[Json] Avoid writing to PipeWriter if IAsyncEnumerable throws before …
BrennanConroy Mar 14, 2025
9bd54e6
update
BrennanConroy Mar 14, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion ...tem.Text.Json/src/System/Text/Json/Serialization/Metadata/JsonTypeInfoOfT.WriteHelpers.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -169,7 +169,6 @@ rootValue is not null &&
try
{
isFinalBlock = EffectiveConverter.WriteCore(writer, rootValue, Options, ref state);
writer.Flush();

if (state.SuppressFlush)
{
Expand All @@ -179,6 +178,7 @@ rootValue is not null &&
}
else
{
writer.Flush();
Copy link
Member

@eiriktsarpalis eiriktsarpalis Mar 14, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is this fixing an unrelated bug related to SuppressFlush handling?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nope, writer.Flush() calls Advance on the underlying PipeWriter which is what we're trying to avoid here. So without the move, [ would be written to the PipeWriter still before observing the exception from the IAsyncEnumerable.

FlushResult result = await pipeWriter.FlushAsync(cancellationToken).ConfigureAwait(false);
if (result.IsCanceled || result.IsCompleted)
{
Expand Down Expand Up @@ -230,6 +230,9 @@ rootValue is not null &&
}
catch
{
// Reset the writer in exception cases as we don't want the writer.Dispose() call to flush any pending bytes.
writer.Reset();
writer.Dispose();
// On exception, walk the WriteStack for any orphaned disposables and try to dispose them.
await state.DisposePendingDisposablesOnExceptionAsync().ConfigureAwait(false);
throw;
Expand Down
94 changes: 93 additions & 1 deletion ....Text.Json/tests/System.Text.Json.Tests/Serialization/JsonSerializerWrapper.Reflection.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -886,11 +886,13 @@ private int ReadExactlyFromSource(byte[] buffer, int offset, int count)
}

// TODO: Deserialize to use PipeReader overloads once implemented
private class AsyncPipelinesSerializerWrapper : JsonSerializerWrapper
private class AsyncPipelinesSerializerWrapper : StreamingJsonSerializerWrapper
{
public override JsonSerializerOptions DefaultOptions => JsonSerializerOptions.Default;
public override bool SupportsNullValueOnDeserialize => true;

public override bool IsAsyncSerializer => true;

public override async Task<T> DeserializeWrapper<T>(string json, JsonSerializerOptions options = null)
{
return await JsonSerializer.DeserializeAsync<T>(new MemoryStream(Encoding.UTF8.GetBytes(json)), options);
Expand All @@ -915,6 +917,31 @@ public override async Task<object> DeserializeWrapper(string json, Type type, Js
return await JsonSerializer.DeserializeAsync(new MemoryStream(Encoding.UTF8.GetBytes(json)), type, context);
}

public override Task<object> DeserializeWrapper(Stream utf8Json, Type returnType, JsonSerializerOptions? options = null)
{
return JsonSerializer.DeserializeAsync(utf8Json, returnType, options).AsTask();
}

public override Task<T> DeserializeWrapper<T>(Stream utf8Json, JsonSerializerOptions? options = null)
{
return JsonSerializer.DeserializeAsync<T>(utf8Json, options).AsTask();
}

public override Task<object> DeserializeWrapper(Stream utf8Json, Type returnType, JsonSerializerContext context)
{
return JsonSerializer.DeserializeAsync(utf8Json, returnType, context).AsTask();
}

public override Task<T> DeserializeWrapper<T>(Stream utf8Json, JsonTypeInfo<T> jsonTypeInfo)
{
return JsonSerializer.DeserializeAsync(utf8Json, jsonTypeInfo).AsTask();
}

public override Task<object> DeserializeWrapper(Stream utf8Json, JsonTypeInfo jsonTypeInfo)
{
return JsonSerializer.DeserializeAsync(utf8Json, jsonTypeInfo).AsTask();
}

public override async Task<string> SerializeWrapper(object value, Type inputType, JsonSerializerOptions options = null)
{
Pipe pipe = new Pipe();
Expand Down Expand Up @@ -969,6 +996,71 @@ public override async Task<string> SerializeWrapper(object value, JsonTypeInfo j
pipe.Reader.AdvanceTo(result.Buffer.End);
return stringResult;
}

public override async Task SerializeWrapper(Stream stream, object value, Type inputType, JsonSerializerOptions? options = null)
{
var writer = PipeWriter.Create(stream);
try
{
await JsonSerializer.SerializeAsync(writer, value, inputType, options);
}
finally
{
await writer.FlushAsync();
}
}

public override async Task SerializeWrapper<T>(Stream stream, T value, JsonSerializerOptions? options = null)
{
var writer = PipeWriter.Create(stream);
try
{
await JsonSerializer.SerializeAsync(writer, value, options);
}
finally
{
await writer.FlushAsync();
}
}

public override async Task SerializeWrapper(Stream stream, object value, Type inputType, JsonSerializerContext context)
{
var writer = PipeWriter.Create(stream);
try
{
await JsonSerializer.SerializeAsync(writer, value, inputType, context);
}
finally
{
await writer.FlushAsync();
}
}

public override async Task SerializeWrapper<T>(Stream stream, T value, JsonTypeInfo<T> jsonTypeInfo)
{
var writer = PipeWriter.Create(stream);
try
{
await JsonSerializer.SerializeAsync(writer, value, jsonTypeInfo);
}
finally
{
await writer.FlushAsync();
}
}

public override async Task SerializeWrapper(Stream stream, object value, JsonTypeInfo jsonTypeInfo)
{
var writer = PipeWriter.Create(stream);
try
{
await JsonSerializer.SerializeAsync(writer, value, jsonTypeInfo);
}
finally
{
await writer.FlushAsync();
}
}
}
}
}
Loading