[API Proposal]: Add formatting support to System.Net.ServerSentEvents

[eiriktsarpalis]

Background and motivation

The System.Net.ServerSentEvents library added support for parsing SSE events. We should also add support for formatting SSE events on the server side, in a way that unlocks work for dotnet/aspnetcore#56172.

API Proposal

namespace System.Net.ServerSentEvents;

// Extends the existing type to support writing
public readonly struct SseItem<T>
{
-   public SseItem(T data, string eventType);
+   public SseItem(T data, string? eventType);

    public T Data { get; }

-   public string EventType { get; }
+   public string? EventType { get; } 

+   public string? EventId { get; init; }
}

public delegate T SseItemParser<out T>(string eventType, ReadOnlySpan<byte> data);
+public delegate byte[] SseItemFormatter<T>(T value);

+public sealed class SseFormatter
+{
+    public static SseFormatter Create<T>(IAsyncEnumerable<SseItem<string>> source);
+    public static SseFormatter Create<T>(IAsyncEnumerable<SseItem<T>> source, SseItemFormatter<T> itemFormatter);
+
+    public Task WriteToStreamAsync(Stream stream, CancellationToken cancellationToken = default);
+    public void WriteToStream(Stream stream, CancellationToken cancellationToken = default);
+}

API Usage

IAsyncEnumerable<SseItem<int>> source = ...;
SseFormatter formatter = SseFormatter.Create<int>(source, value => JsonSerializer.SerializeToBytes(value));
await formatter.WriteToStreamAsync(targetStream, cancellationToken);

Alternative Designs

The current SseItemFormatter delegate uses an allocating design, but which is simpler to use. Alternative designs would involve writing either to a Stream:

public delegate ValueTask SseItemFormatter<T>(Stream stream, T value);

But this would require wrapping the underlying stream in order to intercept potential line breaks. Alternatively, we could use an IBufferWriter<byte> which also composes better with PipeWriter targets:

public delegate void SseItemFormatter<T>(IBufferWriter<byte> writer, T value);

The downside of either approach is that it becomes harder to define formatters from the user's perspective:

SseFormatter formatter = SseFormatter.Create<int>(source, (bufferWriter, value) => 
{ 
    var writer = new Utf8JsonWriter(bufferWriter);
    return JsonSerializer.SerializeToBytes(bufferWriter, value); 
});

Risks

No response

[eiriktsarpalis]

cc @captainsafia @stephentoub

[stephentoub]

• public string EventType { get; }

• public string? EventType { get; }

This will introduce warnings into folks' code. Why is it necessary?

[stephentoub]

• public static SseFormatter Create(IAsyncEnumerable<SseItem> source, SseItemFormatter itemFormatter);

Seems like this design forces a byte[] allocation for every item. What about writing it directly to the destination stream?

[captainsafia]

I noticed that our SseItem struct doesn't define a retry field. Should it since the underlying spec supports it as an optional property?

This will introduce warnings into folks' code. Why is it necessary?

It seems like according to the spec, the event type is actually optional. We may introduce new warnings here but it strikes me as valid given the it's possible for the underlying value to be null. We're a lot more permissive about making non-nullable properties nullable in ASP.NET Core reviews so I tend to be more relax about introducing new warnings if they actually make sense. 😅

[eiriktsarpalis]

Seems like this design forces a byte[] allocation for every item.

I initially experimented with a signature matching that of IUtf8SpanFormattable but realized that it doesn't compose well with how most serializers work. A typical implementation would have to allocate its own byte[] which it then needs to compare against the destination buffer. If it doesn't fit, it would need to be rerun up until the point a matching destination has been provided.

What about writing it directly to the destination stream?

We need a way to intercept line breaks. Maybe it's possible to do so by wrapping the underlying stream but I haven't tested this out. Worth a try you think?

public delegate ValueTask SseItemFormatter<T>(Stream stream, T value);

But not sure how well this would compose with a hypothetical future API that targets PipeWriter. Perhaps using IBufferWriter<byte> might be preferable here:

public delegate void SseItemFormatter<T>(IBufferWriter<byte> writer, T value);

[MihaZupan]

cc: @dotnet/ncl since the bot was lazy

[stephentoub]

It seems like according to the spec, the event type is actually optional

It is optional, and when it's not supplied, implementations are supposed to infer "message". So we exposed SseParser.EventTypeDefault, so that if it's not there in the wire data, the .NET representation of it can provide it explicitly.

If we want to make the ctor parameter optional and have the constructor supply "message" when the event type is null, I'm fine with that. But making the property be nullable doesn't seem warranted; it's just introducing friction. What's the case where someone is going to behave differently based on whether EventType returns null or returns "message"? Presumably we actually make it harder for consumers with this being nullable, as someone wanting to handle the default case needs to check for both null and "message".

[eiriktsarpalis]

Presumably we actually make it harder for consumers with this being nullable, as someone wanting to handle the default case needs to check for both null and "message".

The downside of keeping it non-nullable is that we'd mandating event types to be written on the formatter side. Unless of course we use a separate type to model written events or decide that "message" is the magic string that results in event types being omitted.

[stephentoub]

The downside of keeping it non-nullable is that we'd mandating event types to be written on the formatter side.

Why? If it returns "message", the formatter can choose to elide it. That's true whether or not the property can also return null.

[eiriktsarpalis]

Why? If it returns "message", the formatter can choose to elide it. That's true whether or not the property can also return null.

So we make "message" a magic string?

[stephentoub]

So we make "message" a magic string?

The spec makes it a magic string.

[bartonjs]

Video

• Removed the custom delegate type in favor of Action
• SseItem.EventType will stay non-nullable even though the ctor now accepts null
• Replaced instance SseFormatter with a static class.

namespace System.Net.ServerSentEvents;

public readonly struct SseItem<T>
{
-   public SseItem(T data, string eventType);
+   public SseItem(T data, string? eventType);

namespace System.Net.ServerSentEvents;

// Extends the existing type to support writing
public readonly partial struct SseItem<T>
{
    public string? EventId { get; init; }
}

public static class SseFormatter
{
    public static Task WriteAsync(IAsyncEnumerable<SseItem<string>> source, Stream destination, CancellationToken cancellationToken = default);
    public static Task WriteAsync(IAsyncEnumerable<SseItem<T>> source, Stream destination, Action<T, IBufferWriter<byte>> itemFormatter, CancellationToken cancellationToken = default);
}

[eiriktsarpalis]

I want to propose the following amendments to the API as approved above:

1. The itemFormatter delegate should accept the full SseItem<T> envelope and not just T. This is consistent with the parsing delegate which accepts the event type as input.
2. During review, we stopped short of including an SseItem<T> property representing the retry field in the specification. We already expose the aggregate value in SseParser<T> via the ReconnectionInterval property so I propose we use the identical name and type for the field.

Thus, the amended design looks as follows:

namespace System.Net.ServerSentEvents;

// Extends the existing type to support writing
public readonly partial struct SseItem<T>
{
    public string? EventId { get; init; }
+    public TimeSpan? ReconnectionInterval { get; init; }
}

public static class SseFormatter
{
    public static Task WriteAsync(IAsyncEnumerable<SseItem<string>> source, Stream destination, CancellationToken cancellationToken = default);
-    public static Task WriteAsync(IAsyncEnumerable<SseItem<T>> source, Stream destination, Action<T, IBufferWriter<byte>> itemFormatter, CancellationToken cancellationToken = default);
+   public static Task WriteAsync(IAsyncEnumerable<SseItem<T>> source, Stream destination, Action<SseItem<T>, IBufferWriter<byte>> itemFormatter, CancellationToken cancellationToken = default); 
}