Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 5 additions & 5 deletions src/Directory.Packages.props
Original file line number Diff line number Diff line change
Expand Up @@ -3,9 +3,9 @@
<ManagePackageVersionsCentrally>true</ManagePackageVersionsCentrally>
</PropertyGroup>
<ItemGroup>
<PackageVersion Include="Aspire.Hosting.AppHost" Version="13.1.2" />
<PackageVersion Include="Aspire.Hosting.PostgreSQL" Version="13.1.2" />
<PackageVersion Include="Aspire.Hosting" Version="13.1.2" />
<PackageVersion Include="Aspire.Hosting.AppHost" Version="13.4.6" />
<PackageVersion Include="Aspire.Hosting.PostgreSQL" Version="13.4.6" />
<PackageVersion Include="Aspire.Hosting" Version="13.4.6" />
<PackageVersion Include="AWSSDK.SimpleNotificationService" Version="4.0.3.6" />
<PackageVersion Include="AWSSDK.SQS" Version="4.0.3.5" />
<PackageVersion Include="Azure.Identity" Version="1.13.2" />
Expand All @@ -28,7 +28,7 @@
<PackageVersion Include="JsonSchema.Net" Version="7.3.4" />
<PackageVersion Include="Marten" Version="8.37.0" />
<PackageVersion Include="McMaster.Extensions.CommandLineUtils" Version="4.0.1" />
<PackageVersion Include="MessagePack" Version="2.5.301" />
<PackageVersion Include="MessagePack" Version="2.5.302" />
<PackageVersion Include="Microsoft.Azure.Functions.Extensions" Version="1.1.0" />
<PackageVersion Include="Microsoft.Azure.Functions.Worker.Core" Version="2.0.0" />
<PackageVersion Include="Microsoft.Azure.Functions.Worker.Extensions.Abstractions" Version="1.1.0" />
Expand Down Expand Up @@ -137,7 +137,7 @@
<PackageVersion Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="10.0.0" />
<PackageVersion Include="Microsoft.Extensions.Hosting.Abstractions" Version="10.0.0" />
<PackageVersion Include="Microsoft.Extensions.Http" Version="10.0.0" />
<PackageVersion Include="Microsoft.Extensions.Logging" Version="10.0.0" />
<PackageVersion Include="Microsoft.Extensions.Logging" Version="10.0.8" />
<PackageVersion Include="Microsoft.Extensions.ObjectPool" Version="10.0.0" />
<PackageVersion Include="Microsoft.Extensions.Options" Version="10.0.0" />
<PackageVersion Include="Microsoft.Extensions.Primitives" Version="10.0.0" />
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,100 @@
using System.Threading.Channels;
using Microsoft.Extensions.Logging;

namespace HotChocolate.Fusion.Aspire;

/// <summary>
/// Schedules schema recomposition runs for a single gateway. Triggers that arrive while a run
/// is already pending are coalesced into that run, and two runs for the same gateway never
/// overlap.
/// </summary>
internal sealed class GatewayRecompositionWorker
{
private readonly Channel<bool> _triggers = Channel.CreateBounded<bool>(
new BoundedChannelOptions(1)
{
FullMode = BoundedChannelFullMode.DropWrite
});
private readonly string _gatewayName;
private readonly Func<CancellationToken, Task> _recomposeAsync;
private readonly TimeSpan _debounceDelay;
private readonly TimeProvider _timeProvider;
private readonly ILogger _logger;

public GatewayRecompositionWorker(
string gatewayName,
Func<CancellationToken, Task> recomposeAsync,
TimeSpan debounceDelay,
TimeProvider timeProvider,
ILogger logger)
{
ArgumentException.ThrowIfNullOrEmpty(gatewayName);
ArgumentNullException.ThrowIfNull(recomposeAsync);
ArgumentOutOfRangeException.ThrowIfLessThan(debounceDelay, TimeSpan.Zero);
ArgumentNullException.ThrowIfNull(timeProvider);
ArgumentNullException.ThrowIfNull(logger);

_gatewayName = gatewayName;
_recomposeAsync = recomposeAsync;
_debounceDelay = debounceDelay;
_timeProvider = timeProvider;
_logger = logger;
}

/// <summary>
/// Requests a schema recomposition for the gateway. When a recomposition is already
/// pending, the request is absorbed by the pending run.
/// </summary>
public void TriggerRecomposition() => _triggers.Writer.TryWrite(true);

/// <summary>
/// Processes recomposition requests until the token signals shutdown. A failing
/// recomposition is logged and does not terminate the worker.
/// </summary>
public async Task RunAsync(CancellationToken stoppingToken)
{
try
{
while (await _triggers.Reader.WaitToReadAsync(stoppingToken))
{
_triggers.Reader.TryRead(out _);

if (_debounceDelay > TimeSpan.Zero)
{
await Task.Delay(_debounceDelay, _timeProvider, stoppingToken);
}

// Requests that arrived while waiting are covered by the run that starts now.
_triggers.Reader.TryRead(out _);

try
{
await _recomposeAsync(stoppingToken);
}
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
{
return;
}
catch (Exception exception)
{
_logger.LogError(
exception,
"Schema recomposition failed for {ResourceName}: {Error}",
_gatewayName,
exception.Message);
}
}
}
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
{
// The application is shutting down.
}
catch (Exception exception)
{
_logger.LogError(
exception,
"The schema recomposition worker for {ResourceName} stopped unexpectedly.",
_gatewayName);
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,9 @@
<ItemGroup>
<PackageReference Include="Aspire.Hosting" />
<PackageReference Include="MessagePack" />
<PackageReference Include="Microsoft.Extensions.Logging" />
<!-- Aspire.Hosting 13.4.x requires Microsoft.Extensions.Logging >= 10.0.8, above the net9.0 central pin. -->
<PackageReference Include="Microsoft.Extensions.Logging" VersionOverride="10.0.8" Condition="'$(TargetFramework)' == 'net9.0'" />
<PackageReference Include="Microsoft.Extensions.Logging" Condition="'$(TargetFramework)' != 'net9.0'" />
</ItemGroup>

<ItemGroup>
Expand Down
Loading
Loading