5835751495

pending completion

Build # 5835751495

Build Type

push

github

Committed by

lucaslorentz

Commit Message

Add husky and apply some code fixes

Run Details

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

2286 of 2792 relevant lines covered (81.88%)

143.14 hits per line

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

60.75

/src/LLL.DurableTask.Server.Grpc.Client/GrpcClientOrchestrationServiceClient.cs

﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using DurableTask.Core;
using DurableTask.Core.Query;
using DurableTaskGrpc;
using Google.Protobuf.WellKnownTypes;
using Grpc.Core;
using LLL.DurableTask.Core;

namespace LLL.DurableTask.Server.Client;

public partial class GrpcClientOrchestrationService :
    IOrchestrationServiceClient,
    IOrchestrationServiceQueryClient,
    IOrchestrationServicePurgeClient,
    IOrchestrationServiceFeaturesClient,
    IOrchestrationServiceRewindClient
{
    public async Task<OrchestrationFeature[]> GetFeatures()
    {
        var response = await _client.GetFeaturesAsync(new Empty());
        return response.Features
            .Select(f => (OrchestrationFeature)f)
            .ToArray();
    }
    public async Task CreateTaskOrchestrationAsync(TaskMessage creationMessage)
    {
        await CreateTaskOrchestrationAsync(creationMessage, Array.Empty<OrchestrationStatus>());
    }

    public async Task CreateTaskOrchestrationAsync(TaskMessage creationMessage, OrchestrationStatus[] dedupeStatuses)
    {
        var request = new CreateTaskOrchestrationRequest
        {
            CreationMessage = _options.DataConverter.Serialize(creationMessage)
        };

        if (dedupeStatuses != null)
            request.DedupeStatuses.AddRange(dedupeStatuses.Select(s => (int)s));

        await _client.CreateTaskOrchestrationAsync(request);
    }

    public async Task<string> GetOrchestrationHistoryAsync(string instanceId, string executionId)
    {
        var request = new GetOrchestrationHistoryRequest
        {
            InstanceId = instanceId,
            ExecutionId = executionId
        };

        var response = await _client.GetOrchestrationHistoryAsync(request);

        return response.History;
    }

    public async Task<IList<OrchestrationState>> GetOrchestrationStateAsync(string instanceId, bool allExecutions)
    {
        var request = new GetOrchestrationInstanceStateRequest
        {
            InstanceId = instanceId,
            AllExecutions = allExecutions
        };

        var response = await _client.GetOrchestrationInstanceStateAsync(request);

        return response.States
            .Select(s => _options.DataConverter.Deserialize<OrchestrationState>(s))
            .ToArray();
    }

    public async Task<OrchestrationState> GetOrchestrationStateAsync(string instanceId, string executionId)
    {
        var request = new GetOrchestrationStateRequest
        {
            InstanceId = instanceId,
            ExecutionId = executionId
        };

        var response = await _client.GetOrchestrationStateAsync(request);

        return string.IsNullOrEmpty(response.State) ? null : _options.DataConverter.Deserialize<OrchestrationState>(response.State);
    }

    public async Task ForceTerminateTaskOrchestrationAsync(string instanceId, string reason)
    {
        var request = new ForceTerminateTaskOrchestrationRequest
        {
            InstanceId = instanceId,
            Reason = reason
        };

        await _client.ForceTerminateTaskOrchestrationAsync(request);
    }

    public async Task RewindTaskOrchestrationAsync(string instanceId, string reason)
    {
        var request = new RewindTaskOrchestrationRequest
        {
            InstanceId = instanceId,
            Reason = reason
        };

        await _client.RewindTaskOrchestrationAsync(request);
    }

    public async Task PurgeOrchestrationHistoryAsync(DateTime thresholdDateTimeUtc, OrchestrationStateTimeRangeFilterType timeRangeFilterType)
    {
        var request = new PurgeOrchestrationHistoryRequest
        {
            ThresholdDateTimeUtc = ToTimestamp(thresholdDateTimeUtc),
            TimeRangeFilterType = (OrchestrationTimeFilterType)timeRangeFilterType
        };

        await _client.PurgeOrchestrationHistoryAsync(request);
    }

    public async Task SendTaskOrchestrationMessageAsync(TaskMessage message)
    {
        await SendTaskOrchestrationMessageBatchAsync(message);
    }

    public async Task SendTaskOrchestrationMessageBatchAsync(params TaskMessage[] messages)
    {
        var request = new SendTaskOrchestrationMessageBatchRequest
        {
            Messages = { messages.Select(_options.DataConverter.Serialize) }
        };

        await _client.SendTaskOrchestrationMessageBatchAsync(request);
    }

    public async Task<OrchestrationState> WaitForOrchestrationAsync(string instanceId, string executionId, TimeSpan timeout, CancellationToken cancellationToken)
    {
        var request = new WaitForOrchestrationRequest
        {
            InstanceId = instanceId,
            ExecutionId = executionId,
            Timeout = Duration.FromTimeSpan(timeout)
        };

        var callOptions = new CallOptions(cancellationToken: cancellationToken);

        var response = await _client.WaitForOrchestrationAsync(request, callOptions);

        return string.IsNullOrEmpty(response.State) ? null : _options.DataConverter.Deserialize<OrchestrationState>(response.State);
    }

    public async Task<OrchestrationQueryResult> GetOrchestrationWithQueryAsync(OrchestrationQuery query, CancellationToken cancellationToken)
    {
        var request = new GetOrchestrationWithQueryRequest();

        if (query.RuntimeStatus != null)
            request.RuntimeStatus.AddRange(query.RuntimeStatus.Select(s => (int)s));

        request.CreatedTimeFrom = ToTimestamp(query.CreatedTimeFrom);
        request.CreatedTimeTo = ToTimestamp(query.CreatedTimeTo);

        if (query.TaskHubNames != null)
            request.TaskHubNames.AddRange(query.TaskHubNames);

        request.PageSize = query.PageSize;
        request.ContinuationToken = query.ContinuationToken;
        request.InstanceIdPrefix = query.InstanceIdPrefix;
        request.FetchInputsAndOutputs = query.FetchInputsAndOutputs;

        if (query is ExtendedOrchestrationQuery extendedQuery)
        {
            request.NamePrefix = extendedQuery.NamePrefix;
            request.CompletedTimeFrom = ToTimestamp(extendedQuery.CompletedTimeFrom);
            request.CompletedTimeTo = ToTimestamp(extendedQuery.CompletedTimeTo);
            request.IncludePreviousExecutions = extendedQuery.IncludePreviousExecutions;
            foreach (var kv in extendedQuery.Tags)
                request.Tags.Add(kv.Key, kv.Value);
        }

        var callOptions = new CallOptions(cancellationToken: cancellationToken);

        var response = await _client.GetOrchestrationWithQueryAsync(request, callOptions);

        var orchestrationsState = response.OrchestrationState
            .Select(s => _options.DataConverter.Deserialize<OrchestrationState>(s))
            .ToArray();

        return new OrchestrationQueryResult(orchestrationsState, response.ContinuationToken);
    }

    public async Task<PurgeResult> PurgeInstanceStateAsync(string instanceId)
    {
        var request = new PurgeInstanceHistoryRequest
        {
            InstanceId = instanceId
        };

        var result = await _client.PurgeInstanceHistoryAsync(request);

        return new PurgeResult(result.InstancesDeleted);
    }

    public async Task<PurgeResult> PurgeInstanceStateAsync(PurgeInstanceFilter filter)
    {
        var request = new PurgeInstanceHistoryRequest
        {
            CreatedTimeFrom = ToTimestamp(filter.CreatedTimeFrom),
            CreatedTimeTo = ToTimestamp(filter.CreatedTimeTo),
        };

        if (filter.RuntimeStatus != null)
            request.RuntimeStatus.AddRange(filter.RuntimeStatus.Select(s => (int)s));

        var result = await _client.PurgeInstanceHistoryAsync(request);

        return new PurgeResult(result.InstancesDeleted);
    }
}

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 DurableTask.Core.Query;
8	using DurableTaskGrpc;
9	using Google.Protobuf.WellKnownTypes;
10	using Grpc.Core;
11	using LLL.DurableTask.Core;
12
13	namespace LLL.DurableTask.Server.Client;
14
15	public partial class GrpcClientOrchestrationService :
16	IOrchestrationServiceClient,
17	IOrchestrationServiceQueryClient,
18	IOrchestrationServicePurgeClient,
19	IOrchestrationServiceFeaturesClient,
20	IOrchestrationServiceRewindClient
21	{
22	public async Task<OrchestrationFeature[]> GetFeatures()
23	{
24	var response = await _client.GetFeaturesAsync(new Empty());	×
25	return response.Features	×
26	.Select(f => (OrchestrationFeature)f)	×
27	.ToArray();	×
28	}
29	public async Task CreateTaskOrchestrationAsync(TaskMessage creationMessage)
30	{
31	await CreateTaskOrchestrationAsync(creationMessage, Array.Empty<OrchestrationStatus>());	×
32	}
33
34	public async Task CreateTaskOrchestrationAsync(TaskMessage creationMessage, OrchestrationStatus[] dedupeStatuses)
35	{
36	var request = new CreateTaskOrchestrationRequest	12✔
37	{	12✔
38	CreationMessage = _options.DataConverter.Serialize(creationMessage)	12✔
39	};	12✔
40
41	if (dedupeStatuses != null)	12✔
42	request.DedupeStatuses.AddRange(dedupeStatuses.Select(s => (int)s));	×
43
44	await _client.CreateTaskOrchestrationAsync(request);	12✔
45	}
46
47	public async Task<string> GetOrchestrationHistoryAsync(string instanceId, string executionId)
48	{
49	var request = new GetOrchestrationHistoryRequest	×
50	{	×
51	InstanceId = instanceId,	×
52	ExecutionId = executionId	×
53	};	×
54
55	var response = await _client.GetOrchestrationHistoryAsync(request);	×
56
57	return response.History;	×
58	}
59
60	public async Task<IList<OrchestrationState>> GetOrchestrationStateAsync(string instanceId, bool allExecutions)
61	{
62	var request = new GetOrchestrationInstanceStateRequest	2✔
63	{	2✔
64	InstanceId = instanceId,	2✔
65	AllExecutions = allExecutions	2✔
66	};	2✔
67
68	var response = await _client.GetOrchestrationInstanceStateAsync(request);	2✔
69
70	return response.States	2✔
71	.Select(s => _options.DataConverter.Deserialize<OrchestrationState>(s))	3✔
72	.ToArray();	2✔
73	}
74
75	public async Task<OrchestrationState> GetOrchestrationStateAsync(string instanceId, string executionId)
76	{
77	var request = new GetOrchestrationStateRequest	1✔
78	{	1✔
79	InstanceId = instanceId,	1✔
80	ExecutionId = executionId	1✔
81	};	1✔
82
83	var response = await _client.GetOrchestrationStateAsync(request);	1✔
84
85	return string.IsNullOrEmpty(response.State) ? null : _options.DataConverter.Deserialize<OrchestrationState>(response.State);	1✔
86	}
87
88	public async Task ForceTerminateTaskOrchestrationAsync(string instanceId, string reason)
89	{
90	var request = new ForceTerminateTaskOrchestrationRequest	×
91	{	×
92	InstanceId = instanceId,	×
93	Reason = reason	×
94	};	×
95
96	await _client.ForceTerminateTaskOrchestrationAsync(request);	×
97	}
98
99	public async Task RewindTaskOrchestrationAsync(string instanceId, string reason)
100	{
101	var request = new RewindTaskOrchestrationRequest	×
102	{	×
103	InstanceId = instanceId,	×
104	Reason = reason	×
105	};	×
106
107	await _client.RewindTaskOrchestrationAsync(request);	×
108	}
109
110	public async Task PurgeOrchestrationHistoryAsync(DateTime thresholdDateTimeUtc, OrchestrationStateTimeRangeFilterType timeRangeFilterType)
111	{
112	var request = new PurgeOrchestrationHistoryRequest	×
113	{	×
114	ThresholdDateTimeUtc = ToTimestamp(thresholdDateTimeUtc),	×
115	TimeRangeFilterType = (OrchestrationTimeFilterType)timeRangeFilterType	×
116	};	×
117
118	await _client.PurgeOrchestrationHistoryAsync(request);	×
119	}
120
121	public async Task SendTaskOrchestrationMessageAsync(TaskMessage message)
122	{
123	await SendTaskOrchestrationMessageBatchAsync(message);	2✔
124	}
125
126	public async Task SendTaskOrchestrationMessageBatchAsync(params TaskMessage[] messages)
127	{
128	var request = new SendTaskOrchestrationMessageBatchRequest	2✔
129	{	2✔
130	Messages = { messages.Select(_options.DataConverter.Serialize) }	2✔
131	};	2✔
132
133	await _client.SendTaskOrchestrationMessageBatchAsync(request);	2✔
134	}
135
136	public async Task<OrchestrationState> WaitForOrchestrationAsync(string instanceId, string executionId, TimeSpan timeout, CancellationToken cancellationToken)
137	{
138	var request = new WaitForOrchestrationRequest	17✔
139	{	17✔
140	InstanceId = instanceId,	17✔
141	ExecutionId = executionId,	17✔
142	Timeout = Duration.FromTimeSpan(timeout)	17✔
143	};	17✔
144
145	var callOptions = new CallOptions(cancellationToken: cancellationToken);	17✔
146
147	var response = await _client.WaitForOrchestrationAsync(request, callOptions);	17✔
148
149	return string.IsNullOrEmpty(response.State) ? null : _options.DataConverter.Deserialize<OrchestrationState>(response.State);	17✔
150	}
151
152	public async Task<OrchestrationQueryResult> GetOrchestrationWithQueryAsync(OrchestrationQuery query, CancellationToken cancellationToken)
153	{
154	var request = new GetOrchestrationWithQueryRequest();	1✔
155
156	if (query.RuntimeStatus != null)	1✔
157	request.RuntimeStatus.AddRange(query.RuntimeStatus.Select(s => (int)s));	×
158
159	request.CreatedTimeFrom = ToTimestamp(query.CreatedTimeFrom);	1✔
160	request.CreatedTimeTo = ToTimestamp(query.CreatedTimeTo);	1✔
161
162	if (query.TaskHubNames != null)	1✔
163	request.TaskHubNames.AddRange(query.TaskHubNames);	×
164
165	request.PageSize = query.PageSize;	1✔
166	request.ContinuationToken = query.ContinuationToken;	1✔
167	request.InstanceIdPrefix = query.InstanceIdPrefix;	1✔
168	request.FetchInputsAndOutputs = query.FetchInputsAndOutputs;	1✔
169
170	if (query is ExtendedOrchestrationQuery extendedQuery)	1✔
171	{
172	request.NamePrefix = extendedQuery.NamePrefix;	1✔
173	request.CompletedTimeFrom = ToTimestamp(extendedQuery.CompletedTimeFrom);	1✔
174	request.CompletedTimeTo = ToTimestamp(extendedQuery.CompletedTimeTo);	1✔
175	request.IncludePreviousExecutions = extendedQuery.IncludePreviousExecutions;	1✔
176	foreach (var kv in extendedQuery.Tags)	6✔
177	request.Tags.Add(kv.Key, kv.Value);	2✔
178	}
179
180	var callOptions = new CallOptions(cancellationToken: cancellationToken);	1✔
181
182	var response = await _client.GetOrchestrationWithQueryAsync(request, callOptions);	1✔
183
184	var orchestrationsState = response.OrchestrationState	1✔
185	.Select(s => _options.DataConverter.Deserialize<OrchestrationState>(s))	2✔
186	.ToArray();	1✔
187
188	return new OrchestrationQueryResult(orchestrationsState, response.ContinuationToken);	1✔
189	}
190
191	public async Task<PurgeResult> PurgeInstanceStateAsync(string instanceId)
192	{
193	var request = new PurgeInstanceHistoryRequest	1✔
194	{	1✔
195	InstanceId = instanceId	1✔
196	};	1✔
197
198	var result = await _client.PurgeInstanceHistoryAsync(request);	1✔
199
200	return new PurgeResult(result.InstancesDeleted);	1✔
201	}
202
203	public async Task<PurgeResult> PurgeInstanceStateAsync(PurgeInstanceFilter filter)
204	{
205	var request = new PurgeInstanceHistoryRequest	×
206	{	×
207	CreatedTimeFrom = ToTimestamp(filter.CreatedTimeFrom),	×
208	CreatedTimeTo = ToTimestamp(filter.CreatedTimeTo),	×
209	};	×
210
211	if (filter.RuntimeStatus != null)	×
212	request.RuntimeStatus.AddRange(filter.RuntimeStatus.Select(s => (int)s));	×
213
214	var result = await _client.PurgeInstanceHistoryAsync(request);	×
215
216	return new PurgeResult(result.InstancesDeleted);	×
217	}
218	}

lucaslorentz / durabletask-extensions / 5835751495

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