var url = "http://localhost:9999";
var builder = WebApplication.CreateBuilder(args);
builder.WebHost
    .UseKestrel(kestrel=> kestrel.ConfigureEndpointDefaults(listen=>listen.Protocols = HttpProtocols.Http1AndHttp2AndHttp3))
    .UseUrls(url);
var app = builder.Build();
app.MapPost("/", httpContext=> HandleRequestAsync(httpContext, async (request, writer) => {
    Console.WriteLine($"[Server]Receive request message: {request}");
    await writer.WriteStringAsync(request);
}));
await app.StartAsync();

await SendStreamRequestAsync(url, ["foo", "bar", "baz", "qux"], reply => {
    Console.WriteLine($"[Client]Receive reply message: {reply}\n");
    return Task.CompletedTask;
});

static async Task HandleRequestAsync(HttpContext httpContext, Func<string, PipeWriter, Task> handler)
{
    var reader = httpContext.Request.BodyReader;
    var writer = httpContext.Response.BodyWriter;
    while (true)
    {
        var result = await reader.ReadAsync();
        var buffer = result.Buffer;
        while (TryReadMessage(ref buffer, out var message))
        {
            await handler(message, writer);
        }
        reader.AdvanceTo(buffer.Start, buffer.End);
        if (result.IsCompleted)
        {
            break;
        }
    }
}

static bool TryReadMessage(ref ReadOnlySequence<byte> buffer, [NotNullWhen(true)]out string? message)
{
    var reader = new SequenceReader<byte>(buffer);
    if (!reader.TryReadLittleEndian(out int length))
    {
        message = default;
        return false;
    }

    message = Encoding.UTF8.GetString(buffer.Slice(4, length));
    buffer = buffer.Slice(length + 4);
    return true;
}

public static class Extensions
{
    public static ValueTask<FlushResult> WriteStringAsync(this PipeWriter writer, string content)
    {
        var length = Encoding.UTF8.GetByteCount(content);
        var span = writer.GetSpan(4 + length);
        BitConverter.TryWriteBytes(span, length);
        Encoding.UTF8.GetBytes(content, span.Slice(4));
        writer.Advance(4 + length);
        return writer.FlushAsync();
    }
}

public class StreamContent(StreamContentWriter writer) : HttpContent
{
    private readonly StreamContentWriter _writer = writer;
    protected override Task SerializeToStreamAsync(Stream stream, TransportContext? context) 
        => _writer.SetOutputStream(stream).WaitAsync();
    protected override bool TryComputeLength(out long length) => (length = -1) != -1;
}

public class StreamContentWriter
{
    private readonly TaskCompletionSource<Stream> _streamSetSource = new();
    private readonly TaskCompletionSource _streamEndSource = new();
    public StreamContentWriter SetOutputStream(Stream outputStream)
    {
        _streamSetSource.SetResult(outputStream);
        return this;
    }

    public async Task WriteAsync(string content)
    {
        var stream = await _streamSetSource.Task;
        await PipeWriter.Create(stream).WriteStringAsync(content);
    }

    public void Complete() => _streamEndSource.SetResult();
    public Task WaitAsync() => _streamEndSource.Task;
}

static async Task SendStreamRequestAsync(string url,string[] lines, Func<string, Task> handler)
{
    using var httpClient = new HttpClient();
    var writer = new StreamContentWriter();
    var request = new HttpRequestMessage(HttpMethod.Post, url)
    {
        Version = HttpVersion.Version20,
        VersionPolicy = HttpVersionPolicy.RequestVersionExact,
        Content = new StreamingWeb.StreamContent(writer)
    };
    var task = httpClient.SendAsync(request, HttpCompletionOption.ResponseHeadersRead);
    _ = Task.Run(async () =>
    {
        var response = await task;
        var reader = PipeReader.Create(await response.Content.ReadAsStreamAsync());
        while (true)
        {
            var result = await reader.ReadAsync();
            var buffer = result.Buffer;
            while (TryReadMessage(ref buffer, out var message))
            {
                await handler(message);
            }
            reader.AdvanceTo(buffer.Start, buffer.End);
            if (result.IsCompleted)
            {
                break;
            }
        }
    });

    foreach (string line in lines)
    {
        await writer.WriteAsync($"{line} ({DateTimeOffset.UtcNow})");
        await Task.Delay(1000);
    }
    writer.Complete();
}