-
Notifications
You must be signed in to change notification settings - Fork 5
/
SearchClientFunctions.cs
104 lines (95 loc) · 4.97 KB
/
SearchClientFunctions.cs
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
using System;
using System.Net.Http;
using System.Threading.Tasks;
using Atlas.Client.Models.Search.Results.Matching;
using Atlas.Common.ApplicationInsights;
using Atlas.Functions.DurableFunctions.Search.Orchestration;
using Atlas.Functions.Models;
using Microsoft.Azure.Functions.Worker;
using Microsoft.Azure.Functions.Worker.Http;
using Microsoft.DurableTask.Client;
using Newtonsoft.Json;
namespace Atlas.Functions.DurableFunctions.Search.Client
{
/// <summary>
/// See https://docs.microsoft.com/en-us/azure/azure-functions/durable/durable-functions-types-features-overview for the types of durable functions available
/// We are using:
/// - Client = entry points.
/// - Orchestrator = orchestration. DETERMINISTIC (https://docs.microsoft.com/en-us/azure/azure-functions/durable/durable-functions-code-constraints)
/// - Activity = business logic. IDEMPOTENT (may be called multiple times, only at least once execution is guaranteed by Azure)
/// </summary>
public class SearchClientFunctions
{
private readonly ILogger logger;
public SearchClientFunctions(ILogger logger)
{
this.logger = logger;
}
[Function(nameof(Search))]
public async Task Search(
[ServiceBusTrigger(
"%AtlasFunction:MessagingServiceBus:MatchingResultsTopic%",
"%AtlasFunction:MessagingServiceBus:MatchingResultsSubscription%",
Connection = "AtlasFunction:MessagingServiceBus:ConnectionString"
)]
MatchingResultsNotification resultsNotification,
[DurableClient] DurableTaskClient starter,
DateTime enqueuedTimeUtc)
{
var searchId = resultsNotification.SearchRequestId;
if (!resultsNotification.SearchRequest?.RunMatchPrediction ?? false)
{
logger.SendTrace($"Match prediction for search request '{searchId}' was not requested, so will not be run.");
// TODO: ATLAS-493: Make the matching only results usable.
return;
}
var instanceId = await starter.ScheduleNewOrchestrationInstanceAsync(
nameof(SearchOrchestrationFunctions.SearchOrchestrator),
input: new SearchOrchestratorParameters{ MatchingResultsNotification = resultsNotification, InitiationTime = enqueuedTimeUtc });
try
{
logger.SendTrace($"Started match prediction orchestration with ID = '{searchId}'. Orchestration Instance: {instanceId}");
}
catch (Exception)
{
// This function cannot be allowed to fail post-orchestration scheduling, as it would then retry, and we cannot schedule more than one orchestrator with the same id.
// We are only doing logging past this point, so if it fails we just swallow exceptions.
}
}
[Function(nameof(RepeatSearch))]
public async Task RepeatSearch(
[ServiceBusTrigger(
"%AtlasFunction:MessagingServiceBus:RepeatSearchMatchingResultsTopic%",
"%AtlasFunction:MessagingServiceBus:RepeatSearchMatchingResultsSubscription%",
Connection = "AtlasFunction:MessagingServiceBus:ConnectionString"
)]
MatchingResultsNotification resultsNotification,
[DurableClient] DurableTaskClient starter)
{
var repeatSearchId = resultsNotification.RepeatSearchRequestId;
if (!resultsNotification.SearchRequest?.RunMatchPrediction ?? false)
{
logger.SendTrace($"Match prediction for repeat search request '{repeatSearchId}' was not requested, so will not be run.");
// TODO: ATLAS-493: Make the matching only results usable.
return;
}
var instanceId = await starter.ScheduleNewOrchestrationInstanceAsync(nameof(SearchOrchestrationFunctions.RepeatSearchOrchestrator), input: resultsNotification);
try
{
logger.SendTrace($"Started match prediction orchestration with Repeat Search ID = '{repeatSearchId}'. Orchestration Instance: {instanceId}");
}
catch (Exception)
{
// This function cannot be allowed to fail post-orchestration scheduling, as it would then retry, and we cannot schedule more than one orchestrator with the same id.
// We are only doing logging past this point, so if it fails we just swallow exceptions.
}
}
[Function(nameof(GetStatusCheckEndpoints))]
public HttpResponseData GetStatusCheckEndpoints([HttpTrigger(AuthorizationLevel.Function, "get", Route = $"match-prediction-status/{{{nameof(instanceId)}}}")] HttpRequestData request,
string instanceId,
[DurableClient] DurableTaskClient orchestrationClient)
{
return orchestrationClient.CreateCheckStatusResponse(request, instanceId);
}
}
}