wait-for-callback パターンを使用した Simple Message Queue (旧称：MNS) の統合 - CloudFlow

このトピックでは、タスクステップで wait-for-callback パターンを使用して Simple Message Queue (formerly MNS) トピックを統合し、メッセージをパブリッシュする方法について説明します。Simple Message Queue (formerly MNS)

仕組み

アプリケーションをデプロイすると、フローは次のように実行されます：

フローが開始され、タスクステップが MNS トピックにメッセージをパブリッシュします。メッセージ本文には、タスクステップの TaskToken が含まれます。
タスクステップは一時停止し、コールバックを待ちます。
Simple Message Queue (formerly MNS) のトピックは、メッセージを受信すると、HTTP プッシュを介してメッセージと TaskToken を Function Compute (FC) 関数の HTTP トリガーに送信し、関数をトリガーして実行します。
FC 関数は TaskToken を取得し、ReportTaskSucceeded API を呼び出してタスクのステータスを報告します。
フローは実行を再開します。

アプリケーションのデプロイ

Serverless Workflow コンソールにログインします。
[フロー] ページで、[フローの作成] をクリックします。
[フローの作成] ページで、[サンプルプロジェクト] > [Simple Message Queue (formerly MNS) トピックのタスクステップオーケストレーション] テンプレートを選択し、[次へ] をクリックします。
[アプリケーションの作成] ページで、必須パラメーターを設定し、[デプロイ] をクリックします。
- [アプリケーション名]：アプリケーションのカスタム名を入力します。名前はアカウント内で一意である必要があります。
- [トピック名]：トピックのカスタム名を入力します。指定した Simple Message Queue (formerly MNS) トピックが存在しない場合は、自動的に作成されます。
[デプロイ] をクリックすると、ページにアプリケーション用に作成されたすべてのリソースが表示され、その後リソーススタックの詳細ページにリダイレクトされます。 [概要] タブでは、リソースリストで各リソースの作成の進捗を確認できます。リソースのステータスには、作成中や初期化完了などがあります。
フローを実行します。
次の入力でフローを実行します：
```
{
   "messageBody": "hello world"
}
```
実行が成功すると、そのステータスを確認できます。

実行詳細ページでは、ステータスが成功と表示されます。ビジュアルフローでは、開始から mns-topic-task を経て終了までのパス全体が正常に実行されたことが示されます。

アプリケーションコード

このフローは Simple Message Queue (formerly MNS) トピックをオーケストレーションします。
タスクステップのコールバック用の TaskToken は、後続のコールバックのためにメッセージの MessageBody にカプセル化されます。outputMappings セクションは、ReportTaskSucceeded によって設定された output を読み取ります。
```
version: v1
type: flow
steps:
 - type: task
 name: mns-topic-task
 resourceArn: acs:mns:::/topics/<topic>/messages
 pattern: waitForCallback
 inputMappings:
 - target: messageBody
 source: $input.messageBody
 - target: taskToken
 source: $context.task.token
 outputMappings:
 - target: status
 source: $local.status
 serviceParams:
 MessageBody: $
```

この FC 関数は、タスクステップにコールバックを送信します。

この関数は、MessageBody にカプセル化された TaskToken を読み取り、output を {"status":"success"} に設定することでタスクのステータスを報告します。

def handler(environ, start_response):
 # リクエストボディを取得
 try:
 request_body_size = int(environ.get('CONTENT_LENGTH',
0))
 except ValueError:
 request_body_size = 0
 request_body =
environ['wsgi.input'].read(request_body_size)
 print('Request body:
{}'.format(request_body))
 body = json.loads(request_body)
 message_body_str =
body['Message']
 # メッセージ本文から MessageBody と TaskToken を読み取り
 message_body =
json.loads(message_body_str)
 task_token =
message_body['taskToken']
 ori_message_body =
message_body['messageBody']
 print('Task token: {}\norigin message
body: {}'.format(task_token, ori_message_body))
 # STS トークンを使用して fnf クライアントを初期化
 context = environ['fc.context']
 creds = context.credentials
 sts_creds =
StsTokenCredential(creds.access_key_id, creds.access_key_secret, creds.security_token)
 fnf_client =
AcsClient(credential=sts_creds, region_id=context.region)
 # サーバーレスワークフローにタスクの成功を報告
 req =
ReportTaskSucceededRequest()
 req.set_TaskToken(task_token)
 req.set_Output('{"status":
"success"}')
 resp =
fnf_client.do_action_with_exception(req)
 print('Report task response:
{}'.format(resp))
 # HTTP リクエストへの応答
 status = '200 OK'
 response_headers = [('Content-type',
'text/plain')]
 start_response(status,
response_headers)
 return [b'OK']

CloudFlow:Simple Message Queue (formerly MNS)

仕組み

アプリケーションのデプロイ

アプリケーションコード

関連情報