zyberzero/frame-processor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

8.3 Wrap full pipeline in lock

Browse Source 
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Fritiof Hedman
2026-06-07 16:05:02 +02:00
parent e1c35d7423
commit d6ebf89468
3 changed files with 33 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
src/FrameProcessor/Controllers/FramesController.cs
View File

@@ -1,3 +1,4 @@
using FrameProcessor.Concurrency;
using FrameProcessor.Configuration;
using FrameProcessor.Domain;
using FrameProcessor.ImagePipeline;
@@ -17,19 +18,22 @@ public sealed class FramesController : ControllerBase
private readonly ImageStore _store;
private readonly MqttPublisher _mqtt;
private readonly IImageUrlFetcher _urlFetcher;
private readonly FrameLockProvider _locks;
public FramesController(
FramesRegistry frames,
IImagePipeline pipeline,
ImageStore store,
MqttPublisher mqtt,
IImageUrlFetcher urlFetcher)
IImageUrlFetcher urlFetcher,
FrameLockProvider locks)
{
_frames = frames;
_pipeline = pipeline;
_store = store;
_mqtt = mqtt;
_urlFetcher = urlFetcher;
_locks = locks;
}
[HttpPost("{name}/image")]
@@ -46,13 +50,16 @@ public sealed class FramesController : ControllerBase
return BadRequest(new { error = "Missing 'image' file part." });
}
byte[] pngBytes;
await using (var stream = file.OpenReadStream())
using (await _locks.AcquireAsync(frame.Name, cancellationToken).ConfigureAwait(false))
{
pngBytes = _pipeline.Process(stream, frame);
}
byte[] pngBytes;
await using (var stream = file.OpenReadStream())
{
pngBytes = _pipeline.Process(stream, frame);
}
return await FinishUploadAsync(frame, pngBytes, cancellationToken).ConfigureAwait(false);
return await FinishUploadAsync(frame, pngBytes, cancellationToken).ConfigureAwait(false);
}
}
[HttpPost("{name}/image-url")]
@@ -73,23 +80,26 @@ public sealed class FramesController : ControllerBase
return BadRequest(new { error = "Missing or invalid 'url'." });
}
Stream source;
try
using (await _locks.AcquireAsync(frame.Name, cancellationToken).ConfigureAwait(false))
{
source = await _urlFetcher.FetchAsync(uri, cancellationToken).ConfigureAwait(false);
}
catch (ImageFetchException ex)
{
return StatusCode(StatusCodes.Status502BadGateway, new { error = ex.Message });
}
Stream source;
try
{
source = await _urlFetcher.FetchAsync(uri, cancellationToken).ConfigureAwait(false);
}
catch (ImageFetchException ex)
{
return StatusCode(StatusCodes.Status502BadGateway, new { error = ex.Message });
}
byte[] pngBytes;
await using (source)
{
pngBytes = _pipeline.Process(source, frame);
}
byte[] pngBytes;
await using (source)
{
pngBytes = _pipeline.Process(source, frame);
}
return await FinishUploadAsync(frame, pngBytes, cancellationToken).ConfigureAwait(false);
return await FinishUploadAsync(frame, pngBytes, cancellationToken).ConfigureAwait(false);
}
}
private async Task<IActionResult> FinishUploadAsync(Frame frame, byte[] pngBytes, CancellationToken cancellationToken)

2
src/FrameProcessor/Program.cs
View File

@@ -1,3 +1,4 @@
using FrameProcessor.Concurrency;
using FrameProcessor.Configuration;
using FrameProcessor.ImagePipeline;
using FrameProcessor.Middleware;
@@ -44,6 +45,7 @@ builder.Services.AddSingleton<IImagePipeline, FrameProcessor.ImagePipeline.Image
builder.Services.AddSingleton<ImageStore>();
builder.Services.AddSingleton<MqttPublisher>();
builder.Services.AddHostedService(sp => sp.GetRequiredService<MqttPublisher>());
builder.Services.AddSingleton<FrameLockProvider>();
builder.Services.AddHttpClient<IImageUrlFetcher, ImageUrlFetcher>((sp, client) =>
{