Skip to content

Commit 3058527

Browse files
lucaslorentzlucaslorentz
committed
Add nats
1 parent 40bf025 commit 3058527

7 files changed

+130
-17
lines changed

‎docker-compose.yml

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,13 @@ services:
4949
timeout: 5s
5050
retries: 5
5151

52+
nats:
53+
image: nats
54+
ports:
55+
- 4222:4222
56+
- 6222:6222
57+
- 8222:8222
58+
5259
volumes:
5360
mysql:
5461
sqlserver:

‎src/LLL.DurableTask.EFCore/EFCoreOrchestrationOptions.cs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,6 @@ public class EFCoreOrchestrationOptions
1616
public TimeSpan OrchestrationLockTimeout { get; set; } = TimeSpan.FromMinutes(1);
1717
public TimeSpan ActivtyLockTimeout { get; set; } = TimeSpan.FromMinutes(1);
1818
public TimeSpan FetchNewMessagesPollingTimeout { get; set; } = TimeSpan.FromSeconds(10);
19-
public int DelayInSecondsAfterFailure { get; set; } = 5;
19+
public int DelayInSecondsAfterFailure { get; set; } = 0;
2020
}
2121
}

‎src/LLL.DurableTask.EFCore/EFCoreOrchestrationService.cs

Lines changed: 42 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@
1313
using Microsoft.EntityFrameworkCore;
1414
using Microsoft.Extensions.Logging;
1515
using Microsoft.Extensions.Options;
16+
using NATS.Client;
1617

1718
namespace LLL.DurableTask.EFCore
1819
{
@@ -30,6 +31,8 @@ public partial class EFCoreOrchestrationService :
3031
private readonly ILogger<EFCoreOrchestrationService> _logger;
3132

3233
private CancellationTokenSource _stopCts = new CancellationTokenSource();
34+
private IConnection _connection;
35+
private IAsyncSubscription _subscription;
3336

3437
public EFCoreOrchestrationService(
3538
IOptions<EFCoreOrchestrationOptions> options,
@@ -49,6 +52,12 @@ public EFCoreOrchestrationService(
4952
_instanceMapper = instanceMapper;
5053
_executionMapper = executionMapper;
5154
_logger = logger;
55+
56+
57+
var cf = new ConnectionFactory();
58+
_connection = cf.CreateConnection();
59+
_subscription = _connection.SubscribeAsync(">");
60+
_subscription.Start();
5261
}
5362

5463
public int TaskOrchestrationDispatcherCount => _options.TaskOrchestrationDispatcherCount;
@@ -152,7 +161,8 @@ public async Task<TaskOrchestrationWorkItem> LockNextTaskOrchestrationWorkItemAs
152161
instance,
153162
execution,
154163
runtimeState,
155-
_stopCts.Token);
164+
_stopCts.Token,
165+
_subscription);
156166

157167
var messages = await session.FetchNewMessagesAsync(dbContext);
158168

@@ -179,7 +189,10 @@ public async Task<TaskOrchestrationWorkItem> LockNextTaskOrchestrationWorkItemAs
179189
r => r != null,
180190
receiveTimeout,
181191
_options.PollingInterval,
182-
stoppableCancellationToken);
192+
stoppableCancellationToken,
193+
BackoffPollingHelper.CreateNatsWaitUntilSignal(
194+
_subscription,
195+
orchestrations.Select(nv => $"orchestration.{QueueMapper.ToQueue(nv)}.*").ToHashSet()));
183196
}
184197

185198
public async Task<TaskActivityWorkItem> LockNextTaskActivityWorkItem(TimeSpan receiveTimeout, CancellationToken cancellationToken)
@@ -215,7 +228,10 @@ public async Task<TaskActivityWorkItem> LockNextTaskActivityWorkItem(
215228
x => x != null,
216229
receiveTimeout,
217230
_options.PollingInterval,
218-
stoppableCancellationToken);
231+
stoppableCancellationToken,
232+
BackoffPollingHelper.CreateNatsWaitUntilSignal(
233+
_subscription,
234+
activities.Select(nv => $"activitiy.{QueueMapper.ToQueue(nv)}").ToHashSet()));
219235
}
220236

221237
public async Task ReleaseTaskOrchestrationWorkItemAsync(TaskOrchestrationWorkItem workItem)
@@ -398,16 +414,32 @@ await _dbContextExtensions.WithinTransaction(dbContext, async () =>
398414
session.RuntimeState = newOrchestrationRuntimeState;
399415
session.ClearMessages();
400416
}
417+
418+
// Notify
419+
foreach (var executionStartedEvent in orchestratorMessages.Select(m => m.Event).OfType<ExecutionStartedEvent>())
420+
{
421+
_connection.Publish($"orchestration.{QueueMapper.ToQueue(executionStartedEvent.Name, executionStartedEvent.Version)}.{executionStartedEvent.OrchestrationInstance.InstanceId}",
422+
Array.Empty<byte>());
423+
}
424+
425+
foreach (var taskSheduledEvent in outboundMessages.Select(m => m.Event).OfType<TaskScheduledEvent>())
426+
{
427+
_connection.Publish($"activitiy.{QueueMapper.ToQueue(taskSheduledEvent.Name, taskSheduledEvent.Version)}", Array.Empty<byte>());
428+
}
429+
430+
_connection.Publish($"history.{workItem.InstanceId}", Array.Empty<byte>());
431+
432+
_connection.Flush();
401433
}
402434

403435
public async Task CompleteTaskActivityWorkItemAsync(TaskActivityWorkItem workItem, TaskMessage responseMessage)
404436
{
437+
var (id, lockId, orchestrationQueue) = ParseTaskActivityWorkItemId(workItem.Id);
438+
405439
using (var dbContext = _dbContextFactory.CreateDbContext())
406440
{
407441
await _dbContextExtensions.WithinTransaction(dbContext, async () =>
408442
{
409-
var (id, lockId, orchestrationQueue) = ParseTaskActivityWorkItemId(workItem.Id);
410-
411443
var activityMessage = await dbContext.ActivityMessages
412444
.FirstAsync(w => w.Id == id && w.LockId == lockId);
413445

@@ -423,6 +455,11 @@ await _dbContextExtensions.WithinTransaction(dbContext, async () =>
423455
await dbContext.SaveChangesAsync();
424456
});
425457
}
458+
459+
// Notify
460+
_connection.Publish($"orchestration.{orchestrationQueue}.{workItem.TaskMessage.OrchestrationInstance.InstanceId}", Array.Empty<byte>());
461+
462+
_connection.Flush();
426463
}
427464

428465
private async Task SendTaskOrchestrationMessagesAsync(

‎src/LLL.DurableTask.EFCore/EFCoreOrchestrationServiceClient.cs

Lines changed: 10 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -29,10 +29,10 @@ public Task CreateTaskOrchestrationAsync(TaskMessage creationMessage)
2929

3030
public async Task CreateTaskOrchestrationAsync(TaskMessage creationMessage, OrchestrationStatus[] dedupeStatuses)
3131
{
32+
var executionStartedEvent = creationMessage.Event as ExecutionStartedEvent;
33+
3234
using (var dbContext = _dbContextFactory.CreateDbContext())
3335
{
34-
var executionStartedEvent = creationMessage.Event as ExecutionStartedEvent;
35-
3636
var instanceId = creationMessage.OrchestrationInstance.InstanceId;
3737
var executionId = creationMessage.OrchestrationInstance.ExecutionId;
3838

@@ -75,6 +75,8 @@ public async Task CreateTaskOrchestrationAsync(TaskMessage creationMessage, Orch
7575

7676
await dbContext.SaveChangesAsync();
7777
}
78+
79+
_connection.Publish($"orchestration.{QueueMapper.ToQueue(executionStartedEvent.Name, executionStartedEvent.Version)}.{executionStartedEvent.OrchestrationInstance.InstanceId}", Array.Empty<byte>());
7880
}
7981

8082
public Task ForceTerminateTaskOrchestrationAsync(string instanceId, string reason)
@@ -95,6 +97,7 @@ public async Task<string> GetOrchestrationHistoryAsync(string instanceId, string
9597
var events = await dbContext.Events
9698
.Where(e => e.ExecutionId == executionId)
9799
.OrderBy(e => e.SequenceNumber)
100+
.AsNoTracking()
98101
.ToArrayAsync();
99102

100103
return $"[{string.Join(",", events.Select(e => e.Content))}]";
@@ -190,7 +193,10 @@ public async Task<OrchestrationState> WaitForOrchestrationAsync(
190193
s => IsFinalExecutionStatus(s.OrchestrationStatus),
191194
timeout,
192195
_options.PollingInterval,
193-
stoppableCancellationToken);
196+
stoppableCancellationToken,
197+
BackoffPollingHelper.CreateNatsWaitUntilSignal(
198+
_subscription,
199+
new HashSet<string> { $"history.{instanceId}" }));
194200

195201
if (!IsFinalExecutionStatus(state.OrchestrationStatus))
196202
return null;
@@ -220,6 +226,7 @@ public async Task<OrchestrationQueryResult> GetOrchestrationWithQueryAsync(Orche
220226
.OrderByDescending(x => x.CreatedTime)
221227
.ThenByDescending(x => x.InstanceId)
222228
.Take(query.PageSize + 1)
229+
.AsNoTracking()
223230
.ToArrayAsync();
224231

225232
var pageInstances = instances

‎src/LLL.DurableTask.EFCore/EFCoreOrchestrationSession.cs

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
using LLL.DurableTask.EFCore.Entities;
88
using LLL.DurableTask.EFCore.Polling;
99
using Microsoft.EntityFrameworkCore;
10+
using NATS.Client;
1011

1112
namespace LLL.DurableTask.EFCore
1213
{
@@ -16,21 +17,24 @@ public class EFCoreOrchestrationSession : IOrchestrationSession
1617

1718
private readonly IDbContextFactory<OrchestrationDbContext> _dbContextFactory;
1819
private readonly CancellationToken _stopCancellationToken;
20+
private readonly IAsyncSubscription _subscription;
1921

2022
public EFCoreOrchestrationSession(
2123
EFCoreOrchestrationOptions options,
2224
IDbContextFactory<OrchestrationDbContext> dbContextFactory,
2325
Instance instance,
2426
Execution execution,
2527
OrchestrationRuntimeState runtimeState,
26-
CancellationToken stopCancellationToken)
28+
CancellationToken stopCancellationToken,
29+
IAsyncSubscription subscription)
2730
{
2831
_options = options;
2932
_dbContextFactory = dbContextFactory;
3033
Instance = instance;
3134
Execution = execution;
3235
RuntimeState = runtimeState;
3336
_stopCancellationToken = stopCancellationToken;
37+
_subscription = subscription;
3438
}
3539

3640
public Instance Instance { get; }
@@ -55,7 +59,10 @@ public async Task<IList<TaskMessage>> FetchNewOrchestrationMessagesAsync(
5559
x => x == null || x.Count > 0,
5660
_options.FetchNewMessagesPollingTimeout,
5761
_options.PollingInterval,
58-
_stopCancellationToken);
62+
_stopCancellationToken,
63+
BackoffPollingHelper.CreateNatsWaitUntilSignal(
64+
_subscription,
65+
new HashSet<string> { $"orchestration.{Instance.LastQueue}.{Instance.InstanceId}" }));
5966
}
6067

6168
public async Task<IList<TaskMessage>> FetchNewMessagesAsync(

‎src/LLL.DurableTask.EFCore/LLL.DurableTask.EFCore.csproj

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414
<PackageReference Include="Microsoft.EntityFrameworkCore" Version="7.0.2" />
1515
<PackageReference Include="Microsoft.EntityFrameworkCore.Relational" Version="7.0.2" />
1616
<PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.1.1" PrivateAssets="all" />
17+
<PackageReference Include="NATS.Client" Version="1.0.1" />
1718
</ItemGroup>
1819

1920
<ItemGroup>

‎src/LLL.DurableTask.EFCore/Polling/BackoffPollingHelper.cs

Lines changed: 60 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,11 @@
11
using System;
2+
using System.Collections.Generic;
23
using System.Diagnostics;
4+
using System.Linq;
5+
using System.Runtime.CompilerServices;
36
using System.Threading;
47
using System.Threading.Tasks;
8+
using NATS.Client;
59

610
namespace LLL.DurableTask.EFCore.Polling
711
{
@@ -12,7 +16,11 @@ public static async Task<T> PollAsync<T>(
1216
Func<T, bool> shouldAcceptValue,
1317
TimeSpan timeout,
1418
PollingIntervalOptions interval,
15-
CancellationToken cancellationToken)
19+
CancellationToken cancellationToken,
20+
Func<CancellationToken, Task> waitFunction = null,
21+
[CallerMemberName] string memberName = "",
22+
[CallerFilePath] string fileName = "",
23+
[CallerLineNumber] int lineNumber = 0)
1624
{
1725
T value;
1826

@@ -22,13 +30,29 @@ public static async Task<T> PollAsync<T>(
2230
{
2331
cancellationToken.ThrowIfCancellationRequested();
2432

25-
value = await valueProvider();
33+
using var waitCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
34+
var waitTask = (waitFunction ?? WaitUntilCancellation)(waitCts.Token);
2635

27-
if (shouldAcceptValue(value)
28-
|| stopwatch.Elapsed >= timeout)
29-
break;
36+
try
37+
{
38+
value = await valueProvider();
3039

31-
await Task.Delay(CalculateDelay(interval, count++));
40+
if (shouldAcceptValue(value)
41+
|| stopwatch.Elapsed >= timeout)
42+
break;
43+
44+
waitCts.CancelAfter(CalculateDelay(interval, count++));
45+
46+
try
47+
{
48+
await waitTask;
49+
}
50+
catch (OperationCanceledException) { }
51+
}
52+
finally
53+
{
54+
waitCts.Cancel();
55+
}
3256
} while (stopwatch.Elapsed < timeout);
3357

3458
return value;
@@ -38,5 +62,35 @@ private static int CalculateDelay(PollingIntervalOptions interval, int count)
3862
{
3963
return (int)Math.Min(interval.Initial * Math.Pow(interval.Factor, count), interval.Max);
4064
}
65+
66+
public static async Task WaitUntilCancellation(CancellationToken cancellationToken)
67+
{
68+
var tcs = new TaskCompletionSource();
69+
using var registration = cancellationToken.Register(() => tcs.TrySetCanceled());
70+
await tcs.Task;
71+
}
72+
73+
public static Func<CancellationToken, Task> CreateNatsWaitUntilSignal(IAsyncSubscription subscription, HashSet<string> subjects)
74+
{
75+
return async (cancellationToken) =>
76+
{
77+
var tcs = new TaskCompletionSource();
78+
void OnMessage(object o, MsgHandlerEventArgs e)
79+
{
80+
if (subjects.Contains(e.Message.Subject) || subjects.Any(s => e.Message.Subject.StartsWith(s)))
81+
tcs.TrySetResult();
82+
};
83+
subscription.MessageHandler += OnMessage;
84+
using var registration = cancellationToken.Register(() => tcs.TrySetCanceled());
85+
try
86+
{
87+
await tcs.Task;
88+
}
89+
finally
90+
{
91+
subscription.MessageHandler -= OnMessage;
92+
}
93+
};
94+
}
4195
}
4296
}

0 commit comments

Comments
 (0)