Minimal api stream sample

aspnetcore/fundamentals/minimal-apis/bindStreamPipeReader/7.0-samples/PipeStreamToBackgroundQueue/BackgroundQueueService.cs

@@ -0,0 +1,40 @@
+using System.Text.Json;
+using System.Threading.Channels;
+
+namespace BackgroundQueueService;
+
+class BackgroundQueue : BackgroundService
+{
+    private readonly Channel<ReadOnlyMemory<byte>> _queue;
+    private readonly ILogger<BackgroundQueue> _logger;
+
+    public BackgroundQueue(Channel<ReadOnlyMemory<byte>> queue, ILogger<BackgroundQueue> logger)
+    {
+        _queue = queue;
+        _logger = logger;
+    }
+
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        await foreach (var dataStream in _queue.Reader.ReadAllAsync(stoppingToken))
+        {
+            try
+            {    
+                var person = JsonSerializer.Deserialize<Person>(dataStream.Span)!;
+                 _logger.LogInformation($"{person.Name} is {person.Age} years and from {person.Country}");
+                // you could do something else with the data
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError(ex.Message);
+            }
+        }
+    }
+}
+
+class Person
+{
+    public string Name { get; set; } = String.Empty;
+    public int Age { get; set; }
+    public string Country { get; set; } = String.Empty;
+}

aspnetcore/fundamentals/minimal-apis/bindStreamPipeReader/7.0-samples/PipeStreamToBackgroundQueue/PipeStreamToBackgroundQueue.csproj

@@ -0,0 +1,9 @@
+<Project Sdk="Microsoft.NET.Sdk.Web">
+
+  <PropertyGroup>
+    <TargetFramework>net7.0</TargetFramework>
+    <Nullable>enable</Nullable>
+    <ImplicitUsings>enable</ImplicitUsings>
+  </PropertyGroup>
+
+</Project>

aspnetcore/fundamentals/minimal-apis/bindStreamPipeReader/7.0-samples/PipeStreamToBackgroundQueue/Program.cs

@@ -0,0 +1,59 @@
+using System.Threading.Channels;
+using BackgroundQueueService;
+
+var builder = WebApplication.CreateBuilder(args);
+// The max memory we're willing to use for the upload endpoint on this instance
+var maxMemory = 500 * 1024 * 1024;
+
+// The max size of a single message, we're staying below the default LOH size of 85K
+var maxMessageSize = 80 * 1024;
+
+// The max size of the queue based on those restrictions
+var maxQueueSize = maxMemory / maxMessageSize;
+
+// Create a channel to send data to the background queue.
+builder.Services.AddSingleton<Channel<ReadOnlyMemory<byte>>>((_) => Channel.CreateBounded<ReadOnlyMemory<byte>>(maxQueueSize));
+
+// Create a background queue service.
+builder.Services.AddHostedService<BackgroundQueue>();
+var app = builder.Build();
+
+// curl --request POST 'http://localhost:5256/register' --header 'Content-Type: application/json' --data-raw '{ "Name":"Samson", "Age": 23, "Country":"Nigeria" }'
+app.MapPost("/register", async (HttpRequest req, Stream body, Channel<ReadOnlyMemory<byte>> queue) =>
+{
+    if (req.ContentLength is not null && req.ContentLength > maxMessageSize)
+    {
+        // Message size exceeded
+        return Results.BadRequest();
+    }
+
+    // We're not above the message size and we have a content length, or
+    // we're a chunked request and we're going to read up to the maxMessageSize + 1. 
+    // We add one to the message size so that we can detect when a chunked request body
+    // is bigger than our configured max.
+    var readSize = (int?)req.ContentLength ?? (maxMessageSize + 1);
+
+    var buffer = new byte[readSize];
+
+    // Read at least that many bytes from the body
+    var read = await body.ReadAtLeastAsync(buffer, readSize, throwOnEndOfStream: false);
+
+    // We read more than the max, so this is a bad request
+    if (read > maxMessageSize)
+    {
+        // Message size exceeded
+        return Results.BadRequest();
+    }
+
+    // Attempt to send the buffer to the background queue.
+    if (queue.Writer.TryWrite(buffer.AsMemory(0..read)))
+    {
+        return Results.Accepted();
+    }
+
+    // We couldn't accept the message since we're overloaded
+    return Results.StatusCode(StatusCodes.Status429TooManyRequests);
+});
+
+
+app.Run();

aspnetcore/fundamentals/minimal-apis/bindStreamPipeReader/7.0-samples/PipeStreamToBackgroundQueue/appsettings.Development.json

@@ -0,0 +1,8 @@
+{
+  "Logging": {
+    "LogLevel": {
+      "Default": "Information",
+      "Microsoft.AspNetCore": "Warning"
+    }
+  }
+}

aspnetcore/fundamentals/minimal-apis/bindStreamPipeReader/7.0-samples/PipeStreamToBackgroundQueue/appsettings.json

@@ -0,0 +1,9 @@
+{
+  "Logging": {
+    "LogLevel": {
+      "Default": "Information",
+      "Microsoft.AspNetCore": "Warning"
+    }
+  },
+  "AllowedHosts": "*"
+}