Durable Functions 中的監視器案例 - 天氣監看員範例

發行項
06/01/2023

監視模式是指工作流程中的彈性「週期性」程序，例如，輪詢直到符合特定條件。本文會說明使用 Durable Functions 來實作監視的範例。

必要條件

C#
JavaScript

案例概觀

此範例會監視某地點的目前天氣狀況，並在天空放晴時用簡訊對使用者發出警示。您可以使用一般的計時器觸發函式來檢查天氣並傳送警示。不過，這個方法有一個問題，那就是存留期管理。如果應傳送的警示只有一個，則監視器必須在偵測到天氣晴朗後自行停用。監視模式除了可以結束自身的執行外，還有其他優點：

監視器會依間隔 (而非排程) 來執行：計時器觸發程序每小時「執行」一次；監視器會在動作之間「等候」一小時。除非有指定，否則監視器的動作不會重疊，對於長時間執行的工作來說，這一點很重要。
監視器可具有動態間隔：等待時間可以根據某些條件來變更。
監視器可於某些條件成立時終止，或由其他程序加以終止。
監視器可以採用參數。此範例會示範如何將相同的天氣監視程序套用至任何要求的地點和電話號碼。
監視器具有擴充性。由於每個監視器都是協調流程執行個體，您可以建立多個監視器，而不必建立新函式或定義多個程式碼。
監視器可輕鬆整合至大型工作流程。監視器可以是更複雜之協調流程函式的一個區段，也可以是子協調流程。

組態

設定 Twilio 整合

這個範例牽涉到使用 Twilio 服務來將 SMS 訊息傳送至行動電話。 Azure Functions 已透過 Twilio 繫結支援 Twilio，範例會使用該功能。

首先您需要的是 Twilio 帳戶。您可以在 https://www.twilio.com/try-twilio 免費建立一個帳戶。一旦您擁有帳戶，將下列三個應用程式設定新增至您的函式應用程式。

應用程式設定名稱	值描述
TwilioAccountSid	Twilio 帳戶的 SID
TwilioAuthToken	Twilio 帳戶的驗證權杖
TwilioPhoneNumber	與您的 Twilio 帳戶相關聯的電話號碼。這是用來傳送 SMS 訊息。

設定 Weather Underground 整合

這個範例涉及使用 Weather Underground API 來檢查某地點的目前天氣狀況。

您首先需要的是 Weather Underground 帳戶。您可以在以下位置免費建立一個帳戶：https://www.wunderground.com/signup。擁有帳戶後，您就需要取得 API 金鑰。您可以造訪 https://www.wunderground.com/weather/api 並選取 [金鑰設定]，來取得此金鑰。 Stratus Developer 是免費的方案，且足已執行此範例。

擁有 API 金鑰後，將下列應用程式設定新增至您的函式應用程式。

應用程式設定名稱	值描述
WeatherUndergroundApiKey	您的 Weather Underground API 金鑰。

函式

本文說明範例應用程式中的函式如下：

E3_Monitor：定期呼叫 E3_GetIsClear 的協調器函式。如果 E3_GetIsClear 傳回 true，此函式會呼叫 E3_SendGoodWeatherAlert。
E3_GetIsClear：活動函式，用來檢查某地點目前天氣狀況。
E3_SendGoodWeatherAlert：會透過 Twilio 傳送手機簡訊的活動函式。

[FunctionName("E3_Monitor")]
public static async Task Run([OrchestrationTrigger] IDurableOrchestrationContext monitorContext, ILogger log)
{
    MonitorRequest input = monitorContext.GetInput<MonitorRequest>();
    if (!monitorContext.IsReplaying) { log.LogInformation($"Received monitor request. Location: {input?.Location}. Phone: {input?.Phone}."); }

    VerifyRequest(input);

    DateTime endTime = monitorContext.CurrentUtcDateTime.AddHours(6);
    if (!monitorContext.IsReplaying) { log.LogInformation($"Instantiating monitor for {input.Location}. Expires: {endTime}."); }

    while (monitorContext.CurrentUtcDateTime < endTime)
    {
        // Check the weather
        if (!monitorContext.IsReplaying) { log.LogInformation($"Checking current weather conditions for {input.Location} at {monitorContext.CurrentUtcDateTime}."); }

        bool isClear = await monitorContext.CallActivityAsync<bool>("E3_GetIsClear", input.Location);

        if (isClear)
        {
            // It's not raining! Or snowing. Or misting. Tell our user to take advantage of it.
            if (!monitorContext.IsReplaying) { log.LogInformation($"Detected clear weather for {input.Location}. Notifying {input.Phone}."); }

            await monitorContext.CallActivityAsync("E3_SendGoodWeatherAlert", input.Phone);
            break;
        }
        else
        {
            // Wait for the next checkpoint
            var nextCheckpoint = monitorContext.CurrentUtcDateTime.AddMinutes(30);
            if (!monitorContext.IsReplaying) { log.LogInformation($"Next check for {input.Location} at {nextCheckpoint}."); }

            await monitorContext.CreateTimer(nextCheckpoint, CancellationToken.None);
        }
    }

    log.LogInformation($"Monitor expiring.");
}

[Deterministic]
private static void VerifyRequest(MonitorRequest request)
{
    if (request == null)
    {
        throw new ArgumentNullException(nameof(request), "An input object is required.");
    }

    if (request.Location == null)
    {
        throw new ArgumentNullException(nameof(request.Location), "A location input is required.");
    }

    if (string.IsNullOrEmpty(request.Phone))
    {
        throw new ArgumentNullException(nameof(request.Phone), "A phone number input is required.");
    }
}

協調器需要一個要監視的位置，以及在該位置是否變得明確時要傳送訊息至其中的電話號碼。此資料會以強型別 MonitorRequest 物件的形式傳遞至協調器。

E3_Monitor 函式會使用協調器函式的標準 function.json。

{
  "bindings": [
    {
      "name": "context",
      "type": "orchestrationTrigger",
      "direction": "in"
    }
  ],
  "disabled": false
}

以下是實作函式的程式碼：

const df = require("durable-functions");
const moment = require("moment");

module.exports = df.orchestrator(function* (context) {
    const input = context.df.getInput();
    context.log(
        "Received monitor request. location: " +
            (input ? input.location : undefined) +
            ". phone: " +
            (input ? input.phone : undefined) +
            "."
    );

    verifyRequest(input);

    const endTime = moment.utc(context.df.currentUtcDateTime).add(6, "h");
    context.log(
        "Instantiating monitor for " +
            input.location.city +
            ", " +
            input.location.state +
            ". Expires: " +
            endTime +
            "."
    );

    while (moment.utc(context.df.currentUtcDateTime).isBefore(endTime)) {
        // Check the weather
        context.log(
            "Checking current weather conditions for " +
                input.location.city +
                ", " +
                input.location.state +
                " at " +
                context.df.currentUtcDateTime +
                "."
        );
        const isClear = yield context.df.callActivity("E3_GetIsClear", input.location);

        if (isClear) {
            // It's not raining! Or snowing. Or misting. Tell our user to take advantage of it.
            context.log(
                "Detected clear weather for " +
                    input.location.city +
                    ", " +
                    input.location.state +
                    ". Notifying " +
                    input.phone +
                    "."
            );

            yield context.df.callActivity("E3_SendGoodWeatherAlert", input.phone);
            break;
        } else {
            // Wait for the next checkpoint
            const nextCheckpoint = moment.utc(context.df.currentUtcDateTime).add(30, "s");
            context.log(
                "Next check for " +
                    input.location.city +
                    ", " +
                    input.location.state +
                    " at " +
                    nextCheckpoint.toString()
            );

            yield context.df.createTimer(nextCheckpoint.toDate()); // accomodate cancellation tokens
        }
    }

    context.log("Monitor expiring.");
});

function verifyRequest(request) {
    if (!request) {
        throw new Error("An input object is required.");
    }
    if (!request.location) {
        throw new Error("A location input is required.");
    }
    if (!request.phone) {
        throw new Error("A phone number input is required.");
    }
}

此協調器函式會執行下列動作：

取得 MonitorRequest，其中包含要監視的「地點」以及要用來作為簡訊通知傳送目的地的「電話號碼」。
判斷監視器的到期時間。為求簡單明瞭，這個範例會使用硬式編碼值。
呼叫 E3_GetIsClear 來判斷所要求之地點的天空是否晴朗。
如果天氣晴朗，則呼叫 E3_SendGoodWeatherAlert 以將簡訊通知傳送至要求的電話號碼。
建立長期計時器以在下一個輪詢間隔繼續進行協調流程。為求簡單明瞭，這個範例會使用硬式編碼值。
持續執行直到目前的國際標準時間超過監視器的到期時間，或傳送簡訊通知之後。

透過多次呼叫協調器函式，多個協調器執行個體可以同步執行。您可以指定要監視的地點以及要作為簡訊通知傳送目的地的電話號碼。最後，請注意協調器函式在等候計時器時「不會」執行，因此您無須支付其費用。

E3_GetIsClear 活動函式

如同其他範例一樣，協助程式活動函式是使用 activityTrigger 觸發程序繫結的一般函式。 E3_GetIsClear 函式會使用 Weather Underground API 取得目前的天氣狀況，並判斷天空是否晴朗。

C#
JavaScript

[FunctionName("E3_GetIsClear")]
public static async Task<bool> GetIsClear([ActivityTrigger] Location location)
{
    var currentConditions = await WeatherUnderground.GetCurrentConditionsAsync(location);
    return currentConditions.Equals(WeatherCondition.Clear);
}

function.json 定義如下：

{
  "bindings": [
    {
      "name": "location",
      "type": "activityTrigger",
      "direction": "in"
    }
  ],
  "disabled": false
}

實作如下。

const request = require("request");

const clearWeatherConditions = [
    "Overcast",
    "Clear",
    "Partly Cloudy",
    "Mostly Cloudy",
    "Scattered Clouds",
];

module.exports = function (context, location) {
    getCurrentConditions(location)
        .then(function (data) {
            const isClear = clearWeatherConditions.includes(data.weather);
            context.done(null, isClear);
        })
        .catch(function (err) {
            context.log(`E3_GetIsClear encountered an error: ${err}`);
            context.done(err);
        });
};

function getCurrentConditions(location) {
    return new Promise(function (resolve, reject) {
        const options = {
            url: `https://api.wunderground.com/api/${process.env["WeatherUndergroundApiKey"]}/conditions/q/${location.state}/${location.city}.json`,
            method: "GET",
            json: true,
        };
        request(options, function (err, res, body) {
            if (err) {
                reject(err);
            }
            if (body.error) {
                reject(body.error);
            }
            if (body.response.error) {
                reject(body.response.error);
            }
            resolve(body.current_observation);
        });
    });
}

E3_SendGoodWeatherAlert 活動函式

E3_SendGoodWeatherAlert 函式會使用 Twilio 繫結來傳送手機簡訊，通知使用者這是散步的好時機。

C#
JavaScript

    [FunctionName("E3_SendGoodWeatherAlert")]
    public static void SendGoodWeatherAlert(
        [ActivityTrigger] string phoneNumber,
        ILogger log,
        [TwilioSms(AccountSidSetting = "TwilioAccountSid", AuthTokenSetting = "TwilioAuthToken", From = "%TwilioPhoneNumber%")]
            out CreateMessageOptions message)
    {
        message = new CreateMessageOptions(new PhoneNumber(phoneNumber));
        message.Body = $"The weather's clear outside! Go take a walk!";
    }

internal class WeatherUnderground
{
    private static readonly HttpClient httpClient = new HttpClient();
    private static IReadOnlyDictionary<string, WeatherCondition> weatherMapping = new Dictionary<string, WeatherCondition>()
    {
        { "Clear", WeatherCondition.Clear },
        { "Overcast", WeatherCondition.Clear },
        { "Cloudy", WeatherCondition.Clear },
        { "Clouds", WeatherCondition.Clear },
        { "Drizzle", WeatherCondition.Precipitation },
        { "Hail", WeatherCondition.Precipitation },
        { "Ice", WeatherCondition.Precipitation },
        { "Mist", WeatherCondition.Precipitation },
        { "Precipitation", WeatherCondition.Precipitation },
        { "Rain", WeatherCondition.Precipitation },
        { "Showers", WeatherCondition.Precipitation },
        { "Snow", WeatherCondition.Precipitation },
        { "Spray", WeatherCondition.Precipitation },
        { "Squall", WeatherCondition.Precipitation },
        { "Thunderstorm", WeatherCondition.Precipitation },
    };

    internal static async Task<WeatherCondition> GetCurrentConditionsAsync(Location location)
    {
        var apiKey = Environment.GetEnvironmentVariable("WeatherUndergroundApiKey");
        if (string.IsNullOrEmpty(apiKey))
        {
            throw new InvalidOperationException("The WeatherUndergroundApiKey environment variable was not set.");
        }

        var callString = string.Format("http://api.wunderground.com/api/{0}/conditions/q/{1}/{2}.json", apiKey, location.State, location.City);
        var response = await httpClient.GetAsync(callString);
        var conditions = await response.Content.ReadAsAsync<JObject>();

        JToken currentObservation;
        if (!conditions.TryGetValue("current_observation", out currentObservation))
        {
            JToken error = conditions.SelectToken("response.error");

            if (error != null)
            {
                throw new InvalidOperationException($"API returned an error: {error}.");
            }
            else
            {
                throw new ArgumentException("Could not find weather for this location. Try being more specific.");
            }
        }

        return MapToWeatherCondition((string)(currentObservation as JObject).GetValue("weather"));
    }

    private static WeatherCondition MapToWeatherCondition(string weather)
    {
        foreach (var pair in weatherMapping)
        {
            if (weather.Contains(pair.Key))
            {
                return pair.Value;
            }
        }

        return WeatherCondition.Other;
    }
}

注意

您必須安裝 Microsoft.Azure.WebJobs.Extensions.Twilio Nuget 套件，才能執行範例程式碼。

其 function.json 很簡單：

{
    "bindings": [
      {
        "name": "phoneNumber",
        "type": "activityTrigger",
        "direction": "in"
      },
      {
        "type": "twilioSms",
        "name": "message",
        "from": "%TwilioPhoneNumber%",
        "accountSidSetting": "TwilioAccountSid",
        "authTokenSetting": "TwilioAuthToken",
        "direction": "out"
      }
    ],
    "disabled": false
  }

以下是傳送手機簡訊的程式碼：

module.exports = function (context, phoneNumber) {
    context.bindings.message = {
        body: `The weather's clear outside! Go take a walk!`,
        to: phoneNumber,
    };

    context.done();
};

執行範例

使用範例中所包含的 HTTP 觸發函式，您可以透過傳送下列 HTTP POST 要求來啟動協調流程：

POST https://{host}/orchestrators/E3_Monitor
Content-Length: 77
Content-Type: application/json

{ "location": { "city": "Redmond", "state": "WA" }, "phone": "+1425XXXXXXX" }

HTTP/1.1 202 Accepted
Content-Type: application/json; charset=utf-8
Location: https://{host}/runtime/webhooks/durabletask/instances/f6893f25acf64df2ab53a35c09d52635?taskHub=SampleHubVS&connection=Storage&code={SystemKey}
RetryAfter: 10

{"id": "f6893f25acf64df2ab53a35c09d52635", "statusQueryGetUri": "https://{host}/runtime/webhooks/durabletask/instances/f6893f25acf64df2ab53a35c09d52635?taskHub=SampleHubVS&connection=Storage&code={systemKey}", "sendEventPostUri": "https://{host}/runtime/webhooks/durabletask/instances/f6893f25acf64df2ab53a35c09d52635/raiseEvent/{eventName}?taskHub=SampleHubVS&connection=Storage&code={systemKey}", "terminatePostUri": "https://{host}/runtime/webhooks/durabletask/instances/f6893f25acf64df2ab53a35c09d52635/terminate?reason={text}&taskHub=SampleHubVS&connection=Storage&code={systemKey}"}

E3_Monitor 執行個體會啟動並查詢所要求地點的目前天氣狀況。如果天氣晴朗，它會呼叫活動函式來傳送警示。否則，它會設定計時器。當計時器時間結束時，協調流程會恢復執行。

在 Azure Functions 入口網站中查看函式記錄，就能看到協調流程的活動。

2018-03-01T01:14:41.649 Function started (Id=2d5fcadf-275b-4226-a174-f9f943c90cd1)
2018-03-01T01:14:42.741 Started orchestration with ID = '1608200bb2ce4b7face5fc3b8e674f2e'.
2018-03-01T01:14:42.780 Function completed (Success, Id=2d5fcadf-275b-4226-a174-f9f943c90cd1, Duration=1111ms)
2018-03-01T01:14:52.765 Function started (Id=b1b7eb4a-96d3-4f11-a0ff-893e08dd4cfb)
2018-03-01T01:14:52.890 Received monitor request. Location: Redmond, WA. Phone: +1425XXXXXXX.
2018-03-01T01:14:52.895 Instantiating monitor for Redmond, WA. Expires: 3/1/2018 7:14:52 AM.
2018-03-01T01:14:52.909 Checking current weather conditions for Redmond, WA at 3/1/2018 1:14:52 AM.
2018-03-01T01:14:52.954 Function completed (Success, Id=b1b7eb4a-96d3-4f11-a0ff-893e08dd4cfb, Duration=189ms)
2018-03-01T01:14:53.226 Function started (Id=80a4cb26-c4be-46ba-85c8-ea0c6d07d859)
2018-03-01T01:14:53.808 Function completed (Success, Id=80a4cb26-c4be-46ba-85c8-ea0c6d07d859, Duration=582ms)
2018-03-01T01:14:53.967 Function started (Id=561d0c78-ee6e-46cb-b6db-39ef639c9a2c)
2018-03-01T01:14:53.996 Next check for Redmond, WA at 3/1/2018 1:44:53 AM.
2018-03-01T01:14:54.030 Function completed (Success, Id=561d0c78-ee6e-46cb-b6db-39ef639c9a2c, Duration=62ms)

一旦達到了協調流程的逾時或偵測到天氣放晴，協調流程就會完成。您也可以使用其他函式內的 terminate API，或叫用上述 202 回應中所參考的 terminatePostUri HTTP POST Webhook。若要使用 Webhook，請將 {text} 取代為提早終止的原因。 HTTP POST URL 看起來大致如下所示：

POST https://{host}/runtime/webhooks/durabletask/instances/f6893f25acf64df2ab53a35c09d52635/terminate?reason=Because&taskHub=SampleHubVS&connection=Storage&code={systemKey}

下一步

這個範例已示範如何使用 Durable Functions 與長期計時器和條件式邏輯來監視外部來源的狀態。下一個範例會說明如何使用外部事件和長期計時器來處理人為互動。

執行人為互動範例