此页面由 Cloud Translation API 翻译。

使用回调和 Eventarc 触发器等待事件

您的工作流可能需要等待外部进程。您可以使用 HTTP 回调来等待其他服务向回调端点发出请求；该请求将继续执行工作流。您还可以等待使用轮询。

本教程演示了如何使用 HTTP 回调和 Eventarc 触发器等待事件或 Pub/Sub 消息，而不是使用轮询。虽然您可以使用事件或 Pub/Sub 消息触发工作流，但您可能希望暂停该执行，等待另一个事件发生后再继续。例如，某个事件触发工作流来启动进程，但工作流需要等待另一个事件来表明进程已完成。您可以通过让一个工作流回调另一个工作流来实现此目的。

创建 Firestore 数据库

Firestore 将数据存储在包含映射到值的字段的文档中。这些文档存储在集合中，而集合是文档中的容器，可用于整理数据和构建查询。详细了解 Firestore。

请注意，每个 Google Cloud 项目只能有一个 Firestore 数据库。如果您需要创建新数据库，请完成以下步骤。

控制台

在 Google Cloud 控制台中，前往 Firestore 开始使用页面。

前往“开始使用”页面
点击选择原生模式。

如需有关选择数据库模式的指导以及逐个功能的比较情况，请参阅选择原生模式或 Datastore 模式。
在选择位置列表中，选择 nam5（美国）。

该位置同时适用于 Google Cloud 项目中的 Firestore 数据库和 App Engine 应用。创建数据库后，您将无法更改位置。
点击创建数据库。

gcloud

如需创建 Firestore 数据库，您必须先创建 App Engine 应用，然后运行 gcloud firestore databases create 命令：

gcloud app create --region=us-central
gcloud firestore databases create --region=us-central

您可以忽略 us-central is not a valid Firestore location 警告。App Engine 和 Firestore 支持相同的位置，但 App Engine us-central（爱荷华）区域位置与 Firestore nam5（美国）多区域位置相对应。

创建 Pub/Sub 主题

本教程使用 Pub/Sub 作为事件来源。创建一个 Pub/Sub 主题，以便您可以向其发布消息。详细了解如何创建和管理主题。

控制台

在 Google Cloud 控制台中，前往 Pub/Sub 主题页面。

打开“主题”。
点击 创建主题。
在主题 ID 字段中，输入 topic-callback。
接受其他默认设置。
点击创建主题。

gcloud

要创建主题，请运行 gcloud pubsub topics create 命令：

gcloud pubsub topics create topic-callback

创建 Cloud Storage 存储桶

本教程使用 Cloud Storage 作为事件来源。创建一个 Cloud Storage 存储桶，以便您可以向其中上传文件。详细了解如何创建存储分区。

控制台

在 Google Cloud 控制台中，前往 Cloud Storage 存储分区页面。

转到 Cloud Storage
点击创建。
在存储桶的名称中，输入 PROJECT_ID-bucket-callback。

项目 ID 用于 callback-event-sample 工作流中，以标识相应存储桶。
点击继续。
对于位置类型，请选择区域，然后选择 us-central1（爱荷华）。
接受其他默认设置。
点击创建。

gcloud

如需创建存储桶，请运行 gcloud storage buckets create 命令：

gcloud storage buckets create gs://PROJECT_ID-bucket-callback \
    --location=us-central1

项目 ID 用于 callback-event-sample 工作流中，以标识相应存储桶。

创建事件来源后，您可以部署事件接收器工作流。

部署监听事件的工作流

当消息发布到 Pub/Sub 主题或文件上传到 Cloud Storage 存储桶时，系统会触发 callback-event-listener 工作流。工作流接收事件，从 Firestore 数据库中检索相应回调的详细信息，然后向回调端点发送 HTTP 请求。

控制台

在 Google Cloud 控制台中，前往 Workflows 页面：

转到 Workflows
点击创建。
输入新工作流的名称：callback-event-listener。
在区域列表中，选择 us-central1。
选择您之前创建的服务账号。
点击下一步。

在工作流编辑器中，输入工作流的定义：

main:
  params: [event]
  steps:
    - log_event:
        call: sys.log
        args:
          text: ${event}
          severity: INFO
    - init:
        assign:
          - database_root: ${"projects/" + sys.get_env("GOOGLE_CLOUD_PROJECT_ID") + "/databases/(default)/documents/callbacks/"}
          - event_source_tokens: ${text.split(event.source, "/")}
          - event_source_len: ${len(event_source_tokens)}
          - event_source: ${event_source_tokens[event_source_len - 1]}
          - doc_name: ${database_root + event_source}
    - get_document_for_event_source:
        try:
          call: googleapis.firestore.v1.projects.databases.documents.get
          args:
            name: ${doc_name}
          result: document
        except:
            as: e
            steps:
                - known_errors:
                    switch:
                    - condition: ${e.code == 404}
                      return: ${"No callbacks for event source " + event_source}
                - unhandled_exception:
                    raise: ${e}
    - process_callback_urls:
        steps:
          - check_fields_exist:
              switch:
              - condition: ${not("fields" in document)}
                return: ${"No callbacks for event source " + event_source}
              - condition: true
                next: processFields
          - processFields:
              for:
                  value: key
                  in: ${keys(document.fields)}
                  steps:
                      - extract_callback_url:
                          assign:
                              - callback_url: ${document.fields[key]["stringValue"]}
                      - log_callback_url:
                          call: sys.log
                          args:
                            text: ${"Calling back url " + callback_url}
                            severity: INFO
                      - http_post:
                          call: http.post
                          args:
                              url: ${callback_url}
                              auth:
                                  type: OAuth2
                              body:
                                  event: ${event}

点击部署。

gcloud

为工作流创建源代码文件：
```
touch callback-event-listener.yaml
```

在文本编辑器中，将以下工作流复制到您的源代码文件中：

main:
  params: [event]
  steps:
    - log_event:
        call: sys.log
        args:
          text: ${event}
          severity: INFO
    - init:
        assign:
          - database_root: ${"projects/" + sys.get_env("GOOGLE_CLOUD_PROJECT_ID") + "/databases/(default)/documents/callbacks/"}
          - event_source_tokens: ${text.split(event.source, "/")}
          - event_source_len: ${len(event_source_tokens)}
          - event_source: ${event_source_tokens[event_source_len - 1]}
          - doc_name: ${database_root + event_source}
    - get_document_for_event_source:
        try:
          call: googleapis.firestore.v1.projects.databases.documents.get
          args:
            name: ${doc_name}
          result: document
        except:
            as: e
            steps:
                - known_errors:
                    switch:
                    - condition: ${e.code == 404}
                      return: ${"No callbacks for event source " + event_source}
                - unhandled_exception:
                    raise: ${e}
    - process_callback_urls:
        steps:
          - check_fields_exist:
              switch:
              - condition: ${not("fields" in document)}
                return: ${"No callbacks for event source " + event_source}
              - condition: true
                next: processFields
          - processFields:
              for:
                  value: key
                  in: ${keys(document.fields)}
                  steps:
                      - extract_callback_url:
                          assign:
                              - callback_url: ${document.fields[key]["stringValue"]}
                      - log_callback_url:
                          call: sys.log
                          args:
                            text: ${"Calling back url " + callback_url}
                            severity: INFO
                      - http_post:
                          call: http.post
                          args:
                              url: ${callback_url}
                              auth:
                                  type: OAuth2
                              body:
                                  event: ${event}

输入以下命令以部署工作流：

gcloud workflows deploy callback-event-listener \
    --source=callback-event-listener.yaml \
    --location=us-central1 \
    --service-account=SERVICE_ACCOUNT_NAME@PROJECT_ID.iam.gserviceaccount.com

将 SERVICE_ACCOUNT_NAME 替换为您之前创建的服务账号的名称。

部署等待事件的工作流

callback-event-sample 工作流将其回调详细信息存储在 Firestore 数据库中，暂停执行，然后等待特定事件发生。

控制台

在 Google Cloud 控制台中，前往 Workflows 页面：

转到 Workflows
点击创建。
输入新工作流的名称：callback-event-sample。
在区域列表中，选择 us-central1。
选择您之前创建的服务账号。
点击下一步。

在工作流编辑器中，输入工作流的定义：

main:
  steps:
    - init:
        assign:
          - pubsub_topic: topic-callback
          - storage_bucket: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID") + "-bucket-callback"}
    - await_pubsub_message:
        call: await_callback_event
        args:
          event_source: ${pubsub_topic}
        result: pubsub_event
    - await_storage_bucket:
        call: await_callback_event
        args:
          event_source: ${storage_bucket}
        result: storage_event
    - return_events:
        return:
            pubsub_event: ${pubsub_event}
            storage_event: ${storage_event}

await_callback_event:
    params: [event_source]
    steps:
        - init:
            assign:
              - database_root: ${"projects/" + sys.get_env("GOOGLE_CLOUD_PROJECT_ID") + "/databases/(default)/documents/callbacks/"}
              - doc_name: ${database_root + event_source}
              - execution_id: ${sys.get_env("GOOGLE_CLOUD_WORKFLOW_EXECUTION_ID")}
              - firestore_key: ${"exec_" + text.split(execution_id, "-")[0]}
        - create_callback:
            call: events.create_callback_endpoint
            args:
              http_callback_method: POST
            result: callback_details
        - save_callback_url:
            call: googleapis.firestore.v1.projects.databases.documents.patch
            args:
              name: ${doc_name}
              updateMask:
                fieldPaths: ["${firestore_key}"]
              body:
                fields:
                  ${firestore_key}:
                    stringValue: ${callback_details.url}
        - log_and_await_callback:
            try:
              steps:
                - log_await_start:
                    call: sys.log
                    args:
                      severity: INFO
                      data: ${"Started waiting 1hr for an event from source " + event_source}
                - await_callback:
                    call: events.await_callback
                    args:
                      callback: ${callback_details}
                      timeout: 3600
                    result: callback_request
                - log_await_stop:
                    call: sys.log
                    args:
                      severity: INFO
                      data: ${"Stopped waiting for an event from source " + event_source}
            except:
                as: e
                steps:
                    - log_error:
                        call: sys.log
                        args:
                            severity: "ERROR"
                            text: ${"Received error " + e.message}
        - delete_callback_url:
            call: googleapis.firestore.v1.projects.databases.documents.patch
            args:
              name: ${doc_name}
              updateMask:
                fieldPaths: ["${firestore_key}"]
        - check_null_event:
            switch:
              - condition: ${callback_request == null}
                return: null
        - log_await_result:
            call: sys.log
            args:
              severity: INFO
              data: ${callback_request.http_request.body.event}
        - return_event:
            return: ${callback_request.http_request.body.event}

点击部署。

gcloud

为工作流创建源代码文件：
```
touch callback-event-sample.yaml
```

在文本编辑器中，将以下工作流复制到您的源代码文件中：

main:
  steps:
    - init:
        assign:
          - pubsub_topic: topic-callback
          - storage_bucket: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID") + "-bucket-callback"}
    - await_pubsub_message:
        call: await_callback_event
        args:
          event_source: ${pubsub_topic}
        result: pubsub_event
    - await_storage_bucket:
        call: await_callback_event
        args:
          event_source: ${storage_bucket}
        result: storage_event
    - return_events:
        return:
            pubsub_event: ${pubsub_event}
            storage_event: ${storage_event}

await_callback_event:
    params: [event_source]
    steps:
        - init:
            assign:
              - database_root: ${"projects/" + sys.get_env("GOOGLE_CLOUD_PROJECT_ID") + "/databases/(default)/documents/callbacks/"}
              - doc_name: ${database_root + event_source}
              - execution_id: ${sys.get_env("GOOGLE_CLOUD_WORKFLOW_EXECUTION_ID")}
              - firestore_key: ${"exec_" + text.split(execution_id, "-")[0]}
        - create_callback:
            call: events.create_callback_endpoint
            args:
              http_callback_method: POST
            result: callback_details
        - save_callback_url:
            call: googleapis.firestore.v1.projects.databases.documents.patch
            args:
              name: ${doc_name}
              updateMask:
                fieldPaths: ["${firestore_key}"]
              body:
                fields:
                  ${firestore_key}:
                    stringValue: ${callback_details.url}
        - log_and_await_callback:
            try:
              steps:
                - log_await_start:
                    call: sys.log
                    args:
                      severity: INFO
                      data: ${"Started waiting 1hr for an event from source " + event_source}
                - await_callback:
                    call: events.await_callback
                    args:
                      callback: ${callback_details}
                      timeout: 3600
                    result: callback_request
                - log_await_stop:
                    call: sys.log
                    args:
                      severity: INFO
                      data: ${"Stopped waiting for an event from source " + event_source}
            except:
                as: e
                steps:
                    - log_error:
                        call: sys.log
                        args:
                            severity: "ERROR"
                            text: ${"Received error " + e.message}
        - delete_callback_url:
            call: googleapis.firestore.v1.projects.databases.documents.patch
            args:
              name: ${doc_name}
              updateMask:
                fieldPaths: ["${firestore_key}"]
        - check_null_event:
            switch:
              - condition: ${callback_request == null}
                return: null
        - log_await_result:
            call: sys.log
            args:
              severity: INFO
              data: ${callback_request.http_request.body.event}
        - return_event:
            return: ${callback_request.http_request.body.event}

输入以下命令以部署工作流：

gcloud workflows deploy callback-event-sample \
    --source=callback-event-sample.yaml \
    --location=us-central1 \
    --service-account=SERVICE_ACCOUNT_NAME@PROJECT_ID.iam.gserviceaccount.com

将 SERVICE_ACCOUNT_NAME 替换为您之前创建的服务账号的名称。

创建 Eventarc 触发器以路由 Pub/Sub 事件

借助 Eventarc 触发器，您可以通过为触发器指定过滤条件（包括事件来源和目标工作流）来路由事件。创建一个 Eventarc 触发器，以便在向 Pub/Sub 主题发布消息后执行 callback-event-listener 工作流。详细了解如何触发工作流。

控制台

在 Google Cloud 控制台中，前往 Eventarc 页面。

转到 Eventarc
点击 创建触发器。
输入触发器名称。

例如 trigger-pubsub-events-listener。
在事件提供方列表中，选择 Cloud Pub/Sub。
在事件列表中，选择自定义下的 google.cloud.pubsub.topic.v1.messagePublished。
在选择 Cloud Pub/Sub 主题列表中，选择您之前创建的主题。
在区域列表中，选择 us-central1（爱荷华）。
如果出现提示，请向 Pub/Sub 服务账号授予 iam.serviceAccountTokenCreator 角色。
选择您之前创建的服务账号。
在事件目的地列表中，选择 Workflows。
在选择工作流列表中，选择 callback-event-listener 工作流。
点击创建。

gcloud

如需创建触发器，请运行 gcloud eventarc triggers create 命令：

gcloud eventarc triggers create trigger-pubsub-events-listener \
    --location=us-central1 \
    --destination-workflow=callback-event-listener \
    --destination-workflow-location=us-central1 \
    --event-filters="type=google.cloud.pubsub.topic.v1.messagePublished" \
    --transport-topic=topic-callback \
    --service-account=SERVICE_ACCOUNT_NAME@PROJECT_ID.iam.gserviceaccount.com

事件会被转换并作为运行时参数传递给工作流执行。请注意，新触发器最多可能需要 2 分钟才会变为有效状态。

创建 Eventarc 触发器以路由 Cloud Storage 事件

借助 Eventarc 触发器，您可以通过为触发器指定过滤条件（包括事件来源和目标工作流）来路由事件。创建一个 Eventarc 触发器，以便在将文件上传到 Cloud Storage 存储桶后执行 callback-event-listener 工作流。详细了解如何触发工作流。

控制台

在 Google Cloud 控制台中，前往 Eventarc 页面。

转到 Eventarc
点击 创建触发器。
输入触发器名称。

例如 trigger-storage-events-listener。
在事件提供方列表中，选择 Cloud Storage。
在事件列表中，选择直接下的 google.cloud.storage.object.v1.finalized。
在存储桶列表中，浏览并选择您之前创建的存储桶。
在区域列表中，根据您的 Cloud Storage 存储桶，接受默认值 us-central1（爱荷华）。
如果出现提示，请向 Pub/Sub 服务账号授予 iam.serviceAccountTokenCreator 角色。
选择您之前创建的服务账号。
在事件目的地列表中，选择 Workflows。
在选择工作流列表中，选择 callback-event-listener 工作流。
点击创建。

gcloud

如需创建触发器，请运行 gcloud eventarc triggers create 命令：

gcloud eventarc triggers create trigger-storage-events-listener \
    --location=us-central1 \
    --destination-workflow=callback-event-listener \
    --destination-workflow-location=us-central1 \
    --event-filters="type=google.cloud.storage.object.v1.finalized" \
    --event-filters="bucket=PROJECT_ID-bucket-callback" \
    --service-account=SERVICE_ACCOUNT_NAME@PROJECT_ID.iam.gserviceaccount.com

事件会被转换并作为运行时参数传递给工作流执行。请注意，新触发器最多可能需要 2 分钟才会变为有效状态。

执行主要工作流

执行某个工作流会运行与该工作流关联的当前工作流定义。执行 callback-event-sample 工作流。这是主要工作流，它会等待特定事件发生，只有在次要工作流发出适当的回调请求时才会继续执行。

控制台

在 Google Cloud 控制台中，前往 Workflows 页面。

转到 Workflows
在工作流页面上，点击 callback-event-sample 工作流以转到其详情页面。
在工作流详细信息页面上，选择执行。
再次点击执行。

工作流执行开始。执行运行时，您应该会看到执行状态为 Running，以及类似于以下内容的日志条目：Started waiting 1hr for an event from source topic-callback。

gcloud

如需执行工作流，请运行 gcloud workflows run 命令：

gcloud workflows run callback-event-sample \
    --location=us-central1

工作流执行开始。执行运行时，您应该会看到类似于以下内容的执行状态：

Waiting for execution [a848a164-268a-449c-b2fe-396f32f2ed66] to complete...working...

生成事件并检查执行状态

您可以生成事件、查看日志条目并检查工作流执行状态，以确认结果是否符合预期。

发布消息

向您之前创建的 Pub/Sub 主题发布消息。

控制台

在 Google Cloud 控制台中，前往 Pub/Sub 主题页面。

打开“主题”。
点击 topic-callback。
点击消息标签页。
点击发布消息。
在消息正文字段中，输入 Hello World。
点击发布。

gcloud

要发布消息，请使用 gcloud pubsub topics publish 命令：

gcloud pubsub topics publish topic-callback \
    --message="Hello World"

上传对象

将文件上传到您之前创建的 Cloud Storage 存储桶。

控制台

在 Google Cloud 控制台中，转到 Cloud Storage 存储桶页面。

进入“存储桶”
点击您之前创建的存储桶的名称。
在对象标签页中，执行以下任一操作：
- 将所需文件从桌面或文件管理器拖放到 Google Cloud 控制台的主窗格中。
- 点击上传文件，选择要上传的文件，然后点击打开。

gcloud

如需上传文件，请运行 gcloud storage cp 命令：

gcloud storage cp OBJECT_LOCATION gs://PROJECT_ID-bucket-callback/

将 OBJECT_LOCATION 替换为对象的本地路径。例如 random.txt。

查看日志条目和执行状态

确认 callback-event-sample 工作流已成功完成。

控制台

在 Google Cloud 控制台中，前往 Workflows 页面。

转到 Workflows
在工作流页面上，点击 callback-event-sample 工作流以转到其详情页面。
在工作流详细信息页面上，如需检索特定执行的详细信息，请点击相应的执行 ID。

执行状态应为成功，并且在“输出”窗格中，您应该会看到收到的 Pub/Sub 和 Cloud Storage 事件。

gcloud

过滤日志条目并以 JSON 格式返回输出：

gcloud logging read "resource.type=workflows.googleapis.com/Workflow AND textPayload:calling OR textPayload:waiting" \
    --format=json

查找类似如下内容的日志条目：

"textPayload": "Stopped waiting for an event from source..."
"textPayload": "Calling back url https://workflowexecutions.googleapis.com/v1/projects/..."
"textPayload": "Started waiting 1hr for an event from source..."

检查最后一次执行尝试的状态：

gcloud workflows executions wait-last

结果应如下所示：

Using cached execution name: projects/1085953646031/locations/us-central1/workflows/callback-event-sample/executions/79929e4e-82c1-4da1-b068-f828034c01b7
Waiting for execution [79929e4e-82c1-4da1-b068-f828034c01b7] to complete...done.
[...]
state: SUCCEEDED