5835662074

pending completion

Build # 5835662074

Build Type

Pull #28

github

Committed by

lucaslorentz

Commit Message

Add husky and apply some code fixes

Pull Request Pull Request #28: Add husky and apply code fixes

Run Details

2502 of 2502 new or added lines in 91 files covered. (100.0%)

2297 of 2792 relevant lines covered (82.27%)

141.58 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

88.89

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

﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using DurableTask.Core;
using LLL.DurableTask.EFCore.Entities;
using LLL.DurableTask.EFCore.Polling;
using Microsoft.EntityFrameworkCore;

namespace LLL.DurableTask.EFCore;

public class EFCoreOrchestrationSession : IOrchestrationSession
{
    private readonly EFCoreOrchestrationOptions _options;

    private readonly IDbContextFactory<OrchestrationDbContext> _dbContextFactory;
    private readonly CancellationToken _stopCancellationToken;

    public EFCoreOrchestrationSession(
        EFCoreOrchestrationOptions options,
        IDbContextFactory<OrchestrationDbContext> dbContextFactory,
        Instance instance,
        Execution execution,
        OrchestrationRuntimeState runtimeState,
        CancellationToken stopCancellationToken)
    {
        _options = options;
        _dbContextFactory = dbContextFactory;
        Instance = instance;
        Execution = execution;
        RuntimeState = runtimeState;
        _stopCancellationToken = stopCancellationToken;
    }

    public Instance Instance { get; }
    public Execution Execution { get; set; }
    public OrchestrationRuntimeState RuntimeState { get; set; }
    public List<OrchestrationMessage> Messages { get; } = new List<OrchestrationMessage>();

    public bool Released { get; set; }

    public async Task<IList<TaskMessage>> FetchNewOrchestrationMessagesAsync(
        TaskOrchestrationWorkItem workItem)
    {
        return await BackoffPollingHelper.PollAsync(async () =>
        {
            using var dbContext = _dbContextFactory.CreateDbContext();
            var messages = await FetchNewMessagesAsync(dbContext);
            await dbContext.SaveChangesAsync();
            return messages;
        },
        x => x == null || x.Count > 0,
        _options.FetchNewMessagesPollingTimeout,
        _options.PollingInterval,
        _stopCancellationToken);
    }

    public async Task<IList<TaskMessage>> FetchNewMessagesAsync(
        OrchestrationDbContext dbContext,
        CancellationToken cancellationToken = default)
    {
        var newDbMessages = await dbContext.OrchestrationMessages
            .Where(w => w.AvailableAt <= DateTime.UtcNow
                && w.InstanceId == Instance.InstanceId
                && w.Instance.LockId == Instance.LockId // Ensure we still own the lock
                && !Messages.Contains(w))
            .OrderBy(w => w.AvailableAt)
            .ThenBy(w => w.SequenceNumber)
            .AsNoTracking()
            .ToArrayAsync(cancellationToken);

        var messagesToDiscard = newDbMessages
            .Where(m => m.ExecutionId != null && m.ExecutionId != Instance.LastExecutionId)
            .ToArray();

        if (messagesToDiscard.Length > 0)
        {
            foreach (var message in messagesToDiscard)
            {
                dbContext.OrchestrationMessages.Attach(message);
                dbContext.OrchestrationMessages.Remove(message);
            }

            newDbMessages = newDbMessages
                .Except(messagesToDiscard)
                .ToArray();
        }

        Messages.AddRange(newDbMessages);

        var deserializedMessages = newDbMessages
            .Select(w => _options.DataConverter.Deserialize<TaskMessage>(w.Message))
            .ToList();

        return deserializedMessages;
    }

    public void ClearMessages()
    {
        Messages.Clear();
    }
}

1	using System;
2	using System.Collections.Generic;
3	using System.Linq;
4	using System.Threading;
5	using System.Threading.Tasks;
6	using DurableTask.Core;
7	using LLL.DurableTask.EFCore.Entities;
8	using LLL.DurableTask.EFCore.Polling;
9	using Microsoft.EntityFrameworkCore;
10
11	namespace LLL.DurableTask.EFCore;
12
13	public class EFCoreOrchestrationSession : IOrchestrationSession
14	{
15	private readonly EFCoreOrchestrationOptions _options;
16
17	private readonly IDbContextFactory<OrchestrationDbContext> _dbContextFactory;
18	private readonly CancellationToken _stopCancellationToken;
19
20	public EFCoreOrchestrationSession(	136✔
21	EFCoreOrchestrationOptions options,	136✔
22	IDbContextFactory<OrchestrationDbContext> dbContextFactory,	136✔
23	Instance instance,	136✔
24	Execution execution,	136✔
25	OrchestrationRuntimeState runtimeState,	136✔
26	CancellationToken stopCancellationToken)	136✔
27	{
28	_options = options;	136✔
29	_dbContextFactory = dbContextFactory;	136✔
30	Instance = instance;	136✔
31	Execution = execution;	136✔
32	RuntimeState = runtimeState;	136✔
33	_stopCancellationToken = stopCancellationToken;	136✔
34	}
35
36	public Instance Instance { get; }	1,930✔
37	public Execution Execution { get; set; }	563✔
38	public OrchestrationRuntimeState RuntimeState { get; set; }	332✔
39	public List<OrchestrationMessage> Messages { get; } = new List<OrchestrationMessage>();	1,544✔
40
41	public bool Released { get; set; }	228✔
42
43	public async Task<IList<TaskMessage>> FetchNewOrchestrationMessagesAsync(
44	TaskOrchestrationWorkItem workItem)
45	{
46	return await BackoffPollingHelper.PollAsync(async () =>	82✔
47	{	82✔
48	using var dbContext = _dbContextFactory.CreateDbContext();	356✔
49	var messages = await FetchNewMessagesAsync(dbContext);	356✔
50	await dbContext.SaveChangesAsync();	356✔
51	return messages;	356✔
52	},	82✔
53	x => x == null \|\| x.Count > 0,	356✔
54	_options.FetchNewMessagesPollingTimeout,	82✔
55	_options.PollingInterval,	82✔
56	_stopCancellationToken);	82✔
57	}
58
59	public async Task<IList<TaskMessage>> FetchNewMessagesAsync(
60	OrchestrationDbContext dbContext,
61	CancellationToken cancellationToken = default)
62	{
63	var newDbMessages = await dbContext.OrchestrationMessages	410✔
64	.Where(w => w.AvailableAt <= DateTime.UtcNow	410✔
65	&& w.InstanceId == Instance.InstanceId	410✔
66	&& w.Instance.LockId == Instance.LockId // Ensure we still own the lock	410✔
67	&& !Messages.Contains(w))	410✔
68	.OrderBy(w => w.AvailableAt)	410✔
69	.ThenBy(w => w.SequenceNumber)	410✔
70	.AsNoTracking()	410✔
71	.ToArrayAsync(cancellationToken);	410✔
72
73	var messagesToDiscard = newDbMessages	410✔
74	.Where(m => m.ExecutionId != null && m.ExecutionId != Instance.LastExecutionId)	680✔
75	.ToArray();	410✔
76
77	if (messagesToDiscard.Length > 0)	410✔
78	{
79	foreach (var message in messagesToDiscard)	×
80	{
81	dbContext.OrchestrationMessages.Attach(message);	×
82	dbContext.OrchestrationMessages.Remove(message);	×
83	}
84
85	newDbMessages = newDbMessages	×
86	.Except(messagesToDiscard)	×
87	.ToArray();	×
88	}
89
90	Messages.AddRange(newDbMessages);	410✔
91
92	var deserializedMessages = newDbMessages	410✔
93	.Select(w => _options.DataConverter.Deserialize<TaskMessage>(w.Message))	680✔
94	.ToList();	410✔
95
96	return deserializedMessages;	410✔
97	}
98
99	public void ClearMessages()
100	{
101	Messages.Clear();	196✔
102	}
103	}

lucaslorentz / durabletask-extensions / 5835662074

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous