mahdium/HoolIt
1
0
Fork 0
mirror of https://github.com/mmahdium/HoolIt.git synced 2026-08-12 10:42:49 +03:30
Code Issues Packages Projects Releases Wiki Activity

Refactor SSE implementation to improve efficiency, replace ConcurrentDictionary structure, and optimize response handling.

Browse Source
This commit is contained in:
mahdium
2026-02-15 11:01:54 +03:30
parent 3f1303097e
commit 284b58299f
2 changed files with 72 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
HoolIt.sln.DotSettings.user
+2 -1
View File
@@ -1,3 +1,4 @@
<wpf:ResourceDictionary xml:space="preserve" xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml" xmlns:s="clr-namespace:System;assembly=mscorlib" xmlns:ss="urn:shemas-jetbrains-com:settings-storage-xaml" xmlns:wpf="http://schemas.microsoft.com/winfx/2006/xaml/presentation">
<s:String x:Key="/Default/CodeInspection/ExcludedFiles/FilesAndFoldersToSkip2/=7020124F_002D9FFC_002D4AC3_002D8F3D_002DAAB8E0240759_002Ff_003ADebugger_002Ecs_002Fl_003A_002E_002E_003F_002E_002E_003F_002Econfig_003FJetBrains_003FRider2025_002E3_003Fresharper_002Dhost_003FSourcesCache_003Ff9d2f95d72fa884d8b6ddefc717c56da3657fbb2d5fb683656c3589eb6587_003FDebugger_002Ecs/@EntryIndexedValue">ForceIncluded</s:String>
<s:String x:Key="/Default/CodeInspection/ExcludedFiles/FilesAndFoldersToSkip2/=7020124F_002D9FFC_002D4AC3_002D8F3D_002DAAB8E0240759_002Ff_003AHttpResponseStream_002Ecs_002Fl_003A_002E_002E_003F_002E_002E_003F_002E_002E_003F_002E_002E_003F_002E_002E_003Fhome_003Fmahdium_003F_002Econfig_003FJetBrains_003FRider2024_002E3_003Fresharper_002Dhost_003FSourcesCache_003F912bd5c687f4cf55e0daddfb3f8eecd859debac3856d3a98f1a2ad5208413bd_003FHttpResponseStream_002Ecs/@EntryIndexedValue">ForceIncluded</s:String></wpf:ResourceDictionary>
<s:String x:Key="/Default/CodeInspection/ExcludedFiles/FilesAndFoldersToSkip2/=7020124F_002D9FFC_002D4AC3_002D8F3D_002DAAB8E0240759_002Ff_003AHttpResponseStream_002Ecs_002Fl_003A_002E_002E_003F_002E_002E_003F_002E_002E_003F_002E_002E_003F_002E_002E_003Fhome_003Fmahdium_003F_002Econfig_003FJetBrains_003FRider2024_002E3_003Fresharper_002Dhost_003FSourcesCache_003F912bd5c687f4cf55e0daddfb3f8eecd859debac3856d3a98f1a2ad5208413bd_003FHttpResponseStream_002Ecs/@EntryIndexedValue">ForceIncluded</s:String>
<s:String x:Key="/Default/CodeInspection/ExcludedFiles/FilesAndFoldersToSkip2/=7020124F_002D9FFC_002D4AC3_002D8F3D_002DAAB8E0240759_002Ff_003AIHttpResponseBodyFeature_002Ecs_002Fl_003A_002E_002E_003F_002E_002E_003F_002Econfig_003FJetBrains_003FRider2025_002E3_003Fresharper_002Dhost_003FDecompilerCache_003Fdecompiler_003F0be67315c9494f8ba5a3f8abd54817d1ed10_003F40_003Fc9c44723_003FIHttpResponseBodyFeature_002Ecs/@EntryIndexedValue">ForceIncluded</s:String></wpf:ResourceDictionary>
HoolIt/Program.cs
+70 -72
View File
@@ -1,6 +1,5 @@
using System.Buffers;
using System.Collections.Concurrent;
using System.Runtime.CompilerServices;
using System.Text;
using System.Text.Json;
using System.Text.Json.Serialization;
using System.Threading.Channels;
@@ -17,97 +16,96 @@ var app = builder.Build();
app.Urls.Clear();
app.Urls.Add("http://0.0.0.0:5030");
var subscribers = new ConcurrentDictionary<string, ConcurrentDictionary<ChannelWriter<string>, byte>>();
var cancellationSources =
new ConcurrentDictionary<string, CancellationTokenSource>(); // To manage cancellation per feedId
var topicSubscribers = new ConcurrentDictionary<string, ConcurrentDictionary<Guid, ChannelWriter<byte[]>>>();
app.MapGet("/", () => Results.Redirect("https://github.com/mmahdium/HoolIt"));
// HAPI!
// https://github.com/jheising/HAPI
var createApi = app.MapGroup("/dweet/for");
createApi.MapGet("/{feedId}", async (HttpContext context, string feedId) =>
createApi.MapGet("/{feedId}", (HttpContext context, string feedId) =>
{
var queryDataDic = context.Request.Query.ToDictionary(k => k.Key, v => v.Value[0]);
var dweet = new Dweet
{
Content = queryDataDic,
Content = context.Request.Query.ToDictionary(k => k.Key, v => v.Value[0])!,
Created = DateTime.UtcNow,
Thing = feedId
};
try
var utf8Bytes = JsonSerializer.SerializeToUtf8Bytes(dweet, AppJsonSerializerContext.Default.Dweet);
if (topicSubscribers.TryGetValue(feedId, out var subscribers))
{
var jsonQueryData = JsonSerializer.Serialize(dweet, AppJsonSerializerContext.Default.Dweet);
var writers = subscribers.Values.ToArray();
if (subscribers.TryGetValue(feedId, out var feedSubs))
foreach (var writer in feedSubs.Keys)
await writer.WriteAsync(jsonQueryData);
return Results.Ok(new AddDweetSucceededResponse
{ This = "succeeded", By = "dweeting", The = "dweet", With = dweet });
foreach (var w in writers) w.TryWrite(utf8Bytes);
}
catch (Exception e)
return Results.Ok(new AddDweetSucceededResponse
{
return Results.Json(
new AddDweetFailedResponse
{
This = "failed", With = "WeMessedUp",
Because = "IDK, we couldnt dweet it. Report it at: https://github.com/mmahdium/HoolIt/issues"
}, statusCode: 500);
}
This = "succeeded",
By = "dweeting",
The = "dweet",
With = dweet
});
});
// Subscribe endpoint
var getLiveDataApi = app.MapGroup("/listen/for/dweets/from");
getLiveDataApi.MapGet("/{feedId}",
async (HttpContext context, string feedId, IHostApplicationLifetime appLifetime,
CancellationToken reqCancellationToken) =>
getLiveDataApi.MapGet("/{feedId}", async (HttpContext context, string feedId, CancellationToken reqCancellationToken) =>
{
context.Response.StatusCode = 200;
context.Response.Headers.ContentType = "text/plain; charset=utf-8";
context.Response.Headers.CacheControl = "no-cache";
context.Response.Headers["X-Content-Type-Options"] = "nosniff";
context.Features.Get<Microsoft.AspNetCore.Http.Features.IHttpResponseBodyFeature>()?
.DisableBuffering();
var channel = Channel.CreateBounded<byte[]>(new BoundedChannelOptions(512) // tune buffer count
{
context.Response.StatusCode = 200;
context.Response.Headers.ContentType = "text/plain";
context.Response.Headers.CacheControl = "no-cache";
context.Response.Headers["X-Content-Type-Options"] = "nosniff";
// Disable response buffering
context.Features.Get<Microsoft.AspNetCore.Http.Features.IHttpResponseBodyFeature>()?
.DisableBuffering();
var channel = Channel.CreateUnbounded<string>();
var writer = channel.Writer;
var feedSubs = subscribers.GetOrAdd(feedId, _ => new ConcurrentDictionary<ChannelWriter<string>, byte>());
feedSubs.TryAdd(writer, 0);
//Console.WriteLine($"Added subscriber to feed {feedId}");
try
{
var reader = channel.Reader;
while (!reqCancellationToken.IsCancellationRequested &&
await reader.WaitToReadAsync(reqCancellationToken))
while (reader.TryRead(out var msg))
{
await context.Response.WriteAsync(msg + "\n", reqCancellationToken);
await context.Response.Body.FlushAsync(reqCancellationToken);
}
}
catch (OperationCanceledException)
{
//Console.WriteLine($"Cancellation requested for feed {feedId}");
}
finally
{
if (subscribers.TryGetValue(feedId, out var list))
{
feedSubs.TryRemove(writer, out _);
//Console.WriteLine($"Removed subscriber from feed {feedId}");
if (list.Count == 0) subscribers.TryRemove(feedId, out _);
}
}
SingleReader = true,
SingleWriter = true,
FullMode = BoundedChannelFullMode.DropOldest
});
var writer = channel.Writer;
var subscribers = topicSubscribers.GetOrAdd(feedId, _ => new ConcurrentDictionary<Guid, ChannelWriter<byte[]>>());
var subscriberId = Guid.CreateVersion7(DateTimeOffset.UtcNow);
subscribers.TryAdd(subscriberId, writer);
try
{
var reader = channel.Reader;
var bodyWriter = context.Response.BodyWriter;
while (await reader.WaitToReadAsync(reqCancellationToken))
while (reader.TryRead(out var msgBytes))
{
var memory = bodyWriter.GetMemory(msgBytes.Length + 1); // +1 for newline
msgBytes.CopyTo(memory);
memory.Span[msgBytes.Length] = (byte)'\n';
bodyWriter.Advance(msgBytes.Length + 1);
await bodyWriter.FlushAsync(reqCancellationToken);
}
}
catch (OperationCanceledException)
{
// client disconnected
}
finally
{
if (topicSubscribers.TryGetValue(feedId, out var existing))
{
existing.TryRemove(subscriberId, out var removedWriter);
removedWriter?.TryComplete();
if (existing.IsEmpty) topicSubscribers.TryRemove(feedId, out _);
}
}
});
app.Run();
[JsonSerializable(typeof(Dweet))]