FHIR Pub/Sub 通知

本页介绍了如何在 FHIR 存储区中发生临床事件时使用 Pub/Sub 接收通知。

您可以将 Pub/Sub 通知用于多种用例,包括触发对新数据的下游处理或分析。例如,机器学习模型可以在有新数据可供训练时接收通知,并生成数据洞见或预测结果,以便改进患者护理。

概览

当 FHIR 存储区中创建、更新、修补或删除 FHIR 资源时,您可以接收 Pub/Sub 通知。当从 Cloud Storage 导入 FHIR 资源时,Cloud Healthcare API 不会发送通知。

如需接收通知,您必须创建 Pub/Sub 主题和订阅,然后配置 FHIR 存储区以向该主题发送通知。

下图展示了在使用 fhir.create 方法在 FHIR 存储区中创建 FHIR 资源时,如何创建和传送 Pub/Sub 通知。更新、修补或删除 FHIR 资源时,步骤也是一样的。

FHIR Pub/Sub 通知。

图 1. 针对 FHIR 存储区中的更改使用 Pub/Sub 通知。

图 1 显示了以下步骤:

  1. 调用方发出 fhirStores.fhir.create 请求以创建 FHIR 资源。
  2. FHIR 存储区会接收请求,创建一条 Pub/Sub 消息,然后将其发送到 FHIR 存储区上配置的 Pub/Sub 主题。
  3. Pub/Sub 会将消息转发给附加到该主题的订阅。
  4. 订阅者会收到来自订阅的通知,其采用 Pub/Sub 消息形式。每个订阅可以有一个或多个订阅者(提高并行性)。

通知配置

您可以在 FHIR 存储区的 FhirNotificationConfig 对象中配置 Pub/Sub 通知及其行为。每个 FHIR 存储区可以配置一个 FhirNotificationConfig

下表介绍了 FhirNotificationConfig 对象中的字段。

字段 说明 示例
pubsubTopic 要附加到 FHIR 存储区的 Pub/Sub 主题。系统会向指定的主题发送通知。 projects/my-project/topics/my-topic
sendFullResource 是否在通知中包含已创建、更新或修补的 FHIR 资源的完整内容。此字段对删除 FHIR 资源时发送的通知没有影响。如需包含已删除资源的完整内容,请将 sendPreviousResourceOnDelete 设置为 true true
sendPreviousResourceOnDelete 是否在通知中包含已删除的 FHIR 资源的完整内容。此字段对创建、更新或修补 FHIR 资源时发送的通知没有影响。 true

通知格式和内容

每个 Pub/Sub 通知都包含一个 message 对象,用于存储与临床事件相关的信息。message 对象类似于以下内容:

{
  "message": {
    "attributes": {
      "action": "ACTION",
      "lastUpdatedTime": "RFC_1123_FORMAT_DATETIME",
      "payloadType": "PAYLOAD_TYPE",
      "resourceType": "FHIR_RESOURCE_TYPE",
      "storeName": "projects/PROJECT_ID/locations/LOCATION/datasets/DATASET_ID/fhirStores/FHIR_STORE_ID",
      "versionId": "VERSION_ID"
    },
    "data": "BASE_64_ENCODED_DATA",
    "messageId": "MESSAGE_ID",
    "publishTime": "YYYY-MM-DDTHH:MM:SS+ZZ:ZZ"
  }
}

如需了解每条 Pub/Sub 消息中包含的其他字段,请参阅 ReceivedMessagePubsubMessage

下表介绍了 attributes 对象中的每个字段:

属性 说明 示例
action 在 FHIR 资源上发生的操作。可能的值包括:
  • CreateResource
  • PatchResource
  • UpdateResource
  • DeleteResource
CreateResource
resourceType 已修改的 FHIR 资源的类型。可能的值包括 Cloud Healthcare API 中支持的任何 FHIR 资源类型 Patient
payloadType 消息的载荷类型,NameOnlyFullResource FullResource
storeName 发生操作的 FHIR 存储区的完整资源名称。 projects/my-project/locations/us/datasets/my-dataset/fhirStores/my-fhir-store
lastUpdatedTime 最近一次修改 FHIR 资源的时间戳。时间戳使用 RFC 1123 格式。 Mon, 01 Jan 2020 00:00:00 UTC
versionId 发生操作的 FHIR 资源的最新版本的 ID。如需详细了解版本 ID,请参阅列出 FHIR 资源版本 MTY4MzA2MDQzOTI5NjIxMDAwMA

下表介绍了 message 对象中的其余字段:

字段 说明 示例
data FHIR 资源名称或 FHIR 资源内容的 base 64 编码字符串,具体取决于 FhirNotificationConfig 中指定的值。
messageId Pub/Sub 消息的标识符。
publishTime Pub/Sub 服务器发布消息的时间。

指定要包含在通知中的信息

Pub/Sub 通知(如通知格式和内容中所详述)包含一组标准字段。您可以在每条通知中包含完整的 FHIR 资源,也可以仅包含其名称。资源名称包含资源的完整路径和资源 ID,格式如下:

projects/PROJECT_ID/locations/LOCATION/datasets/DATASET_ID/fhirStores/FHIR_STORE_ID/fhir/RESOURCE_TYPE/RESOURCE_ID

FHIR 资源信息以 base 64 编码的字符串形式存储在 data 字段中。

通过添加 FHIR 资源的完整内容,您无需单独查询 Pub/Sub 和 Cloud Healthcare API 即可获取资源详细信息。

获取 FHIR 资源名称

如需在创建、更新或修补 FHIR 资源时仅包含 FHIR 资源的名称,请将 sendFullResource 设置为 false。如需在删除 FHIR 资源时仅包含名称,请将 sendPreviousResourceOnDelete 设置为 false

查看通知时,message 对象类似于以下内容:

{
  "message": {
    "attributes": {
      "action": "{CreateResource|PatchResource|UpdateResource|DeleteResource}",
      "lastUpdatedTime": "RFC_1123_FORMAT_DATETIME",
      "payloadType": "NameOnly",
      "resourceType": "FHIR_RESOURCE_TYPE",
      "storeName": "projects/PROJECT_ID/locations/LOCATION/datasets/DATASET_ID/fhirStores/FHIR_STORE_ID",
      "versionId": "VERSION_ID"
    },
    "data": "BASE64_ENCODED_FHIR_RESOURCE_NAME",
    "messageId": "MESSAGE_ID",
    "publishTime": "YYYY-MM-DDTHH:MM:SS+ZZ:ZZ"
  }
}

请注意通知中的以下事项:

  • payloadType 字段设置为 NameOnly,以指示请求的以下信息:

    • 对于创建、更新和修补操作,sendFullResource 设为 false
    • 对于删除操作,sendPreviousResourceOnDelete 会设置为 false
  • data 字段中仅包含 FHIR 资源名称。该名称会编码为 base 64 编码的字符串。

获取创建、更新或修补的 FHIR 资源内容

如需在创建、更新或修补 FHIR 资源时添加该资源的完整内容,请将 sendFullResource 设置为 true

如果您删除 FHIR 资源,则此行为不适用。如果您删除了 FHIR 资源,并且 sendFullResource 设置为 true,但 sendPreviousResourceOnDelete 设置为 false,则通知与仅检索 FHIR 资源名称 时相同。如需在删除 FHIR 资源时包含 FHIR 资源内容,请参阅获取已删除的 FHIR 资源内容

查看通知时,message 对象类似于以下内容:

{
  "message": {
    "attributes": {
      "action": "{CreateResource|PatchResource|UpdateResource}",
      "lastUpdatedTime": "RFC_1123_FORMAT_DATETIME",
      "payloadType": "FullResource",
      "resourceType": "FHIR_RESOURCE_TYPE",
      "storeName": "projects/PROJECT_ID/locations/LOCATION/datasets/DATASET_ID/fhirStores/FHIR_STORE_ID",
      "versionId": "VERSION_ID"
    },
    "data": "BASE64_ENCODED_FHIR_RESOURCE_CONTENTS",
    "messageId": "MESSAGE_ID",
    "publishTime": "YYYY-MM-DDTHH:MM:SS+ZZ:ZZ"
  }
}

请注意通知中的以下事项:

  • payloadType 设置为 FullResource 表示 sendFullResource 设置为 true。FHIR 资源的完整内容作为 base 64 编码的字符串包含在 data 字段中。
  • data 字段包含以 base 64 编码的字符串形式的 FHIR 资源内容。

获取已删除的 FHIR 资源内容

如需在删除 FHIR 资源时包含其完整内容,请将 sendPreviousResourceOnDelete 设为 true

查看通知时,message 对象类似于以下内容:

{
  "message": {
    "attributes": {
      "action": "DeleteResource",
      "lastUpdatedTime": "RFC_1123_FORMAT_DATETIME",
      "payloadType": "FullResource",
      "resourceType": "FHIR_RESOURCE_TYPE",
      "storeName": "projects/PROJECT_ID/locations/LOCATION/datasets/DATASET_ID/fhirStores/FHIR_STORE_ID",
      "versionId": "VERSION_ID"
    },
    "data": "BASE64_ENCODED_FHIR_RESOURCE_CONTENTS",
    "messageId": "MESSAGE_ID",
    "publishTime": "YYYY-MM-DDTHH:MM:SS+ZZ:ZZ"
  }
}

请记下以下字段中的值:

  • 即使 sendFullResource 设置为 falsepayloadType 也会设置为 FullResource

    FHIR 资源的完整内容作为 base 64 编码的字符串包含在 data 字段中。

  • data 字段包含 FHIR 资源的内容(在资源被删除之前,该内容为 base 64 编码的字符串)。

配置和查看 FHIR 通知

以下示例展示了如何在 FHIR 存储区中创建 FHIR 资源时查看生成的 Pub/Sub 通知。

准备工作

在配置和使用 Pub/Sub 通知之前,请先完成以下部分:

查看 Pub/Sub 配额

熟悉 Pub/Sub 配额和限制。如需了解如何查看配额、申请更多配额,以及配额用尽时会发生什么情况,请参阅使用配额

启用 Pub/Sub API

在 Google Cloud 控制台中,启用 Pub/Sub API:

启用 API

配置 Pub/Sub 权限

要允许将消息从 Cloud Healthcare API 发布到 Pub/Sub,您必须将 pubsub.publisher 角色添加到项目的 Cloud Healthcare Service Agent 服务账号。如需查看添加所需角色的步骤,请参阅 DICOM、FHIR 和 HL7v2 存储区 Pub/Sub 权限

创建 Pub/Sub 主题

如需创建主题,请参阅创建主题

单个数据存储区可以有自己的 Pub/Sub 主题,或者多个数据存储区可以共享同一主题。

指定 Pub/Sub 主题时,请使用以下格式:

projects/PROJECT_ID/topics/TOPIC_NAME

PROJECT_ID 是您的 Google Cloud 项目 ID,TOPIC_NAME 是 Pub/Sub 主题的名称。

创建 Pub/Sub 订阅

如需接收发布到主题的消息,您需要创建 Pub/Sub 订阅。每个 Pub/Sub 主题都需要至少一个 Pub/Sub 订阅。订阅将主题与负责接收和处理发布到该主题的消息的订阅者应用关联起来。

如需创建订阅并将其附加到 Pub/Sub 主题,请参阅创建订阅

创建或修改 FHIR 存储区

使用已配置的 FhirNotificationConfig 对象创建修改 FHIR 存储区。

以下示例展示了如何修改现有 FHIR 存储区。sendFullResourcesendPreviousResourceOnDelete 字段设置为 true,这意味着在创建、更新、修补或删除资源时,通知会包含完整的 FHIR 资源内容。

REST

如需修改 FHIR 存储区,请使用 projects.locations.datasets.fhirStores.patch 方法。

在使用任何请求数据之前,请先进行以下替换:

  • PROJECT_ID:您的 Google Cloud 项目的 ID
  • LOCATION:数据集位置
  • DATASET_ID:FHIR 存储区的父数据集
  • FHIR_STORE_ID:FHIR 存储区 ID
  • PUBSUB_TOPIC_ID:Pub/Sub 主题 ID

请求 JSON 正文:

{
  "notificationConfigs": [
    {
      "pubsubTopic": "projects/PROJECT_ID/topics/PUBSUB_TOPIC_ID",
      "sendFullResource": true,
      "sendPreviousResourceOnDelete": true
    }
  ]
}

如需发送请求,请选择以下方式之一:

将请求正文保存在名为 request.json 的文件中。在终端中运行以下命令,在当前目录中创建或覆盖此文件:

cat > request.json << 'EOF'
{
  "notificationConfigs": [
    {
      "pubsubTopic": "projects/PROJECT_ID/topics/PUBSUB_TOPIC_ID",
      "sendFullResource": true,
      "sendPreviousResourceOnDelete": true
    }
  ]
}
EOF

然后,执行以下命令以发送 REST 请求:

curl -X PATCH \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json; charset=utf-8" \
-d @request.json \
"https://healthcare.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/datasets/DATASET_ID/fhirStores/FHIR_STORE_ID?updateMask=notificationConfigs"

将请求正文保存在名为 request.json 的文件中。在终端中运行以下命令,在当前目录中创建或覆盖此文件:

@'
{
  "notificationConfigs": [
    {
      "pubsubTopic": "projects/PROJECT_ID/topics/PUBSUB_TOPIC_ID",
      "sendFullResource": true,
      "sendPreviousResourceOnDelete": true
    }
  ]
}
'@  | Out-File -FilePath request.json -Encoding utf8

然后,执行以下命令以发送 REST 请求:

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

Invoke-WebRequest `
-Method PATCH `
-Headers $headers `
-ContentType: "application/json; charset=utf-8" `
-InFile request.json `
-Uri "https://healthcare.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/datasets/DATASET_ID/fhirStores/FHIR_STORE_ID?updateMask=notificationConfigs" | Select-Object -Expand Content

您应该收到类似以下内容的 JSON 响应:

{
  "name": "projects/PROJECT_ID/locations/LOCATION/datasets/DATASET_ID/fhirStores/FHIR_STORE_ID",
  "version": "R4",
  "notificationConfigs": [
    {
      "pubsubTopic": "projects/PROJECT_ID/topics/PUBSUB_TOPIC_ID",
      "sendFullResource": true,
      "sendPreviousResourceOnDelete": true
    }
  ]
}

创建 FHIR 资源

在 FHIR 存储区中创建 FHIR 资源。该请求会导致 Cloud Healthcare API 向配置的 Pub/Sub 主题发布消息。

查看 Pub/Sub 通知

查看发布到 Pub/Sub 主题的消息。在 FHIR 存储区中创建患者资源时,系统会生成以下消息。

在示例输出中,FHIR 资源的内容位于 data 字段中的 base64 编码字符串中。您必须对 base64 编码的值进行解码才能获取内容。 大多数平台和操作系统都具备用于解码 base64 文本的工具。

RESTgcloud

如需查看发布到 Pub/Sub 主题的消息,请使用 projects.subscriptions.pull 方法。以下示例使用 ?maxMessages=10 查询参数指定请求中返回的消息数量上限。您可以根据需要调整此值。

在使用任何请求数据之前,请先进行以下替换:

  • PROJECT_ID:您的 Google Cloud 项目的 ID
  • PUBSUB_SUBSCRIPTION_ID:附加到 FHIR 存储区中配置的 Pub/Sub 主题的订阅的 ID

如需发送请求,请选择以下方式之一:

执行以下命令:

curl -X POST \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json; charset=utf-8" \
-d "" \
"https://pubsub.googleapis.com/v1/projects/PROJECT_ID/subscriptions/PUBSUB_SUBSCRIPTION_ID:pull?maxMessages=10"

执行以下命令:

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

Invoke-WebRequest `
-Method POST `
-Headers $headers `
-Uri "https://pubsub.googleapis.com/v1/projects/PROJECT_ID/subscriptions/PUBSUB_SUBSCRIPTION_ID:pull?maxMessages=10" | Select-Object -Expand Content

您应该收到类似以下内容的 JSON 响应:

{
  "receivedMessages": [
    {
      "ackId": "RFAGFixdRkhRNxkIaFEOT14jPzUgKEUaAggUBXx9cEFLdVhUcGhRDRlyfWB9bQ5GAgpGWixfURsHaE5tdR",
      "message": {
        "data": "wogICJiaXJ0aERhdGUiOiAiMTk3MC0wMS0wMSIsCiAgImdlbmRlciI6ICJmZW1hbGUiLAogICJpZCI6ICIyYmMwODg4Yi00MGRmLTQwYzctOWRhYi0wMzc4YTFiZWE0MGIiLAogICJtZXRhIjogewogICAgImxhc3RVcGRhdGVkIjogIjIwMjMtMDUtMDJUMjA6NDc6MTkuMjk2MjEwKzAwOjAwIiwKICAgICJ2ZXJzaW9uSWQiOiAiTVRZNE16QTJNRFF6T1RJNU5qSXhNREF3TUEiCiAgfSwKICAibmFtZSI6IFsKICAgIHsKICAgICAgImZhbWlseSI6ICJTbWl0aCIsCiAgICAgICJnaXZlbiI6IFsKICAgICAgICAiRGFyY3kiCiAgICAgIF0sCiAgICAgICJ1c2UiOiAib2ZmaWNpYWwiCiAgICB9CiAgXSwKICAicmVzb3VyY2VUeXBlIjogIlBhdGllbnQiCn0=",
        "attributes": {
          "storeName": "projects/PROJECT_ID/locations/LOCATION/datasets/DATASET_ID/fhirStores/FHIR_STORE_ID",
          "action": "CreateResource",
          "versionId": "MTY4MzA2MDQzOTI5NjIxMDAwMA",
          "resourceType": "Patient",
          "lastUpdatedTime": "Mon, 01 Jan 2020 00:00:00 UTC",
          "payloadType": "FullResource"
        },
        "messageId": "7586159156345265",
        "publishTime": "YYYY-MM-DDTHH:MM:SS+ZZ:ZZ"
      }
    }
  ]
}

如需查看发布到 Pub/Sub 主题的消息,请运行 gcloud pubsub subscriptions pull 命令。

此示例使用了以下 Google Cloud CLI 标志:

在使用下面的命令数据之前,请先进行以下替换:

  • PROJECT_ID:您的 Google Cloud 项目的 ID
  • PUBSUB_SUBSCRIPTION_ID:附加到 FHIR 存储区中配置的 Pub/Sub 主题的订阅的 ID

执行以下命令:

Linux、macOS 或 Cloud Shell

gcloud pubsub subscriptions pull \
    projects/PROJECT_ID/subscriptions/PUBSUB_SUBSCRIPTION_ID \
    --auto-ack \
    --format=json
gcloud pubsub subscriptions pull `
    projects/PROJECT_ID/subscriptions/PUBSUB_SUBSCRIPTION_ID `
    --auto-ack `
    --format=json
gcloud pubsub subscriptions pull ^
    projects/PROJECT_ID/subscriptions/PUBSUB_SUBSCRIPTION_ID ^
    --auto-ack ^
    --format=json

您应该会收到类似如下所示的响应:

[
  {
    "ackId": "RFAGFixdRkhRNxkIaFEOT14jPzUgKEUaAggUBXx9cEFLdVhUcGhRDRlyfWB9bQ5GAgpGWixfURsHaE5tdR",
    "ackStatus": "SUCCESS",
    "message": {
      "attributes": {
        "action": "CreateResource",
        "lastUpdatedTime": "Mon, 01 Jan 2020 00:00:00 UTC",
        "payloadType": "FullResource",
        "resourceType": "Patient",
        "storeName": "projects/PROJECT_ID/locations/LOCATION/datasets/DATASET_ID/fhirStores/FHIR_STORE_ID",
        "versionId": "MTY4MzA2MDQzOTI5NjIxMDAwMA"
      },
      "data": "wogICJiaXJ0aERhdGUiOiAiMTk3MC0wMS0wMSIsCiAgImdlbmRlciI6ICJmZW1hbGUiLAogICJpZCI6ICIyYmMwODg4Yi00MGRmLTQwYzctOWRhYi0wMzc4YTFiZWE0MGIiLAogICJtZXRhIjogewogICAgImxhc3RVcGRhdGVkIjogIjIwMjMtMDUtMDJUMjA6NDc6MTkuMjk2MjEwKzAwOjAwIiwKICAgICJ2ZXJzaW9uSWQiOiAiTVRZNE16QTJNRFF6T1RJNU5qSXhNREF3TUEiCiAgfSwKICAibmFtZSI6IFsKICAgIHsKICAgICAgImZhbWlseSI6ICJTbWl0aCIsCiAgICAgICJnaXZlbiI6IFsKICAgICAgICAiRGFyY3kiCiAgICAgIF0sCiAgICAgICJ1c2UiOiAib2ZmaWNpYWwiCiAgICB9CiAgXSwKICAicmVzb3VyY2VUeXBlIjogIlBhdGllbnQiCn0=",
      "messageId": "7586159156345265",
      "publishTime": "YYYY-MM-DDTHH:MM:SS+ZZ:ZZ"
    }
  }
]

FHIR 资源过大或流量过高时的行为

如果 FHIR 资源的大小过大,或者 Cloud Healthcare API 服务器的流量过高,attributes 字段可能只包含资源名称,而不是完整的资源内容。即使 sendFullResourcesendPreviousResourceOnDelete 设置为 true,也会发生这种行为。

如需验证 Pub/Sub 通知是否包含完整的 FHIR 资源,请查看通知中的响应,然后检查 payloadType 字段。如果 payloadType 设置为 nameOnly,则 attributes 字段未完全填充 FHIR 资源数据。然后,您必须手动从 FHIR 存储区(而不是从 Pub/Sub 消息)中获取 FHIR 资源的内容。

Cloud Healthcare API 和 Pub/Sub 消息存储政策

为确保 Cloud Healthcare API 数据与 Pub/Sub 消息中的关联数据位于同一区域,您必须设置 Pub/Sub 消息存储政策

您必须为数据存储区中配置的 Pub/Sub 主题明确设置消息存储政策,才能确保数据保留在同一区域中。例如,如果 Cloud Healthcare API 数据集和 FHIR 存储区位于 us-central1,则消息存储政策必须仅允许 us-central1 区域。

如需配置消息存储政策,请参阅配置消息存储区政策

排查遗漏的 Pub/Sub 消息问题

如果无法将通知发布到 Pub/Sub,则 Cloud Logging 会记录错误。如需了解详情,请参阅在 Cloud Logging 中查看错误日志

如果错误生成率超过限制,则超出此限制的错误不会提交到 Cloud Logging。

使用 NotificationConfig 配置查看 FHIR 通知(已废弃)

FhirStore 资源包含一个 NotificationConfig 对象,您可以在其中指定 Pub/Sub 主题。对 FHIR 资源的更改始终会在 Pub/Sub 消息的 data 字段中包含以下标识符:

projects/PROJECT_ID/locations/LOCATION/datasets/DATASET_ID/fhirStores/FHIR_STORE_ID/fhir/RESOURCE_TYPE/RESOURCE_ID

以下这组键值对始终包含在消息的 attributes 字段中:

属性名称 可能的值 示例 说明
action
  • CreateResource
  • PatchResource
  • UpdateResource
  • DeleteResource
CreateResource 刚刚发生的事件的类型。
resourceType 任何 FHIR 资源类型 Patient 已修改的资源类型。

后续步骤