5805868292

pending completion

Build # 5805868292

Build Type

push

github

Committed by

lucaslorentz

Commit Message

Search by completed time

Run Details

12 of 12 new or added lines in 5 files covered. (100.0%)

2311 of 2794 relevant lines covered (82.71%)

145.34 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, new OrchestrationStatus[0]);
        }

        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, new OrchestrationStatus[0]);	×
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	14✔
139	{	14✔
140	InstanceId = instanceId,	14✔
141	ExecutionId = executionId,	14✔
142	Timeout = Duration.FromTimeSpan(timeout)	14✔
143	};	14✔
144
145	var callOptions = new CallOptions(cancellationToken: cancellationToken);	14✔
146
147	var response = await _client.WaitForOrchestrationAsync(request, callOptions);	14✔
148
149	return string.IsNullOrEmpty(response.State) ? null : _options.DataConverter.Deserialize<OrchestrationState>(response.State);	14✔
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	}
219	}

lucaslorentz / durabletask-extensions / 5805868292

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