本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。 # 使用 AWS IoT Device SDK 與 Greengrass 核、其他元件和通訊 AWS IoT Core 核心裝置上執行的元件可以使用中的 AWS IoT Greengrass 核心程序間通訊 (IPC) 程式庫 AWS IoT Device SDK ，與核和其他 Greengrass AWS IoT Greengrass 元件進行通訊。若要開發和執行使用 IPC 的自訂元件，您必須使用 AWS IoT Device SDK 連線到 AWS IoT Greengrass Core IPC 服務並執行 IPC 操作。 IPC 介面支援兩種類型的操作： + **請求/回應** 元件會將請求傳送至 IPC 服務，並接收包含請求結果的回應。 + **訂閱** 元件會將訂閱請求傳送至 IPC 服務，並預期事件訊息串流回應。元件提供訂閱處理常式，可處理事件訊息、錯誤和串流關閉。 AWS IoT Device SDK 包含處理常式界面，具有每個 IPC 操作的正確回應和事件類型。如需詳細資訊，請參閱[訂閱 IPC 事件串流](#ipc-subscribe-operations)。 **Topics** + [IPC 用戶端版本](#ipc-client-versions) + [支援用於程序間通訊SDKs](#ipc-requirements) + [連線至 AWS IoT Greengrass Core IPC 服務](#ipc-service-connect) + [授權元件執行 IPC 操作](#ipc-authorization-policies) + [訂閱 IPC 事件串流](#ipc-subscribe-operations) + [IPC 最佳實務](#ipc-best-practices) + [發佈/訂閱本機訊息](ipc-publish-subscribe.md) + [發佈/訂閱 AWS IoT Core MQTT 訊息](ipc-iot-core-mqtt.md) + [與元件生命週期互動](ipc-component-lifecycle.md) + [與元件組態互動](ipc-component-configuration.md) + [擷取秘密值](ipc-secret-manager.md) + [與本機陰影互動](ipc-local-shadows.md) + [管理本機部署和元件](ipc-local-deployments-components.md) + [驗證和授權用戶端裝置](ipc-client-device-auth.md) ## IPC 用戶端版本在 Java 和 Python SDKs的更新版本中， AWS IoT Greengrass 提供了 IPC 用戶端的改良版本，稱為 IPC 用戶端 V2。IPC 用戶端 V2： + 減少您需要寫入以使用 IPC 操作的程式碼量，並協助避免 IPC 用戶端 V1 可能發生的常見錯誤。 + 在單獨的執行緒中呼叫訂閱處理常式回呼，因此您現在可以在訂閱處理常式回呼中執行封鎖程式碼，包括額外的 IPC 函數呼叫。IPC 用戶端 V1 使用相同的執行緒與 IPC 伺服器通訊，並呼叫訂閱處理常式回呼。 + 可讓您使用 Lambda 表達式 (Java) 或函數 (Python) 呼叫訂閱操作。IPC 用戶端 V1 需要您定義訂閱處理常式類別。 + 提供每個 IPC 操作的同步和非同步版本。IPC 用戶端 V1 僅提供每個操作的非同步版本。我們建議您使用 IPC 用戶端 V2 來利用這些改進。不過，本文件和某些線上內容中的許多範例只會示範如何使用 IPC 用戶端 V1。您可以使用下列範例和教學課程來查看使用 IPC 用戶端 V2 的範例元件： + [PublishToTopic 範例](ipc-publish-subscribe.md#ipc-operation-publishtotopic-examples) + [SubscribeToTopic 範例](ipc-publish-subscribe.md#ipc-operation-subscribetotopic-examples) + [教學課程：開發可延遲元件更新的 Greengrass 元件](defer-component-updates-tutorial.md) + [教學課程：透過 MQTT 與本機 IoT 裝置互動](client-devices-tutorial.md) 目前， AWS IoT Device SDK for C\$1\$1 v2 僅支援 IPC 用戶端 V1。 ## 支援用於程序間通訊SDKs AWS IoT Greengrass 核心 IPC 程式庫包含在下列 AWS IoT Device SDK 版本中。 | SDK | 最低版本 | Usage | | --- | --- | --- | | [AWS IoT Device SDK for Java v2](https://github.com/aws/aws-iot-device-sdk-java-v2) | v1.6.0 | 請參閱 [AWS IoT Device SDK 用於 Java v2 (IPC 用戶端 V2)](#ipc-java-v2) | | [AWS IoT Device SDK for Python v2](https://github.com/aws/aws-iot-device-sdk-python-v2) | 1.9.0 版 | 請參閱 [AWS IoT Device SDK 適用於 Python v2 (IPC 用戶端 V2)](#ipc-python-v2) | | [AWS IoT Device SDK 適用於 C\$1\$1 v2](https://github.com/aws/aws-iot-device-sdk-cpp-v2) | 1.17.0 版 | 請參閱 [AWS IoT Device SDK 用於 C\$1\$1 v2](#ipc-cpp) | | [AWS IoT Device SDK for JavaScript v2](https://github.com/aws/aws-iot-device-sdk-js-v2) | v1.12.0 | 請參閱 [AWS IoT Device SDK 適用於 JavaScript v2 (IPC 用戶端 V1)](#ipc-nodejs) | ## 連線至 AWS IoT Greengrass Core IPC 服務若要在自訂元件中使用程序間通訊，您必須建立與 AWS IoT Greengrass 核心軟體執行之 IPC 伺服器通訊端的連線。完成下列任務，以您選擇的 AWS IoT Device SDK 語言下載並使用。 ### AWS IoT Device SDK 用於 Java v2 (IPC 用戶端 V2) **使用 AWS IoT Device SDK for Java v2 (IPC 用戶端 V2)** 1. 下載 [AWS IoT Device SDK for Java v2](https://github.com/aws/aws-iot-device-sdk-java-v2) (v1.6.0 或更新版本）。 1. 執行下列其中一項以在元件中執行自訂程式碼： + 將元件建置為包含的 JAR 檔案 AWS IoT Device SDK，然後在元件配方中執行此 JAR 檔案。 + 將 AWS IoT Device SDK JAR 定義為元件成品，並在元件配方中執行應用程式時將該成品新增至 classpath。 1. 使用下列程式碼來建立 IPC 用戶端。 ``` try (GreengrassCoreIPCClientV2 ipcClient = GreengrassCoreIPCClientV2.builder().build()) { // Use client. } catch (Exception e) { LOGGER.log(Level.SEVERE, "Exception occurred when using IPC.", e); System.exit(1); } ``` ### AWS IoT Device SDK 適用於 Python v2 (IPC 用戶端 V2) **使用 AWS IoT Device SDK for Python v2 (IPC 用戶端 V2)** 1. 下載 [AWS IoT Device SDK for Python](https://github.com/aws/aws-iot-device-sdk-python-v2) (v1.9.0 或更新版本）。 1. 將開發套件的[安裝步驟](https://github.com/aws/aws-iot-device-sdk-python-v2#installation)新增至元件配方中的安裝生命週期。 1. 建立與 AWS IoT Greengrass Core IPC 服務的連線。使用下列程式碼來建立 IPC 用戶端。 ``` from awsiot.greengrasscoreipc.clientv2 import GreengrassCoreIPCClientV2 try: ipc_client = GreengrassCoreIPCClientV2() # Use IPC client. except Exception: print('Exception occurred when using IPC.', file=sys.stderr) traceback.print_exc() exit(1) ``` ### AWS IoT Device SDK 用於 C\$1\$1 v2 若要建置適用於 C\$1\$1 的 AWS IoT Device SDK v2，裝置必須具有下列工具： + C\$1\$1 11 或更新版本 + CMake 3.1 或更新版本 + 下列其中一個編譯器： + GCC 4.8 或更新版本 + Clang 3.9 或更新版本 + MSVC 2015 或更新版本 **使用 AWS IoT Device SDK 適用於 C\$1\$1 v2 的** 1. 下載 [AWS IoT Device SDK for C\$1\$1 v2](https://github.com/aws/aws-iot-device-sdk-cpp-v2) (v1.17.0 或更新版本）。 1. 遵循 [README 中的安裝說明](https://github.com/aws/aws-iot-device-sdk-cpp-v2#Installation)，從來源建置 AWS IoT Device SDK 適用於 C\$1\$1 v2 的。 1. 在 C\$1\$1 建置工具中，連結您在上一個步驟中建置的 Greengrass IPC 程式庫 `AWS::GreengrassIpc-cpp`。下列`CMakeLists.txt`範例會將 Greengrass IPC 程式庫連結至您使用 CMake 建置的專案。 ``` cmake_minimum_required(VERSION 3.1) project (greengrassv2_pubsub_subscriber) file(GLOB MAIN_SRC "*.h" "*.cpp" ) add_executable(${PROJECT_NAME} ${MAIN_SRC}) set_target_properties(${PROJECT_NAME} PROPERTIES LINKER_LANGUAGE CXX CXX_STANDARD 11) find_package(aws-crt-cpp PATHS ~/sdk-cpp-workspace/build) find_package(EventstreamRpc-cpp PATHS ~/sdk-cpp-workspace/build) find_package(GreengrassIpc-cpp PATHS ~/sdk-cpp-workspace/build) target_link_libraries(${PROJECT_NAME} AWS::GreengrassIpc-cpp) ``` 1. 在元件程式碼中，建立與 AWS IoT Greengrass Core IPC 服務的連線，以建立 IPC 用戶端 (`Aws::Greengrass::GreengrassCoreIpcClient`)。您必須定義 IPC 連線生命週期處理常式，以處理 IPC 連線、中斷連線和錯誤事件。下列範例會建立 IPC 用戶端和 IPC 連線生命週期處理常式，在 IPC 用戶端連線、中斷連線和遇到錯誤時列印。 ``` #include #include #include using namespace Aws::Crt; using namespace Aws::Greengrass; class IpcClientLifecycleHandler : public ConnectionLifecycleHandler { void OnConnectCallback() override { std::cout << "OnConnectCallback" << std::endl; } void OnDisconnectCallback(RpcError error) override { std::cout << "OnDisconnectCallback: " << error.StatusToString() << std::endl; exit(-1); } bool OnErrorCallback(RpcError error) override { std::cout << "OnErrorCallback: " << error.StatusToString() << std::endl; return true; } }; int main() { // Create the IPC client. ApiHandle apiHandle(g_allocator); Io::EventLoopGroup eventLoopGroup(1); Io::DefaultHostResolver socketResolver(eventLoopGroup, 64, 30); Io::ClientBootstrap bootstrap(eventLoopGroup, socketResolver); IpcClientLifecycleHandler ipcLifecycleHandler; GreengrassCoreIpcClient ipcClient(bootstrap); auto connectionStatus = ipcClient.Connect(ipcLifecycleHandler).get(); if (!connectionStatus) { std::cerr << "Failed to establish IPC connection: " << connectionStatus.StatusToString() << std::endl; exit(-1); } // Use the IPC client to create an operation request. // Activate the operation request. auto activate = operation.Activate(request, nullptr); activate.wait(); // Wait for Greengrass Core to respond to the request. auto responseFuture = operation.GetResult(); if (responseFuture.wait_for(std::chrono::seconds(timeout)) == std::future_status::timeout) { std::cerr << "Operation timed out while waiting for response from Greengrass Core." << std::endl; exit(-1); } // Check the result of the request. auto response = responseFuture.get(); if (response) { std::cout << "Successfully published to topic: " << topic << std::endl; } else { // An error occurred. std::cout << "Failed to publish to topic: " << topic << std::endl; auto errorType = response.GetResultType(); if (errorType == OPERATION_ERROR) { auto *error = response.GetOperationError(); std::cout << "Operation error: " << error->GetMessage().value() << std::endl; } else { std::cout << "RPC error: " << response.GetRpcError() << std::endl; } exit(-1); } return 0; } ``` 1. 若要在元件中執行自訂程式碼，請將程式碼建置為二進位成品，然後在元件配方中執行二進位成品。將成品的`Execute`許可設定為 `OWNER`，讓 AWS IoT Greengrass Core 軟體執行二進位成品。元件配方的`Manifests`區段可能看起來類似下列範例。 ------ #### [ JSON ] ``` { ... "Manifests": [ { "Lifecycle": { "Run": "{artifacts:path}/greengrassv2_pubsub_subscriber" }, "Artifacts": [ { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.PubSubSubscriberCpp/1.0.0/greengrassv2_pubsub_subscriber", "Permission": { "Execute": "OWNER" } } ] } ] } ``` ------ #### [ YAML ] ``` ... Manifests: - Lifecycle: Run: {artifacts:path}/greengrassv2_pubsub_subscriber Artifacts: - URI: s3://amzn-s3-demo-bucket/artifacts/com.example.PubSubSubscriberCpp/1.0.0/greengrassv2_pubsub_subscriber Permission: Execute: OWNER ``` ------ ### AWS IoT Device SDK 適用於 JavaScript v2 (IPC 用戶端 V1) 若要建置 AWS IoT Device SDK 適用於 JavaScript v2 的以搭配 NodeJS 使用，裝置必須具備下列工具： + NodeJS 10.0 或更新版本 + 執行 `node -v`以檢查節點版本。 + CMake 3.1 或更新版本 **使用 AWS IoT Device SDK for JavaScript v2 (IPC 用戶端 V1)** 1. 下載 [AWS IoT Device SDK for JavaScript v2](https://github.com/aws/aws-iot-device-sdk-js-v2) (v1.12.10 或更新版本）。 1. 遵循 [README 中的安裝說明](https://github.com/aws/aws-iot-device-sdk-js-v2/tree/v1.12.1#installation)，從來源建置 AWS IoT Device SDK 適用於 JavaScript v2 的。 1. 建立與 AWS IoT Greengrass Core IPC 服務的連線。完成下列步驟以建立 IPC 用戶端並建立連線。 1. 使用下列程式碼來建立 IPC 用戶端。 ``` import * as greengrascoreipc from 'aws-iot-device-sdk-v2'; let client = greengrascoreipc.createClient(); ``` 1. 使用下列程式碼，從您的元件建立與 Greengrass 核的連線。 ``` await client.connect(); ``` ## 授權元件執行 IPC 操作若要允許自訂元件使用某些 IPC 操作，您必須定義*授權政策*，允許元件對特定資源執行操作。每個授權政策都會定義操作清單，以及政策允許的資源清單。例如，發佈/訂閱訊息 IPC 服務會定義主題資源的發佈和訂閱操作。您可以使用`*`萬用字元來允許存取所有操作或所有資源。您可以使用`accessControl`組態參數來定義授權政策，您可以在元件配方中或部署元件時加以設定。`accessControl` 物件會將 IPC 服務識別符映射至授權政策清單。您可以為每個 IPC 服務定義多個授權政策來控制存取。每個授權政策都有一個政策 ID，在所有元件中必須是唯一的。 **提示** 若要建立唯一的政策 IDs，您可以結合元件名稱、IPC 服務名稱和計數器。例如，名為的元件`com.example.HelloWorld`可能會使用下列 IDs 定義兩個發佈/訂閱授權政策： `com.example.HelloWorld:pubsub:1` `com.example.HelloWorld:pubsub:2` 授權政策使用以下格式。此物件是`accessControl`組態參數。 ------ #### [ JSON ] ``` { "IPC service identifier": { "policyId": { "policyDescription": "description", "operations": [ "operation1", "operation2" ], "resources": [ "resource1", "resource2" ] } } } ``` ------ #### [ YAML ] ``` IPC service identifier: policyId: policyDescription: description operations: - operation1 - operation2 resources: - resource1 - resource2 ``` ------ ### 授權政策中的萬用字元您可以在 IPC 授權政策的 `resources`元素中使用`*`萬用字元，以允許存取單一授權政策中的多個資源。 + 在所有版本的 [Greengrass 核](greengrass-nucleus-component.md)中，您可以指定單一`*`字元做為資源，以允許存取所有資源。 + 在 [Greengrass 核](greengrass-nucleus-component.md) v2.6.0 和更新版本中，您可以在資源中指定`*`字元，以符合字元的任意組合。例如，您可以指定 `factory/1/devices/Thermostat*/status` 允許存取工廠中所有恆溫器裝置的狀態主題，其中每個裝置的名稱開頭為 `Thermostat`。當您定義 AWS IoT Core MQTT IPC 服務的授權政策時，您也可以使用 MQTT 萬用字元 (`+` 和 `#`) 來比對多個資源。如需詳細資訊，請參閱 [MQTT IPC 授權政策中的 AWS IoT Core MQTT 萬用字元](ipc-iot-core-mqtt.md#ipc-iot-core-mqtt-authorization-mqtt-wildcards)。 ### 授權政策中的配方變數如果您使用 [Greengrass 核 v](greengrass-nucleus-component.md)2.6.0 或更新版本，並將 Greengrass 核的 [interpolateComponentConfiguration](greengrass-nucleus-component.md#greengrass-nucleus-component-configuration-interpolate-component-configuration) 組態選項設定為 `true`，您可以在授權政策中使用`{iot:thingName}`[配方變數](component-recipe-reference.md#recipe-variables)。當您需要包含核心裝置名稱的授權政策時，例如 MQTT 主題或裝置影子，您可以使用此配方變數來設定一組核心裝置的單一授權政策。例如，您可以允許元件存取陰影 IPC 操作的下列資源。 ``` $aws/things/{iot:thingName}/shadow/ ``` ### 授權政策中的特殊字元若要在授權政策中指定文字`*`或`?`字元，您必須使用逸出序列。下列逸出序列會指示 AWS IoT Greengrass Core 軟體使用常值，而非字元的特殊意義。例如， `*` 字元是符合任何字元組合的[萬用字元](#ipc-authorization-policy-wildcards)。 | 文字字元 | 逸出序列 | 備註 | | --- | --- | --- | | `*` | `${*}` | | | `?` | `${?}` | AWS IoT Greengrass 目前不支援萬`?`用字元，符合任何單一字元。 | | `$` | `${$}` | 使用此逸出序列來比對包含的資源`${`。例如，若要符合名為的資源`${resourceName}`，您必須指定 `${$}{resourceName}`。否則，若要比對包含的資源`$`，您可以使用常值 `$`，例如允許存取開頭為的主題`$aws`。 | ### 授權政策範例您可以參考下列授權政策範例，協助您設定元件的授權政策。 **Example 具有授權政策的元件配方範例** 下列範例元件配方包含 `accessControl` 物件，可定義授權政策。此政策授權 `com.example.HelloWorld`元件發佈至 `test/topic`主題。 ``` { "RecipeFormatVersion": "2020-01-25", "ComponentName": "com.example.HelloWorld", "ComponentVersion": "1.0.0", "ComponentDescription": "A component that publishes messages.", "ComponentPublisher": "Amazon", "ComponentConfiguration": { "DefaultConfiguration": { "accessControl": { "aws.greengrass.ipc.pubsub": { "com.example.HelloWorld:pubsub:1": { "policyDescription": "Allows access to publish to test/topic.", "operations": [ "aws.greengrass#PublishToTopic" ], "resources": [ "test/topic" ] } } } } }, "Manifests": [ { "Lifecycle": { "Run": "java -jar {artifacts:path}/HelloWorld.jar" } } ] } ``` ``` --- RecipeFormatVersion: '2020-01-25' ComponentName: com.example.HelloWorld ComponentVersion: '1.0.0' ComponentDescription: A component that publishes messages. ComponentPublisher: Amazon ComponentConfiguration: DefaultConfiguration: accessControl: aws.greengrass.ipc.pubsub: "com.example.HelloWorld:pubsub:1": policyDescription: Allows access to publish to test/topic. operations: - "aws.greengrass#PublishToTopic" resources: - "test/topic" Manifests: - Lifecycle: Run: |- java -jar {artifacts:path}/HelloWorld.jar ``` **Example 使用授權政策更新元件組態的範例** 部署中的下列範例組態更新指定使用定義授權政策的 `accessControl` 物件來設定元件。此政策授權 `com.example.HelloWorld`元件發佈至 `test/topic`主題。 **要合併的組態** ``` { "accessControl": { "aws.greengrass.ipc.pubsub": { "com.example.HelloWorld:pubsub:1": { "policyDescription": "Allows access to publish to test/topic.", "operations": [ "aws.greengrass#PublishToTopic" ], "resources": [ "test/topic" ] } } } } ``` 下列命令會建立核心裝置的部署。 ``` aws greengrassv2 create-deployment --cli-input-json file://hello-world-deployment.json ``` `hello-world-deployment.json` 檔案包含下列 JSON 文件。 ``` { "targetArn": "arn:aws:iot:us-west-2:123456789012:thing/MyGreengrassCore", "deploymentName": "Deployment for MyGreengrassCore", "components": { "com.example.HelloWorld": { "componentVersion": "1.0.0", "configurationUpdate": { "merge": "{\"accessControl\":{\"aws.greengrass.ipc.pubsub\":{\"com.example.HelloWorld:pubsub:1\":{\"policyDescription\":\"Allows access to publish to test/topic.\",\"operations\":[\"aws.greengrass#PublishToTopic\"],\"resources\":[\"test/topic\"]}}}}" } } } } ``` 下列 [Greengrass CLI](greengrass-cli-component.md) 命令會在核心裝置上建立本機部署。 ``` sudo greengrass-cli deployment create \ --recipeDir recipes \ --artifactDir artifacts \ --merge "com.example.HelloWorld=1.0.0" \ --update-config hello-world-configuration.json ``` `hello-world-configuration.json` 檔案包含下列 JSON 文件。 ``` { "com.example.HelloWorld": { "MERGE": { "accessControl": { "aws.greengrass.ipc.pubsub": { "com.example.HelloWorld:pubsub:1": { "policyDescription": "Allows access to publish to test/topic.", "operations": [ "aws.greengrass#PublishToTopic" ], "resources": [ "test/topic" ] } } } } } } ``` ## 訂閱 IPC 事件串流您可以使用 IPC 操作來訂閱 Greengrass 核心裝置上的事件串流。若要使用訂閱操作，請定義*訂閱處理常*式並建立 IPC 服務的請求。然後，每次核心裝置將事件訊息串流到您的元件時，IPC 用戶端都會執行訂閱處理常式的函數。您可以關閉訂閱以停止處理事件訊息。若要這麼做，請在您用來開啟訂閱的訂閱操作物件上呼叫 `closeStream()`(Java)、 `close()` (Python) 或 `Close()`(C\$1\$1)。 AWS IoT Greengrass Core IPC 服務支援下列訂閱操作： + [SubscribeToTopic](ipc-publish-subscribe.md#ipc-operation-subscribetotopic) + [SubscribeToIoTCore](ipc-iot-core-mqtt.md#ipc-operation-subscribetoiotcore) + [SubscribeToComponentUpdates](ipc-component-lifecycle.md#ipc-operation-subscribetocomponentupdates) + [SubscribeToConfigurationUpdate](ipc-component-configuration.md#ipc-operation-subscribetoconfigurationupdate) + [SubscribeToValidateConfigurationUpdates](ipc-component-configuration.md#ipc-operation-subscribetovalidateconfigurationupdates) **Topics** + [定義訂閱處理常式](#ipc-define-subscription-handlers) + [範例訂閱處理常式](#ipc-subscription-handler-examples) ### 定義訂閱處理常式若要定義訂閱處理常式，請定義處理事件訊息、錯誤和串流關閉的回呼函數。如果您使用 IPC 用戶端 V1，則必須在類別中定義這些函數。如果您使用適用於 Java 和 Python SDKs 較新版本的 IPC 用戶端 V2，您可以定義這些函數，而無需建立訂閱處理常式類別。 ------ #### [ Java ] 如果您使用 IPC 用戶端 V1，則必須實作一般`software.amazon.awssdk.eventstreamrpc.StreamResponseHandler`界面。*StreamEventType* 是訂閱操作的事件訊息類型。定義下列函數來處理事件訊息、錯誤和串流關閉。如果您使用 IPC 用戶端 V2，您可以在訂閱處理常式類別之外定義這些函數，或使用 [lambda 運算式](https://docs.oracle.com/javase/tutorial/java/javaOO/lambdaexpressions.html)。 `void onStreamEvent(StreamEventType event)` IPC 用戶端在收到事件訊息時呼叫的回呼，例如 MQTT 訊息或元件更新通知。 `boolean onStreamError(Throwable error)` IPC 用戶端在發生串流錯誤時呼叫的回呼。傳回 true 以關閉訂閱串流，因為發生錯誤，或傳回 false 以保持串流開啟。 `void onStreamClosed()` IPC 用戶端在串流關閉時呼叫的回呼。 ------ #### [ Python ] 如果您使用 IPC 用戶端 V1，則必須擴展對應至訂閱操作的串流回應處理常式類別。 AWS IoT Device SDK 包含每個訂閱操作的訂閱處理常式類別。*StreamEventType* 是訂閱操作的事件訊息類型。定義下列函數來處理事件訊息、錯誤和串流關閉。如果您使用 IPC 用戶端 V2，您可以在訂閱處理常式類別之外定義這些函數，或使用 [lambda 運算式](https://docs.python.org/3/tutorial/controlflow.html#lambda-expressions)。 `def on_stream_event(self, event: StreamEventType) -> None` IPC 用戶端在收到事件訊息時呼叫的回呼，例如 MQTT 訊息或元件更新通知。 `def on_stream_error(self, error: Exception) -> bool` IPC 用戶端在發生串流錯誤時呼叫的回呼。傳回 true 以關閉訂閱串流，因為發生錯誤，或傳回 false 以保持串流開啟。 `def on_stream_closed(self) -> None` IPC 用戶端在串流關閉時呼叫的回呼。 ------ #### [ C\$1\$1 (IPC client V1) ] 實作衍生自對應至訂閱操作之串流回應處理常式類別的類別。 AWS IoT Device SDK 包含每個訂閱操作的訂閱處理常式基本類別。*StreamEventType* 是訂閱操作的事件訊息類型。定義下列函數來處理事件訊息、錯誤和串流關閉。 `void OnStreamEvent(StreamEventType *event)` IPC 用戶端在收到事件訊息時呼叫的回呼，例如 MQTT 訊息或元件更新通知。 `bool OnStreamError(OperationError *error)` IPC 用戶端在發生串流錯誤時呼叫的回呼。傳回 true 以關閉訂閱串流，因為發生錯誤，或傳回 false 以保持串流開啟。 `void OnStreamClosed()` IPC 用戶端在串流關閉時呼叫的回呼。 ------ #### [ JavaScript ] 實作衍生自對應至訂閱操作之串流回應處理常式類別的類別。 AWS IoT Device SDK 包含每個訂閱操作的訂閱處理常式基本類別。*StreamEventType* 是訂閱操作的事件訊息類型。定義下列函數來處理事件訊息、錯誤和串流關閉。 `on(event: 'ended', listener: StreamingOperationEndedListener)` IPC 用戶端在串流關閉時呼叫的回呼。 `on(event: 'streamError', listener: StreamingRpcErrorListener)` IPC 用戶端在發生串流錯誤時呼叫的回呼。傳回 true 以關閉訂閱串流，因為發生錯誤，或傳回 false 以保持串流開啟。 `on(event: 'message', listener: (message: InboundMessageType) => void)` IPC 用戶端在收到事件訊息時呼叫的回呼，例如 MQTT 訊息或元件更新通知。 ------ ### 範例訂閱處理常式下列範例示範如何使用 [SubscribeToTopic](ipc-publish-subscribe.md#ipc-operation-subscribetotopic)操作和訂閱處理常式來訂閱本機發佈/訂閱訊息。 ------ #### [ Java (IPC client V2) ] **Example 範例：訂閱本機發佈/訂閱訊息** ``` package com.aws.greengrass.docs.samples.ipc; import software.amazon.awssdk.aws.greengrass.GreengrassCoreIPCClientV2; import software.amazon.awssdk.aws.greengrass.SubscribeToTopicResponseHandler; import software.amazon.awssdk.aws.greengrass.model.*; import java.nio.charset.StandardCharsets; import java.util.Optional; public class SubscribeToTopicV2 { public static void main(String[] args) { String topic = args[0]; try (GreengrassCoreIPCClientV2 ipcClient = GreengrassCoreIPCClientV2.builder().build()) { SubscribeToTopicRequest request = new SubscribeToTopicRequest().withTopic(topic); GreengrassCoreIPCClientV2.StreamingResponse response = ipcClient.subscribeToTopic(request, SubscribeToTopicV2::onStreamEvent, Optional.of(SubscribeToTopicV2::onStreamError), Optional.of(SubscribeToTopicV2::onStreamClosed)); SubscribeToTopicResponseHandler responseHandler = response.getHandler(); System.out.println("Successfully subscribed to topic: " + topic); // Keep the main thread alive, or the process will exit. try { while (true) { Thread.sleep(10000); } } catch (InterruptedException e) { System.out.println("Subscribe interrupted."); } // To stop subscribing, close the stream. responseHandler.closeStream(); } catch (Exception e) { if (e.getCause() instanceof UnauthorizedError) { System.err.println("Unauthorized error while publishing to topic: " + topic); } else { System.err.println("Exception occurred when using IPC."); } e.printStackTrace(); System.exit(1); } } public static void onStreamEvent(SubscriptionResponseMessage subscriptionResponseMessage) { try { BinaryMessage binaryMessage = subscriptionResponseMessage.getBinaryMessage(); String message = new String(binaryMessage.getMessage(), StandardCharsets.UTF_8); String topic = binaryMessage.getContext().getTopic(); System.out.printf("Received new message on topic %s: %s%n", topic, message); } catch (Exception e) { System.err.println("Exception occurred while processing subscription response " + "message."); e.printStackTrace(); } } public static boolean onStreamError(Throwable error) { System.err.println("Received a stream error."); error.printStackTrace(); return false; // Return true to close stream, false to keep stream open. } public static void onStreamClosed() { System.out.println("Subscribe to topic stream closed."); } } ``` ------ #### [ Python (IPC client V2) ] **Example 範例：訂閱本機發佈/訂閱訊息** ``` import sys import time import traceback from awsiot.greengrasscoreipc.clientv2 import GreengrassCoreIPCClientV2 from awsiot.greengrasscoreipc.model import ( SubscriptionResponseMessage, UnauthorizedError ) def main(): args = sys.argv[1:] topic = args[0] try: ipc_client = GreengrassCoreIPCClientV2() # Subscription operations return a tuple with the response and the operation. _, operation = ipc_client.subscribe_to_topic(topic=topic, on_stream_event=on_stream_event, on_stream_error=on_stream_error, on_stream_closed=on_stream_closed) print('Successfully subscribed to topic: ' + topic) # Keep the main thread alive, or the process will exit. try: while True: time.sleep(10) except InterruptedError: print('Subscribe interrupted.') # To stop subscribing, close the stream. operation.close() except UnauthorizedError: print('Unauthorized error while subscribing to topic: ' + topic, file=sys.stderr) traceback.print_exc() exit(1) except Exception: print('Exception occurred', file=sys.stderr) traceback.print_exc() exit(1) def on_stream_event(event: SubscriptionResponseMessage) -> None: try: message = str(event.binary_message.message, 'utf-8') topic = event.binary_message.context.topic print('Received new message on topic %s: %s' % (topic, message)) except: traceback.print_exc() def on_stream_error(error: Exception) -> bool: print('Received a stream error.', file=sys.stderr) traceback.print_exc() return False # Return True to close stream, False to keep stream open. def on_stream_closed() -> None: print('Subscribe to topic stream closed.') if __name__ == '__main__': main() ``` ------ #### [ C\$1\$1 (IPC client V1) ] **Example 範例：訂閱本機發佈/訂閱訊息** ``` #include #include #include using namespace Aws::Crt; using namespace Aws::Greengrass; class SubscribeResponseHandler : public SubscribeToTopicStreamHandler { public: virtual ~SubscribeResponseHandler() {} private: void OnStreamEvent(SubscriptionResponseMessage *response) override { auto jsonMessage = response->GetJsonMessage(); if (jsonMessage.has_value() && jsonMessage.value().GetMessage().has_value()) { auto messageString = jsonMessage.value().GetMessage().value().View().WriteReadable(); // Handle JSON message. } else { auto binaryMessage = response->GetBinaryMessage(); if (binaryMessage.has_value() && binaryMessage.value().GetMessage().has_value()) { auto messageBytes = binaryMessage.value().GetMessage().value(); std::string messageString(messageBytes.begin(), messageBytes.end()); // Handle binary message. } } } bool OnStreamError(OperationError *error) override { // Handle error. return false; // Return true to close stream, false to keep stream open. } void OnStreamClosed() override { // Handle close. } }; class IpcClientLifecycleHandler : public ConnectionLifecycleHandler { void OnConnectCallback() override { // Handle connection to IPC service. } void OnDisconnectCallback(RpcError error) override { // Handle disconnection from IPC service. } bool OnErrorCallback(RpcError error) override { // Handle IPC service connection error. return true; } }; int main() { ApiHandle apiHandle(g_allocator); Io::EventLoopGroup eventLoopGroup(1); Io::DefaultHostResolver socketResolver(eventLoopGroup, 64, 30); Io::ClientBootstrap bootstrap(eventLoopGroup, socketResolver); IpcClientLifecycleHandler ipcLifecycleHandler; GreengrassCoreIpcClient ipcClient(bootstrap); auto connectionStatus = ipcClient.Connect(ipcLifecycleHandler).get(); if (!connectionStatus) { std::cerr << "Failed to establish IPC connection: " << connectionStatus.StatusToString() << std::endl; exit(-1); } String topic("my/topic"); int timeout = 10; SubscribeToTopicRequest request; request.SetTopic(topic); //SubscribeResponseHandler streamHandler; auto streamHandler = MakeShared(DefaultAllocator()); auto operation = ipcClient.NewSubscribeToTopic(streamHandler); auto activate = operation->Activate(request, nullptr); activate.wait(); auto responseFuture = operation->GetResult(); if (responseFuture.wait_for(std::chrono::seconds(timeout)) == std::future_status::timeout) { std::cerr << "Operation timed out while waiting for response from Greengrass Core." << std::endl; exit(-1); } auto response = responseFuture.get(); if (!response) { // Handle error. auto errorType = response.GetResultType(); if (errorType == OPERATION_ERROR) { auto *error = response.GetOperationError(); (void)error; // Handle operation error. } else { // Handle RPC error. } exit(-1); } // Keep the main thread alive, or the process will exit. while (true) { std::this_thread::sleep_for(std::chrono::seconds(10)); } operation->Close(); return 0; } ``` ------ #### [ JavaScript ] **Example 範例：訂閱本機發佈/訂閱訊息** ``` import * as greengrasscoreipc from "aws-iot-device-sdk-v2/dist/greengrasscoreipc"; import {SubscribeToTopicRequest, SubscriptionResponseMessage} from "aws-iot-device-sdk-v2/dist/greengrasscoreipc/model"; import {RpcError} from "aws-iot-device-sdk-v2/dist/eventstream_rpc"; class SubscribeToTopic { private ipcClient : greengrasscoreipc.Client private readonly topic : string; constructor() { // define your own constructor, e.g. this.topic = ""; this.subscribeToTopic().then(r => console.log("Started workflow")); } private async subscribeToTopic() { try { this.ipcClient = await getIpcClient(); const subscribeToTopicRequest : SubscribeToTopicRequest = { topic: this.topic, } const streamingOperation = this.ipcClient.subscribeToTopic(subscribeToTopicRequest, undefined); // conditionally apply options streamingOperation.on("message", (message: SubscriptionResponseMessage) => { // parse the message depending on your use cases, e.g. if(message.binaryMessage && message.binaryMessage.message) { const receivedMessage = message.binaryMessage?.message.toString(); } }); streamingOperation.on("streamError", (error : RpcError) => { // define your own error handling logic }) streamingOperation.on("ended", () => { // define your own logic }) await streamingOperation.activate(); // Keep the main thread alive, or the process will exit. await new Promise((resolve) => setTimeout(resolve, 10000)) } catch (e) { // parse the error depending on your use cases throw e } } } export async function getIpcClient(){ try { const ipcClient = greengrasscoreipc.createClient(); await ipcClient.connect() .catch(error => { // parse the error depending on your use cases throw error; }); return ipcClient } catch (err) { // parse the error depending on your use cases throw err } } // starting point const subscribeToTopic = new SubscribeToTopic(); ``` ------ #### [ Rust ] **Example 範例：訂閱本機發佈/訂閱訊息** ``` use gg_sdk::{Sdk, SubscribeToTopicPayload}; use std::{thread, time::Duration}; fn main() { let sdk = Sdk::init(); sdk.connect().expect("Failed to establish IPC connection"); let topic = "my/topic"; let callback = |topic: &str, payload: SubscribeToTopicPayload| match payload { SubscribeToTopicPayload::Binary(message) => { let message = String::from_utf8_lossy(message); println!("Received new message on topic {topic}: {message}"); } SubscribeToTopicPayload::Json(_) => { println!("Received new message on topic {topic}: (JSON message)"); } }; let _sub = sdk .subscribe_to_topic(topic, &callback) .expect("Failed to subscribe to topic"); println!("Successfully subscribed to topic: {topic}"); // Keep the main thread alive, or the process will exit. loop { thread::sleep(Duration::from_secs(10)); } } ``` ------ #### [ C ] **Example 範例：訂閱本機發佈/訂閱訊息** ``` #include #include #include #include #include #include #include #include #include static void on_subscription_response( void *ctx, GgBuffer topic, GgObject payload, GgIpcSubscriptionHandle handle ) { (void) ctx; (void) handle; if (gg_obj_type(payload) == GG_TYPE_BUF) { GgBuffer message = gg_obj_into_buf(payload); printf( "Received new message on topic %.*s: %.*s\n", (int) topic.len, topic.data, (int) message.len, message.data ); } else { assert(gg_obj_type(payload) == GG_TYPE_MAP); printf( "Received new message on topic %.*s: (JSON message)\n", (int) topic.len, topic.data ); } } int main(void) { gg_sdk_init(); GgError err = ggipc_connect(); if (err != GG_ERR_OK) { fprintf(stderr, "Failed to establish IPC connection.\n"); exit(-1); } GgBuffer topic = GG_STR("my/topic"); GgIpcSubscriptionHandle handle; err = ggipc_subscribe_to_topic( topic, on_subscription_response, NULL, &handle ); if (err != GG_ERR_OK) { fprintf( stderr, "Failed to subscribe to topic: %.*s\n", (int) topic.len, topic.data ); exit(-1); } printf( "Successfully subscribed to topic: %.*s\n", (int) topic.len, topic.data ); // Keep the main thread alive, or the process will exit. while (1) { sleep(10); } // To stop subscribing, close the stream. ggipc_close_subscription(handle); } ``` ------ #### [ C\$1\$1 (Component SDK) ] **Example 範例：訂閱本機發佈/訂閱訊息** ``` #include #include #include #include #include class ResponseHandler : public gg::ipc::LocalTopicCallback { void operator()( std::string_view topic, gg::Object payload, gg::ipc::Subscription &handle ) override { (void) handle; if (payload.index() == GG_TYPE_BUF) { std::cout << "Received new message on topic " << topic << ": " << get(payload) << "\n"; } else { assert(payload.index() == GG_TYPE_MAP); std::cout << "Received new message on topic " << topic << ": (JSON message)\n"; } } }; int main() { auto &client = gg::ipc::Client::get(); auto error = client.connect(); if (error) { std::cerr << "Failed to establish IPC connection.\n"; exit(-1); } std::string_view topic = "my/topic"; static ResponseHandler handler; error = client.subscribe_to_topic(topic, handler); if (error) { std::cerr << "Failed to subscribe to topic: " << topic << "\n"; exit(-1); } std::cout << "Successfully subscribed to topic: " << topic << "\n"; // Keep the main thread alive, or the process will exit. while (1) { sleep(10); } } ``` ------ ## IPC 最佳實務在自訂元件中使用 IPC 的最佳實務在 IPC 用戶端 V1 和 IPC 用戶端 V2 之間有所不同。遵循您使用的 IPC 用戶端版本的最佳實務。 ------ #### [ IPC client V2 ] IPC 用戶端 V2 會在單獨的執行緒中執行回呼函數，因此相較於 IPC 用戶端 V1，當您使用 IPC 和寫入訂閱處理常式函數時，必須遵循的準則較少。 + **重複使用一個 IPC 用戶端** 建立 IPC 用戶端之後，請保持開啟狀態，並重複使用於所有 IPC 操作。建立多個用戶端會使用額外的資源，並可能導致資源洩漏。 + **處理例外狀況** IPC 用戶端 V2 會記錄訂閱處理常式函數中未攔截的例外狀況。您應該在處理常式函數中擷取例外狀況，以處理程式碼中發生的錯誤。 ------ #### [ IPC client V1 ] IPC 用戶端 V1 使用與 IPC 伺服器通訊的單一執行緒，並呼叫訂閱處理常式。當您寫入訂閱處理常式函數時，必須考慮此同步行為。 + **重複使用一個 IPC 用戶端** 建立 IPC 用戶端之後，請保持開啟狀態，並重複使用於所有 IPC 操作。建立多個用戶端會使用額外的資源，並可能導致資源洩漏。 + **以非同步方式執行封鎖碼** IPC 用戶端 V1 無法在執行緒遭到封鎖時傳送新請求或處理新事件訊息。您應該在從處理常式函數執行的個別執行緒中執行封鎖碼。封鎖程式碼包含`sleep`呼叫、持續執行的迴圈，以及需要時間才能完成的同步 I/O 請求。 + **以非同步方式傳送新的 IPC 請求** IPC 用戶端 V1 無法從訂閱處理常式函數內傳送新請求，因為如果您等待回應，請求會封鎖處理常式函數。您應該在從處理常式函數執行的個別執行緒中傳送 IPC 請求。 + **處理例外狀況** IPC 用戶端 V1 不會處理訂閱處理常式函數中未攔截的例外狀況。如果您的處理常式函數擲回例外狀況，訂閱會關閉，而且例外狀況不會出現在您的元件日誌中。您應該在處理常式函數中擷取例外狀況，以保持訂閱開啟，並記錄程式碼中發生的錯誤。 ------ # 發佈/訂閱本機訊息發佈/訂閱 (pubsub) 訊息可讓您傳送和接收主題的訊息。元件可以將訊息發佈至主題，以傳送訊息至其他元件。然後，訂閱該主題的元件可以對其接收的訊息採取行動。 **注意** 您無法使用此發佈/訂閱 IPC 服務來發佈或訂閱 AWS IoT Core MQTT。如需如何與 AWS IoT Core MQTT 交換訊息的詳細資訊，請參閱 [發佈/訂閱 AWS IoT Core MQTT 訊息](ipc-iot-core-mqtt.md)。 **Topics** + [最低 SDK 版本](#ipc-publish-subscribe-sdk-versions) + [Authorization](#ipc-publish-subscribe-authorization) + [PublishToTopic](#ipc-operation-publishtotopic) + [SubscribeToTopic](#ipc-operation-subscribetotopic) + [範例](#ipc-publish-subscribe-examples) ## 最低 SDK 版本下表列出在本機主題之間發佈和訂閱訊息時 AWS IoT Device SDK ，您必須使用的最低版本。 | SDK | 最低版本 | | --- | --- | | [AWS IoT Device SDK for Java v2](https://github.com/aws/aws-iot-device-sdk-java-v2) | v1.2.10 | | [AWS IoT Device SDK for Python v2](https://github.com/aws/aws-iot-device-sdk-python-v2) | 1.5.3 版 | | [AWS IoT Device SDK 適用於 C\$1\$1 v2](https://github.com/aws/aws-iot-device-sdk-cpp-v2) | 1.17.0 版 | | [適用於 JavaScript 的 AWS IoT Device SDK ](https://github.com/aws/aws-iot-device-sdk-js-v2) v2 | v1.12.0 | ## Authorization 若要在自訂元件中使用本機發佈/訂閱訊息，您必須定義授權政策，允許您的元件向主題傳送和接收訊息。如需定義授權政策的資訊，請參閱 [授權元件執行 IPC 操作](interprocess-communication.md#ipc-authorization-policies)。發佈/訂閱訊息的授權政策具有下列屬性。 **IPC 服務識別符：** `aws.greengrass.ipc.pubsub` | 作業 | Description | Resources | | --- | --- | --- | | `aws.greengrass#PublishToTopic` | 允許元件發佈訊息到您指定的主題。 | 主題字串，例如 `test/topic`。使用 `*`來比對主題中任意字元組合。此主題字串不支援 MQTT 主題萬用字元 (`#` 和 `+`)。 | | `aws.greengrass#SubscribeToTopic` | 允許元件訂閱您指定主題的訊息。 | 主題字串，例如 `test/topic`。使用 `*`來比對主題中任意字元組合。在 [Greengrass nucleus](greengrass-nucleus-component.md) v2.6.0 和更新版本中，您可以訂閱包含 MQTT 主題萬用字元 (`#` 和 `+`) 的主題。此主題字串支援 MQTT 主題萬用字元做為文字字元。例如，如果元件的授權政策授予的存取權`test/topic/#`，則該元件可以訂閱 `test/topic/#`，但無法訂閱 `test/topic/filter`。 | | `*` | 允許元件發佈和訂閱您指定主題的訊息。 | 主題字串，例如 `test/topic`。使用 `*`來比對主題中任意字元組合。在 [Greengrass nucleus](greengrass-nucleus-component.md) v2.6.0 和更新版本中，您可以訂閱包含 MQTT 主題萬用字元 (`#` 和 `+`) 的主題。此主題字串支援 MQTT 主題萬用字元做為文字字元。例如，如果元件的授權政策授予的存取權`test/topic/#`，則該元件可以訂閱 `test/topic/#`，但無法訂閱 `test/topic/filter`。 | ### 授權政策範例您可以參考下列授權政策範例，協助您設定元件的授權政策。 **Example 範例授權政策** 下列範例授權政策允許元件發佈和訂閱所有主題。 ``` { "accessControl": { "aws.greengrass.ipc.pubsub": { "com.example.MyLocalPubSubComponent:pubsub:1": { "policyDescription": "Allows access to publish/subscribe to all topics.", "operations": [ "aws.greengrass#PublishToTopic", "aws.greengrass#SubscribeToTopic" ], "resources": [ "*" ] } } } } ``` ## PublishToTopic 發布訊息至主題。 ### 請求此操作的請求具有下列參數： `topic` 要發佈訊息的主題。 `publishMessage` (Python：`publish_message`) 要發佈的訊息。此物件 `PublishMessage`包含下列資訊。您必須指定其中一個 `jsonMessage`和 `binaryMessage`。 `jsonMessage` (Python：`json_message`) （選用） JSON 訊息。此物件 `JsonMessage`包含下列資訊： `message` JSON 訊息做為物件。 `context` 訊息的內容，例如發佈訊息的主題。此功能適用於 [Greengrass 核元件](greengrass-nucleus-component.md)的 v2.6.0 和更新版本。下表列出 AWS IoT Device SDK 您必須用來存取訊息內容的最低版本。 [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/zh_tw/greengrass/v2/developerguide/ipc-publish-subscribe.html) AWS IoT Greengrass 核心軟體在 `PublishToTopic`和 `SubscribeToTopic`操作中使用相同的訊息物件。 AWS IoT Greengrass 核心軟體會在您訂閱時於訊息中設定此內容物件，並在您發佈的訊息中忽略此內容物件。此物件 `MessageContext`包含下列資訊： `topic` 發佈訊息的主題。 `binaryMessage` (Python：`binary_message`) （選用）二進位訊息。此物件 `BinaryMessage`包含下列資訊： `message` Blob 形式的二進位訊息。 `context` 訊息的內容，例如發佈訊息的主題。此功能適用於 [Greengrass 核元件](greengrass-nucleus-component.md)的 v2.6.0 和更新版本。下表列出 AWS IoT Device SDK 您必須用來存取訊息內容的最低版本。 [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/zh_tw/greengrass/v2/developerguide/ipc-publish-subscribe.html) AWS IoT Greengrass 核心軟體在 `PublishToTopic`和 `SubscribeToTopic`操作中使用相同的訊息物件。 AWS IoT Greengrass 核心軟體會在您訂閱時於訊息中設定此內容物件，並在您發佈的訊息中忽略此內容物件。此物件 `MessageContext`包含下列資訊： `topic` 發佈訊息的主題。 ### 回應此操作不會在其回應中提供任何資訊。 ### 範例下列範例示範如何在自訂元件程式碼中呼叫此操作。 ------ #### [ Java (IPC client V2) ] **Example 範例：發佈二進位訊息** ``` package com.aws.greengrass.docs.samples.ipc; import software.amazon.awssdk.aws.greengrass.GreengrassCoreIPCClientV2; import software.amazon.awssdk.aws.greengrass.model.BinaryMessage; import software.amazon.awssdk.aws.greengrass.model.PublishMessage; import software.amazon.awssdk.aws.greengrass.model.PublishToTopicRequest; import software.amazon.awssdk.aws.greengrass.model.PublishToTopicResponse; import software.amazon.awssdk.aws.greengrass.model.UnauthorizedError; import java.nio.charset.StandardCharsets; public class PublishToTopicV2 { public static void main(String[] args) { String topic = args[0]; String message = args[1]; try (GreengrassCoreIPCClientV2 ipcClient = GreengrassCoreIPCClientV2.builder().build()) { PublishToTopicV2.publishBinaryMessageToTopic(ipcClient, topic, message); System.out.println("Successfully published to topic: " + topic); } catch (Exception e) { if (e.getCause() instanceof UnauthorizedError) { System.err.println("Unauthorized error while publishing to topic: " + topic); } else { System.err.println("Exception occurred when using IPC."); } e.printStackTrace(); System.exit(1); } } public static PublishToTopicResponse publishBinaryMessageToTopic( GreengrassCoreIPCClientV2 ipcClient, String topic, String message) throws InterruptedException { BinaryMessage binaryMessage = new BinaryMessage().withMessage(message.getBytes(StandardCharsets.UTF_8)); PublishMessage publishMessage = new PublishMessage().withBinaryMessage(binaryMessage); PublishToTopicRequest publishToTopicRequest = new PublishToTopicRequest().withTopic(topic).withPublishMessage(publishMessage); return ipcClient.publishToTopic(publishToTopicRequest); } } ``` ------ #### [ Python (IPC client V2) ] **Example 範例：發佈二進位訊息** ``` import sys import traceback from awsiot.greengrasscoreipc.clientv2 import GreengrassCoreIPCClientV2 from awsiot.greengrasscoreipc.model import ( PublishMessage, BinaryMessage ) def main(): args = sys.argv[1:] topic = args[0] message = args[1] try: ipc_client = GreengrassCoreIPCClientV2() publish_binary_message_to_topic(ipc_client, topic, message) print('Successfully published to topic: ' + topic) except Exception: print('Exception occurred', file=sys.stderr) traceback.print_exc() exit(1) def publish_binary_message_to_topic(ipc_client, topic, message): binary_message = BinaryMessage(message=bytes(message, 'utf-8')) publish_message = PublishMessage(binary_message=binary_message) return ipc_client.publish_to_topic(topic=topic, publish_message=publish_message) if __name__ == '__main__': main() ``` ------ #### [ C\$1\$1 (IPC client V1) ] **Example 範例：發佈二進位訊息** ``` #include #include #include using namespace Aws::Crt; using namespace Aws::Greengrass; class IpcClientLifecycleHandler : public ConnectionLifecycleHandler { void OnConnectCallback() override { // Handle connection to IPC service. } void OnDisconnectCallback(RpcError error) override { // Handle disconnection from IPC service. } bool OnErrorCallback(RpcError error) override { // Handle IPC service connection error. return true; } }; int main() { ApiHandle apiHandle(g_allocator); Io::EventLoopGroup eventLoopGroup(1); Io::DefaultHostResolver socketResolver(eventLoopGroup, 64, 30); Io::ClientBootstrap bootstrap(eventLoopGroup, socketResolver); IpcClientLifecycleHandler ipcLifecycleHandler; GreengrassCoreIpcClient ipcClient(bootstrap); auto connectionStatus = ipcClient.Connect(ipcLifecycleHandler).get(); if (!connectionStatus) { std::cerr << "Failed to establish IPC connection: " << connectionStatus.StatusToString() << std::endl; exit(-1); } String topic("my/topic"); String message("Hello, World!"); int timeout = 10; PublishToTopicRequest request; Vector messageData({message.begin(), message.end()}); BinaryMessage binaryMessage; binaryMessage.SetMessage(messageData); PublishMessage publishMessage; publishMessage.SetBinaryMessage(binaryMessage); request.SetTopic(topic); request.SetPublishMessage(publishMessage); auto operation = ipcClient.NewPublishToTopic(); auto activate = operation->Activate(request, nullptr); activate.wait(); auto responseFuture = operation->GetResult(); if (responseFuture.wait_for(std::chrono::seconds(timeout)) == std::future_status::timeout) { std::cerr << "Operation timed out while waiting for response from Greengrass Core." << std::endl; exit(-1); } auto response = responseFuture.get(); if (!response) { // Handle error. auto errorType = response.GetResultType(); if (errorType == OPERATION_ERROR) { auto *error = response.GetOperationError(); (void)error; // Handle operation error. } else { // Handle RPC error. } } return 0; } ``` ------ #### [ JavaScript ] **Example 範例：發佈二進位訊息** ``` import * as greengrasscoreipc from "aws-iot-device-sdk-v2/dist/greengrasscoreipc"; import {BinaryMessage, PublishMessage, PublishToTopicRequest} from "aws-iot-device-sdk-v2/dist/greengrasscoreipc/model"; class PublishToTopic { private ipcClient : greengrasscoreipc.Client private readonly topic : string; private readonly messageString : string; constructor() { // define your own constructor, e.g. this.topic = ""; this.messageString = ""; this.publishToTopic().then(r => console.log("Started workflow")); } private async publishToTopic() { try { this.ipcClient = await getIpcClient(); const binaryMessage : BinaryMessage = { message: this.messageString } const publishMessage : PublishMessage = { binaryMessage: binaryMessage } const request : PublishToTopicRequest = { topic: this.topic, publishMessage: publishMessage } this.ipcClient.publishToTopic(request).finally(() => console.log(`Published message ${publishMessage.binaryMessage?.message} to topic`)) } catch (e) { // parse the error depending on your use cases throw e } } } export async function getIpcClient(){ try { const ipcClient = greengrasscoreipc.createClient(); await ipcClient.connect() .catch(error => { // parse the error depending on your use cases throw error; }); return ipcClient } catch (err) { // parse the error depending on your use cases throw err } } // starting point const publishToTopic = new PublishToTopic(); ``` ------ #### [ Rust ] **Example 範例：發佈二進位訊息** ``` use gg_sdk::Sdk; fn main() { let sdk = Sdk::init(); sdk.connect().expect("Failed to establish IPC connection"); let message = b"Hello, World"; let topic = "my/topic"; sdk.publish_to_topic_binary(topic, message) .expect("Failed to publish to topic"); println!("Successfully published to topic: {topic}"); } ``` ------ #### [ C ] **Example 範例：發佈二進位訊息** ``` #include #include #include #include #include int main(void) { gg_sdk_init(); GgError err = ggipc_connect(); if (err != GG_ERR_OK) { fprintf(stderr, "Failed to establish IPC connection.\n"); exit(-1); } GgBuffer message = GG_STR("Hello, World"); GgBuffer topic = GG_STR("my/topic"); err = ggipc_publish_to_topic_binary(topic, message); if (err != GG_ERR_OK) { fprintf( stderr, "Failed to publish to topic: %.*s\n", (int) topic.len, topic.data ); exit(-1); } printf( "Successfully published to topic: %.*s\n", (int) topic.len, topic.data ); } ``` ------ #### [ C\$1\$1 (Component SDK) ] **Example 範例：發佈二進位訊息** ``` #include #include int main() { auto &client = gg::ipc::Client::get(); auto error = client.connect(); if (error) { std::cerr << "Failed to establish IPC connection.\n"; exit(-1); } std::string_view message = "Hello, World"; std::string_view topic = "my/topic"; error = client.publish_to_topic(topic, message); if (error) { std::cerr << "Failed to publish to topic: " << topic << "\n"; exit(-1); } std::cout << "Successfully published to topic: " << topic << "\n"; } ``` ------ ## SubscribeToTopic 訂閱主題上的訊息。此操作是您訂閱事件訊息串流的訂閱操作。若要使用此操作，請使用處理事件訊息、錯誤和串流關閉的函數來定義串流回應處理常式。如需詳細資訊，請參閱[訂閱 IPC 事件串流](interprocess-communication.md#ipc-subscribe-operations)。 **事件訊息類型：** `SubscriptionResponseMessage` ### 請求此操作的請求具有下列參數： `topic` 要訂閱的主題。在 [Greengrass 核](greengrass-nucleus-component.md) v2.6.0 和更新版本中，此主題支援 MQTT 主題萬用字元 (`#` 和 `+`)。 `receiveMode` (Python：`receive_mode`) （選用）指定元件是否從本身接收訊息的行為。您可以變更此行為，以允許元件對自己的訊息採取動作。預設行為取決於主題是否包含 MQTT 萬用字元。您可以從以下選項中選擇： + `RECEIVE_ALL_MESSAGES` – 接收符合主題的所有訊息，包括來自訂閱元件的訊息。當您訂閱不包含 MQTT 萬用字元的主題時，此模式是預設選項。 + `RECEIVE_MESSAGES_FROM_OTHERS` – 接收符合主題的所有訊息，但來自訂閱元件的訊息除外。當您訂閱包含 MQTT 萬用字元的主題時，此模式是預設選項。此功能適用於 [Greengrass 核元件](greengrass-nucleus-component.md)的 v2.6.0 和更新版本。下表列出 AWS IoT Device SDK 您必須用來設定接收模式的最低版本。 [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/zh_tw/greengrass/v2/developerguide/ipc-publish-subscribe.html) ### 回應此操作的回應包含下列資訊： `messages` 訊息串流。此物件 `SubscriptionResponseMessage`包含下列資訊。每個訊息都包含 `jsonMessage`或 `binaryMessage`。 `jsonMessage` (Python：`json_message`) （選用） JSON 訊息。此物件 `JsonMessage`包含下列資訊： `message` JSON 訊息做為物件。 `context` 訊息的內容，例如發佈訊息的主題。此功能適用於 [Greengrass 核元件](greengrass-nucleus-component.md)的 v2.6.0 和更新版本。下表列出 AWS IoT Device SDK 您必須用來存取訊息內容的最低版本。 [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/zh_tw/greengrass/v2/developerguide/ipc-publish-subscribe.html) AWS IoT Greengrass 核心軟體在 `PublishToTopic`和 `SubscribeToTopic`操作中使用相同的訊息物件。 AWS IoT Greengrass 核心軟體會在您訂閱時於訊息中設定此內容物件，並在您發佈的訊息中忽略此內容物件。此物件 `MessageContext`包含下列資訊： `topic` 發佈訊息的主題。 `binaryMessage` (Python：`binary_message`) （選用）二進位訊息。此物件 `BinaryMessage`包含下列資訊： `message` Blob 形式的二進位訊息。 `context` 訊息的內容，例如發佈訊息的主題。此功能適用於 [Greengrass 核元件](greengrass-nucleus-component.md)的 v2.6.0 和更新版本。下表列出 AWS IoT Device SDK 您必須用來存取訊息內容的最低版本。 [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/zh_tw/greengrass/v2/developerguide/ipc-publish-subscribe.html) AWS IoT Greengrass 核心軟體在 `PublishToTopic`和 `SubscribeToTopic`操作中使用相同的訊息物件。 AWS IoT Greengrass 核心軟體會在您訂閱時於訊息中設定此內容物件，並在您發佈的訊息中忽略此內容物件。此物件 `MessageContext`包含下列資訊： `topic` 發佈訊息的主題。 `topicName` (Python：`topic_name`) 發佈訊息的主題。此屬性目前未使用。在 [Greengrass 核](greengrass-nucleus-component.md) v2.6.0 和更新版本中，您可以從取得 `(jsonMessage|binaryMessage).context.topic`值`SubscriptionResponseMessage`，以取得訊息發佈的主題。 ### 範例下列範例示範如何在自訂元件程式碼中呼叫此操作。 ------ #### [ Java (IPC client V2) ] **Example 範例：訂閱本機發佈/訂閱訊息** ``` package com.aws.greengrass.docs.samples.ipc; import software.amazon.awssdk.aws.greengrass.GreengrassCoreIPCClientV2; import software.amazon.awssdk.aws.greengrass.SubscribeToTopicResponseHandler; import software.amazon.awssdk.aws.greengrass.model.*; import java.nio.charset.StandardCharsets; import java.util.Optional; public class SubscribeToTopicV2 { public static void main(String[] args) { String topic = args[0]; try (GreengrassCoreIPCClientV2 ipcClient = GreengrassCoreIPCClientV2.builder().build()) { SubscribeToTopicRequest request = new SubscribeToTopicRequest().withTopic(topic); GreengrassCoreIPCClientV2.StreamingResponse response = ipcClient.subscribeToTopic(request, SubscribeToTopicV2::onStreamEvent, Optional.of(SubscribeToTopicV2::onStreamError), Optional.of(SubscribeToTopicV2::onStreamClosed)); SubscribeToTopicResponseHandler responseHandler = response.getHandler(); System.out.println("Successfully subscribed to topic: " + topic); // Keep the main thread alive, or the process will exit. try { while (true) { Thread.sleep(10000); } } catch (InterruptedException e) { System.out.println("Subscribe interrupted."); } // To stop subscribing, close the stream. responseHandler.closeStream(); } catch (Exception e) { if (e.getCause() instanceof UnauthorizedError) { System.err.println("Unauthorized error while publishing to topic: " + topic); } else { System.err.println("Exception occurred when using IPC."); } e.printStackTrace(); System.exit(1); } } public static void onStreamEvent(SubscriptionResponseMessage subscriptionResponseMessage) { try { BinaryMessage binaryMessage = subscriptionResponseMessage.getBinaryMessage(); String message = new String(binaryMessage.getMessage(), StandardCharsets.UTF_8); String topic = binaryMessage.getContext().getTopic(); System.out.printf("Received new message on topic %s: %s%n", topic, message); } catch (Exception e) { System.err.println("Exception occurred while processing subscription response " + "message."); e.printStackTrace(); } } public static boolean onStreamError(Throwable error) { System.err.println("Received a stream error."); error.printStackTrace(); return false; // Return true to close stream, false to keep stream open. } public static void onStreamClosed() { System.out.println("Subscribe to topic stream closed."); } } ``` ------ #### [ Python (IPC client V2) ] **Example 範例：訂閱本機發佈/訂閱訊息** ``` import sys import time import traceback from awsiot.greengrasscoreipc.clientv2 import GreengrassCoreIPCClientV2 from awsiot.greengrasscoreipc.model import ( SubscriptionResponseMessage, UnauthorizedError ) def main(): args = sys.argv[1:] topic = args[0] try: ipc_client = GreengrassCoreIPCClientV2() # Subscription operations return a tuple with the response and the operation. _, operation = ipc_client.subscribe_to_topic(topic=topic, on_stream_event=on_stream_event, on_stream_error=on_stream_error, on_stream_closed=on_stream_closed) print('Successfully subscribed to topic: ' + topic) # Keep the main thread alive, or the process will exit. try: while True: time.sleep(10) except InterruptedError: print('Subscribe interrupted.') # To stop subscribing, close the stream. operation.close() except UnauthorizedError: print('Unauthorized error while subscribing to topic: ' + topic, file=sys.stderr) traceback.print_exc() exit(1) except Exception: print('Exception occurred', file=sys.stderr) traceback.print_exc() exit(1) def on_stream_event(event: SubscriptionResponseMessage) -> None: try: message = str(event.binary_message.message, 'utf-8') topic = event.binary_message.context.topic print('Received new message on topic %s: %s' % (topic, message)) except: traceback.print_exc() def on_stream_error(error: Exception) -> bool: print('Received a stream error.', file=sys.stderr) traceback.print_exc() return False # Return True to close stream, False to keep stream open. def on_stream_closed() -> None: print('Subscribe to topic stream closed.') if __name__ == '__main__': main() ``` ------ #### [ C\$1\$1 (IPC client V1) ] **Example 範例：訂閱本機發佈/訂閱訊息** ``` #include #include #include using namespace Aws::Crt; using namespace Aws::Greengrass; class SubscribeResponseHandler : public SubscribeToTopicStreamHandler { public: virtual ~SubscribeResponseHandler() {} private: void OnStreamEvent(SubscriptionResponseMessage *response) override { auto jsonMessage = response->GetJsonMessage(); if (jsonMessage.has_value() && jsonMessage.value().GetMessage().has_value()) { auto messageString = jsonMessage.value().GetMessage().value().View().WriteReadable(); // Handle JSON message. } else { auto binaryMessage = response->GetBinaryMessage(); if (binaryMessage.has_value() && binaryMessage.value().GetMessage().has_value()) { auto messageBytes = binaryMessage.value().GetMessage().value(); std::string messageString(messageBytes.begin(), messageBytes.end()); // Handle binary message. } } } bool OnStreamError(OperationError *error) override { // Handle error. return false; // Return true to close stream, false to keep stream open. } void OnStreamClosed() override { // Handle close. } }; class IpcClientLifecycleHandler : public ConnectionLifecycleHandler { void OnConnectCallback() override { // Handle connection to IPC service. } void OnDisconnectCallback(RpcError error) override { // Handle disconnection from IPC service. } bool OnErrorCallback(RpcError error) override { // Handle IPC service connection error. return true; } }; int main() { ApiHandle apiHandle(g_allocator); Io::EventLoopGroup eventLoopGroup(1); Io::DefaultHostResolver socketResolver(eventLoopGroup, 64, 30); Io::ClientBootstrap bootstrap(eventLoopGroup, socketResolver); IpcClientLifecycleHandler ipcLifecycleHandler; GreengrassCoreIpcClient ipcClient(bootstrap); auto connectionStatus = ipcClient.Connect(ipcLifecycleHandler).get(); if (!connectionStatus) { std::cerr << "Failed to establish IPC connection: " << connectionStatus.StatusToString() << std::endl; exit(-1); } String topic("my/topic"); int timeout = 10; SubscribeToTopicRequest request; request.SetTopic(topic); //SubscribeResponseHandler streamHandler; auto streamHandler = MakeShared(DefaultAllocator()); auto operation = ipcClient.NewSubscribeToTopic(streamHandler); auto activate = operation->Activate(request, nullptr); activate.wait(); auto responseFuture = operation->GetResult(); if (responseFuture.wait_for(std::chrono::seconds(timeout)) == std::future_status::timeout) { std::cerr << "Operation timed out while waiting for response from Greengrass Core." << std::endl; exit(-1); } auto response = responseFuture.get(); if (!response) { // Handle error. auto errorType = response.GetResultType(); if (errorType == OPERATION_ERROR) { auto *error = response.GetOperationError(); (void)error; // Handle operation error. } else { // Handle RPC error. } exit(-1); } // Keep the main thread alive, or the process will exit. while (true) { std::this_thread::sleep_for(std::chrono::seconds(10)); } operation->Close(); return 0; } ``` ------ #### [ JavaScript ] **Example 範例：訂閱本機發佈/訂閱訊息** ``` import * as greengrasscoreipc from "aws-iot-device-sdk-v2/dist/greengrasscoreipc"; import {SubscribeToTopicRequest, SubscriptionResponseMessage} from "aws-iot-device-sdk-v2/dist/greengrasscoreipc/model"; import {RpcError} from "aws-iot-device-sdk-v2/dist/eventstream_rpc"; class SubscribeToTopic { private ipcClient : greengrasscoreipc.Client private readonly topic : string; constructor() { // define your own constructor, e.g. this.topic = ""; this.subscribeToTopic().then(r => console.log("Started workflow")); } private async subscribeToTopic() { try { this.ipcClient = await getIpcClient(); const subscribeToTopicRequest : SubscribeToTopicRequest = { topic: this.topic, } const streamingOperation = this.ipcClient.subscribeToTopic(subscribeToTopicRequest, undefined); // conditionally apply options streamingOperation.on("message", (message: SubscriptionResponseMessage) => { // parse the message depending on your use cases, e.g. if(message.binaryMessage && message.binaryMessage.message) { const receivedMessage = message.binaryMessage?.message.toString(); } }); streamingOperation.on("streamError", (error : RpcError) => { // define your own error handling logic }) streamingOperation.on("ended", () => { // define your own logic }) await streamingOperation.activate(); // Keep the main thread alive, or the process will exit. await new Promise((resolve) => setTimeout(resolve, 10000)) } catch (e) { // parse the error depending on your use cases throw e } } } export async function getIpcClient(){ try { const ipcClient = greengrasscoreipc.createClient(); await ipcClient.connect() .catch(error => { // parse the error depending on your use cases throw error; }); return ipcClient } catch (err) { // parse the error depending on your use cases throw err } } // starting point const subscribeToTopic = new SubscribeToTopic(); ``` ------ #### [ Rust ] **Example 範例：訂閱本機發佈/訂閱訊息** ``` use gg_sdk::{Sdk, SubscribeToTopicPayload}; use std::{thread, time::Duration}; fn main() { let sdk = Sdk::init(); sdk.connect().expect("Failed to establish IPC connection"); let topic = "my/topic"; let callback = |topic: &str, payload: SubscribeToTopicPayload| match payload { SubscribeToTopicPayload::Binary(message) => { let message = String::from_utf8_lossy(message); println!("Received new message on topic {topic}: {message}"); } SubscribeToTopicPayload::Json(_) => { println!("Received new message on topic {topic}: (JSON message)"); } }; let _sub = sdk .subscribe_to_topic(topic, &callback) .expect("Failed to subscribe to topic"); println!("Successfully subscribed to topic: {topic}"); // Keep the main thread alive, or the process will exit. loop { thread::sleep(Duration::from_secs(10)); } } ``` ------ #### [ C ] **Example 範例：訂閱本機發佈/訂閱訊息** ``` #include #include #include #include #include #include #include #include #include static void on_subscription_response( void *ctx, GgBuffer topic, GgObject payload, GgIpcSubscriptionHandle handle ) { (void) ctx; (void) handle; if (gg_obj_type(payload) == GG_TYPE_BUF) { GgBuffer message = gg_obj_into_buf(payload); printf( "Received new message on topic %.*s: %.*s\n", (int) topic.len, topic.data, (int) message.len, message.data ); } else { assert(gg_obj_type(payload) == GG_TYPE_MAP); printf( "Received new message on topic %.*s: (JSON message)\n", (int) topic.len, topic.data ); } } int main(void) { gg_sdk_init(); GgError err = ggipc_connect(); if (err != GG_ERR_OK) { fprintf(stderr, "Failed to establish IPC connection.\n"); exit(-1); } GgBuffer topic = GG_STR("my/topic"); GgIpcSubscriptionHandle handle; err = ggipc_subscribe_to_topic( topic, on_subscription_response, NULL, &handle ); if (err != GG_ERR_OK) { fprintf( stderr, "Failed to subscribe to topic: %.*s\n", (int) topic.len, topic.data ); exit(-1); } printf( "Successfully subscribed to topic: %.*s\n", (int) topic.len, topic.data ); // Keep the main thread alive, or the process will exit. while (1) { sleep(10); } // To stop subscribing, close the stream. ggipc_close_subscription(handle); } ``` ------ #### [ C\$1\$1 (Component SDK) ] **Example 範例：訂閱本機發佈/訂閱訊息** ``` #include #include #include #include #include class ResponseHandler : public gg::ipc::LocalTopicCallback { void operator()( std::string_view topic, gg::Object payload, gg::ipc::Subscription &handle ) override { (void) handle; if (payload.index() == GG_TYPE_BUF) { std::cout << "Received new message on topic " << topic << ": " << get(payload) << "\n"; } else { assert(payload.index() == GG_TYPE_MAP); std::cout << "Received new message on topic " << topic << ": (JSON message)\n"; } } }; int main() { auto &client = gg::ipc::Client::get(); auto error = client.connect(); if (error) { std::cerr << "Failed to establish IPC connection.\n"; exit(-1); } std::string_view topic = "my/topic"; static ResponseHandler handler; error = client.subscribe_to_topic(topic, handler); if (error) { std::cerr << "Failed to subscribe to topic: " << topic << "\n"; exit(-1); } std::cout << "Successfully subscribed to topic: " << topic << "\n"; // Keep the main thread alive, or the process will exit. while (1) { sleep(10); } } ``` ------ ## 範例使用下列範例來了解如何在元件中使用發佈/訂閱 IPC 服務。 ### 發佈/訂閱發佈者範例 (Java、IPC 用戶端 V1) 下列範例配方允許元件發佈至所有主題。 ------ #### [ JSON ] ``` { "RecipeFormatVersion": "2020-01-25", "ComponentName": "com.example.PubSubPublisherJava", "ComponentVersion": "1.0.0", "ComponentDescription": "A component that publishes messages.", "ComponentPublisher": "Amazon", "ComponentConfiguration": { "DefaultConfiguration": { "accessControl": { "aws.greengrass.ipc.pubsub": { "com.example.PubSubPublisherJava:pubsub:1": { "policyDescription": "Allows access to publish to all topics.", "operations": [ "aws.greengrass#PublishToTopic" ], "resources": [ "*" ] } } } } }, "Manifests": [ { "Lifecycle": { "Run": "java -jar {artifacts:path}/PubSubPublisher.jar" } } ] } ``` ------ #### [ YAML ] ``` --- RecipeFormatVersion: '2020-01-25' ComponentName: com.example.PubSubPublisherJava ComponentVersion: '1.0.0' ComponentDescription: A component that publishes messages. ComponentPublisher: Amazon ComponentConfiguration: DefaultConfiguration: accessControl: aws.greengrass.ipc.pubsub: 'com.example.PubSubPublisherJava:pubsub:1': policyDescription: Allows access to publish to all topics. operations: - 'aws.greengrass#PublishToTopic' resources: - '*' Manifests: - Lifecycle: Run: |- java -jar {artifacts:path}/PubSubPublisher.jar ``` ------ 下列範例 Java 應用程式示範如何使用發佈/訂閱 IPC 服務將訊息發佈至其他元件。 ``` /* Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. * SPDX-License-Identifier: Apache-2.0 */ package com.example.ipc.pubsub; import software.amazon.awssdk.aws.greengrass.GreengrassCoreIPCClient; import software.amazon.awssdk.aws.greengrass.model.*; import software.amazon.awssdk.eventstreamrpc.EventStreamRPCConnection; import java.nio.charset.StandardCharsets; import java.util.Optional; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; public class PubSubPublisher { public static void main(String[] args) { String message = "Hello from the pub/sub publisher (Java)."; String topic = "test/topic/java"; try (EventStreamRPCConnection eventStreamRPCConnection = IPCUtils.getEventStreamRpcConnection()) { GreengrassCoreIPCClient ipcClient = new GreengrassCoreIPCClient(eventStreamRPCConnection); while (true) { PublishToTopicRequest publishRequest = new PublishToTopicRequest(); PublishMessage publishMessage = new PublishMessage(); BinaryMessage binaryMessage = new BinaryMessage(); binaryMessage.setMessage(message.getBytes(StandardCharsets.UTF_8)); publishMessage.setBinaryMessage(binaryMessage); publishRequest.setPublishMessage(publishMessage); publishRequest.setTopic(topic); CompletableFuture futureResponse = ipcClient .publishToTopic(publishRequest, Optional.empty()).getResponse(); try { futureResponse.get(10, TimeUnit.SECONDS); System.out.println("Successfully published to topic: " + topic); } catch (TimeoutException e) { System.err.println("Timeout occurred while publishing to topic: " + topic); } catch (ExecutionException e) { if (e.getCause() instanceof UnauthorizedError) { System.err.println("Unauthorized error while publishing to topic: " + topic); } else { System.err.println("Execution exception while publishing to topic: " + topic); } throw e; } Thread.sleep(5000); } } catch (InterruptedException e) { System.out.println("Publisher interrupted."); } catch (Exception e) { System.err.println("Exception occurred when using IPC."); e.printStackTrace(); System.exit(1); } } } ``` ### 發佈/訂閱訂閱者範例 (Java、IPC 用戶端 V1) 下列範例配方允許元件訂閱所有主題。 ------ #### [ JSON ] ``` { "RecipeFormatVersion": "2020-01-25", "ComponentName": "com.example.PubSubSubscriberJava", "ComponentVersion": "1.0.0", "ComponentDescription": "A component that subscribes to messages.", "ComponentPublisher": "Amazon", "ComponentConfiguration": { "DefaultConfiguration": { "accessControl": { "aws.greengrass.ipc.pubsub": { "com.example.PubSubSubscriberJava:pubsub:1": { "policyDescription": "Allows access to subscribe to all topics.", "operations": [ "aws.greengrass#SubscribeToTopic" ], "resources": [ "*" ] } } } } }, "Manifests": [ { "Lifecycle": { "Run": "java -jar {artifacts:path}/PubSubSubscriber.jar" } } ] } ``` ------ #### [ YAML ] ``` --- RecipeFormatVersion: '2020-01-25' ComponentName: com.example.PubSubSubscriberJava ComponentVersion: '1.0.0' ComponentDescription: A component that subscribes to messages. ComponentPublisher: Amazon ComponentConfiguration: DefaultConfiguration: accessControl: aws.greengrass.ipc.pubsub: 'com.example.PubSubSubscriberJava:pubsub:1': policyDescription: Allows access to subscribe to all topics. operations: - 'aws.greengrass#SubscribeToTopic' resources: - '*' Manifests: - Lifecycle: Run: |- java -jar {artifacts:path}/PubSubSubscriber.jar ``` ------ 下列範例 Java 應用程式示範如何使用發佈/訂閱 IPC 服務來訂閱其他元件的訊息。 ``` /* Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. * SPDX-License-Identifier: Apache-2.0 */ package com.example.ipc.pubsub; import software.amazon.awssdk.aws.greengrass.GreengrassCoreIPCClient; import software.amazon.awssdk.aws.greengrass.SubscribeToTopicResponseHandler; import software.amazon.awssdk.aws.greengrass.model.SubscribeToTopicRequest; import software.amazon.awssdk.aws.greengrass.model.SubscribeToTopicResponse; import software.amazon.awssdk.aws.greengrass.model.SubscriptionResponseMessage; import software.amazon.awssdk.aws.greengrass.model.UnauthorizedError; import software.amazon.awssdk.eventstreamrpc.EventStreamRPCConnection; import software.amazon.awssdk.eventstreamrpc.StreamResponseHandler; import java.nio.charset.StandardCharsets; import java.util.Optional; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; public class PubSubSubscriber { public static void main(String[] args) { String topic = "test/topic/java"; try (EventStreamRPCConnection eventStreamRPCConnection = IPCUtils.getEventStreamRpcConnection()) { GreengrassCoreIPCClient ipcClient = new GreengrassCoreIPCClient(eventStreamRPCConnection); SubscribeToTopicRequest subscribeRequest = new SubscribeToTopicRequest(); subscribeRequest.setTopic(topic); SubscribeToTopicResponseHandler operationResponseHandler = ipcClient .subscribeToTopic(subscribeRequest, Optional.of(new SubscribeResponseHandler())); CompletableFuture futureResponse = operationResponseHandler.getResponse(); try { futureResponse.get(10, TimeUnit.SECONDS); System.out.println("Successfully subscribed to topic: " + topic); } catch (TimeoutException e) { System.err.println("Timeout occurred while subscribing to topic: " + topic); throw e; } catch (ExecutionException e) { if (e.getCause() instanceof UnauthorizedError) { System.err.println("Unauthorized error while subscribing to topic: " + topic); } else { System.err.println("Execution exception while subscribing to topic: " + topic); } throw e; } // Keep the main thread alive, or the process will exit. try { while (true) { Thread.sleep(10000); } } catch (InterruptedException e) { System.out.println("Subscribe interrupted."); } } catch (Exception e) { System.err.println("Exception occurred when using IPC."); e.printStackTrace(); System.exit(1); } } private static class SubscribeResponseHandler implements StreamResponseHandler { @Override public void onStreamEvent(SubscriptionResponseMessage subscriptionResponseMessage) { try { String message = new String(subscriptionResponseMessage.getBinaryMessage() .getMessage(), StandardCharsets.UTF_8); System.out.println("Received new message: " + message); } catch (Exception e) { e.printStackTrace(); } } @Override public boolean onStreamError(Throwable error) { System.err.println("Received a stream error."); error.printStackTrace(); return false; // Return true to close stream, false to keep stream open. } @Override public void onStreamClosed() { System.out.println("Subscribe to topic stream closed."); } } } ``` ### 發佈/訂閱發佈者範例 (Python、IPC 用戶端 V1) 下列範例配方允許元件發佈至所有主題。 ------ #### [ JSON ] ``` { "RecipeFormatVersion": "2020-01-25", "ComponentName": "com.example.PubSubPublisherPython", "ComponentVersion": "1.0.0", "ComponentDescription": "A component that publishes messages.", "ComponentPublisher": "Amazon", "ComponentConfiguration": { "DefaultConfiguration": { "accessControl": { "aws.greengrass.ipc.pubsub": { "com.example.PubSubPublisherPython:pubsub:1": { "policyDescription": "Allows access to publish to all topics.", "operations": [ "aws.greengrass#PublishToTopic" ], "resources": [ "*" ] } } } } }, "Manifests": [ { "Platform": { "os": "linux" }, "Lifecycle": { "install": "python3 -m pip install --user awsiotsdk", "Run": "python3 -u {artifacts:path}/pubsub_publisher.py" } }, { "Platform": { "os": "windows" }, "Lifecycle": { "install": "py -3 -m pip install --user awsiotsdk", "Run": "py -3 -u {artifacts:path}/pubsub_publisher.py" } } ] } ``` ------ #### [ YAML ] ``` --- RecipeFormatVersion: '2020-01-25' ComponentName: com.example.PubSubPublisherPython ComponentVersion: 1.0.0 ComponentDescription: A component that publishes messages. ComponentPublisher: Amazon ComponentConfiguration: DefaultConfiguration: accessControl: aws.greengrass.ipc.pubsub: com.example.PubSubPublisherPython:pubsub:1: policyDescription: Allows access to publish to all topics. operations: - aws.greengrass#PublishToTopic resources: - "*" Manifests: - Platform: os: linux Lifecycle: install: python3 -m pip install --user awsiotsdk Run: python3 -u {artifacts:path}/pubsub_publisher.py - Platform: os: windows Lifecycle: install: py -3 -m pip install --user awsiotsdk Run: py -3 -u {artifacts:path}/pubsub_publisher.py ``` ------ 下列範例 Python 應用程式示範如何使用發佈/訂閱 IPC 服務將訊息發佈至其他元件。 ``` import concurrent.futures import sys import time import traceback import awsiot.greengrasscoreipc from awsiot.greengrasscoreipc.model import ( PublishToTopicRequest, PublishMessage, BinaryMessage, UnauthorizedError ) topic = "test/topic/python" message = "Hello from the pub/sub publisher (Python)." TIMEOUT = 10 try: ipc_client = awsiot.greengrasscoreipc.connect() while True: request = PublishToTopicRequest() request.topic = topic publish_message = PublishMessage() publish_message.binary_message = BinaryMessage() publish_message.binary_message.message = bytes(message, "utf-8") request.publish_message = publish_message operation = ipc_client.new_publish_to_topic() operation.activate(request) future_response = operation.get_response() try: future_response.result(TIMEOUT) print('Successfully published to topic: ' + topic) except concurrent.futures.TimeoutError: print('Timeout occurred while publishing to topic: ' + topic, file=sys.stderr) except UnauthorizedError as e: print('Unauthorized error while publishing to topic: ' + topic, file=sys.stderr) raise e except Exception as e: print('Exception while publishing to topic: ' + topic, file=sys.stderr) raise e time.sleep(5) except InterruptedError: print('Publisher interrupted.') except Exception: print('Exception occurred when using IPC.', file=sys.stderr) traceback.print_exc() exit(1) ``` ### 發佈/訂閱訂閱者範例 (Python、IPC 用戶端 V1) 下列範例配方允許元件訂閱所有主題。 ------ #### [ JSON ] ``` { "RecipeFormatVersion": "2020-01-25", "ComponentName": "com.example.PubSubSubscriberPython", "ComponentVersion": "1.0.0", "ComponentDescription": "A component that subscribes to messages.", "ComponentPublisher": "Amazon", "ComponentConfiguration": { "DefaultConfiguration": { "accessControl": { "aws.greengrass.ipc.pubsub": { "com.example.PubSubSubscriberPython:pubsub:1": { "policyDescription": "Allows access to subscribe to all topics.", "operations": [ "aws.greengrass#SubscribeToTopic" ], "resources": [ "*" ] } } } } }, "Manifests": [ { "Platform": { "os": "linux" }, "Lifecycle": { "install": "python3 -m pip install --user awsiotsdk", "Run": "python3 -u {artifacts:path}/pubsub_subscriber.py" } }, { "Platform": { "os": "windows" }, "Lifecycle": { "install": "py -3 -m pip install --user awsiotsdk", "Run": "py -3 -u {artifacts:path}/pubsub_subscriber.py" } } ] } ``` ------ #### [ YAML ] ``` --- RecipeFormatVersion: '2020-01-25' ComponentName: com.example.PubSubSubscriberPython ComponentVersion: 1.0.0 ComponentDescription: A component that subscribes to messages. ComponentPublisher: Amazon ComponentConfiguration: DefaultConfiguration: accessControl: aws.greengrass.ipc.pubsub: com.example.PubSubSubscriberPython:pubsub:1: policyDescription: Allows access to subscribe to all topics. operations: - aws.greengrass#SubscribeToTopic resources: - "*" Manifests: - Platform: os: linux Lifecycle: install: python3 -m pip install --user awsiotsdk Run: python3 -u {artifacts:path}/pubsub_subscriber.py - Platform: os: windows Lifecycle: install: py -3 -m pip install --user awsiotsdk Run: py -3 -u {artifacts:path}/pubsub_subscriber.py ``` ------ 下列範例 Python 應用程式示範如何使用發佈/訂閱 IPC 服務來訂閱其他元件的訊息。 ``` import concurrent.futures import sys import time import traceback import awsiot.greengrasscoreipc import awsiot.greengrasscoreipc.client as client from awsiot.greengrasscoreipc.model import ( SubscribeToTopicRequest, SubscriptionResponseMessage, UnauthorizedError ) topic = "test/topic/python" TIMEOUT = 10 class StreamHandler(client.SubscribeToTopicStreamHandler): def __init__(self): super().__init__() def on_stream_event(self, event: SubscriptionResponseMessage) -> None: try: message = str(event.binary_message.message, "utf-8") print("Received new message: " + message) except: traceback.print_exc() def on_stream_error(self, error: Exception) -> bool: print("Received a stream error.", file=sys.stderr) traceback.print_exc() return False # Return True to close stream, False to keep stream open. def on_stream_closed(self) -> None: print('Subscribe to topic stream closed.') try: ipc_client = awsiot.greengrasscoreipc.connect() request = SubscribeToTopicRequest() request.topic = topic handler = StreamHandler() operation = ipc_client.new_subscribe_to_topic(handler) operation.activate(request) future_response = operation.get_response() try: future_response.result(TIMEOUT) print('Successfully subscribed to topic: ' + topic) except concurrent.futures.TimeoutError as e: print('Timeout occurred while subscribing to topic: ' + topic, file=sys.stderr) raise e except UnauthorizedError as e: print('Unauthorized error while subscribing to topic: ' + topic, file=sys.stderr) raise e except Exception as e: print('Exception while subscribing to topic: ' + topic, file=sys.stderr) raise e # Keep the main thread alive, or the process will exit. try: while True: time.sleep(10) except InterruptedError: print('Subscribe interrupted.') except Exception: print('Exception occurred when using IPC.', file=sys.stderr) traceback.print_exc() exit(1) ``` ### 發佈/訂閱發佈者範例 (C\$1\$1、IPC 用戶端 V1) 下列範例配方允許元件發佈至所有主題。 ------ #### [ JSON ] ``` { "RecipeFormatVersion": "2020-01-25", "ComponentName": "com.example.PubSubPublisherCpp", "ComponentVersion": "1.0.0", "ComponentDescription": "A component that publishes messages.", "ComponentPublisher": "Amazon", "ComponentConfiguration": { "DefaultConfiguration": { "accessControl": { "aws.greengrass.ipc.pubsub": { "com.example.PubSubPublisherCpp:pubsub:1": { "policyDescription": "Allows access to publish to all topics.", "operations": [ "aws.greengrass#PublishToTopic" ], "resources": [ "*" ] } } } } }, "Manifests": [ { "Lifecycle": { "Run": "{artifacts:path}/greengrassv2_pubsub_publisher" }, "Artifacts": [ { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.PubSubPublisherCpp/1.0.0/greengrassv2_pubsub_publisher", "Permission": { "Execute": "OWNER" } } ] } ] } ``` ------ #### [ YAML ] ``` --- RecipeFormatVersion: '2020-01-25' ComponentName: com.example.PubSubPublisherCpp ComponentVersion: 1.0.0 ComponentDescription: A component that publishes messages. ComponentPublisher: Amazon ComponentConfiguration: DefaultConfiguration: accessControl: aws.greengrass.ipc.pubsub: com.example.PubSubPublisherCpp:pubsub:1: policyDescription: Allows access to publish to all topics. operations: - aws.greengrass#PublishToTopic resources: - "*" Manifests: - Lifecycle: Run: "{artifacts:path}/greengrassv2_pubsub_publisher" Artifacts: - URI: s3://amzn-s3-demo-bucket/artifacts/com.example.PubSubPublisherCpp/1.0.0/greengrassv2_pubsub_publisher Permission: Execute: OWNER ``` ------ 下列 C\$1\$1 應用程式範例示範如何使用發佈/訂閱 IPC 服務將訊息發佈至其他元件。 ``` #include #include #include using namespace Aws::Crt; using namespace Aws::Greengrass; class IpcClientLifecycleHandler : public ConnectionLifecycleHandler { void OnConnectCallback() override { std::cout << "OnConnectCallback" << std::endl; } void OnDisconnectCallback(RpcError error) override { std::cout << "OnDisconnectCallback: " << error.StatusToString() << std::endl; exit(-1); } bool OnErrorCallback(RpcError error) override { std::cout << "OnErrorCallback: " << error.StatusToString() << std::endl; return true; } }; int main() { String message("Hello from the pub/sub publisher (C++)."); String topic("test/topic/cpp"); int timeout = 10; ApiHandle apiHandle(g_allocator); Io::EventLoopGroup eventLoopGroup(1); Io::DefaultHostResolver socketResolver(eventLoopGroup, 64, 30); Io::ClientBootstrap bootstrap(eventLoopGroup, socketResolver); IpcClientLifecycleHandler ipcLifecycleHandler; GreengrassCoreIpcClient ipcClient(bootstrap); auto connectionStatus = ipcClient.Connect(ipcLifecycleHandler).get(); if (!connectionStatus) { std::cerr << "Failed to establish IPC connection: " << connectionStatus.StatusToString() << std::endl; exit(-1); } while (true) { PublishToTopicRequest request; Vector messageData({message.begin(), message.end()}); BinaryMessage binaryMessage; binaryMessage.SetMessage(messageData); PublishMessage publishMessage; publishMessage.SetBinaryMessage(binaryMessage); request.SetTopic(topic); request.SetPublishMessage(publishMessage); auto operation = ipcClient.NewPublishToTopic(); auto activate = operation->Activate(request, nullptr); activate.wait(); auto responseFuture = operation->GetResult(); if (responseFuture.wait_for(std::chrono::seconds(timeout)) == std::future_status::timeout) { std::cerr << "Operation timed out while waiting for response from Greengrass Core." << std::endl; exit(-1); } auto response = responseFuture.get(); if (response) { std::cout << "Successfully published to topic: " << topic << std::endl; } else { // An error occurred. std::cout << "Failed to publish to topic: " << topic << std::endl; auto errorType = response.GetResultType(); if (errorType == OPERATION_ERROR) { auto *error = response.GetOperationError(); std::cout << "Operation error: " << error->GetMessage().value() << std::endl; } else { std::cout << "RPC error: " << response.GetRpcError() << std::endl; } exit(-1); } std::this_thread::sleep_for(std::chrono::seconds(5)); } return 0; } ``` ### 發佈/訂閱訂閱者範例 (C\$1\$1、IPC 用戶端 V1) 下列範例配方允許元件訂閱所有主題。 ------ #### [ JSON ] ``` { "RecipeFormatVersion": "2020-01-25", "ComponentName": "com.example.PubSubSubscriberCpp", "ComponentVersion": "1.0.0", "ComponentDescription": "A component that subscribes to messages.", "ComponentPublisher": "Amazon", "ComponentConfiguration": { "DefaultConfiguration": { "accessControl": { "aws.greengrass.ipc.pubsub": { "com.example.PubSubSubscriberCpp:pubsub:1": { "policyDescription": "Allows access to subscribe to all topics.", "operations": [ "aws.greengrass#SubscribeToTopic" ], "resources": [ "*" ] } } } } }, "Manifests": [ { "Lifecycle": { "Run": "{artifacts:path}/greengrassv2_pub_sub_subscriber" }, "Artifacts": [ { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.PubSubSubscriberCpp/1.0.0/greengrassv2_pub_sub_subscriber", "Permission": { "Execute": "OWNER" } } ] } ] } ``` ------ #### [ YAML ] ``` --- RecipeFormatVersion: '2020-01-25' ComponentName: com.example.PubSubSubscriberCpp ComponentVersion: 1.0.0 ComponentDescription: A component that subscribes to messages. ComponentPublisher: Amazon ComponentConfiguration: DefaultConfiguration: accessControl: aws.greengrass.ipc.pubsub: com.example.PubSubSubscriberCpp:pubsub:1: policyDescription: Allows access to subscribe to all topics. operations: - aws.greengrass#SubscribeToTopic resources: - "*" Manifests: - Lifecycle: Run: "{artifacts:path}/greengrassv2_pub_sub_subscriber" Artifacts: - URI: s3://amzn-s3-demo-bucket/artifacts/com.example.PubSubSubscriberCpp/1.0.0/greengrassv2_pub_sub_subscriber Permission: Execute: OWNER ``` ------ 下列 C\$1\$1 應用程式範例示範如何使用發佈/訂閱 IPC 服務來訂閱其他元件的訊息。 ``` #include #include #include using namespace Aws::Crt; using namespace Aws::Greengrass; class SubscribeResponseHandler : public SubscribeToTopicStreamHandler { public: virtual ~SubscribeResponseHandler() {} private: void OnStreamEvent(SubscriptionResponseMessage *response) override { auto jsonMessage = response->GetJsonMessage(); if (jsonMessage.has_value() && jsonMessage.value().GetMessage().has_value()) { auto messageString = jsonMessage.value().GetMessage().value().View().WriteReadable(); std::cout << "Received new message: " << messageString << std::endl; } else { auto binaryMessage = response->GetBinaryMessage(); if (binaryMessage.has_value() && binaryMessage.value().GetMessage().has_value()) { auto messageBytes = binaryMessage.value().GetMessage().value(); std::string messageString(messageBytes.begin(), messageBytes.end()); std::cout << "Received new message: " << messageString << std::endl; } } } bool OnStreamError(OperationError *error) override { std::cout << "Received an operation error: "; if (error->GetMessage().has_value()) { std::cout << error->GetMessage().value(); } std::cout << std::endl; return false; // Return true to close stream, false to keep stream open. } void OnStreamClosed() override { std::cout << "Subscribe to topic stream closed." << std::endl; } }; class IpcClientLifecycleHandler : public ConnectionLifecycleHandler { void OnConnectCallback() override { std::cout << "OnConnectCallback" << std::endl; } void OnDisconnectCallback(RpcError error) override { std::cout << "OnDisconnectCallback: " << error.StatusToString() << std::endl; exit(-1); } bool OnErrorCallback(RpcError error) override { std::cout << "OnErrorCallback: " << error.StatusToString() << std::endl; return true; } }; int main() { String topic("test/topic/cpp"); int timeout = 10; ApiHandle apiHandle(g_allocator); Io::EventLoopGroup eventLoopGroup(1); Io::DefaultHostResolver socketResolver(eventLoopGroup, 64, 30); Io::ClientBootstrap bootstrap(eventLoopGroup, socketResolver); IpcClientLifecycleHandler ipcLifecycleHandler; GreengrassCoreIpcClient ipcClient(bootstrap); auto connectionStatus = ipcClient.Connect(ipcLifecycleHandler).get(); if (!connectionStatus) { std::cerr << "Failed to establish IPC connection: " << connectionStatus.StatusToString() << std::endl; exit(-1); } SubscribeToTopicRequest request; request.SetTopic(topic); auto streamHandler = MakeShared(DefaultAllocator()); auto operation = ipcClient.NewSubscribeToTopic(streamHandler); auto activate = operation->Activate(request, nullptr); activate.wait(); auto responseFuture = operation->GetResult(); if (responseFuture.wait_for(std::chrono::seconds(timeout)) == std::future_status::timeout) { std::cerr << "Operation timed out while waiting for response from Greengrass Core." << std::endl; exit(-1); } auto response = responseFuture.get(); if (response) { std::cout << "Successfully subscribed to topic: " << topic << std::endl; } else { // An error occurred. std::cout << "Failed to subscribe to topic: " << topic << std::endl; auto errorType = response.GetResultType(); if (errorType == OPERATION_ERROR) { auto *error = response.GetOperationError(); std::cout << "Operation error: " << error->GetMessage().value() << std::endl; } else { std::cout << "RPC error: " << response.GetRpcError() << std::endl; } exit(-1); } // Keep the main thread alive, or the process will exit. while (true) { std::this_thread::sleep_for(std::chrono::seconds(10)); } operation->Close(); return 0; } ``` ### 發佈/訂閱發佈者範例 (Rust) 下列範例配方允許元件發佈至所有主題。 ``` { "RecipeFormatVersion": "2020-01-25", "ComponentName": "com.example.PubSubPublisherRust", "ComponentVersion": "1.0.0", "ComponentDescription": "A component that publishes messages.", "ComponentPublisher": "Amazon", "ComponentConfiguration": { "DefaultConfiguration": { "accessControl": { "aws.greengrass.ipc.pubsub": { "com.example.PubSubPublisherRust:pubsub:1": { "policyDescription": "Allows access to publish to all topics.", "operations": ["aws.greengrass#PublishToTopic"], "resources": ["*"] } } } } }, "Manifests": [ { "Platform": { "os": "linux", "runtime": "*" }, "Lifecycle": { "run": "{artifacts:path}/publish_to_topic" }, "Artifacts": [ { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.PubSubPublisherRust/1.0.0/publish_to_topic", "Permission": { "Execute": "OWNER" } } ] } ] } ``` 下列範例 Rust 應用程式示範如何使用發佈/訂閱 IPC 服務將訊息發佈至其他元件。 ``` use gg_sdk::Sdk; fn main() { let sdk = Sdk::init(); sdk.connect().expect("Failed to establish IPC connection"); let message = b"Hello, World"; let topic = "my/topic"; sdk.publish_to_topic_binary(topic, message) .expect("Failed to publish to topic"); println!("Successfully published to topic: {topic}"); } ``` ### 發佈/訂閱訂閱者範例 (Rust) 下列範例配方允許元件訂閱所有主題。 ``` { "RecipeFormatVersion": "2020-01-25", "ComponentName": "com.example.PubSubSubscriberRust", "ComponentVersion": "1.0.0", "ComponentDescription": "A component that subscribes to messages.", "ComponentPublisher": "Amazon", "ComponentConfiguration": { "DefaultConfiguration": { "accessControl": { "aws.greengrass.ipc.pubsub": { "com.example.PubSubSubscriberRust:pubsub:1": { "policyDescription": "Allows access to subscribe to all topics.", "operations": ["aws.greengrass#SubscribeToTopic"], "resources": ["*"] } } } } }, "Manifests": [ { "Platform": { "os": "linux", "runtime": "*" }, "Lifecycle": { "run": "{artifacts:path}/subscribe_to_topic" }, "Artifacts": [ { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.PubSubSubscriberRust/1.0.0/subscribe_to_topic", "Permission": { "Execute": "OWNER" } } ] } ] } ``` 下列範例 Rust 應用程式示範如何使用發佈/訂閱 IPC 服務來訂閱來自其他元件的訊息。 ``` use gg_sdk::{Sdk, SubscribeToTopicPayload}; use std::{thread, time::Duration}; fn main() { let sdk = Sdk::init(); sdk.connect().expect("Failed to establish IPC connection"); let topic = "my/topic"; let callback = |topic: &str, payload: SubscribeToTopicPayload| match payload { SubscribeToTopicPayload::Binary(message) => { let message = String::from_utf8_lossy(message); println!("Received new message on topic {topic}: {message}"); } SubscribeToTopicPayload::Json(_) => { println!("Received new message on topic {topic}: (JSON message)"); } }; let _sub = sdk .subscribe_to_topic(topic, &callback) .expect("Failed to subscribe to topic"); println!("Successfully subscribed to topic: {topic}"); // Keep the main thread alive, or the process will exit. loop { thread::sleep(Duration::from_secs(10)); } } ``` ### 發佈/訂閱發佈者範例 (C) 下列範例配方允許元件發佈至所有主題。 ``` { "RecipeFormatVersion": "2020-01-25", "ComponentName": "com.example.PubSubPublisherC", "ComponentVersion": "1.0.0", "ComponentDescription": "A component that publishes messages.", "ComponentPublisher": "Amazon", "ComponentConfiguration": { "DefaultConfiguration": { "accessControl": { "aws.greengrass.ipc.pubsub": { "com.example.PubSubPublisherC:pubsub:1": { "policyDescription": "Allows access to publish to all topics.", "operations": ["aws.greengrass#PublishToTopic"], "resources": ["*"] } } } } }, "Manifests": [ { "Platform": { "os": "linux", "runtime": "*" }, "Lifecycle": { "run": "{artifacts:path}/sample_publish_to_topic" }, "Artifacts": [ { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.PubSubPublisherC/1.0.0/sample_publish_to_topic", "Permission": { "Execute": "OWNER" } } ] } ] } ``` 下列範例 C 應用程式示範如何使用發佈/訂閱 IPC 服務將訊息發佈至其他元件。 ``` #include #include #include #include #include int main(void) { gg_sdk_init(); GgError err = ggipc_connect(); if (err != GG_ERR_OK) { fprintf(stderr, "Failed to establish IPC connection.\n"); exit(-1); } GgBuffer message = GG_STR("Hello, World"); GgBuffer topic = GG_STR("my/topic"); err = ggipc_publish_to_topic_binary(topic, message); if (err != GG_ERR_OK) { fprintf( stderr, "Failed to publish to topic: %.*s\n", (int) topic.len, topic.data ); exit(-1); } printf( "Successfully published to topic: %.*s\n", (int) topic.len, topic.data ); } ``` ### 發佈/訂閱訂閱者範例 (C) 下列範例配方允許元件訂閱所有主題。 ``` { "RecipeFormatVersion": "2020-01-25", "ComponentName": "com.example.PubSubSubscriberC", "ComponentVersion": "1.0.0", "ComponentDescription": "A component that subscribes to messages.", "ComponentPublisher": "Amazon", "ComponentConfiguration": { "DefaultConfiguration": { "accessControl": { "aws.greengrass.ipc.pubsub": { "com.example.PubSubSubscriberC:pubsub:1": { "policyDescription": "Allows access to subscribe to all topics.", "operations": ["aws.greengrass#SubscribeToTopic"], "resources": ["*"] } } } } }, "Manifests": [ { "Platform": { "os": "linux", "runtime": "*" }, "Lifecycle": { "run": "{artifacts:path}/sample_subscribe_to_topic" }, "Artifacts": [ { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.PubSubSubscriberC/1.0.0/sample_subscribe_to_topic", "Permission": { "Execute": "OWNER" } } ] } ] } ``` 下列範例 C 應用程式示範如何使用發佈/訂閱 IPC 服務來訂閱來自其他元件的訊息。 ``` #include #include #include #include #include #include #include #include #include static void on_subscription_response( void *ctx, GgBuffer topic, GgObject payload, GgIpcSubscriptionHandle handle ) { (void) ctx; (void) handle; if (gg_obj_type(payload) == GG_TYPE_BUF) { GgBuffer message = gg_obj_into_buf(payload); printf( "Received new message on topic %.*s: %.*s\n", (int) topic.len, topic.data, (int) message.len, message.data ); } else { assert(gg_obj_type(payload) == GG_TYPE_MAP); printf( "Received new message on topic %.*s: (JSON message)\n", (int) topic.len, topic.data ); } } int main(void) { gg_sdk_init(); GgError err = ggipc_connect(); if (err != GG_ERR_OK) { fprintf(stderr, "Failed to establish IPC connection.\n"); exit(-1); } GgBuffer topic = GG_STR("my/topic"); GgIpcSubscriptionHandle handle; err = ggipc_subscribe_to_topic( topic, on_subscription_response, NULL, &handle ); if (err != GG_ERR_OK) { fprintf( stderr, "Failed to subscribe to topic: %.*s\n", (int) topic.len, topic.data ); exit(-1); } printf( "Successfully subscribed to topic: %.*s\n", (int) topic.len, topic.data ); // Keep the main thread alive, or the process will exit. while (1) { sleep(10); } // To stop subscribing, close the stream. ggipc_close_subscription(handle); } ``` ### 發佈/訂閱發佈者範例 (C\$1\$1、元件 SDK) 下列範例配方允許元件發佈至所有主題。 ``` { "RecipeFormatVersion": "2020-01-25", "ComponentName": "com.example.PubSubPublisherCpp", "ComponentVersion": "1.0.0", "ComponentDescription": "A component that publishes messages.", "ComponentPublisher": "Amazon", "ComponentConfiguration": { "DefaultConfiguration": { "accessControl": { "aws.greengrass.ipc.pubsub": { "com.example.PubSubPublisherCpp:pubsub:1": { "policyDescription": "Allows access to publish to all topics.", "operations": ["aws.greengrass#PublishToTopic"], "resources": ["*"] } } } } }, "Manifests": [ { "Platform": { "os": "linux", "runtime": "*" }, "Lifecycle": { "run": "{artifacts:path}/sample_cpp_publish_to_topic" }, "Artifacts": [ { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.PubSubPublisherCpp/1.0.0/sample_cpp_publish_to_topic", "Permission": { "Execute": "OWNER" } } ] } ] } ``` 下列 C\$1\$1 應用程式範例示範如何使用發佈/訂閱 IPC 服務將訊息發佈至其他元件。 ``` #include #include int main() { auto &client = gg::ipc::Client::get(); auto error = client.connect(); if (error) { std::cerr << "Failed to establish IPC connection.\n"; exit(-1); } std::string_view message = "Hello, World"; std::string_view topic = "my/topic"; error = client.publish_to_topic(topic, message); if (error) { std::cerr << "Failed to publish to topic: " << topic << "\n"; exit(-1); } std::cout << "Successfully published to topic: " << topic << "\n"; } ``` ### 發佈/訂閱訂閱者範例 (C\$1\$1、元件 SDK) 下列範例配方允許元件訂閱所有主題。 ``` { "RecipeFormatVersion": "2020-01-25", "ComponentName": "com.example.PubSubSubscriberCpp", "ComponentVersion": "1.0.0", "ComponentDescription": "A component that subscribes to messages.", "ComponentPublisher": "Amazon", "ComponentConfiguration": { "DefaultConfiguration": { "accessControl": { "aws.greengrass.ipc.pubsub": { "com.example.PubSubSubscriberCpp:pubsub:1": { "policyDescription": "Allows access to subscribe to all topics.", "operations": ["aws.greengrass#SubscribeToTopic"], "resources": ["*"] } } } } }, "Manifests": [ { "Platform": { "os": "linux", "runtime": "*" }, "Lifecycle": { "run": "{artifacts:path}/sample_cpp_subscribe_to_topic" }, "Artifacts": [ { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.PubSubSubscriberCpp/1.0.0/sample_cpp_subscribe_to_topic", "Permission": { "Execute": "OWNER" } } ] } ] } ``` 下列 C\$1\$1 應用程式範例示範如何使用發佈/訂閱 IPC 服務來訂閱來自其他元件的訊息。 ``` #include #include #include #include #include class ResponseHandler : public gg::ipc::LocalTopicCallback { void operator()( std::string_view topic, gg::Object payload, gg::ipc::Subscription &handle ) override { (void) handle; if (payload.index() == GG_TYPE_BUF) { std::cout << "Received new message on topic " << topic << ": " << get(payload) << "\n"; } else { assert(payload.index() == GG_TYPE_MAP); std::cout << "Received new message on topic " << topic << ": (JSON message)\n"; } } }; int main() { auto &client = gg::ipc::Client::get(); auto error = client.connect(); if (error) { std::cerr << "Failed to establish IPC connection.\n"; exit(-1); } std::string_view topic = "my/topic"; static ResponseHandler handler; error = client.subscribe_to_topic(topic, handler); if (error) { std::cerr << "Failed to subscribe to topic: " << topic << "\n"; exit(-1); } std::cout << "Successfully subscribed to topic: " << topic << "\n"; // Keep the main thread alive, or the process will exit. while (1) { sleep(10); } } ``` # 發佈/訂閱 AWS IoT Core MQTT 訊息 AWS IoT Core MQTT 訊息 IPC 服務可讓您在之間傳送和接收 MQTT 訊息 AWS IoT Core。元件可以將訊息發佈到 AWS IoT Core 並訂閱主題，以對來自其他來源的 MQTT 訊息採取行動。如需 MQTT AWS IoT Core 實作的詳細資訊，請參閱《 *AWS IoT Core 開發人員指南*》中的 [MQTT](https://docs.aws.amazon.com/iot/latest/developerguide/mqtt.html)。 **注意** 此 MQTT 訊息 IPC 服務可讓您與交換訊息 AWS IoT Core。如需如何在元件之間交換訊息的詳細資訊，請參閱 [發佈/訂閱本機訊息](ipc-publish-subscribe.md)。 **Topics** + [最低 SDK 版本](#ipc-iot-core-mqtt-sdk-versions) + [Authorization](#ipc-iot-core-mqtt-authorization) + [PublishToIoTCore](#ipc-operation-publishtoiotcore) + [SubscribeToIoTCore](#ipc-operation-subscribetoiotcore) + [範例](#ipc-iot-core-mqtt-examples) ## 最低 SDK 版本下表列出 AWS IoT Device SDK 您必須用來發佈和訂閱 MQTT 訊息的最小版本 AWS IoT Core。 | SDK | 最低版本 | | --- | --- | | [AWS IoT Device SDK for Java v2](https://github.com/aws/aws-iot-device-sdk-java-v2) | v1.2.10 | | [AWS IoT Device SDK for Python v2](https://github.com/aws/aws-iot-device-sdk-python-v2) | 1.5.3 版 | | [AWS IoT Device SDK 適用於 C\$1\$1 v2](https://github.com/aws/aws-iot-device-sdk-cpp-v2) | 1.17.0 版 | | [AWS IoT Device SDK for JavaScript v2](https://github.com/aws/aws-iot-device-sdk-js-v2) | v1.12.0 | ## Authorization 若要在自訂元件中使用 AWS IoT Core MQTT 訊息，您必須定義授權政策，以允許您的元件傳送和接收主題的訊息。如需定義授權政策的資訊，請參閱 [授權元件執行 IPC 操作](interprocess-communication.md#ipc-authorization-policies)。 AWS IoT Core MQTT 訊息的授權政策具有下列屬性。 **IPC 服務識別符：** `aws.greengrass.ipc.mqttproxy` | 作業 | Description | Resources | | --- | --- | --- | | `aws.greengrass#PublishToIoTCore` | 允許元件在您指定的 MQTT 主題 AWS IoT Core 上發佈訊息至。 | 主題字串，例如 `test/topic`或 `*` 允許存取所有主題。您可以使用 MQTT 主題萬用字元 (`#` 和 `+`) 來比對多個資源。 | | `aws.greengrass#SubscribeToIoTCore` | 允許元件在您指定的主題 AWS IoT Core 上訂閱來自的訊息。 | 主題字串，例如 `test/topic`或 `*` 允許存取所有主題。您可以使用 MQTT 主題萬用字元 (`#` 和 `+`) 來比對多個資源。 | | `*` | 允許元件發佈和訂閱您指定主題的 AWS IoT Core MQTT 訊息。 | 主題字串，例如 `test/topic`或 `*` 允許存取所有主題。您可以使用 MQTT 主題萬用字元 (`#` 和 `+`) 來比對多個資源。 | ### MQTT 授權政策中的 AWS IoT Core MQTT 萬用字元您可以在 MQTT IPC 授權政策中使用 AWS IoT Core MQTT 萬用字元。元件可以發佈和訂閱與您在授權政策中允許的主題篩選條件相符的主題。例如，如果元件的授權政策授予的存取權`test/topic/#`，則該元件可以訂閱 `test/topic/#`，而且可以發佈和訂閱 `test/topic/filter`。 ### AWS IoT Core MQTT 授權政策中的配方變數如果您使用 [Greengrass 核](greengrass-nucleus-component.md)的 v2.6.0 或更新版本，您可以在授權政策中使用`{iot:thingName}`配方變數。此功能可讓您為一組核心裝置設定單一授權政策，其中每個核心裝置只能存取包含自己的名稱的主題。例如，您可以允許元件存取下列主題資源。 ``` devices/{iot:thingName}/messages ``` 如需詳細資訊，請參閱[配方變數](component-recipe-reference.md#recipe-variables)及[在合併更新中使用配方變數](update-component-configurations.md#merge-configuration-update-recipe-variables)。 ### 授權政策範例您可以參考下列授權政策範例，協助您設定元件的授權政策。 **Example 具有無限制存取的授權政策範例** 下列範例授權政策允許元件發佈和訂閱所有主題。 ``` { "accessControl": { "aws.greengrass.ipc.mqttproxy": { "com.example.MyIoTCorePubSubComponent:mqttproxy:1": { "policyDescription": "Allows access to publish/subscribe to all topics.", "operations": [ "aws.greengrass#PublishToIoTCore", "aws.greengrass#SubscribeToIoTCore" ], "resources": [ "*" ] } } } } ``` ``` --- accessControl: aws.greengrass.ipc.mqttproxy: com.example.MyIoTCorePubSubComponent:mqttproxy:1: policyDescription: Allows access to publish/subscribe to all topics. operations: - aws.greengrass#PublishToIoTCore - aws.greengrass#SubscribeToIoTCore resources: - "*" ``` **Example 具有有限存取權的授權政策範例** 下列範例授權政策允許元件發佈和訂閱兩個名為 `factory/1/events`和的主題`factory/1/actions`。 ``` { "accessControl": { "aws.greengrass.ipc.mqttproxy": { "com.example.MyIoTCorePubSubComponent:mqttproxy:1": { "policyDescription": "Allows access to publish/subscribe to factory 1 topics.", "operations": [ "aws.greengrass#PublishToIoTCore", "aws.greengrass#SubscribeToIoTCore" ], "resources": [ "factory/1/actions", "factory/1/events" ] } } } } ``` ``` --- accessControl: aws.greengrass.ipc.mqttproxy: "com.example.MyIoTCorePubSubComponent:mqttproxy:1": policyDescription: Allows access to publish/subscribe to factory 1 topics. operations: - aws.greengrass#PublishToIoTCore - aws.greengrass#SubscribeToIoTCore resources: - factory/1/actions - factory/1/events ``` **Example 一組核心裝置的範例授權政策** 此範例使用適用於 [Greengrass 核元件](greengrass-nucleus-component.md) v2.6.0 和更新版本的功能。Greengrass 核 v2.6.0 在元件組態中新增了對大多數[配方變數](component-recipe-reference.md#recipe-variables)的支援`{iot:thingName}`，例如。下列範例授權政策允許元件發佈和訂閱主題，其中包含執行元件的核心裝置名稱。 ``` { "accessControl": { "aws.greengrass.ipc.mqttproxy": { "com.example.MyIoTCorePubSubComponent:mqttproxy:1": { "policyDescription": "Allows access to publish/subscribe to all topics.", "operations": [ "aws.greengrass#PublishToIoTCore", "aws.greengrass#SubscribeToIoTCore" ], "resources": [ "factory/1/devices/{iot:thingName}/controls" ] } } } } ``` ``` --- accessControl: aws.greengrass.ipc.mqttproxy: "com.example.MyIoTCorePubSubComponent:mqttproxy:1": policyDescription: Allows access to publish/subscribe to all topics. operations: - aws.greengrass#PublishToIoTCore - aws.greengrass#SubscribeToIoTCore resources: - factory/1/devices/{iot:thingName}/controls ``` ## PublishToIoTCore 在主題 AWS IoT Core 上將 MQTT 訊息發佈至。當您發佈 MQTT 訊息到時 AWS IoT Core，每秒有 100 筆交易的配額。如果您超過此配額，訊息會排入佇列，以便在 Greengrass 裝置上處理。也有每秒 512 Kb 的資料配額，以及每秒 20，000 個發佈的全帳戶配額（部分為 2，000 AWS 區域)。如需中 MQTT 訊息中介裝置限制的詳細資訊 AWS IoT Core，請參閱[AWS IoT Core 訊息中介裝置和通訊協定限制和配額](https://docs.aws.amazon.com/general/latest/gr/iot-core.html#message-broker-limits)。如果您超過這些配額，Greengrass 裝置會將訊息發佈限制為 AWS IoT Core。訊息會存放在記憶體中的多工緩衝處理程式中。根據預設，配置給多工緩衝處理站的記憶體為 2.5 Mb。如果多工緩衝處理程式填滿，則會拒絕新訊息。您可以增加多工緩衝處理的大小。如需詳細資訊，請參閱 [Greengrass 核](greengrass-nucleus-component.md) 文件中的 [Configuration](greengrass-nucleus-component.md#greengrass-nucleus-component-configuration)。為了避免填入多工緩衝處理程式並需要增加配置的記憶體，請將發佈請求限制為每秒不超過 100 個請求。當您的應用程式需要以更高的速率或更大的訊息傳送訊息時，請考慮使用 [串流管理員](stream-manager-component.md) 傳送訊息至 Kinesis Data Streams。串流管理員元件旨在將大量資料傳輸至 AWS 雲端。如需詳細資訊，請參閱[管理 Greengrass 核心裝置上的資料串流](manage-data-streams.md)。 ### 請求此操作的請求具有下列參數： `topicName` (Python：`topic_name`) 要發佈訊息的主題。 `qos` 要使用的 MQTT QoS。此列舉 `QOS`具有下列值： + `AT_MOST_ONCE` – QoS 0。MQTT 訊息最多傳遞一次。 + `AT_LEAST_ONCE` – QoS 1。MQTT 訊息至少傳遞一次。 `payload` （選用）作為 Blob 的訊息承載。使用 MQTT 5 [Greengrass 核](greengrass-nucleus-component.md)時，下列功能適用於 2.10.0 版和更新版本。當您使用 MQTT 3.1.1 時，會忽略這些功能。下表列出您必須用來存取這些功能的 AWS IoT 裝置 SDK 最低版本。 | SDK | 最低版本 | | --- | --- | | [適用於 Python 的 AWS IoT Device SDK](https://github.com/aws/aws-iot-device-sdk-python-v2) v2 | 1.15.0 版 | | [適用於 JAVA 的 AWS IoT Device SDK](https://github.com/aws/aws-iot-device-sdk-java-v2) v2 | 1.13.0 版 | | [適用於 C\$1\$1 的 AWS IoT Device SDK](https://github.com/aws/aws-iot-device-sdk-cpp-v2) v2 | 1.24.0 版 | | [適用於 JavaScript 的 AWS IoT Device SDK ](https://github.com/aws/aws-iot-device-sdk-js-v2) v2 | 1.13.0 版 | `payloadFormat` （選用）訊息承載的格式。如果您未設定 `payloadFormat`，則會假設類型為 `BYTES`。列舉具有下列值： + `BYTES` – 承載的內容是二進位 Blob。 + `UTF8` – 承載的內容是 UTF8 字元字串。 `retain` （選用）指出是否要在發佈`true`時將 MQTT 保留選項設定為。 `userProperties` （選用）要傳送的應用程式特定`UserProperty`物件清單。`UserProperty` 物件的定義如下： ``` UserProperty: key: string value: string ``` `messageExpiryIntervalSeconds` （選用）訊息過期並由伺服器刪除之前的秒數。如果未設定此值，則訊息不會過期。 `correlationData` （選用）新增至請求的資訊，可用於將請求與回應建立關聯。 `responseTopic` （選用）應該用於回應訊息的主題。 `contentType` （選用）訊息內容類型的應用程式特定識別符。 ### 回應此操作不會在其回應中提供任何資訊。 ### 範例下列範例示範如何在自訂元件程式碼中呼叫此操作。 ------ #### [ Java (IPC client V2) ] **Example 範例：發佈訊息** ``` package com.aws.greengrass.docs.samples.ipc; import software.amazon.awssdk.aws.greengrass.GreengrassCoreIPCClientV2; import software.amazon.awssdk.aws.greengrass.model.PublishToIoTCoreRequest; import software.amazon.awssdk.aws.greengrass.model.QOS; import java.nio.charset.StandardCharsets; public class PublishToIoTCore { public static void main(String[] args) { String topic = args[0]; String message = args[1]; QOS qos = QOS.get(args[2]); try (GreengrassCoreIPCClientV2 ipcClientV2 = GreengrassCoreIPCClientV2.builder().build()) { ipcClientV2.publishToIoTCore(new PublishToIoTCoreRequest() .withTopicName(topic) .withPayload(message.getBytes(StandardCharsets.UTF_8)) .withQos(qos)); System.out.println("Successfully published to topic: " + topic); } catch (Exception e) { System.err.println("Exception occurred."); e.printStackTrace(); System.exit(1); } } } ``` ------ #### [ Python (IPC client V2) ] **Example 範例：發佈訊息** 此範例假設您使用 AWS IoT Device SDK 適用於 Python v2 的 1.5.4 版或更新版本。 ``` import awsiot.greengrasscoreipc.clientv2 as clientV2 topic = 'my/topic' qos = '1' payload = 'Hello, World' ipc_client = clientV2.GreengrassCoreIPCClientV2() resp = ipc_client.publish_to_iot_core(topic_name=topic, qos=qos, payload=payload) ipc_client.close() ``` ------ #### [ Java (IPC client V1) ] **Example 範例：發佈訊息** 此範例使用 `IPCUtils`類別來建立與 AWS IoT Greengrass Core IPC 服務的連線。如需詳細資訊，請參閱[連線至 AWS IoT Greengrass Core IPC 服務](interprocess-communication.md#ipc-service-connect)。 ``` package com.aws.greengrass.docs.samples.ipc; import com.aws.greengrass.docs.samples.ipc.util.IPCUtils; import software.amazon.awssdk.aws.greengrass.GreengrassCoreIPCClient; import software.amazon.awssdk.aws.greengrass.PublishToIoTCoreResponseHandler; import software.amazon.awssdk.aws.greengrass.model.PublishToIoTCoreRequest; import software.amazon.awssdk.aws.greengrass.model.PublishToIoTCoreResponse; import software.amazon.awssdk.aws.greengrass.model.QOS; import software.amazon.awssdk.aws.greengrass.model.UnauthorizedError; import software.amazon.awssdk.eventstreamrpc.EventStreamRPCConnection; import java.nio.charset.StandardCharsets; import java.util.Optional; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; public class PublishToIoTCore { public static final int TIMEOUT_SECONDS = 10; public static void main(String[] args) { String topic = args[0]; String message = args[1]; QOS qos = QOS.get(args[2]); try (EventStreamRPCConnection eventStreamRPCConnection = IPCUtils.getEventStreamRpcConnection()) { GreengrassCoreIPCClient ipcClient = new GreengrassCoreIPCClient(eventStreamRPCConnection); PublishToIoTCoreResponseHandler responseHandler = PublishToIoTCore.publishBinaryMessageToTopic(ipcClient, topic, message, qos); CompletableFuture futureResponse = responseHandler.getResponse(); try { futureResponse.get(TIMEOUT_SECONDS, TimeUnit.SECONDS); System.out.println("Successfully published to topic: " + topic); } catch (TimeoutException e) { System.err.println("Timeout occurred while publishing to topic: " + topic); } catch (ExecutionException e) { if (e.getCause() instanceof UnauthorizedError) { System.err.println("Unauthorized error while publishing to topic: " + topic); } else { throw e; } } } catch (InterruptedException e) { System.out.println("IPC interrupted."); } catch (ExecutionException e) { System.err.println("Exception occurred when using IPC."); e.printStackTrace(); System.exit(1); } } public static PublishToIoTCoreResponseHandler publishBinaryMessageToTopic(GreengrassCoreIPCClient greengrassCoreIPCClient, String topic, String message, QOS qos) { PublishToIoTCoreRequest publishToIoTCoreRequest = new PublishToIoTCoreRequest(); publishToIoTCoreRequest.setTopicName(topic); publishToIoTCoreRequest.setPayload(message.getBytes(StandardCharsets.UTF_8)); publishToIoTCoreRequest.setQos(qos); return greengrassCoreIPCClient.publishToIoTCore(publishToIoTCoreRequest, Optional.empty()); } } ``` ------ #### [ Python (IPC client V1) ] **Example 範例：發佈訊息** 此範例假設您使用 AWS IoT Device SDK 適用於 Python v2 的 1.5.4 版或更新版本。 ``` import awsiot.greengrasscoreipc import awsiot.greengrasscoreipc.client as client from awsiot.greengrasscoreipc.model import ( QOS, PublishToIoTCoreRequest ) TIMEOUT = 10 ipc_client = awsiot.greengrasscoreipc.connect() topic = "my/topic" message = "Hello, World" qos = QOS.AT_LEAST_ONCE request = PublishToIoTCoreRequest() request.topic_name = topic request.payload = bytes(message, "utf-8") request.qos = qos operation = ipc_client.new_publish_to_iot_core() operation.activate(request) future_response = operation.get_response() future_response.result(TIMEOUT) ``` ------ #### [ C\$1\$1 (IPC client V1) ] **Example 範例：發佈訊息** ``` #include #include #include using namespace Aws::Crt; using namespace Aws::Greengrass; class IpcClientLifecycleHandler : public ConnectionLifecycleHandler { void OnConnectCallback() override { // Handle connection to IPC service. } void OnDisconnectCallback(RpcError error) override { // Handle disconnection from IPC service. } bool OnErrorCallback(RpcError error) override { // Handle IPC service connection error. return true; } }; int main() { ApiHandle apiHandle(g_allocator); Io::EventLoopGroup eventLoopGroup(1); Io::DefaultHostResolver socketResolver(eventLoopGroup, 64, 30); Io::ClientBootstrap bootstrap(eventLoopGroup, socketResolver); IpcClientLifecycleHandler ipcLifecycleHandler; GreengrassCoreIpcClient ipcClient(bootstrap); auto connectionStatus = ipcClient.Connect(ipcLifecycleHandler).get(); if (!connectionStatus) { std::cerr << "Failed to establish IPC connection: " << connectionStatus.StatusToString() << std::endl; exit(-1); } String message("Hello, World!"); String topic("my/topic"); QOS qos = QOS_AT_MOST_ONCE; int timeout = 10; PublishToIoTCoreRequest request; Vector messageData({message.begin(), message.end()}); request.SetTopicName(topic); request.SetPayload(messageData); request.SetQos(qos); auto operation = ipcClient.NewPublishToIoTCore(); auto activate = operation->Activate(request, nullptr); activate.wait(); auto responseFuture = operation->GetResult(); if (responseFuture.wait_for(std::chrono::seconds(timeout)) == std::future_status::timeout) { std::cerr << "Operation timed out while waiting for response from Greengrass Core." << std::endl; exit(-1); } auto response = responseFuture.get(); if (!response) { // Handle error. auto errorType = response.GetResultType(); if (errorType == OPERATION_ERROR) { auto *error = response.GetOperationError(); (void)error; // Handle operation error. } else { // Handle RPC error. } } return 0; } ``` ------ #### [ JavaScript ] **Example 範例：發佈訊息** ``` import * as greengrasscoreipc from "aws-iot-device-sdk-v2/dist/greengrasscoreipc"; import {QOS, PublishToIoTCoreRequest} from "aws-iot-device-sdk-v2/dist/greengrasscoreipc/model"; class PublishToIoTCore { private ipcClient: greengrasscoreipc.Client private readonly topic: string; constructor() { // define your own constructor, e.g. this.topic = ""; this.publishToIoTCore().then(r => console.log("Started workflow")); } private async publishToIoTCore() { try { const request: PublishToIoTCoreRequest = { topicName: this.topic, qos: QOS.AT_LEAST_ONCE, // you can change this depending on your use case } this.ipcClient = await getIpcClient(); await this.ipcClient.publishToIoTCore(request); } catch (e) { // parse the error depending on your use cases throw e } } } export async function getIpcClient(){ try { const ipcClient = greengrasscoreipc.createClient(); await ipcClient.connect() .catch(error => { // parse the error depending on your use cases throw error; }); return ipcClient } catch (err) { // parse the error depending on your use cases throw err } } // starting point const publishToIoTCore = new PublishToIoTCore(); ``` ------ #### [ Rust ] **Example 範例：發佈訊息** ``` use gg_sdk::{Qos, Sdk}; fn main() { let sdk = Sdk::init(); sdk.connect().expect("Failed to establish IPC connection"); let message = b"Hello, World"; let topic = "my/topic"; let qos = Qos::AtLeastOnce; sdk.publish_to_iot_core(topic, message, qos) .expect("Failed to publish to topic"); println!("Successfully published to topic: {topic}"); } ``` ------ #### [ C ] **Example 範例：發佈訊息** ``` #include #include #include #include #include int main(void) { gg_sdk_init(); GgError err = ggipc_connect(); if (err != GG_ERR_OK) { fprintf(stderr, "Failed to establish IPC connection.\n"); exit(-1); } GgBuffer message = GG_STR("Hello, World"); GgBuffer topic = GG_STR("my/topic"); uint8_t qos = 1; err = ggipc_publish_to_iot_core(topic, message, qos); if (err != GG_ERR_OK) { fprintf( stderr, "Failed to publish to topic: %.*s\n", (int) topic.len, topic.data ); exit(-1); } printf( "Successfully published to topic: %.*s\n", (int) topic.len, topic.data ); } ``` ------ #### [ C\$1\$1 (Component SDK) ] **Example 範例：發佈訊息** ``` #include #include int main() { auto &client = gg::ipc::Client::get(); auto error = client.connect(); if (error) { std::cerr << "Failed to establish IPC connection.\n"; exit(-1); } std::string_view message = "Hello, World"; std::string_view topic = "my/topic"; uint8_t qos = 1; error = client.publish_to_iot_core(topic, message, qos); if (error) { std::cerr << "Failed to publish to topic: " << topic << "\n"; exit(-1); } std::cout << "Successfully published to topic: " << topic << "\n"; } ``` ------ ## SubscribeToIoTCore 在主題或主題篩選條件 AWS IoT Core 上，從訂閱 MQTT 訊息。 AWS IoT Greengrass 核心軟體會在元件生命週期結束時移除訂閱。此操作是您訂閱事件訊息串流的訂閱操作。若要使用此操作，請使用處理事件訊息、錯誤和串流關閉的函數來定義串流回應處理常式。如需詳細資訊，請參閱[訂閱 IPC 事件串流](interprocess-communication.md#ipc-subscribe-operations)。 **事件訊息類型：** `IoTCoreMessage` ### 請求此操作的請求具有下列參數： `topicName` (Python：`topic_name`) 要訂閱的主題。您可以使用 MQTT 主題萬用字元 (`#` 和 `+`) 來訂閱多個主題。 `qos` 要使用的 MQTT QoS。此列舉 `QOS`具有下列值： + `AT_MOST_ONCE` – QoS 0。MQTT 訊息最多傳遞一次。 + `AT_LEAST_ONCE` – QoS 1。MQTT 訊息至少傳遞一次。 ### 回應此操作的回應包含下列資訊： `messages` MQTT 訊息的串流。此物件 `IoTCoreMessage`包含下列資訊： `message` MQTT 訊息。此物件 `MQTTMessage`包含下列資訊： `topicName` (Python：`topic_name`) 發佈訊息的主題。 `payload` （選用）作為 Blob 的訊息承載。使用 MQTT 5 [Greengrass 核](greengrass-nucleus-component.md)時，下列功能適用於 2.10.0 版和更新版本。當您使用 MQTT 3.1.1 時，會忽略這些功能。下表列出您必須用來存取這些功能的 AWS IoT 裝置 SDK 最低版本。 [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/zh_tw/greengrass/v2/developerguide/ipc-iot-core-mqtt.html) `payloadFormat` （選用）訊息承載的格式。如果您未設定 `payloadFormat`，則會假設類型為 `BYTES`。列舉具有下列值： + `BYTES` – 承載的內容是二進位 Blob。 + `UTF8` – 承載的內容是 UTF8 字元字串。 `retain` （選用）指出是否要在發佈`true`時將 MQTT 保留選項設定為。 `userProperties` （選用）要傳送的應用程式特定`UserProperty`物件清單。`UserProperty` 物件的定義如下： ``` UserProperty: key: string value: string ``` `messageExpiryIntervalSeconds` （選用）訊息過期並由伺服器刪除之前的秒數。如果未設定此值，則訊息不會過期。 `correlationData` （選用）新增至請求的資訊，可用於將請求與回應建立關聯。 `responseTopic` （選用）應該用於回應訊息的主題。 `contentType` （選用）訊息內容類型的應用程式特定識別符。 ### 範例下列範例示範如何在自訂元件程式碼中呼叫此操作。 ------ #### [ Java (IPC client V2) ] **Example 範例：訂閱訊息** ``` package com.aws.greengrass.docs.samples.ipc; import software.amazon.awssdk.aws.greengrass.GreengrassCoreIPCClientV2; import software.amazon.awssdk.aws.greengrass.SubscribeToIoTCoreResponseHandler; import software.amazon.awssdk.aws.greengrass.model.QOS; import software.amazon.awssdk.aws.greengrass.model.IoTCoreMessage; import software.amazon.awssdk.aws.greengrass.model.SubscribeToIoTCoreRequest; import software.amazon.awssdk.aws.greengrass.model.SubscribeToIoTCoreResponse; import java.nio.charset.StandardCharsets; import java.util.Optional; import java.util.function.Consumer; import java.util.function.Function; public class SubscribeToIoTCore { public static void main(String[] args) { String topic = args[0]; QOS qos = QOS.get(args[1]); Consumer onStreamEvent = ioTCoreMessage -> System.out.printf("Received new message on topic %s: %s%n", ioTCoreMessage.getMessage().getTopicName(), new String(ioTCoreMessage.getMessage().getPayload(), StandardCharsets.UTF_8)); Optional> onStreamError = Optional.of(e -> { System.err.println("Received a stream error."); e.printStackTrace(); return false; }); Optional onStreamClosed = Optional.of(() -> System.out.println("Subscribe to IoT Core stream closed.")); try (GreengrassCoreIPCClientV2 ipcClientV2 = GreengrassCoreIPCClientV2.builder().build()) { SubscribeToIoTCoreRequest request = new SubscribeToIoTCoreRequest() .withTopicName(topic) .withQos(qos); GreengrassCoreIPCClientV2.StreamingResponse streamingResponse = ipcClientV2.subscribeToIoTCore(request, onStreamEvent, onStreamError, onStreamClosed); streamingResponse.getResponse(); System.out.println("Successfully subscribed to topic: " + topic); // Keep the main thread alive, or the process will exit. while (true) { Thread.sleep(10000); } // To stop subscribing, close the stream. streamingResponse.getHandler().closeStream(); } catch (InterruptedException e) { System.out.println("Subscribe interrupted."); } catch (Exception e) { System.err.println("Exception occurred."); e.printStackTrace(); System.exit(1); } } } ``` ------ #### [ Python (IPC client V2) ] **Example 範例：訂閱訊息** 此範例假設您使用 AWS IoT Device SDK 適用於 Python v2 的 1.5.4 版或更新版本。 ``` import threading import traceback import awsiot.greengrasscoreipc.clientv2 as clientV2 topic = 'my/topic' qos = '1' def on_stream_event(event): try: topic_name = event.message.topic_name message = str(event.message.payload, 'utf-8') print(f'Received new message on topic {topic_name}: {message}') except: traceback.print_exc() def on_stream_error(error): # Return True to close stream, False to keep stream open. return True def on_stream_closed(): pass ipc_client = clientV2.GreengrassCoreIPCClientV2() resp, operation = ipc_client.subscribe_to_iot_core( topic_name=topic, qos=qos, on_stream_event=on_stream_event, on_stream_error=on_stream_error, on_stream_closed=on_stream_closed ) # Keep the main thread alive, or the process will exit. event = threading.Event() event.wait() # To stop subscribing, close the operation stream. operation.close() ipc_client.close() ``` ------ #### [ Java (IPC client V1) ] **Example 範例：訂閱訊息** 此範例使用 `IPCUtils`類別來建立與 AWS IoT Greengrass Core IPC 服務的連線。如需詳細資訊，請參閱[連線至 AWS IoT Greengrass Core IPC 服務](interprocess-communication.md#ipc-service-connect)。 ``` package com.aws.greengrass.docs.samples.ipc; import com.aws.greengrass.docs.samples.ipc.util.IPCUtils; import software.amazon.awssdk.aws.greengrass.GreengrassCoreIPCClient; import software.amazon.awssdk.aws.greengrass.SubscribeToIoTCoreResponseHandler; import software.amazon.awssdk.aws.greengrass.model.*; import software.amazon.awssdk.eventstreamrpc.EventStreamRPCConnection; import software.amazon.awssdk.eventstreamrpc.StreamResponseHandler; import java.nio.charset.StandardCharsets; import java.util.Optional; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; public class SubscribeToIoTCore { public static final int TIMEOUT_SECONDS = 10; public static void main(String[] args) { String topic = args[0]; QOS qos = QOS.get(args[1]); try (EventStreamRPCConnection eventStreamRPCConnection = IPCUtils.getEventStreamRpcConnection()) { GreengrassCoreIPCClient ipcClient = new GreengrassCoreIPCClient(eventStreamRPCConnection); StreamResponseHandler streamResponseHandler = new SubscriptionResponseHandler(); SubscribeToIoTCoreResponseHandler responseHandler = SubscribeToIoTCore.subscribeToIoTCore(ipcClient, topic, qos, streamResponseHandler); CompletableFuture futureResponse = responseHandler.getResponse(); try { futureResponse.get(TIMEOUT_SECONDS, TimeUnit.SECONDS); System.out.println("Successfully subscribed to topic: " + topic); } catch (TimeoutException e) { System.err.println("Timeout occurred while subscribing to topic: " + topic); } catch (ExecutionException e) { if (e.getCause() instanceof UnauthorizedError) { System.err.println("Unauthorized error while subscribing to topic: " + topic); } else { throw e; } } // Keep the main thread alive, or the process will exit. try { while (true) { Thread.sleep(10000); } } catch (InterruptedException e) { System.out.println("Subscribe interrupted."); } // To stop subscribing, close the stream. responseHandler.closeStream(); } catch (InterruptedException e) { System.out.println("IPC interrupted."); } catch (ExecutionException e) { System.err.println("Exception occurred when using IPC."); e.printStackTrace(); System.exit(1); } } public static SubscribeToIoTCoreResponseHandler subscribeToIoTCore(GreengrassCoreIPCClient greengrassCoreIPCClient, String topic, QOS qos, StreamResponseHandler streamResponseHandler) { SubscribeToIoTCoreRequest subscribeToIoTCoreRequest = new SubscribeToIoTCoreRequest(); subscribeToIoTCoreRequest.setTopicName(topic); subscribeToIoTCoreRequest.setQos(qos); return greengrassCoreIPCClient.subscribeToIoTCore(subscribeToIoTCoreRequest, Optional.of(streamResponseHandler)); } public static class SubscriptionResponseHandler implements StreamResponseHandler { @Override public void onStreamEvent(IoTCoreMessage ioTCoreMessage) { try { String topic = ioTCoreMessage.getMessage().getTopicName(); String message = new String(ioTCoreMessage.getMessage().getPayload(), StandardCharsets.UTF_8); System.out.printf("Received new message on topic %s: %s%n", topic, message); } catch (Exception e) { System.err.println("Exception occurred while processing subscription response " + "message."); e.printStackTrace(); } } @Override public boolean onStreamError(Throwable error) { System.err.println("Received a stream error."); error.printStackTrace(); return false; } @Override public void onStreamClosed() { System.out.println("Subscribe to IoT Core stream closed."); } } } ``` ------ #### [ Python (IPC client V1) ] **Example 範例：訂閱訊息** 此範例假設您使用 AWS IoT Device SDK 適用於 Python v2 的 1.5.4 版或更新版本。 ``` import time import traceback import awsiot.greengrasscoreipc import awsiot.greengrasscoreipc.client as client from awsiot.greengrasscoreipc.model import ( IoTCoreMessage, QOS, SubscribeToIoTCoreRequest ) TIMEOUT = 10 ipc_client = awsiot.greengrasscoreipc.connect() class StreamHandler(client.SubscribeToIoTCoreStreamHandler): def __init__(self): super().__init__() def on_stream_event(self, event: IoTCoreMessage) -> None: try: message = str(event.message.payload, "utf-8") topic_name = event.message.topic_name # Handle message. except: traceback.print_exc() def on_stream_error(self, error: Exception) -> bool: # Handle error. return True # Return True to close stream, False to keep stream open. def on_stream_closed(self) -> None: # Handle close. pass topic = "my/topic" qos = QOS.AT_MOST_ONCE request = SubscribeToIoTCoreRequest() request.topic_name = topic request.qos = qos handler = StreamHandler() operation = ipc_client.new_subscribe_to_iot_core(handler) operation.activate(request) future_response = operation.get_response() future_response.result(TIMEOUT) # Keep the main thread alive, or the process will exit. while True: time.sleep(10) # To stop subscribing, close the operation stream. operation.close() ``` ------ #### [ C\$1\$1 (IPC client V1) ] **Example 範例：訂閱訊息** ``` #include #include #include using namespace Aws::Crt; using namespace Aws::Greengrass; class IoTCoreResponseHandler : public SubscribeToIoTCoreStreamHandler { public: virtual ~IoTCoreResponseHandler() {} private: void OnStreamEvent(IoTCoreMessage *response) override { auto message = response->GetMessage(); if (message.has_value() && message.value().GetPayload().has_value()) { auto messageBytes = message.value().GetPayload().value(); std::string messageString(messageBytes.begin(), messageBytes.end()); std::string topicName = message.value().GetTopicName().value().c_str(); // Handle message. } } bool OnStreamError(OperationError *error) override { // Handle error. return false; // Return true to close stream, false to keep stream open. } void OnStreamClosed() override { // Handle close. } }; class IpcClientLifecycleHandler : public ConnectionLifecycleHandler { void OnConnectCallback() override { // Handle connection to IPC service. } void OnDisconnectCallback(RpcError error) override { // Handle disconnection from IPC service. } bool OnErrorCallback(RpcError error) override { // Handle IPC service connection error. return true; } }; int main() { ApiHandle apiHandle(g_allocator); Io::EventLoopGroup eventLoopGroup(1); Io::DefaultHostResolver socketResolver(eventLoopGroup, 64, 30); Io::ClientBootstrap bootstrap(eventLoopGroup, socketResolver); IpcClientLifecycleHandler ipcLifecycleHandler; GreengrassCoreIpcClient ipcClient(bootstrap); auto connectionStatus = ipcClient.Connect(ipcLifecycleHandler).get(); if (!connectionStatus) { std::cerr << "Failed to establish IPC connection: " << connectionStatus.StatusToString() << std::endl; exit(-1); } String topic("my/topic"); QOS qos = QOS_AT_MOST_ONCE; int timeout = 10; SubscribeToIoTCoreRequest request; request.SetTopicName(topic); request.SetQos(qos); auto streamHandler = MakeShared(DefaultAllocator()); auto operation = ipcClient.NewSubscribeToIoTCore(streamHandler); auto activate = operation->Activate(request, nullptr); activate.wait(); auto responseFuture = operation->GetResult(); if (responseFuture.wait_for(std::chrono::seconds(timeout)) == std::future_status::timeout) { std::cerr << "Operation timed out while waiting for response from Greengrass Core." << std::endl; exit(-1); } auto response = responseFuture.get(); if (!response) { // Handle error. auto errorType = response.GetResultType(); if (errorType == OPERATION_ERROR) { auto *error = response.GetOperationError(); (void)error; // Handle operation error. } else { // Handle RPC error. } exit(-1); } // Keep the main thread alive, or the process will exit. while (true) { std::this_thread::sleep_for(std::chrono::seconds(10)); } operation->Close(); return 0; } ``` ------ #### [ JavaScript ] **Example 範例：訂閱訊息** ``` import * as greengrasscoreipc from "aws-iot-device-sdk-v2/dist/greengrasscoreipc"; import {IoTCoreMessage, QOS, SubscribeToIoTCoreRequest} from "aws-iot-device-sdk-v2/dist/greengrasscoreipc/model"; import {RpcError} from "aws-iot-device-sdk-v2/dist/eventstream_rpc"; class SubscribeToIoTCore { private ipcClient: greengrasscoreipc.Client private readonly topic: string; constructor() { // define your own constructor, e.g. this.topic = ""; this.subscribeToIoTCore().then(r => console.log("Started workflow")); } private async subscribeToIoTCore() { try { const request: SubscribeToIoTCoreRequest = { topicName: this.topic, qos: QOS.AT_LEAST_ONCE, // you can change this depending on your use case } this.ipcClient = await getIpcClient(); const streamingOperation = this.ipcClient.subscribeToIoTCore(request); streamingOperation.on('message', (message: IoTCoreMessage) => { // parse the message depending on your use cases, e.g. if (message.message && message.message.payload) { const receivedMessage = message.message.payload.toString(); } }); streamingOperation.on('streamError', (error : RpcError) => { // define your own error handling logic }); streamingOperation.on('ended', () => { // define your own logic }); await streamingOperation.activate(); // Keep the main thread alive, or the process will exit. await new Promise((resolve) => setTimeout(resolve, 10000)) } catch (e) { // parse the error depending on your use cases throw e } } } export async function getIpcClient(){ try { const ipcClient = greengrasscoreipc.createClient(); await ipcClient.connect() .catch(error => { // parse the error depending on your use cases throw error; }); return ipcClient } catch (err) { // parse the error depending on your use cases throw err } } // starting point const subscribeToIoTCore = new SubscribeToIoTCore(); ``` ------ #### [ Rust ] **Example 範例：訂閱訊息** ``` use gg_sdk::{Qos, Sdk}; use std::{thread, time::Duration}; fn main() { let sdk = Sdk::init(); sdk.connect().expect("Failed to establish IPC connection"); let topic = "my/topic"; let qos = Qos::AtLeastOnce; let callback = |topic: &str, payload: &[u8]| { let message = String::from_utf8_lossy(payload); println!("Received new message on topic {topic}: {message}"); }; let _sub = sdk .subscribe_to_iot_core(topic, qos, &callback) .expect("Failed to subscribe to topic"); println!("Successfully subscribed to topic: {topic}"); // Keep the main thread alive, or the process will exit. loop { thread::sleep(Duration::from_secs(10)); } } ``` ------ #### [ C ] **Example 範例：訂閱訊息** ``` #include #include #include #include #include #include static void on_subscription_response( void *ctx, GgBuffer topic, GgBuffer payload, GgIpcSubscriptionHandle handle ) { (void) ctx; (void) handle; printf( "Received new message on topic %.*s: %.*s\n", (int) topic.len, topic.data, (int) payload.len, payload.data ); } int main(void) { gg_sdk_init(); GgError err = ggipc_connect(); if (err != GG_ERR_OK) { fprintf(stderr, "Failed to establish IPC connection.\n"); exit(-1); } GgBuffer topic = GG_STR("my/topic"); uint8_t qos = 1; GgIpcSubscriptionHandle handle; err = ggipc_subscribe_to_iot_core( topic, qos, on_subscription_response, NULL, &handle ); if (err != GG_ERR_OK) { fprintf( stderr, "Failed to subscribe to topic: %.*s\n", (int) topic.len, topic.data ); exit(-1); } printf( "Successfully subscribed to topic: %.*s\n", (int) topic.len, topic.data ); // Keep the main thread alive, or the process will exit. while (1) { sleep(10); } // To stop subscribing, close the subscription handle. ggipc_close_subscription(handle); } ``` ------ #### [ C\$1\$1 (Component SDK) ] **Example 範例：訂閱訊息** ``` #include #include #include class ResponseHandler : public gg::ipc::IotTopicCallback { void operator()( std::string_view topic, gg::Buffer payload, gg::ipc::Subscription &handle ) override { (void) handle; std::cout << "Received new message on topic " << topic << ": " << payload << "\n"; } }; int main() { auto &client = gg::ipc::Client::get(); auto error = client.connect(); if (error) { std::cerr << "Failed to establish IPC connection.\n"; exit(-1); } std::string_view topic = "my/topic"; uint8_t qos = 1; static ResponseHandler handler; error = client.subscribe_to_iot_core(topic, qos, handler); if (error) { std::cerr << "Failed to subscribe to topic: " << topic << "\n"; exit(-1); } std::cout << "Successfully subscribed to topic: " << topic << "\n"; // Keep the main thread alive, or the process will exit. while (1) { sleep(10); } } ``` ------ ## 範例使用下列範例來了解如何在元件中使用 AWS IoT Core MQTT IPC 服務。 ### 範例 AWS IoT Core MQTT 發佈者 (C\$1\$1、IPC 用戶端 V1) 下列範例配方允許元件發佈至所有主題。 ------ #### [ JSON ] ``` { "RecipeFormatVersion": "2020-01-25", "ComponentName": "com.example.IoTCorePublisherCpp", "ComponentVersion": "1.0.0", "ComponentDescription": "A component that publishes MQTT messages to IoT Core.", "ComponentPublisher": "Amazon", "ComponentConfiguration": { "DefaultConfiguration": { "accessControl": { "aws.greengrass.ipc.mqttproxy": { "com.example.IoTCorePublisherCpp:mqttproxy:1": { "policyDescription": "Allows access to publish to all topics.", "operations": [ "aws.greengrass#PublishToIoTCore" ], "resources": [ "*" ] } } } } }, "Manifests": [ { "Lifecycle": { "Run": "{artifacts:path}/greengrassv2_iotcore_publisher" }, "Artifacts": [ { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.IoTCorePublisherCpp/1.0.0/greengrassv2_iotcore_publisher", "Permission": { "Execute": "OWNER" } } ] } ] } ``` ------ #### [ YAML ] ``` --- RecipeFormatVersion: '2020-01-25' ComponentName: com.example.IoTCorePublisherCpp ComponentVersion: 1.0.0 ComponentDescription: A component that publishes MQTT messages to IoT Core. ComponentPublisher: Amazon ComponentConfiguration: DefaultConfiguration: accessControl: aws.greengrass.ipc.mqttproxy: com.example.IoTCorePublisherCpp:mqttproxy:1: policyDescription: Allows access to publish to all topics. operations: - aws.greengrass#PublishToIoTCore resources: - "*" Manifests: - Lifecycle: Run: "{artifacts:path}/greengrassv2_iotcore_publisher" Artifacts: - URI: s3://amzn-s3-demo-bucket/artifacts/com.example.IoTCorePublisherCpp/1.0.0/greengrassv2_iotcore_publisher Permission: Execute: OWNER ``` ------ 下列 C\$1\$1 應用程式範例示範如何使用 AWS IoT Core MQTT IPC 服務來發佈訊息 AWS IoT Core。 ``` #include #include #include using namespace Aws::Crt; using namespace Aws::Greengrass; class IpcClientLifecycleHandler : public ConnectionLifecycleHandler { void OnConnectCallback() override { std::cout << "OnConnectCallback" << std::endl; } void OnDisconnectCallback(RpcError error) override { std::cout << "OnDisconnectCallback: " << error.StatusToString() << std::endl; exit(-1); } bool OnErrorCallback(RpcError error) override { std::cout << "OnErrorCallback: " << error.StatusToString() << std::endl; return true; } }; int main() { String message("Hello from the Greengrass IPC MQTT publisher (C++)."); String topic("test/topic/cpp"); QOS qos = QOS_AT_LEAST_ONCE; int timeout = 10; ApiHandle apiHandle(g_allocator); Io::EventLoopGroup eventLoopGroup(1); Io::DefaultHostResolver socketResolver(eventLoopGroup, 64, 30); Io::ClientBootstrap bootstrap(eventLoopGroup, socketResolver); IpcClientLifecycleHandler ipcLifecycleHandler; GreengrassCoreIpcClient ipcClient(bootstrap); auto connectionStatus = ipcClient.Connect(ipcLifecycleHandler).get(); if (!connectionStatus) { std::cerr << "Failed to establish IPC connection: " << connectionStatus.StatusToString() << std::endl; exit(-1); } while (true) { PublishToIoTCoreRequest request; Vector messageData({message.begin(), message.end()}); request.SetTopicName(topic); request.SetPayload(messageData); request.SetQos(qos); auto operation = ipcClient.NewPublishToIoTCore(); auto activate = operation->Activate(request, nullptr); activate.wait(); auto responseFuture = operation->GetResult(); if (responseFuture.wait_for(std::chrono::seconds(timeout)) == std::future_status::timeout) { std::cerr << "Operation timed out while waiting for response from Greengrass Core." << std::endl; exit(-1); } auto response = responseFuture.get(); if (response) { std::cout << "Successfully published to topic: " << topic << std::endl; } else { // An error occurred. std::cout << "Failed to publish to topic: " << topic << std::endl; auto errorType = response.GetResultType(); if (errorType == OPERATION_ERROR) { auto *error = response.GetOperationError(); std::cout << "Operation error: " << error->GetMessage().value() << std::endl; } else { std::cout << "RPC error: " << response.GetRpcError() << std::endl; } exit(-1); } std::this_thread::sleep_for(std::chrono::seconds(5)); } return 0; } ``` ### 範例 AWS IoT Core MQTT 訂閱者 (C\$1\$1、IPC 用戶端 V1) 下列範例配方允許元件訂閱所有主題。 ------ #### [ JSON ] ``` { "RecipeFormatVersion": "2020-01-25", "ComponentName": "com.example.IoTCoreSubscriberCpp", "ComponentVersion": "1.0.0", "ComponentDescription": "A component that subscribes to MQTT messages from IoT Core.", "ComponentPublisher": "Amazon", "ComponentConfiguration": { "DefaultConfiguration": { "accessControl": { "aws.greengrass.ipc.mqttproxy": { "com.example.IoTCoreSubscriberCpp:mqttproxy:1": { "policyDescription": "Allows access to subscribe to all topics.", "operations": [ "aws.greengrass#SubscribeToIoTCore" ], "resources": [ "*" ] } } } } }, "Manifests": [ { "Lifecycle": { "Run": "{artifacts:path}/greengrassv2_iotcore_subscriber" }, "Artifacts": [ { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.IoTCoreSubscriberCpp/1.0.0/greengrassv2_iotcore_subscriber", "Permission": { "Execute": "OWNER" } } ] } ] } ``` ------ #### [ YAML ] ``` --- RecipeFormatVersion: '2020-01-25' ComponentName: com.example.IoTCoreSubscriberCpp ComponentVersion: 1.0.0 ComponentDescription: A component that subscribes to MQTT messages from IoT Core. ComponentPublisher: Amazon ComponentConfiguration: DefaultConfiguration: accessControl: aws.greengrass.ipc.mqttproxy: com.example.IoTCoreSubscriberCpp:mqttproxy:1: policyDescription: Allows access to subscribe to all topics. operations: - aws.greengrass#SubscribeToIoTCore resources: - "*" Manifests: - Lifecycle: Run: "{artifacts:path}/greengrassv2_iotcore_subscriber" Artifacts: - URI: s3://amzn-s3-demo-bucket/artifacts/com.example.IoTCoreSubscriberCpp/1.0.0/greengrassv2_iotcore_subscriber Permission: Execute: OWNER ``` ------ 下列 C\$1\$1 應用程式範例示範如何使用 AWS IoT Core MQTT IPC 服務來訂閱訊息 AWS IoT Core。 ``` #include #include #include using namespace Aws::Crt; using namespace Aws::Greengrass; class IoTCoreResponseHandler : public SubscribeToIoTCoreStreamHandler { public: virtual ~IoTCoreResponseHandler() {} private: void OnStreamEvent(IoTCoreMessage *response) override { auto message = response->GetMessage(); if (message.has_value() && message.value().GetPayload().has_value()) { auto messageBytes = message.value().GetPayload().value(); std::string messageString(messageBytes.begin(), messageBytes.end()); std::string messageTopic = message.value().GetTopicName().value().c_str(); std::cout << "Received new message on topic: " << messageTopic << std::endl; std::cout << "Message: " << messageString << std::endl; } } bool OnStreamError(OperationError *error) override { std::cout << "Received an operation error: "; if (error->GetMessage().has_value()) { std::cout << error->GetMessage().value(); } std::cout << std::endl; return false; // Return true to close stream, false to keep stream open. } void OnStreamClosed() override { std::cout << "Subscribe to IoT Core stream closed." << std::endl; } }; class IpcClientLifecycleHandler : public ConnectionLifecycleHandler { void OnConnectCallback() override { std::cout << "OnConnectCallback" << std::endl; } void OnDisconnectCallback(RpcError error) override { std::cout << "OnDisconnectCallback: " << error.StatusToString() << std::endl; exit(-1); } bool OnErrorCallback(RpcError error) override { std::cout << "OnErrorCallback: " << error.StatusToString() << std::endl; return true; } }; int main() { String topic("test/topic/cpp"); QOS qos = QOS_AT_LEAST_ONCE; int timeout = 10; ApiHandle apiHandle(g_allocator); Io::EventLoopGroup eventLoopGroup(1); Io::DefaultHostResolver socketResolver(eventLoopGroup, 64, 30); Io::ClientBootstrap bootstrap(eventLoopGroup, socketResolver); IpcClientLifecycleHandler ipcLifecycleHandler; GreengrassCoreIpcClient ipcClient(bootstrap); auto connectionStatus = ipcClient.Connect(ipcLifecycleHandler).get(); if (!connectionStatus) { std::cerr << "Failed to establish IPC connection: " << connectionStatus.StatusToString() << std::endl; exit(-1); } SubscribeToIoTCoreRequest request; request.SetTopicName(topic); request.SetQos(qos); auto streamHandler = MakeShared(DefaultAllocator()); auto operation = ipcClient.NewSubscribeToIoTCore(streamHandler); auto activate = operation->Activate(request, nullptr); activate.wait(); auto responseFuture = operation->GetResult(); if (responseFuture.wait_for(std::chrono::seconds(timeout)) == std::future_status::timeout) { std::cerr << "Operation timed out while waiting for response from Greengrass Core." << std::endl; exit(-1); } auto response = responseFuture.get(); if (response) { std::cout << "Successfully subscribed to topic: " << topic << std::endl; } else { // An error occurred. std::cout << "Failed to subscribe to topic: " << topic << std::endl; auto errorType = response.GetResultType(); if (errorType == OPERATION_ERROR) { auto *error = response.GetOperationError(); std::cout << "Operation error: " << error->GetMessage().value() << std::endl; } else { std::cout << "RPC error: " << response.GetRpcError() << std::endl; } exit(-1); } // Keep the main thread alive, or the process will exit. while (true) { std::this_thread::sleep_for(std::chrono::seconds(10)); } operation->Close(); return 0; } ``` ### 範例 AWS IoT Core MQTT 發佈者 (Rust) 下列範例配方允許元件發佈至所有主題。 ``` { "RecipeFormatVersion": "2020-01-25", "ComponentName": "com.example.IoTCorePublisherRust", "ComponentVersion": "1.0.0", "ComponentDescription": "A component that publishes MQTT messages to IoT Core.", "ComponentPublisher": "Amazon", "ComponentConfiguration": { "DefaultConfiguration": { "accessControl": { "aws.greengrass.ipc.mqttproxy": { "com.example.IoTCorePublisherRust:mqttproxy:1": { "policyDescription": "Allows access to publish to all topics.", "operations": ["aws.greengrass#PublishToIoTCore"], "resources": ["*"] } } } } }, "Manifests": [ { "Platform": { "os": "linux", "runtime": "*" }, "Lifecycle": { "run": "{artifacts:path}/publish_to_iot_core" }, "Artifacts": [ { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.IoTCorePublisherRust/1.0.0/publish_to_iot_core", "Permission": { "Execute": "OWNER" } } ] } ] } ``` 下列範例 Rust 應用程式示範如何使用 AWS IoT Core MQTT IPC 服務來發佈訊息 AWS IoT Core。 ``` use gg_sdk::{Qos, Sdk}; fn main() { let sdk = Sdk::init(); sdk.connect().expect("Failed to establish IPC connection"); let message = b"Hello, World"; let topic = "my/topic"; let qos = Qos::AtLeastOnce; sdk.publish_to_iot_core(topic, message, qos) .expect("Failed to publish to topic"); println!("Successfully published to topic: {topic}"); } ``` ### 範例 AWS IoT Core MQTT 訂閱者 (Rust) 下列範例配方允許元件訂閱所有主題。 ``` { "RecipeFormatVersion": "2020-01-25", "ComponentName": "com.example.IoTCoreSubscriberRust", "ComponentVersion": "1.0.0", "ComponentDescription": "A component that subscribes to MQTT messages from IoT Core.", "ComponentPublisher": "Amazon", "ComponentConfiguration": { "DefaultConfiguration": { "accessControl": { "aws.greengrass.ipc.mqttproxy": { "com.example.IoTCoreSubscriberRust:mqttproxy:1": { "policyDescription": "Allows access to subscribe to all topics.", "operations": ["aws.greengrass#SubscribeToIoTCore"], "resources": ["*"] } } } } }, "Manifests": [ { "Platform": { "os": "linux", "runtime": "*" }, "Lifecycle": { "run": "{artifacts:path}/subscribe_to_iot_core" }, "Artifacts": [ { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.IoTCoreSubscriberRust/1.0.0/subscribe_to_iot_core", "Permission": { "Execute": "OWNER" } } ] } ] } ``` 下列範例 Rust 應用程式示範如何使用 AWS IoT Core MQTT IPC 服務來訂閱訊息 AWS IoT Core。 ``` use gg_sdk::{Qos, Sdk}; use std::{thread, time::Duration}; fn main() { let sdk = Sdk::init(); sdk.connect().expect("Failed to establish IPC connection"); let topic = "my/topic"; let qos = Qos::AtLeastOnce; let callback = |topic: &str, payload: &[u8]| { let message = String::from_utf8_lossy(payload); println!("Received new message on topic {topic}: {message}"); }; let _sub = sdk .subscribe_to_iot_core(topic, qos, &callback) .expect("Failed to subscribe to topic"); println!("Successfully subscribed to topic: {topic}"); // Keep the main thread alive, or the process will exit. loop { thread::sleep(Duration::from_secs(10)); } } ``` ### 範例 AWS IoT Core MQTT 發佈者 (C) 下列範例配方允許元件發佈至所有主題。 ``` { "RecipeFormatVersion": "2020-01-25", "ComponentName": "com.example.IoTCorePublisherC", "ComponentVersion": "1.0.0", "ComponentDescription": "A component that publishes MQTT messages to IoT Core.", "ComponentPublisher": "Amazon", "ComponentConfiguration": { "DefaultConfiguration": { "accessControl": { "aws.greengrass.ipc.mqttproxy": { "com.example.IoTCorePublisherC:mqttproxy:1": { "policyDescription": "Allows access to publish to all topics.", "operations": ["aws.greengrass#PublishToIoTCore"], "resources": ["*"] } } } } }, "Manifests": [ { "Platform": { "os": "linux", "runtime": "*" }, "Lifecycle": { "run": "{artifacts:path}/sample_publish_to_iot_core" }, "Artifacts": [ { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.IoTCorePublisherC/1.0.0/sample_publish_to_iot_core", "Permission": { "Execute": "OWNER" } } ] } ] } ``` 下列範例 C 應用程式示範如何使用 AWS IoT Core MQTT IPC 服務來發佈訊息 AWS IoT Core。 ``` #include #include #include #include #include int main(void) { gg_sdk_init(); GgError err = ggipc_connect(); if (err != GG_ERR_OK) { fprintf(stderr, "Failed to establish IPC connection.\n"); exit(-1); } GgBuffer message = GG_STR("Hello, World"); GgBuffer topic = GG_STR("my/topic"); uint8_t qos = 1; err = ggipc_publish_to_iot_core(topic, message, qos); if (err != GG_ERR_OK) { fprintf( stderr, "Failed to publish to topic: %.*s\n", (int) topic.len, topic.data ); exit(-1); } printf( "Successfully published to topic: %.*s\n", (int) topic.len, topic.data ); } ``` ### 範例 AWS IoT Core MQTT 訂閱者 (C) 下列範例配方允許元件訂閱所有主題。 ``` { "RecipeFormatVersion": "2020-01-25", "ComponentName": "com.example.IoTCoreSubscriberC", "ComponentVersion": "1.0.0", "ComponentDescription": "A component that subscribes to MQTT messages from IoT Core.", "ComponentPublisher": "Amazon", "ComponentConfiguration": { "DefaultConfiguration": { "accessControl": { "aws.greengrass.ipc.mqttproxy": { "com.example.IoTCoreSubscriberC:mqttproxy:1": { "policyDescription": "Allows access to subscribe to all topics.", "operations": ["aws.greengrass#SubscribeToIoTCore"], "resources": ["*"] } } } } }, "Manifests": [ { "Platform": { "os": "linux", "runtime": "*" }, "Lifecycle": { "run": "{artifacts:path}/sample_subscribe_to_iot_core" }, "Artifacts": [ { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.IoTCoreSubscriberC/1.0.0/sample_subscribe_to_iot_core", "Permission": { "Execute": "OWNER" } } ] } ] } ``` 下列範例 C 應用程式示範如何使用 AWS IoT Core MQTT IPC 服務來訂閱訊息 AWS IoT Core。 ``` #include #include #include #include #include #include static void on_subscription_response( void *ctx, GgBuffer topic, GgBuffer payload, GgIpcSubscriptionHandle handle ) { (void) ctx; (void) handle; printf( "Received new message on topic %.*s: %.*s\n", (int) topic.len, topic.data, (int) payload.len, payload.data ); } int main(void) { gg_sdk_init(); GgError err = ggipc_connect(); if (err != GG_ERR_OK) { fprintf(stderr, "Failed to establish IPC connection.\n"); exit(-1); } GgBuffer topic = GG_STR("my/topic"); uint8_t qos = 1; GgIpcSubscriptionHandle handle; err = ggipc_subscribe_to_iot_core( topic, qos, on_subscription_response, NULL, &handle ); if (err != GG_ERR_OK) { fprintf( stderr, "Failed to subscribe to topic: %.*s\n", (int) topic.len, topic.data ); exit(-1); } printf( "Successfully subscribed to topic: %.*s\n", (int) topic.len, topic.data ); // Keep the main thread alive, or the process will exit. while (1) { sleep(10); } // To stop subscribing, close the subscription handle. ggipc_close_subscription(handle); } ``` ### 範例 AWS IoT Core MQTT 發佈者 (C\$1\$1、元件 SDK) 下列範例配方允許元件發佈至所有主題。 ``` { "RecipeFormatVersion": "2020-01-25", "ComponentName": "com.example.IoTCorePublisherCpp", "ComponentVersion": "1.0.0", "ComponentDescription": "A component that publishes MQTT messages to IoT Core.", "ComponentPublisher": "Amazon", "ComponentConfiguration": { "DefaultConfiguration": { "accessControl": { "aws.greengrass.ipc.mqttproxy": { "com.example.IoTCorePublisherCpp:mqttproxy:1": { "policyDescription": "Allows access to publish to all topics.", "operations": ["aws.greengrass#PublishToIoTCore"], "resources": ["*"] } } } } }, "Manifests": [ { "Platform": { "os": "linux", "runtime": "*" }, "Lifecycle": { "run": "{artifacts:path}/sample_cpp_publish_to_iot_core" }, "Artifacts": [ { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.IoTCorePublisherCpp/1.0.0/sample_cpp_publish_to_iot_core", "Permission": { "Execute": "OWNER" } } ] } ] } ``` 下列 C\$1\$1 應用程式範例示範如何使用 AWS IoT Core MQTT IPC 服務來發佈訊息 AWS IoT Core。 ``` #include #include int main() { auto &client = gg::ipc::Client::get(); auto error = client.connect(); if (error) { std::cerr << "Failed to establish IPC connection.\n"; exit(-1); } std::string_view message = "Hello, World"; std::string_view topic = "my/topic"; uint8_t qos = 1; error = client.publish_to_iot_core(topic, message, qos); if (error) { std::cerr << "Failed to publish to topic: " << topic << "\n"; exit(-1); } std::cout << "Successfully published to topic: " << topic << "\n"; } ``` ### 範例 AWS IoT Core MQTT 訂閱者 (C\$1\$1、元件 SDK) 下列範例配方允許元件訂閱所有主題。 ``` { "RecipeFormatVersion": "2020-01-25", "ComponentName": "com.example.IoTCoreSubscriberCpp", "ComponentVersion": "1.0.0", "ComponentDescription": "A component that subscribes to MQTT messages from IoT Core.", "ComponentPublisher": "Amazon", "ComponentConfiguration": { "DefaultConfiguration": { "accessControl": { "aws.greengrass.ipc.mqttproxy": { "com.example.IoTCoreSubscriberCpp:mqttproxy:1": { "policyDescription": "Allows access to subscribe to all topics.", "operations": ["aws.greengrass#SubscribeToIoTCore"], "resources": ["*"] } } } } }, "Manifests": [ { "Platform": { "os": "linux", "runtime": "*" }, "Lifecycle": { "run": "{artifacts:path}/sample_cpp_subscribe_to_iot_core" }, "Artifacts": [ { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.IoTCoreSubscriberCpp/1.0.0/sample_cpp_subscribe_to_iot_core", "Permission": { "Execute": "OWNER" } } ] } ] } ``` 下列 C\$1\$1 應用程式範例示範如何使用 AWS IoT Core MQTT IPC 服務來訂閱訊息 AWS IoT Core。 ``` #include #include #include class ResponseHandler : public gg::ipc::IotTopicCallback { void operator()( std::string_view topic, gg::Buffer payload, gg::ipc::Subscription &handle ) override { (void) handle; std::cout << "Received new message on topic " << topic << ": " << payload << "\n"; } }; int main() { auto &client = gg::ipc::Client::get(); auto error = client.connect(); if (error) { std::cerr << "Failed to establish IPC connection.\n"; exit(-1); } std::string_view topic = "my/topic"; uint8_t qos = 1; static ResponseHandler handler; error = client.subscribe_to_iot_core(topic, qos, handler); if (error) { std::cerr << "Failed to subscribe to topic: " << topic << "\n"; exit(-1); } std::cout << "Successfully subscribed to topic: " << topic << "\n"; // Keep the main thread alive, or the process will exit. while (1) { sleep(10); } } ``` # 與元件生命週期互動使用元件生命週期 IPC 服務來： + 更新核心裝置上的元件狀態。 + 訂閱元件狀態更新。 + 防止核心在部署期間停止元件以套用更新。 + 暫停和繼續元件程序。 **Topics** + [最低 SDK 版本](#ipc-component-lifecycle-sdk-versions) + [Authorization](#ipc-component-lifecycle-authorization) + [UpdateState](#ipc-operation-updatestate) + [SubscribeToComponentUpdates](#ipc-operation-subscribetocomponentupdates) + [DeferComponentUpdate](#ipc-operation-defercomponentupdate) + [PauseComponent](#ipc-operation-pausecomponent) + [ResumeComponent](#ipc-operation-resumecomponent) ## 最低 SDK 版本下表列出 AWS IoT Device SDK 您必須用來與元件生命週期互動的最低版本。 | SDK | 最低版本 | | --- | --- | | [AWS IoT Device SDK for Java v2](https://github.com/aws/aws-iot-device-sdk-java-v2) | v1.2.10 | | [AWS IoT Device SDK for Python v2](https://github.com/aws/aws-iot-device-sdk-python-v2) | 1.5.3 版 | | [AWS IoT Device SDK 適用於 C\$1\$1 v2](https://github.com/aws/aws-iot-device-sdk-cpp-v2) | 1.17.0 版 | | [AWS IoT Device SDK for JavaScript v2](https://github.com/aws/aws-iot-device-sdk-js-v2) | v1.12.0 | ## Authorization 若要暫停或繼續自訂元件中的其他元件，您必須定義授權政策，以允許元件管理其他元件。如需定義授權政策的資訊，請參閱 [授權元件執行 IPC 操作](interprocess-communication.md#ipc-authorization-policies)。元件生命週期管理的授權政策具有下列屬性。 **IPC 服務識別符：** `aws.greengrass.ipc.lifecycle` | 作業 | Description | Resources | | --- | --- | --- | | `aws.greengrass#PauseComponent` | 允許元件暫停您指定的元件。 | 元件名稱，或 `*` 允許存取所有元件。 | | `aws.greengrass#ResumeComponent` | 允許元件繼續您指定的元件。 | 元件名稱，或 `*` 允許存取所有元件。 | | `*` | 允許元件暫停和繼續您指定的元件。 | 元件名稱，或 `*` 允許存取所有元件。 | ### 授權政策範例您可以參考下列授權政策範例，協助您設定元件的授權政策。 **Example 範例授權政策** 下列範例授權政策允許元件暫停和繼續所有元件。 ``` { "accessControl": { "aws.greengrass.ipc.lifecycle": { "com.example.MyLocalLifecycleComponent:lifecycle:1": { "policyDescription": "Allows access to pause/resume all components.", "operations": [ "aws.greengrass#PauseComponent", "aws.greengrass#ResumeComponent" ], "resources": [ "*" ] } } } } ``` ## UpdateState 更新核心裝置上的元件狀態。 ### 請求此操作的請求具有下列參數： `state` 要設定的狀態。此列舉 `LifecycleState`具有下列值： + `RUNNING` + `ERRORED` ### 回應此操作不會在其回應中提供任何資訊。 ### 範例下列範例示範如何在自訂元件程式碼中呼叫此操作。 ------ #### [ Rust ] **Example 範例：更新狀態** ``` use gg_sdk::{ComponentState, Sdk}; fn main() { let sdk = Sdk::init(); sdk.connect().expect("Failed to establish IPC connection"); // Update component state to RUNNING sdk.update_state(ComponentState::Running) .expect("Failed to update component state"); println!("Successfully updated component state to RUNNING."); } ``` ------ #### [ C ] **Example 範例：更新狀態** ``` #include #include #include #include #include int main(void) { gg_sdk_init(); GgError err = ggipc_connect(); if (err != GG_ERR_OK) { fprintf(stderr, "Failed to establish IPC connection.\n"); exit(-1); } // Update component state to RUNNING err = ggipc_update_state(GG_COMPONENT_STATE_RUNNING); if (err != GG_ERR_OK) { fprintf(stderr, "Failed to update component state.\n"); exit(-1); } printf("Successfully updated component state to RUNNING.\n"); } ``` ------ #### [ C\$1\$1 (Component SDK) ] **Example 範例：更新狀態** ``` #include #include int main() { auto &client = gg::ipc::Client::get(); auto error = client.connect(); if (error) { std::cerr << "Failed to establish IPC connection.\n"; exit(-1); } // Update component state to RUNNING error = client.update_component_state(GG_COMPONENT_STATE_RUNNING); if (error) { std::cerr << "Failed to update component state.\n"; exit(-1); } std::cout << "Successfully updated component state to RUNNING.\n"; } ``` ------ ## SubscribeToComponentUpdates 在 AWS IoT Greengrass Core 軟體更新元件之前訂閱以接收通知。通知會指定是否要在更新時重新啟動核。只有當部署的元件更新政策指定通知元件時，核才會傳送更新通知。預設行為是通知元件。如需詳細資訊，請參閱 [建立部署](create-deployments.md)和您可以在呼叫 [CreateDeployment](https://docs.aws.amazon.com/greengrass/v2/APIReference/API_CreateDeployment.html) 操作時提供的 [DeploymentComponentUpdatePolicy](https://docs.aws.amazon.com/greengrass/v2/APIReference/API_DeploymentComponentUpdatePolicy.html) 物件。 **重要** 本機部署不會在更新之前通知元件。此操作是您訂閱事件訊息串流的訂閱操作。若要使用此操作，請使用處理事件訊息、錯誤和串流關閉的函數來定義串流回應處理常式。如需詳細資訊，請參閱[訂閱 IPC 事件串流](interprocess-communication.md#ipc-subscribe-operations)。 **事件訊息類型：** `ComponentUpdatePolicyEvents` **提示** 您可以遵循教學課程，了解如何開發有條件延遲元件更新的元件。如需詳細資訊，請參閱[教學課程：開發可延遲元件更新的 Greengrass 元件](defer-component-updates-tutorial.md)。 ### 請求此操作的請求沒有任何參數。 ### 回應此操作的回應包含下列資訊： `messages` 通知訊息的串流。此物件 `ComponentUpdatePolicyEvents`包含下列資訊： `preUpdateEvent` (Python：`pre_update_event`) （選用）表示核心想要更新元件的事件。您可以使用 [DeferComponentUpdate](#ipc-operation-defercomponentupdate)操作來回應，以確認或延遲更新，直到您的元件準備好重新啟動為止。此物件 `PreComponentUpdateEvent`包含下列資訊： `deploymentId` (Python：`deployment_id`) 更新元件的 AWS IoT Greengrass 部署 ID。 `isGgcRestarting` (Python：`is_ggc_restarting`) 是否需要重新啟動核心，才能套用更新。 `postUpdateEvent` (Python：`post_update_event`) （選用）指出核心已更新元件的事件。此物件 `PostComponentUpdateEvent`包含下列資訊： `deploymentId` (Python：`deployment_id`) 更新元件的 AWS IoT Greengrass 部署 ID。此功能需要 2.7.0 版或更新版本的 Greengrass 核元件。 ## DeferComponentUpdate 使用確認或延遲您探索的元件更新[SubscribeToComponentUpdates](#ipc-operation-subscribetocomponentupdates)。如果您的元件已準備好讓元件更新繼續，您可以指定在核心再次檢查之前等待的時間量。您也可以使用此操作來告知 nucleus 您的元件已準備好進行更新。如果元件未回應元件更新通知，則 nucleus 會等待您在部署的元件更新政策中指定的時間量。逾時之後，核會繼續進行部署。預設元件更新逾時為 60 秒。如需詳細資訊，請參閱 [建立部署](create-deployments.md)和您可以在呼叫 [CreateDeployment](https://docs.aws.amazon.com/greengrass/v2/APIReference/API_CreateDeployment.html) 操作時提供的 [DeploymentComponentUpdatePolicy](https://docs.aws.amazon.com/greengrass/v2/APIReference/API_DeploymentComponentUpdatePolicy.html) 物件。 **提示** 您可以遵循教學課程，了解如何開發有條件延遲元件更新的元件。如需詳細資訊，請參閱[教學課程：開發可延遲元件更新的 Greengrass 元件](defer-component-updates-tutorial.md)。 ### 請求此操作的請求具有下列參數： `deploymentId` (Python：`deployment_id`) 要延遲的 AWS IoT Greengrass 部署 ID。 `message` （選用）要延遲更新的元件名稱。預設為發出請求的元件名稱。 `recheckAfterMs` (Python：`recheck_after_ms`) 延遲更新的時間量，以毫秒為單位。核會等待這段時間，然後傳送另一個`PreComponentUpdateEvent`您可以使用探索的時間[SubscribeToComponentUpdates](#ipc-operation-subscribetocomponentupdates)。指定 `0` 以確認更新。這會通知 nucleus 您的元件已準備好進行更新。預設為零毫秒，這表示確認更新。 ### 回應此操作不會在其回應中提供任何資訊。 ## PauseComponent 此功能適用於 [Greengrass 核元件](greengrass-nucleus-component.md)的 v2.4.0 和更新版本。目前 AWS IoT Greengrass 不支援 Windows 核心裝置上的此功能。在核心裝置上暫停元件的程序。若要繼續元件，請使用 [ResumeComponent](#ipc-operation-resumecomponent) 操作。您只能暫停一般元件。如果您嘗試暫停任何其他類型的元件，此操作會擲回 `InvalidRequestError`。 **注意** 此操作無法暫停容器化程序，例如 Docker 容器。若要暫停和繼續 Docker 容器，您可以使用 [docker pause](https://docs.docker.com/engine/reference/commandline/pause/) 和 [docker unpause](https://docs.docker.com/engine/reference/commandline/unpause/) 命令。此操作不會暫停元件相依性，或取決於您暫停之元件的元件。當您暫停屬於另一個元件相依性的元件時，請考慮此行為，因為相依元件可能會在其相依性暫停時遇到問題。當您重新啟動或關閉暫停的元件時，例如透過部署，Greengrass 核會繼續該元件並執行其關機生命週期。如需重新啟動元件的詳細資訊，請參閱 [RestartComponent](ipc-local-deployments-components.md#ipc-operation-restartcomponent)。 **重要** 若要使用此操作，您必須定義授權政策，授予使用此操作的許可。如需詳細資訊，請參閱[Authorization](#ipc-component-lifecycle-authorization)。 ### 最低 SDK 版本下表列出 AWS IoT Device SDK 您必須用來暫停和繼續元件的最低版本。 | SDK | 最低版本 | | --- | --- | | [AWS IoT Device SDK for Java v2](https://github.com/aws/aws-iot-device-sdk-java-v2) | 1.4.3 版 | | [AWS IoT Device SDK for Python v2](https://github.com/aws/aws-iot-device-sdk-python-v2) | 1.6.2 版 | | [AWS IoT Device SDK 適用於 C\$1\$1 v2](https://github.com/aws/aws-iot-device-sdk-cpp-v2) | V1.13.1 | | [AWS IoT Device SDK for JavaScript v2](https://github.com/aws/aws-iot-device-sdk-js-v2) | v1.12.0 | ### 請求此操作的請求具有下列參數： `componentName` (Python：`component_name`) 要暫停的元件名稱，必須是一般元件。如需詳細資訊，請參閱[元件類型](develop-greengrass-components.md#component-types)。 ### 回應此操作不會在其回應中提供任何資訊。 ## ResumeComponent 此功能適用於 [Greengrass 核元件](greengrass-nucleus-component.md)的 v2.4.0 和更新版本。目前 AWS IoT Greengrass 不支援 Windows 核心裝置上的此功能。在核心裝置上繼續元件的程序。若要暫停元件，請使用 [PauseComponent](#ipc-operation-pausecomponent) 操作。您只能繼續暫停的元件。如果您嘗試繼續未暫停的元件，此操作會擲回 `InvalidRequestError`。 **重要** 若要使用此操作，您必須定義授予許可的授權政策。如需詳細資訊，請參閱[Authorization](#ipc-component-lifecycle-authorization)。 ### 最低 SDK 版本下表列出 AWS IoT Device SDK 您必須用來暫停和繼續元件的最低版本。 | SDK | 最低版本 | | --- | --- | | [AWS IoT Device SDK for Java v2](https://github.com/aws/aws-iot-device-sdk-java-v2) | 1.4.3 版 | | [AWS IoT Device SDK for Python v2](https://github.com/aws/aws-iot-device-sdk-python-v2) | 1.6.2 版 | | [AWS IoT Device SDK 適用於 C\$1\$1 v2](https://github.com/aws/aws-iot-device-sdk-cpp-v2) | V1.13.1 | | [AWS IoT Device SDK for JavaScript v2](https://github.com/aws/aws-iot-device-sdk-js-v2) | v1.12.0 | ### 請求此操作的請求具有下列參數： `componentName` (Python：`component_name`) 要繼續的元件名稱。 ### 回應此操作不會在其回應中提供任何資訊。 # 與元件組態互動元件組態 IPC 服務可讓您執行下列動作： + 取得並設定元件組態參數。 + 訂閱元件組態更新。 + 在套用核心之前驗證元件組態更新。 **Topics** + [最低 SDK 版本](#ipc-component-configuration-sdk-versions) + [GetConfiguration](#ipc-operation-getconfiguration) + [UpdateConfiguration](#ipc-operation-updateconfiguration) + [SubscribeToConfigurationUpdate](#ipc-operation-subscribetoconfigurationupdate) + [SubscribeToValidateConfigurationUpdates](#ipc-operation-subscribetovalidateconfigurationupdates) + [SendConfigurationValidityReport](#ipc-operation-sendconfigurationvalidityreport) ## 最低 SDK 版本下表列出 AWS IoT Device SDK 您必須用來與元件組態互動的最低版本。 | SDK | 最低版本 | | --- | --- | | [AWS IoT Device SDK for Java v2](https://github.com/aws/aws-iot-device-sdk-java-v2) | v1.2.10 | | [AWS IoT Device SDK for Python v2](https://github.com/aws/aws-iot-device-sdk-python-v2) | 1.5.3 版 | | [AWS IoT Device SDK 適用於 C\$1\$1 v2](https://github.com/aws/aws-iot-device-sdk-cpp-v2) | 1.17.0 版 | | [AWS IoT Device SDK for JavaScript v2](https://github.com/aws/aws-iot-device-sdk-js-v2) | v1.12.0 | ## GetConfiguration 取得核心裝置上元件的組態值。您可以指定要取得組態值的金鑰路徑。 ### 請求此操作的請求具有下列參數： `componentName` (Python：`component_name`) （選用）元件的名稱。預設為發出請求的元件名稱。 `keyPath` (Python：`key_path`) 組態值的金鑰路徑。指定清單，其中每個項目都是組態物件中單一層級的金鑰。例如，指定 `["mqtt", "port"]` 在下列組態`port`中取得的值。 ``` { "mqtt": { "port": 443 } } ``` 若要取得元件的完整組態，請指定空白清單。 ### 回應此操作的回應包含下列資訊： `componentName` (Python：`component_name`) 元件的名稱。 `value` 請求的組態做為物件。 ### 範例下列範例示範如何在自訂元件程式碼中呼叫此操作。 ------ #### [ Rust ] **Example 範例：取得組態** ``` use core::mem::MaybeUninit; use gg_sdk::{Sdk, UnpackedObject}; fn main() { let sdk = Sdk::init(); sdk.connect().expect("Failed to establish IPC connection"); // Get a configuration value at key path ["mqtt", "port"] let mut buf = [MaybeUninit::uninit(); 1024]; let value = sdk .get_config(&["mqtt", "port"], None, &mut buf) .expect("Failed to get configuration"); if let UnpackedObject::I64(port) = value.unpack() { println!("Configuration value: {port}"); } } ``` ------ #### [ C ] **Example 範例：取得組態** ``` #include #include #include #include #include #include #include int main(void) { gg_sdk_init(); GgError err = ggipc_connect(); if (err != GG_ERR_OK) { fprintf(stderr, "Failed to establish IPC connection.\n"); exit(-1); } // Get a configuration value at key path ["mqtt", "port"] uint8_t response_mem[1024]; GgObject value; err = ggipc_get_config( GG_BUF_LIST(GG_STR("mqtt"), GG_STR("port")), NULL, // component_name (NULL = current component) GG_BUF(response_mem), &value ); if (err != GG_ERR_OK) { fprintf(stderr, "Failed to get configuration.\n"); exit(-1); } if (gg_obj_type(value) == GG_TYPE_I64) { printf("Configuration value: %" PRId64 "\n", gg_obj_into_i64(value)); } else if (gg_obj_type(value) == GG_TYPE_BUF) { GgBuffer buf = gg_obj_into_buf(value); printf("Configuration value: %.*s\n", (int) buf.len, buf.data); } else { printf("Configuration value is of unexpected type.\n"); } } ``` ------ #### [ C\$1\$1 (Component SDK) ] **Example 範例：取得組態** ``` #include #include int main() { auto &client = gg::ipc::Client::get(); auto error = client.connect(); if (error) { std::cerr << "Failed to establish IPC connection.\n"; exit(-1); } // Get a configuration value at key path ["mqtt", "port"] std::array key_path = { gg::Buffer { "mqtt" }, gg::Buffer { "port" } }; int64_t value = 0; error = client.get_config(key_path, std::nullopt, value); if (error) { std::cerr << "Failed to get configuration.\n"; exit(-1); } std::cout << "Configuration value: " << value << "\n"; } ``` ------ ## UpdateConfiguration 更新核心裝置上此元件的組態值。 ### 請求此操作的請求具有下列參數： `keyPath` (Python：`key_path`) （選用）要更新的容器節點 ( 物件）的金鑰路徑。指定清單，其中每個項目都是組態物件中單一層級的金鑰。例如，指定金鑰路徑`["mqtt"]`和合併值`{ "port": 443 }`，以在下列組態`port`中設定的值。 ``` { "mqtt": { "port": 443 } } ``` 金鑰路徑必須在組態中指定容器節點（物件）。如果節點不存在於元件的組態中，此操作會建立節點，並將其值設定為中的物件`valueToMerge`。預設為組態物件的根。 `timestamp` 目前的 Unix epoch 時間，以毫秒為單位。此操作使用此時間戳記來解析金鑰的並行更新。如果元件組態中的索引鍵的時間戳記大於請求中的時間戳記，則請求會失敗。 `valueToMerge` (Python：`value_to_merge`) 要在您在中指定的位置合併的組態物件`keyPath`。如需詳細資訊，請參閱[更新元件組態](update-component-configurations.md)。 ### 回應此操作不會在其回應中提供任何資訊。 ### 範例下列範例示範如何在自訂元件程式碼中呼叫此操作。 ------ #### [ Rust ] **Example 範例：更新組態** ``` use gg_sdk::Sdk; fn main() { let sdk = Sdk::init(); sdk.connect().expect("Failed to establish IPC connection"); // Update configuration value at key path ["mqtt", "port"] to 443 sdk.update_config(&["mqtt", "port"], None, 443) .expect("Failed to update configuration"); println!("Successfully updated configuration."); } ``` ------ #### [ C ] **Example 範例：更新組態** ``` #include #include #include #include #include #include int main(void) { gg_sdk_init(); GgError err = ggipc_connect(); if (err != GG_ERR_OK) { fprintf(stderr, "Failed to establish IPC connection.\n"); exit(-1); } // Update configuration value at key path ["mqtt", "port"] to 443 err = ggipc_update_config( GG_BUF_LIST(GG_STR("mqtt"), GG_STR("port")), NULL, // timestamp (NULL = current time) gg_obj_i64(443) ); if (err != GG_ERR_OK) { fprintf(stderr, "Failed to update configuration.\n"); exit(-1); } printf("Successfully updated configuration.\n"); } ``` ------ #### [ C\$1\$1 (Component SDK) ] **Example 範例：更新組態** ``` #include #include int main() { auto &client = gg::ipc::Client::get(); auto error = client.connect(); if (error) { std::cerr << "Failed to establish IPC connection.\n"; exit(-1); } // Update configuration value at key path ["mqtt", "port"] to 443 std::array key_path = { gg::Buffer { "mqtt" }, gg::Buffer { "port" } }; error = client.update_config(key_path, 443); if (error) { std::cerr << "Failed to update configuration.\n"; exit(-1); } std::cout << "Successfully updated configuration.\n"; } ``` ------ ## SubscribeToConfigurationUpdate 訂閱以在元件的組態更新時接收通知。當您訂閱金鑰時，當該金鑰的任何子項更新時，您會收到通知。此操作是您訂閱事件訊息串流的訂閱操作。若要使用此操作，請使用處理事件訊息、錯誤和串流關閉的函數來定義串流回應處理常式。如需詳細資訊，請參閱[訂閱 IPC 事件串流](interprocess-communication.md#ipc-subscribe-operations)。 **事件訊息類型：** `ConfigurationUpdateEvents` ### 請求此操作的請求具有下列參數： `componentName` (Python：`component_name`) （選用）元件的名稱。預設為發出請求的元件名稱。 `keyPath` (Python：`key_path`) 要訂閱之組態值的金鑰路徑。指定清單，其中每個項目都是組態物件中單一層級的金鑰。例如，指定 `["mqtt", "port"]` 在下列組態`port`中取得的值。 ``` { "mqtt": { "port": 443 } } ``` 若要訂閱元件組態中所有值的更新，請指定空白清單。 ### 回應此操作的回應包含下列資訊： `messages` 通知訊息的串流。此物件 `ConfigurationUpdateEvents`包含下列資訊： `configurationUpdateEvent` (Python：`configuration_update_event`) 組態更新事件。此物件 `ConfigurationUpdateEvent`包含下列資訊： `componentName` (Python：`component_name`) 元件的名稱。 `keyPath` (Python：`key_path`) 已更新之組態值的金鑰路徑。 ### 範例下列範例示範如何在自訂元件程式碼中呼叫此操作。 ------ #### [ Rust ] **Example 範例：訂閱組態更新** ``` use gg_sdk::Sdk; use std::{thread, time::Duration}; fn main() { let sdk = Sdk::init(); sdk.connect().expect("Failed to establish IPC connection"); // Subscribe to configuration updates for key path ["mqtt"] let callback = |component_name: &str, key_path: &[&str]| { println!( "Received configuration update for component: {component_name}" ); println!("Key path: {key_path:?}"); }; let _sub = sdk .subscribe_to_configuration_update(None, &["mqtt"], &callback) .expect("Failed to subscribe to configuration updates"); println!("Successfully subscribed to configuration updates."); // Keep the main thread alive, or the process will exit. loop { thread::sleep(Duration::from_secs(10)); } } ``` ------ #### [ C ] **Example 範例：訂閱組態更新** ``` #include #include #include #include #include #include #include static void on_subscription_response( void *ctx, GgBuffer component_name, GgList key_path, GgIpcSubscriptionHandle handle ) { (void) ctx; (void) handle; printf( "Received configuration update for component: %.*s\n", (int) component_name.len, component_name.data ); printf("Key path: ["); for (size_t i = 0; i < key_path.len; i++) { if (i > 0) { printf(", "); } GgObject *obj = &key_path.items[i]; if (gg_obj_type(*obj) == GG_TYPE_BUF) { GgBuffer key = gg_obj_into_buf(*obj); printf("\"%.*s\"", (int) key.len, key.data); } } printf("]\n"); } int main(void) { gg_sdk_init(); GgError err = ggipc_connect(); if (err != GG_ERR_OK) { fprintf(stderr, "Failed to establish IPC connection.\n"); exit(-1); } // Subscribe to configuration updates for key path ["mqtt"] GgIpcSubscriptionHandle handle; err = ggipc_subscribe_to_configuration_update( NULL, // component_name (NULL = current component) GG_BUF_LIST(GG_STR("mqtt")), on_subscription_response, NULL, &handle ); if (err != GG_ERR_OK) { fprintf(stderr, "Failed to subscribe to configuration updates.\n"); exit(-1); } printf("Successfully subscribed to configuration updates.\n"); // Keep the main thread alive, or the process will exit. while (1) { sleep(10); } // To stop subscribing, close the stream. ggipc_close_subscription(handle); } ``` ------ #### [ C\$1\$1 (Component SDK) ] **Example 範例：訂閱組態更新** ``` #include #include #include class ResponseHandler : public gg::ipc::ConfigurationUpdateCallback { void operator()( std::string_view component_name, gg::List key_path, gg::ipc::Subscription &handle ) override { (void) handle; std::cout << "Received configuration update for component: " << component_name << "\n"; std::cout << "Key path: ["; for (size_t i = 0; i < key_path.size(); i++) { if (i > 0) { std::cout << ", "; } std::cout << "\"" << get(key_path[i]) << "\""; } std::cout << "]\n"; } }; int main() { auto &client = gg::ipc::Client::get(); auto error = client.connect(); if (error) { std::cerr << "Failed to establish IPC connection.\n"; exit(-1); } // Subscribe to configuration updates for key path ["mqtt"] std::array key_path = { gg::Buffer { "mqtt" } }; static ResponseHandler handler; error = client.subscribe_to_configuration_update( key_path, std::nullopt, handler ); if (error) { std::cerr << "Failed to subscribe to configuration updates.\n"; exit(-1); } std::cout << "Successfully subscribed to configuration updates.\n"; // Keep the main thread alive, or the process will exit. while (1) { sleep(10); } } ``` ------ ## SubscribeToValidateConfigurationUpdates 訂閱以在此元件的組態更新之前接收通知。這可讓元件驗證其自身組態的更新。使用 [SendConfigurationValidityReport](#ipc-operation-sendconfigurationvalidityreport)操作來告知核心組態是否有效。 **重要** 本機部署不會通知更新元件。此操作是您訂閱事件訊息串流的訂閱操作。若要使用此操作，請使用處理事件訊息、錯誤和串流關閉的函數來定義串流回應處理常式。如需詳細資訊，請參閱[訂閱 IPC 事件串流](interprocess-communication.md#ipc-subscribe-operations)。 **事件訊息類型：** `ValidateConfigurationUpdateEvents` ### 請求此操作的請求沒有任何參數。 ### 回應此操作的回應包含下列資訊： `messages` 通知訊息的串流。此物件 `ValidateConfigurationUpdateEvents`包含下列資訊： `validateConfigurationUpdateEvent` (Python：`validate_configuration_update_event`) 組態更新事件。此物件 `ValidateConfigurationUpdateEvent`包含下列資訊： `deploymentId` (Python：`deployment_id`) 更新元件的 AWS IoT Greengrass 部署 ID。 `configuration` 包含新組態的物件。 ## SendConfigurationValidityReport 告知核心此元件的組態更新是否有效。如果您告訴 nucleus 新組態無效，則部署會失敗。使用 [SubscribeToValidateConfigurationUpdates](#ipc-operation-subscribetovalidateconfigurationupdates)操作來訂閱以驗證組態更新。如果元件未回應驗證組態更新通知，則 nucleus 會等待您在部署的組態驗證政策中指定的時間量。逾時之後，核會繼續進行部署。預設元件驗證逾時為 20 秒。如需詳細資訊，請參閱 [建立部署](create-deployments.md)和您可以在呼叫 [CreateDeployment](https://docs.aws.amazon.com/greengrass/v2/APIReference/API_CreateDeployment.html) 操作時提供的 [DeploymentConfigurationValidationPolicy](https://docs.aws.amazon.com/greengrass/v2/APIReference/API_DeploymentConfigurationValidationPolicy.html) 物件。 ### 請求此操作的請求具有下列參數： `configurationValidityReport` (Python：`configuration_validity_report`) 此報告會告知 nucleus 組態更新是否有效。此物件 `ConfigurationValidityReport`包含下列資訊： `status` 有效性狀態。此列舉 `ConfigurationValidityStatus`具有下列值： + `ACCEPTED` – 組態有效，且核可以套用到此元件。 + `REJECTED` – 組態無效且部署失敗。 `deploymentId` (Python：`deployment_id`) 請求組態更新的 AWS IoT Greengrass 部署 ID。 `message` （選用）報告組態為何無效的訊息。 ### 回應此操作不會在其回應中提供任何資訊。 # 擷取秘密值使用 Secret Manager IPC 服務，從核心裝置上的秘密擷取秘密值。您可以使用[秘密管理員元件](secret-manager-component.md)，將加密的秘密部署到核心裝置。然後，您可以使用 IPC 操作來解密秘密，並在自訂元件中使用其值。 **Topics** + [最低 SDK 版本](#ipc-secret-manager-sdk-versions) + [Authorization](#ipc-secret-manager-authorization) + [GetSecretValue](#ipc-operation-getsecretvalue) + [範例](#ipc-secret-manager-examples) ## 最低 SDK 版本下表列出 AWS IoT Device SDK 您必須用來從核心裝置上的秘密擷取秘密值的最低版本。 | SDK | 最低版本 | | --- | --- | | [AWS IoT Device SDK for Java v2](https://github.com/aws/aws-iot-device-sdk-java-v2) | v1.2.10 | | [AWS IoT Device SDK for Python v2](https://github.com/aws/aws-iot-device-sdk-python-v2) | 1.5.3 版 | | [AWS IoT Device SDK 適用於 C\$1\$1 v2](https://github.com/aws/aws-iot-device-sdk-cpp-v2) | 1.17.0 版 | | [AWS IoT Device SDK for JavaScript v2](https://github.com/aws/aws-iot-device-sdk-js-v2) | v1.12.0 | ## Authorization 若要在自訂元件中使用秘密管理員，您必須定義授權政策，允許元件取得存放在核心裝置上的秘密值。如需定義授權政策的資訊，請參閱 [授權元件執行 IPC 操作](interprocess-communication.md#ipc-authorization-policies)。秘密管理員的授權政策具有下列屬性。 **IPC 服務識別符：** `aws.greengrass.SecretManager` | 作業 | Description | Resources | | --- | --- | --- | | `aws.greengrass#GetSecretValue` 或 `*` | 允許元件取得在核心裝置上加密的秘密值。 | Secrets Manager 秘密 ARN，或`*`允許存取所有秘密。 | ### 授權政策範例您可以參考下列授權政策範例，協助您設定元件的授權政策。 **Example 範例授權政策** 下列範例授權政策允許元件取得核心裝置上任何秘密的值。我們建議您在生產環境中減少授權政策的範圍，以便元件僅擷取其使用的秘密。部署元件時，您可以將`*`萬用字元變更為秘密 ARNs 清單。 ``` { "accessControl": { "aws.greengrass.SecretManager": { "com.example.MySecretComponent:secrets:1": { "policyDescription": "Allows access to a secret.", "operations": [ "aws.greengrass#GetSecretValue" ], "resources": [ "*" ] } } } } ``` ## GetSecretValue 取得您存放在核心裝置上的秘密值。此操作類似於 Secrets Manager 操作，您可以用來取得中的秘密值 AWS 雲端。如需詳細資訊，請參閱 *AWS Secrets Manager API 參考*中的 [GetSecretValue](https://docs.aws.amazon.com/secretsmanager/latest/apireference/API_GetSecretValue.html)。 ### 請求此操作的請求具有下列參數： `refresh` (Python：`refresh`) （選用）：是否要將請求的秘密與來自 AWS Secrets Manager 服務的最新值同步。當設定為 true 時，秘密管理員會請求 AWS Secrets Manager 服務以取得指定秘密標籤的最新值，並將該值傳回為回應。否則，將會傳回儲存在本機的秘密值。此參數無法與請求中的`versionId`參數搭配使用。此參數可搭配 Nucleus 2.13.0 及更高版本使用。 `secretId` (Python：`secret_id`) 要取得的秘密名稱。您可以指定 Amazon Resource Name (ARN) 或秘密的易記名稱。 `versionId` (Python：`version_id`) （選用）要取得的版本 ID。您可指定為 `versionId` 或 `versionStage`。如果您未指定 `versionId`或 `versionStage`，則此操作預設為具有 `AWSCURRENT`標籤的版本。 `versionStage` (Python：`version_stage`) （選用）要取得之版本的預備標籤。您可指定為 `versionId` 或 `versionStage`。如果您未指定 `versionId`或 `versionStage`，則此操作預設為具有 `AWSCURRENT`標籤的版本。 ### 回應此操作的回應包含下列資訊： `secretId` (Python：`secret_id`) 秘密的 ID。 `versionId` (Python：`version_id`) 此版本秘密的 ID。 `versionStage` (Python：`version_stage`) 連接到此版本秘密的預備標籤清單。 `secretValue` (Python：`secret_value`) 此版本秘密的值。此物件 `SecretValue`包含下列資訊。 `secretString` (Python：`secret_string`) 您以字串形式提供給 Secrets Manager 的受保護秘密資訊的解密部分。 `secretBinary` (Python：`secret_binary`) （選用）您以位元組陣列形式提供給 Secrets Manager 做為二進位資料的受保護秘密資訊的解密部分。此屬性包含二進位資料做為 base64 編碼字串。如果您在 Secrets Manager 主控台中建立秘密，則不會使用此屬性。 ### 範例下列範例示範如何在自訂元件程式碼中呼叫此操作。 ------ #### [ Java (IPC client V1) ] **Example 範例：取得秘密值** 此範例使用 `IPCUtils`類別來建立與 AWS IoT Greengrass Core IPC 服務的連線。如需詳細資訊，請參閱[連線至 AWS IoT Greengrass Core IPC 服務](interprocess-communication.md#ipc-service-connect)。 ``` package com.aws.greengrass.docs.samples.ipc; import com.aws.greengrass.docs.samples.ipc.util.IPCUtils; import software.amazon.awssdk.aws.greengrass.GetSecretValueResponseHandler; import software.amazon.awssdk.aws.greengrass.GreengrassCoreIPCClient; import software.amazon.awssdk.aws.greengrass.model.GetSecretValueRequest; import software.amazon.awssdk.aws.greengrass.model.GetSecretValueResponse; import software.amazon.awssdk.aws.greengrass.model.UnauthorizedError; import software.amazon.awssdk.eventstreamrpc.EventStreamRPCConnection; import java.util.Optional; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; public class GetSecretValue { public static final int TIMEOUT_SECONDS = 10; public static void main(String[] args) { String secretArn = args[0]; String versionStage = args[1]; try (EventStreamRPCConnection eventStreamRPCConnection = IPCUtils.getEventStreamRpcConnection()) { GreengrassCoreIPCClient ipcClient = new GreengrassCoreIPCClient(eventStreamRPCConnection); GetSecretValueResponseHandler responseHandler = GetSecretValue.getSecretValue(ipcClient, secretArn, versionStage); CompletableFuture futureResponse = responseHandler.getResponse(); try { GetSecretValueResponse response = futureResponse.get(TIMEOUT_SECONDS, TimeUnit.SECONDS); response.getSecretValue().postFromJson(); String secretString = response.getSecretValue().getSecretString(); System.out.println("Successfully retrieved secret value: " + secretString); } catch (TimeoutException e) { System.err.println("Timeout occurred while retrieving secret: " + secretArn); } catch (ExecutionException e) { if (e.getCause() instanceof UnauthorizedError) { System.err.println("Unauthorized error while retrieving secret: " + secretArn); } else { throw e; } } } catch (InterruptedException e) { System.out.println("IPC interrupted."); } catch (ExecutionException e) { System.err.println("Exception occurred when using IPC."); e.printStackTrace(); System.exit(1); } } public static GetSecretValueResponseHandler getSecretValue(GreengrassCoreIPCClient greengrassCoreIPCClient, String secretArn, String versionStage) { GetSecretValueRequest getSecretValueRequest = new GetSecretValueRequest(); getSecretValueRequest.setSecretId(secretArn); getSecretValueRequest.setVersionStage(versionStage); return greengrassCoreIPCClient.getSecretValue(getSecretValueRequest, Optional.empty()); } } ``` ------ #### [ Python (IPC client V1) ] **Example 範例：取得秘密值** 此範例假設您使用 AWS IoT Device SDK 適用於 Python v2 的 1.5.4 版或更新版本。 ``` import json import awsiot.greengrasscoreipc from awsiot.greengrasscoreipc.model import ( GetSecretValueRequest, GetSecretValueResponse, UnauthorizedError ) secret_id = 'arn:aws:secretsmanager:us-west-2:123456789012:secret:MyGreengrassSecret-abcdef' TIMEOUT = 10 ipc_client = awsiot.greengrasscoreipc.connect() request = GetSecretValueRequest() request.secret_id = secret_id request.version_stage = 'AWSCURRENT' operation = ipc_client.new_get_secret_value() operation.activate(request) future_response = operation.get_response() response = future_response.result(TIMEOUT) secret_json = json.loads(response.secret_value.secret_string) # Handle secret value. ``` ------ #### [ JavaScript ] **Example 範例：取得秘密值** ``` import { GetSecretValueRequest, } from 'aws-iot-device-sdk-v2/dist/greengrasscoreipc/model'; import * as greengrasscoreipc from "aws-iot-device-sdk-v2/dist/greengrasscoreipc"; class GetSecretValue { private readonly secretId : string; private readonly versionStage : string; private ipcClient : greengrasscoreipc.Client constructor() { this.secretId = "" this.versionStage = "" this.getSecretValue().then(r => console.log("Started workflow")); } private async getSecretValue() { try { this.ipcClient = await getIpcClient(); const getSecretValueRequest : GetSecretValueRequest = { secretId: this.secretId, versionStage: this.versionStage, }; const result = await this.ipcClient.getSecretValue(getSecretValueRequest); const secretString = result.secretValue.secretString; console.log("Successfully retrieved secret value: " + secretString) } catch (e) { // parse the error depending on your use cases throw e } } } export async function getIpcClient(){ try { const ipcClient = greengrasscoreipc.createClient(); await ipcClient.connect() .catch(error => { // parse the error depending on your use cases throw error; }); return ipcClient } catch (err) { // parse the error depending on your use cases throw err } } const getSecretValue = new GetSecretValue(); ``` ------ ## 範例使用下列範例來了解如何在元件中使用秘密管理員 IPC 服務。 ### 範例：列印秘密 (Python、IPC 用戶端 V1) 此範例元件會列印您部署到核心裝置的秘密值。 **重要** 此範例元件會列印秘密的值，因此只能與存放測試資料的秘密搭配使用。請勿使用此元件來列印存放重要資訊的秘密值。 **Topics** + [Recipe](#ipc-secret-manager-example-print-secret-python-recipe) + [成品](#ipc-secret-manager-example-print-secret-python-artifacts) + [Usage](#ipc-secret-manager-example-print-secret-python-usage) #### Recipe 下列範例配方會定義秘密 ARN 組態參數，並允許元件取得核心裝置上任何秘密的值。 **注意** 我們建議您在生產環境中減少授權政策的範圍，以便元件僅擷取其使用的秘密。部署元件時，您可以將`*`萬用字元變更為秘密 ARNs 清單。 ------ #### [ JSON ] ``` { "RecipeFormatVersion": "2020-01-25", "ComponentName": "com.example.PrintSecret", "ComponentVersion": "1.0.0", "ComponentDescription": "Prints the value of an AWS Secrets Manager secret.", "ComponentPublisher": "Amazon", "ComponentDependencies": { "aws.greengrass.SecretManager": { "VersionRequirement": "^2.0.0", "DependencyType": "HARD" } }, "ComponentConfiguration": { "DefaultConfiguration": { "SecretArn": "", "accessControl": { "aws.greengrass.SecretManager": { "com.example.PrintSecret:secrets:1": { "policyDescription": "Allows access to a secret.", "operations": [ "aws.greengrass#GetSecretValue" ], "resources": [ "*" ] } } } } }, "Manifests": [ { "Platform": { "os": "linux" }, "Lifecycle": { "install": "python3 -m pip install --user awsiotsdk", "Run": "python3 -u {artifacts:path}/print_secret.py \"{configuration:/SecretArn}\"" } }, { "Platform": { "os": "windows" }, "Lifecycle": { "install": "py -3 -m pip install --user awsiotsdk", "Run": "py -3 -u {artifacts:path}/print_secret.py \"{configuration:/SecretArn}\"" } } ] } ``` ------ #### [ YAML ] ``` --- RecipeFormatVersion: '2020-01-25' ComponentName: com.example.PrintSecret ComponentVersion: 1.0.0 ComponentDescription: Prints the value of a Secrets Manager secret. ComponentPublisher: Amazon ComponentDependencies: aws.greengrass.SecretManager: VersionRequirement: "^2.0.0" DependencyType: HARD ComponentConfiguration: DefaultConfiguration: SecretArn: '' accessControl: aws.greengrass.SecretManager: com.example.PrintSecret:secrets:1: policyDescription: Allows access to a secret. operations: - aws.greengrass#GetSecretValue resources: - "*" Manifests: - Platform: os: linux Lifecycle: install: python3 -m pip install --user awsiotsdk Run: python3 -u {artifacts:path}/print_secret.py "{configuration:/SecretArn}" - Platform: os: windows Lifecycle: install: py -3 -m pip install --user awsiotsdk Run: py -3 -u {artifacts:path}/print_secret.py "{configuration:/SecretArn}" ``` ------ #### 成品下列範例 Python 應用程式示範如何使用 Secret Manager IPC 服務來取得核心裝置上的秘密值。 ``` import concurrent.futures import json import sys import traceback import awsiot.greengrasscoreipc from awsiot.greengrasscoreipc.model import ( GetSecretValueRequest, GetSecretValueResponse, UnauthorizedError ) TIMEOUT = 10 if len(sys.argv) == 1: print('Provide SecretArn in the component configuration.', file=sys.stdout) exit(1) secret_id = sys.argv[1] try: ipc_client = awsiot.greengrasscoreipc.connect() request = GetSecretValueRequest() request.secret_id = secret_id operation = ipc_client.new_get_secret_value() operation.activate(request) future_response = operation.get_response() try: response = future_response.result(TIMEOUT) secret_json = json.loads(response.secret_value.secret_string) print('Successfully got secret: ' + secret_id) print('Secret value: ' + str(secret_json)) except concurrent.futures.TimeoutError: print('Timeout occurred while getting secret: ' + secret_id, file=sys.stderr) except UnauthorizedError as e: print('Unauthorized error while getting secret: ' + secret_id, file=sys.stderr) raise e except Exception as e: print('Exception while getting secret: ' + secret_id, file=sys.stderr) raise e except Exception: print('Exception occurred when using IPC.', file=sys.stderr) traceback.print_exc() exit(1) ``` #### Usage 您可以使用此範例元件搭配[秘密管理員元件](secret-manager-component.md)，在核心裝置上部署和列印秘密的值。 **建立、部署和列印測試秘密** 1. 使用測試資料建立 Secrets Manager 秘密。 ------ #### [ Linux or Unix ] ``` aws secretsmanager create-secret \ --name MyTestGreengrassSecret \ --secret-string '{"my-secret-key": "my-secret-value"}' ``` ------ #### [ Windows Command Prompt (CMD) ] ``` aws secretsmanager create-secret ^ --name MyTestGreengrassSecret ^ --secret-string '{"my-secret-key": "my-secret-value"}' ``` ------ #### [ PowerShell ] ``` aws secretsmanager create-secret ` --name MyTestGreengrassSecret ` --secret-string '{"my-secret-key": "my-secret-value"}' ``` ------ 儲存秘密的 ARN，以便在下列步驟中使用。如需詳細資訊，請參閱*AWS Secrets Manager 《使用者指南*》中的[建立秘密](https://docs.aws.amazon.com/secretsmanager/latest/userguide/manage_create-basic-secret.html)。 1. 使用下列組態合併更新來部署[秘密管理員元件](secret-manager-component.md) (`aws.greengrass.SecretManager`)。指定您先前建立之秘密的 ARN。 ``` { "cloudSecrets": [ { "arn": "arn:aws:secretsmanager:us-west-2:123456789012:secret:MyTestGreengrassSecret-abcdef" } ] } ``` 如需詳細資訊，請參閱 [將 AWS IoT Greengrass 元件部署至裝置](manage-deployments.md)或 [Greengrass CLI 部署命令](gg-cli-deployment.md)。 1. 使用下列組態合併更新，在本節中建立和部署範例元件。指定您先前建立之秘密的 ARN。 ``` { "SecretArn": "arn:aws:secretsmanager:us-west-2:123456789012:secret:MyTestGreengrassSecret", "accessControl": { "aws.greengrass.SecretManager": { "com.example.PrintSecret:secrets:1": { "policyDescription": "Allows access to a secret.", "operations": [ "aws.greengrass#GetSecretValue" ], "resources": [ "arn:aws:secretsmanager:us-west-2:123456789012:secret:MyTestGreengrassSecret-abcdef" ] } } } } ``` 如需詳細資訊，請參閱[建立 AWS IoT Greengrass 元件](create-components.md) 1. 檢視 AWS IoT Greengrass 核心軟體日誌以驗證部署是否成功，並檢視`com.example.PrintSecret`元件日誌以查看列印的秘密值。如需詳細資訊，請參閱[監控 AWS IoT Greengrass 日誌](monitor-logs.md)。 # 與本機陰影互動使用陰影 IPC 服務與裝置上的本機陰影互動。您選擇要互動的裝置可以是您的核心裝置或連線的用戶端裝置。若要使用這些 IPC 操作，請在自訂[元件中包含影子管理員](shadow-manager-component.md)元件做為相依性。然後，您可以在自訂元件中使用 IPC 操作，透過影子管理員與裝置上的本機影子互動。若要讓自訂元件對本機影子狀態的變更做出反應，您也可以使用發佈/訂閱 IPC 服務來訂閱影子事件。如需使用發佈/訂閱服務的詳細資訊，請參閱 [發佈/訂閱本機訊息](ipc-publish-subscribe.md)。 **注意** 若要讓核心裝置與用戶端裝置影子互動，您還必須設定和部署 MQTT 橋接器元件。如需詳細資訊，請參閱[啟用影子管理員以與用戶端裝置通訊](work-with-client-device-shadows.md)。 **Topics** + [最低 SDK 版本](#ipc-local-shadows-sdk-versions) + [Authorization](#ipc-local-shadow-authorization) + [GetThingShadow](#ipc-operation-getthingshadow) + [UpdateThingShadow](#ipc-operation-updatethingshadow) + [DeleteThingShadow](#ipc-operation-deletethingshadow) + [ListNamedShadowsForThing](#ipc-operation-listnamedshadowsforthing) ## 最低 SDK 版本下表列出 AWS IoT Device SDK 您必須用來與本機影子互動的最低版本。 | SDK | 最低版本 | | --- | --- | | [AWS IoT Device SDK for Java v2](https://github.com/aws/aws-iot-device-sdk-java-v2) | 1.4.0 版 | | [AWS IoT Device SDK for Python v2](https://github.com/aws/aws-iot-device-sdk-python-v2) | v1.6.0 | | [AWS IoT Device SDK 適用於 C\$1\$1 v2](https://github.com/aws/aws-iot-device-sdk-cpp-v2) | 1.17.0 版 | | [AWS IoT Device SDK for JavaScript v2](https://github.com/aws/aws-iot-device-sdk-js-v2) | v1.12.0 | ## Authorization 若要在自訂元件中使用陰影 IPC 服務，您必須定義授權政策，允許您的元件與陰影互動。如需定義授權政策的資訊，請參閱 [授權元件執行 IPC 操作](interprocess-communication.md#ipc-authorization-policies)。陰影互動的授權政策具有下列屬性。 **IPC 服務識別符：** `aws.greengrass.ShadowManager` | 作業 | Description | Resources | | --- | --- | --- | | `aws.greengrass#GetThingShadow` | 允許元件擷取物件的陰影。 | 下列其中一個字串： [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/zh_tw/greengrass/v2/developerguide/ipc-local-shadows.html) | | `aws.greengrass#UpdateThingShadow` | 允許元件更新物件的陰影。 | 下列其中一個字串： [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/zh_tw/greengrass/v2/developerguide/ipc-local-shadows.html) | | `aws.greengrass#DeleteThingShadow` | 允許元件刪除物件的陰影。 | 下列其中一個字串： [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/zh_tw/greengrass/v2/developerguide/ipc-local-shadows.html) | | `aws.greengrass#ListNamedShadowsForThing` | 允許元件擷取物件的具名影子清單。 | 物件名稱字串，允許存取物件以列出其影子。使用 `*` 允許存取所有物件。 | **IPC 服務識別符：** `aws.greengrass.ipc.pubsub` | 作業 | Description | Resources | | --- | --- | --- | | `aws.greengrass#SubscribeToTopic` | 允許元件訂閱您指定主題的訊息。 | 下列其中一個主題字串： [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/zh_tw/greengrass/v2/developerguide/ipc-local-shadows.html) 主題字首的值`shadowTopicPrefix`取決於影子的類型： [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/zh_tw/greengrass/v2/developerguide/ipc-local-shadows.html) 使用 `*` 允許存取所有主題。在 [Greengrass nucleus](greengrass-nucleus-component.md) v2.6.0 和更新版本中，您可以訂閱包含 MQTT 主題萬用字元 (`#` 和 `+`) 的主題。此主題字串支援 MQTT 主題萬用字元做為文字字元。例如，如果元件的授權政策授予的存取權`test/topic/#`，則該元件可以訂閱 `test/topic/#`，但無法訂閱 `test/topic/filter`。 | ### 本機影子授權政策中的配方變數如果您使用 [Greengrass 核](greengrass-nucleus-component.md)的 v2.6.0 或更新版本，並將 Greengrass 核的 [interpolateComponentConfiguration](greengrass-nucleus-component.md#greengrass-nucleus-component-configuration-interpolate-component-configuration) 組態選項設定為 `true`，您可以在授權政策中使用`{iot:thingName}`[配方變數](component-recipe-reference.md#recipe-variables)。此功能可讓您為一組核心裝置設定單一授權政策，其中每個核心裝置只能存取自己的影子。例如，您可以允許元件存取陰影 IPC 操作的下列資源。 ``` $aws/things/{iot:thingName}/shadow/ ``` ### 授權政策範例您可以參考下列授權政策範例，協助您設定元件的授權政策。 **Example 範例：允許一組核心裝置與本機影子互動** 此範例使用適用於 [Greengrass 核元件](greengrass-nucleus-component.md) v2.6.0 和更新版本的功能。Greengrass 核 v2.6.0 在元件組態中新增了對大多數[配方變數](component-recipe-reference.md#recipe-variables)的支援`{iot:thingName}`，例如。若要啟用此功能，請將 Greengrass 核的 [interpolateComponentConfiguration](greengrass-nucleus-component.md#greengrass-nucleus-component-configuration-interpolate-component-configuration) 組態選項設定為 `true`。如需適用於所有 Greengrass 核版本的範例，請參閱[單一核心裝置的範例授權政策](#ipc-local-shadows-authorization-example-single-device)。下列範例授權政策允許元件與傳統裝置影子和執行元件`myNamedShadow`之核心裝置的具名影子`com.example.MyShadowInteractionComponent`互動。此政策也允許此元件接收有關這些影子之本機主題的訊息。 ``` { "accessControl": { "aws.greengrass.ShadowManager": { "com.example.MyShadowInteractionComponent:shadow:1": { "policyDescription": "Allows access to shadows", "operations": [ "aws.greengrass#GetThingShadow", "aws.greengrass#UpdateThingShadow", "aws.greengrass#DeleteThingShadow" ], "resources": [ "$aws/things/{iot:thingName}/shadow", "$aws/things/{iot:thingName}/shadow/name/myNamedShadow" ] }, "com.example.MyShadowInteractionComponent:shadow:2": { "policyDescription": "Allows access to things with shadows", "operations": [ "aws.greengrass#ListNamedShadowsForThing" ], "resources": [ "{iot:thingName}" ] } }, "aws.greengrass.ipc.pubsub": { "com.example.MyShadowInteractionComponent:pubsub:1": { "policyDescription": "Allows access to shadow pubsub topics", "operations": [ "aws.greengrass#SubscribeToTopic" ], "resources": [ "$aws/things/{iot:thingName}/shadow/get/accepted", "$aws/things/{iot:thingName}/shadow/name/myNamedShadow/get/accepted" ] } } } } ``` ``` accessControl: aws.greengrass.ShadowManager: 'com.example.MyShadowInteractionComponent:shadow:1': policyDescription: 'Allows access to shadows' operations: - 'aws.greengrass#GetThingShadow' - 'aws.greengrass#UpdateThingShadow' - 'aws.greengrass#DeleteThingShadow' resources: - $aws/things/{iot:thingName}/shadow - $aws/things/{iot:thingName}/shadow/name/myNamedShadow 'com.example.MyShadowInteractionComponent:shadow:2': policyDescription: 'Allows access to things with shadows' operations: - 'aws.greengrass#ListNamedShadowsForThing' resources: - '{iot:thingName}' aws.greengrass.ipc.pubsub: 'com.example.MyShadowInteractionComponent:pubsub:1': policyDescription: 'Allows access to shadow pubsub topics' operations: - 'aws.greengrass#SubscribeToTopic' resources: - $aws/things/{iot:thingName}/shadow/get/accepted - $aws/things/{iot:thingName}/shadow/name/myNamedShadow/get/accepted ``` **Example 範例：允許一組核心裝置與用戶端裝置影子互動** 此功能需要 [Greengrass 核](greengrass-nucleus-component.md) v2.6.0 或更新版本、[陰影管理員](shadow-manager-component.md) v2.2.0 或更新版本，以及 [MQTT 橋接](mqtt-bridge-component.md)器 v2.2.0 或更新版本。您必須設定 MQTT 橋接器，[讓陰影管理員能夠與用戶端裝置通訊](work-with-client-device-shadows.md#enable-shadow-manager-client-devices)。下列範例授權政策允許元件與名稱開頭為的用戶端裝置的所有裝置影子`com.example.MyShadowInteractionComponent`互動`MyClientDevice`。若要讓核心裝置與用戶端裝置影子互動，您還必須設定和部署 MQTT 橋接器元件。如需詳細資訊，請參閱[啟用影子管理員以與用戶端裝置通訊](work-with-client-device-shadows.md)。 ``` { "accessControl": { "aws.greengrass.ShadowManager": { "com.example.MyShadowInteractionComponent:shadow:1": { "policyDescription": "Allows access to shadows", "operations": [ "aws.greengrass#GetThingShadow", "aws.greengrass#UpdateThingShadow", "aws.greengrass#DeleteThingShadow" ], "resources": [ "$aws/things/MyClientDevice*/shadow", "$aws/things/MyClientDevice*/shadow/name/*" ] }, "com.example.MyShadowInteractionComponent:shadow:2": { "policyDescription": "Allows access to things with shadows", "operations": [ "aws.greengrass#ListNamedShadowsForThing" ], "resources": [ "MyClientDevice*" ] } } } } ``` ``` accessControl: aws.greengrass.ShadowManager: 'com.example.MyShadowInteractionComponent:shadow:1': policyDescription: 'Allows access to shadows' operations: - 'aws.greengrass#GetThingShadow' - 'aws.greengrass#UpdateThingShadow' - 'aws.greengrass#DeleteThingShadow' resources: - $aws/things/MyClientDevice*/shadow - $aws/things/MyClientDevice*/shadow/name/* 'com.example.MyShadowInteractionComponent:shadow:2': policyDescription: 'Allows access to things with shadows' operations: - 'aws.greengrass#ListNamedShadowsForThing' resources: - MyClientDevice* ``` **Example 範例：允許單一核心裝置與本機影子互動** 下列範例授權政策允許元件與傳統裝置影子和裝置 `myNamedShadow` 的具名影子`com.example.MyShadowInteractionComponent`互動`MyThingName`。此政策也允許此元件接收有關這些影子之本機主題的訊息。 ``` { "accessControl": { "aws.greengrass.ShadowManager": { "com.example.MyShadowInteractionComponent:shadow:1": { "policyDescription": "Allows access to shadows", "operations": [ "aws.greengrass#GetThingShadow", "aws.greengrass#UpdateThingShadow", "aws.greengrass#DeleteThingShadow" ], "resources": [ "$aws/things/MyThingName/shadow", "$aws/things/MyThingName/shadow/name/myNamedShadow" ] }, "com.example.MyShadowInteractionComponent:shadow:2": { "policyDescription": "Allows access to things with shadows", "operations": [ "aws.greengrass#ListNamedShadowsForThing" ], "resources": [ "MyThingName" ] } }, "aws.greengrass.ipc.pubsub": { "com.example.MyShadowInteractionComponent:pubsub:1": { "policyDescription": "Allows access to shadow pubsub topics", "operations": [ "aws.greengrass#SubscribeToTopic" ], "resources": [ "$aws/things/MyThingName/shadow/get/accepted", "$aws/things/MyThingName/shadow/name/myNamedShadow/get/accepted" ] } } } } ``` ``` accessControl: aws.greengrass.ShadowManager: 'com.example.MyShadowInteractionComponent:shadow:1': policyDescription: 'Allows access to shadows' operations: - 'aws.greengrass#GetThingShadow' - 'aws.greengrass#UpdateThingShadow' - 'aws.greengrass#DeleteThingShadow' resources: - $aws/things/MyThingName/shadow - $aws/things/MyThingName/shadow/name/myNamedShadow 'com.example.MyShadowInteractionComponent:shadow:2': policyDescription: 'Allows access to things with shadows' operations: - 'aws.greengrass#ListNamedShadowsForThing' resources: - MyThingName aws.greengrass.ipc.pubsub: 'com.example.MyShadowInteractionComponent:pubsub:1': policyDescription: 'Allows access to shadow pubsub topics' operations: - 'aws.greengrass#SubscribeToTopic' resources: - $aws/things/MyThingName/shadow/get/accepted - $aws/things/MyThingName/shadow/name/myNamedShadow/get/accepted ``` **Example 範例：允許一組核心裝置對本機陰影狀態變更做出反應** 此範例使用適用於 [Greengrass 核元件](greengrass-nucleus-component.md) v2.6.0 和更新版本的功能。Greengrass 核 v2.6.0 在元件組態中新增了對大多數[配方變數](component-recipe-reference.md#recipe-variables)的支援`{iot:thingName}`，例如。若要啟用此功能，請將 Greengrass 核的 [interpolateComponentConfiguration](greengrass-nucleus-component.md#greengrass-nucleus-component-configuration-interpolate-component-configuration) 組態選項設定為 `true`。如需適用於所有 Greengrass 核版本的範例，請參閱[單一核心裝置的範例授權政策](#interact-with-shadows-react-example-authorization-policy-single-device)。下列範例存取控制政策允許自訂`com.example.MyShadowReactiveComponent`接收傳統裝置影子`/update/delta`主題的訊息，以及`myNamedShadow`每個核心裝置上執行元件的具名影子。 ``` { "accessControl": { "aws.greengrass.ipc.pubsub": { "com.example.MyShadowReactiveComponent:pubsub:1": { "policyDescription": "Allows access to shadow pubsub topics", "operations": [ "aws.greengrass#SubscribeToTopic" ], "resources": [ "$aws/things/{iot:thingName}/shadow/update/delta", "$aws/things/{iot:thingName}/shadow/name/myNamedShadow/update/delta" ] } } } } ``` ``` accessControl: aws.greengrass.ipc.pubsub: "com.example.MyShadowReactiveComponent:pubsub:1": policyDescription: Allows access to shadow pubsub topics operations: - 'aws.greengrass#SubscribeToTopic' resources: - $aws/things/{iot:thingName}/shadow/update/delta - $aws/things/{iot:thingName}/shadow/name/myNamedShadow/update/delta ``` **Example 範例：允許單一核心裝置對本機陰影狀態變更做出反應** 下列範例存取控制政策允許自訂`com.example.MyShadowReactiveComponent`接收傳統裝置影子`/update/delta`主題的訊息，以及裝置 `myNamedShadow`的具名影子`MyThingName`。 ``` { "accessControl": { "aws.greengrass.ipc.pubsub": { "com.example.MyShadowReactiveComponent:pubsub:1": { "policyDescription": "Allows access to shadow pubsub topics", "operations": [ "aws.greengrass#SubscribeToTopic" ], "resources": [ "$aws/things/MyThingName/shadow/update/delta", "$aws/things/MyThingName/shadow/name/myNamedShadow/update/delta" ] } } } } ``` ``` accessControl: aws.greengrass.ipc.pubsub: "com.example.MyShadowReactiveComponent:pubsub:1": policyDescription: Allows access to shadow pubsub topics operations: - 'aws.greengrass#SubscribeToTopic' resources: - $aws/things/MyThingName/shadow/update/delta - $aws/things/MyThingName/shadow/name/myNamedShadow/update/delta ``` ## GetThingShadow 取得指定物件的陰影。 ### 請求此操作的請求具有下列參數： `thingName` (Python：`thing_name`) 物件名稱。類型：`string` `shadowName` (Python：`shadow_name`) 影子的名稱。若要指定物件的傳統陰影，請將此參數設定為空字串 (`""`)。 AWS IoT Greengrass 服務使用`AWSManagedGreengrassV2Deployment`具名影子來管理以個別核心裝置為目標的部署。此具名影子會保留供 AWS IoT Greengrass 服務使用。請勿更新或刪除此具名影子。類型：`string` ### 回應此操作的回應包含下列資訊： `payload` Blob 形式的回應狀態文件。類型：`object`包含下列資訊： `state` 狀態資訊。此物件包含下列資訊。 `desired` 在裝置中請求更新的狀態屬性和值。類型：`map`鍵/值對 `reported` 裝置報告的狀態屬性和值。類型：`map`鍵/值對 `delta` 所需和報告的狀態屬性和值之間的差異。只有在 `desired`和 `reported` 狀態不同時，才會顯示此屬性。類型：`map`鍵/值對 `metadata` `desired` 和 `reported`區段中每個屬性的時間戳記，讓您可以判斷狀態的更新時間。類型：`string` `timestamp` 產生回應的 epoch 日期和時間。類型：`integer` `clientToken` (Python：`clientToken`) 用於比對請求和對應回應的字符類型：`string` `version` 本機影子文件的版本。類型：`integer` ### 錯誤此操作可能會傳回下列錯誤。 `InvalidArgumentsError` 本機影子服務無法驗證請求參數。如果請求包含格式不正確的 JSON 或不支援的字元，就會發生這種情況。 `ResourceNotFoundError` 找不到請求的本機影子文件。 `ServiceError` 發生內部服務錯誤，或對 IPC 服務的請求數量超過影子管理員元件中 `maxLocalRequestsPerSecondPerThing`和 `maxTotalLocalRequestsRate`組態參數中指定的限制。 `UnauthorizedError` 元件的授權政策不包含此操作所需的許可。 ### 範例下列範例示範如何在自訂元件程式碼中呼叫此操作。 ------ #### [ Java (IPC client V1) ] **Example 範例：取得物件影子** 此範例使用 `IPCUtils`類別來建立與 AWS IoT Greengrass Core IPC 服務的連線。如需詳細資訊，請參閱[連線至 AWS IoT Greengrass Core IPC 服務](interprocess-communication.md#ipc-service-connect)。 ``` package com.aws.greengrass.docs.samples.ipc; import com.aws.greengrass.docs.samples.ipc.util.IPCUtils; import software.amazon.awssdk.aws.greengrass.GetThingShadowResponseHandler; import software.amazon.awssdk.aws.greengrass.GreengrassCoreIPCClient; import software.amazon.awssdk.aws.greengrass.model.GetThingShadowRequest; import software.amazon.awssdk.aws.greengrass.model.GetThingShadowResponse; import software.amazon.awssdk.aws.greengrass.model.ResourceNotFoundError; import software.amazon.awssdk.aws.greengrass.model.UnauthorizedError; import software.amazon.awssdk.eventstreamrpc.EventStreamRPCConnection; import java.nio.charset.StandardCharsets; import java.util.Optional; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; public class GetThingShadow { public static final int TIMEOUT_SECONDS = 10; public static void main(String[] args) { // Use the current core device's name if thing name isn't set. String thingName = args[0].isEmpty() ? System.getenv("AWS_IOT_THING_NAME") : args[0]; String shadowName = args[1]; try (EventStreamRPCConnection eventStreamRPCConnection = IPCUtils.getEventStreamRpcConnection()) { GreengrassCoreIPCClient ipcClient = new GreengrassCoreIPCClient(eventStreamRPCConnection); GetThingShadowResponseHandler responseHandler = GetThingShadow.getThingShadow(ipcClient, thingName, shadowName); CompletableFuture futureResponse = responseHandler.getResponse(); try { GetThingShadowResponse response = futureResponse.get(TIMEOUT_SECONDS, TimeUnit.SECONDS); String shadowPayload = new String(response.getPayload(), StandardCharsets.UTF_8); System.out.printf("Successfully got shadow %s/%s: %s%n", thingName, shadowName, shadowPayload); } catch (TimeoutException e) { System.err.printf("Timeout occurred while getting shadow: %s/%s%n", thingName, shadowName); } catch (ExecutionException e) { if (e.getCause() instanceof UnauthorizedError) { System.err.printf("Unauthorized error while getting shadow: %s/%s%n", thingName, shadowName); } else if (e.getCause() instanceof ResourceNotFoundError) { System.err.printf("Unable to find shadow to get: %s/%s%n", thingName, shadowName); } else { throw e; } } } catch (InterruptedException e) { System.out.println("IPC interrupted."); } catch (ExecutionException e) { System.err.println("Exception occurred when using IPC."); e.printStackTrace(); System.exit(1); } } public static GetThingShadowResponseHandler getThingShadow(GreengrassCoreIPCClient greengrassCoreIPCClient, String thingName, String shadowName) { GetThingShadowRequest getThingShadowRequest = new GetThingShadowRequest(); getThingShadowRequest.setThingName(thingName); getThingShadowRequest.setShadowName(shadowName); return greengrassCoreIPCClient.getThingShadow(getThingShadowRequest, Optional.empty()); } } ``` ------ #### [ Python (IPC client V1) ] **Example 範例：取得物件影子** ``` import awsiot.greengrasscoreipc import awsiot.greengrasscoreipc.client as client from awsiot.greengrasscoreipc.model import GetThingShadowRequest TIMEOUT = 10 def sample_get_thing_shadow_request(thingName, shadowName): try: # set up IPC client to connect to the IPC server ipc_client = awsiot.greengrasscoreipc.connect() # create the GetThingShadow request get_thing_shadow_request = GetThingShadowRequest() get_thing_shadow_request.thing_name = thingName get_thing_shadow_request.shadow_name = shadowName # retrieve the GetThingShadow response after sending the request to the IPC server op = ipc_client.new_get_thing_shadow() op.activate(get_thing_shadow_request) fut = op.get_response() result = fut.result(TIMEOUT) return result.payload except InvalidArgumentsError as e: # add error handling ... # except ResourceNotFoundError | UnauthorizedError | ServiceError ``` ------ #### [ JavaScript ] **Example 範例：取得物件影子** ``` import { GetThingShadowRequest } from 'aws-iot-device-sdk-v2/dist/greengrasscoreipc/model'; import * as greengrasscoreipc from 'aws-iot-device-sdk-v2/dist/greengrasscoreipc'; class GetThingShadow { private ipcClient: greengrasscoreipc.Client; private thingName: string; private shadowName: string; constructor() { // Define args parameters here this.thingName = ""; this.shadowName = ""; this.bootstrap(); } async bootstrap() { try { this.ipcClient = await getIpcClient(); } catch (err) { // parse the error depending on your use cases throw err } try { await this.handleGetThingShadowOperation(this.thingName, this.shadowName); } catch (err) { // parse the error depending on your use cases throw err } } async handleGetThingShadowOperation( thingName: string, shadowName: string ) { const request: GetThingShadowRequest = { thingName: thingName, shadowName: shadowName }; const response = await this.ipcClient.getThingShadow(request); } } export async function getIpcClient() { try { const ipcClient = greengrasscoreipc.createClient(); await ipcClient.connect() .catch(error => { // parse the error depending on your use cases throw error; }); return ipcClient } catch (err) { // parse the error depending on your use caseså throw err } } const startScript = new GetThingShadow(); ``` ------ ## UpdateThingShadow 更新指定物件的影子。如果影子不存在，則會建立一個影子。 ### 請求此操作的請求具有下列參數： `thingName` (Python：`thing_name`) 物件名稱。類型：`string` `shadowName` (Python：`shadow_name`) 影子的名稱。若要指定物件的傳統陰影，請將此參數設定為空字串 (`""`)。 AWS IoT Greengrass 服務使用`AWSManagedGreengrassV2Deployment`具名影子來管理以個別核心裝置為目標的部署。此具名影子會保留供 AWS IoT Greengrass 服務使用。請勿更新或刪除此具名影子。類型：`string` `payload` Blob 形式的請求狀態文件。類型：`object`包含下列資訊： `state` 要更新的狀態資訊。此 IPC 操作只會影響指定的欄位。此物件包含下列資訊。一般而言，您會在相同的請求中使用 `desired` 屬性或 `reported` 屬性，但不會同時使用兩者。 `desired` 在裝置中請求更新的狀態屬性和值。類型：`map`鍵/值對 `reported` 裝置報告的狀態屬性和值。類型：`map`鍵/值對 `clientToken` (Python：`client_token`) （選用）用於比對請求的字符，以及用戶端字符對應的回應。類型：`string` `version` （選用）要更新的本機影子文件版本。只有當指定的版本符合其擁有的最新版本時，陰影服務才會處理更新。類型：`integer` ### 回應此操作的回應包含下列資訊： `payload` Blob 形式的回應狀態文件。類型：`object`包含下列資訊： `state` 狀態資訊。此物件包含下列資訊。 `desired` 在裝置中請求更新的狀態屬性和值。類型：`map`鍵/值對 `reported` 裝置報告的狀態屬性和值。類型：`map`鍵/值對 `delta` 裝置報告的狀態屬性和值。類型：`map`鍵/值對 `metadata` `desired` 和 `reported`區段中每個屬性的時間戳記，讓您可以判斷狀態的更新時間。類型：`string` `timestamp` 產生回應的 epoch 日期和時間。類型：`integer` `clientToken` (Python：`client_token`) 用於比對請求和對應回應的字符。類型：`string` `version` 更新完成後的本機影子文件版本。類型：`integer` ### 錯誤此操作可能會傳回下列錯誤。 `ConflictError` 本機影子服務在更新操作期間遇到版本衝突。當請求承載中的版本與最新可用本機影子文件中的版本不相符時，就會發生這種情況。 `InvalidArgumentsError` 本機影子服務無法驗證請求參數。如果請求包含格式不正確的 JSON 或不支援的字元，就會發生這種情況。有效的 `payload`具有下列屬性： + `state` 節點存在，且是包含 `desired`或 `reported` 狀態資訊的物件。 + `desired` 和 `reported`節點是物件或 null。這些物件中至少有一個必須包含有效的狀態資訊。 + `desired` 和 `reported` 物件的深度不能超過八個節點。 + `clientToken` 值的長度不能超過 64 個字元。 + `version` 值必須為 `1` 或更高。 `ServiceError` 發生內部服務錯誤，或對 IPC 服務的請求數量超過影子管理員元件中 `maxLocalRequestsPerSecondPerThing`和 `maxTotalLocalRequestsRate`組態參數中指定的限制。 `UnauthorizedError` 元件的授權政策不包含此操作所需的許可。 ### 範例下列範例示範如何在自訂元件程式碼中呼叫此操作。 ------ #### [ Java (IPC client V1) ] **Example 範例：更新物件影子** 此範例使用 `IPCUtils`類別來建立與 AWS IoT Greengrass Core IPC 服務的連線。如需詳細資訊，請參閱[連線至 AWS IoT Greengrass Core IPC 服務](interprocess-communication.md#ipc-service-connect)。 ``` package com.aws.greengrass.docs.samples.ipc; import com.aws.greengrass.docs.samples.ipc.util.IPCUtils; import software.amazon.awssdk.aws.greengrass.GreengrassCoreIPCClient; import software.amazon.awssdk.aws.greengrass.UpdateThingShadowResponseHandler; import software.amazon.awssdk.aws.greengrass.model.UnauthorizedError; import software.amazon.awssdk.aws.greengrass.model.UpdateThingShadowRequest; import software.amazon.awssdk.aws.greengrass.model.UpdateThingShadowResponse; import software.amazon.awssdk.eventstreamrpc.EventStreamRPCConnection; import java.nio.charset.StandardCharsets; import java.util.Optional; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; public class UpdateThingShadow { public static final int TIMEOUT_SECONDS = 10; public static void main(String[] args) { // Use the current core device's name if thing name isn't set. String thingName = args[0].isEmpty() ? System.getenv("AWS_IOT_THING_NAME") : args[0]; String shadowName = args[1]; byte[] shadowPayload = args[2].getBytes(StandardCharsets.UTF_8); try (EventStreamRPCConnection eventStreamRPCConnection = IPCUtils.getEventStreamRpcConnection()) { GreengrassCoreIPCClient ipcClient = new GreengrassCoreIPCClient(eventStreamRPCConnection); UpdateThingShadowResponseHandler responseHandler = UpdateThingShadow.updateThingShadow(ipcClient, thingName, shadowName, shadowPayload); CompletableFuture futureResponse = responseHandler.getResponse(); try { futureResponse.get(TIMEOUT_SECONDS, TimeUnit.SECONDS); System.out.printf("Successfully updated shadow: %s/%s%n", thingName, shadowName); } catch (TimeoutException e) { System.err.printf("Timeout occurred while updating shadow: %s/%s%n", thingName, shadowName); } catch (ExecutionException e) { if (e.getCause() instanceof UnauthorizedError) { System.err.printf("Unauthorized error while updating shadow: %s/%s%n", thingName, shadowName); } else { throw e; } } } catch (InterruptedException e) { System.out.println("IPC interrupted."); } catch (ExecutionException e) { System.err.println("Exception occurred when using IPC."); e.printStackTrace(); System.exit(1); } } public static UpdateThingShadowResponseHandler updateThingShadow(GreengrassCoreIPCClient greengrassCoreIPCClient, String thingName, String shadowName, byte[] shadowPayload) { UpdateThingShadowRequest updateThingShadowRequest = new UpdateThingShadowRequest(); updateThingShadowRequest.setThingName(thingName); updateThingShadowRequest.setShadowName(shadowName); updateThingShadowRequest.setPayload(shadowPayload); return greengrassCoreIPCClient.updateThingShadow(updateThingShadowRequest, Optional.empty()); } } ``` ------ #### [ Python (IPC client V1) ] **Example 範例：更新物件影子** ``` import awsiot.greengrasscoreipc import awsiot.greengrasscoreipc.client as client from awsiot.greengrasscoreipc.model import UpdateThingShadowRequest TIMEOUT = 10 def sample_update_thing_shadow_request(thingName, shadowName, payload): try: # set up IPC client to connect to the IPC server ipc_client = awsiot.greengrasscoreipc.connect() # create the UpdateThingShadow request update_thing_shadow_request = UpdateThingShadowRequest() update_thing_shadow_request.thing_name = thingName update_thing_shadow_request.shadow_name = shadowName update_thing_shadow_request.payload = payload # retrieve the UpdateThingShadow response after sending the request to the IPC server op = ipc_client.new_update_thing_shadow() op.activate(update_thing_shadow_request) fut = op.get_response() result = fut.result(TIMEOUT) return result.payload except InvalidArgumentsError as e: # add error handling ... # except ConflictError | UnauthorizedError | ServiceError ``` ------ #### [ JavaScript ] **Example 範例：更新物件影子** ``` import { UpdateThingShadowRequest } from 'aws-iot-device-sdk-v2/dist/greengrasscoreipc/model'; import * as greengrasscoreipc from 'aws-iot-device-sdk-v2/dist/greengrasscoreipc'; class UpdateThingShadow { private ipcClient: greengrasscoreipc.Client; private thingName: string; private shadowName: string; private shadowDocumentStr: string; constructor() { // Define args parameters here this.thingName = ""; this.shadowName = ""; this.shadowDocumentStr = ""; this.bootstrap(); } async bootstrap() { try { this.ipcClient = await getIpcClient(); } catch (err) { // parse the error depending on your use cases throw err } try { await this.handleUpdateThingShadowOperation( this.thingName, this.shadowName, this.shadowDocumentStr); } catch (err) { // parse the error depending on your use cases throw err } } async handleUpdateThingShadowOperation( thingName: string, shadowName: string, payloadStr: string ) { const request: UpdateThingShadowRequest = { thingName: thingName, shadowName: shadowName, payload: payloadStr } // make the UpdateThingShadow request const response = await this.ipcClient.updateThingShadow(request); } } export async function getIpcClient() { try { const ipcClient = greengrasscoreipc.createClient(); await ipcClient.connect() .catch(error => { // parse the error depending on your use cases throw error; }); return ipcClient } catch (err) { // parse the error depending on your use cases throw err } } const startScript = new UpdateThingShadow(); ``` ------ ## DeleteThingShadow 刪除特定物件的影子。從影子管理員 v2.0.4 開始，刪除影子會增加版本編號。例如，當您刪除`MyThingShadow`第 1 版的影子時，已刪除影子的版本為 2。如果您接著重新建立名稱為的影子`MyThingShadow`，則該影子的版本為 3。 ### 請求此操作的請求具有下列參數： `thingName` (Python：`thing_name`) 物件名稱。類型：`string` `shadowName` (Python：`shadow_name`) 影子的名稱。若要指定物件的傳統陰影，請將此參數設定為空字串 (`""`)。 AWS IoT Greengrass 服務使用`AWSManagedGreengrassV2Deployment`具名影子來管理以個別核心裝置為目標的部署。此具名影子會保留供 AWS IoT Greengrass 服務使用。請勿更新或刪除此具名影子。類型：`string` ### 回應此操作的回應包含下列資訊： `payload` 空的回應狀態文件。 ### 錯誤此操作可能會傳回下列錯誤。 `InvalidArgumentsError` 本機影子服務無法驗證請求參數。如果請求包含格式不正確的 JSON 或不支援的字元，就會發生這種情況。 `ResourceNotFoundError` 找不到請求的本機影子文件。 `ServiceError` 發生內部服務錯誤，或對 IPC 服務的請求數量超過影子管理員元件中 `maxLocalRequestsPerSecondPerThing`和 `maxTotalLocalRequestsRate`組態參數中指定的限制。 `UnauthorizedError` 元件的授權政策不包含此操作所需的許可。 ### 範例下列範例示範如何在自訂元件程式碼中呼叫此操作。 ------ #### [ Java (IPC client V1) ] **Example 範例：刪除物件影子** 此範例使用 `IPCUtils`類別來建立與 AWS IoT Greengrass Core IPC 服務的連線。如需詳細資訊，請參閱[連線至 AWS IoT Greengrass Core IPC 服務](interprocess-communication.md#ipc-service-connect)。 ``` package com.aws.greengrass.docs.samples.ipc; import com.aws.greengrass.docs.samples.ipc.util.IPCUtils; import software.amazon.awssdk.aws.greengrass.DeleteThingShadowResponseHandler; import software.amazon.awssdk.aws.greengrass.GreengrassCoreIPCClient; import software.amazon.awssdk.aws.greengrass.model.DeleteThingShadowRequest; import software.amazon.awssdk.aws.greengrass.model.DeleteThingShadowResponse; import software.amazon.awssdk.aws.greengrass.model.ResourceNotFoundError; import software.amazon.awssdk.aws.greengrass.model.UnauthorizedError; import software.amazon.awssdk.eventstreamrpc.EventStreamRPCConnection; import java.util.Optional; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; public class DeleteThingShadow { public static final int TIMEOUT_SECONDS = 10; public static void main(String[] args) { // Use the current core device's name if thing name isn't set. String thingName = args[0].isEmpty() ? System.getenv("AWS_IOT_THING_NAME") : args[0]; String shadowName = args[1]; try (EventStreamRPCConnection eventStreamRPCConnection = IPCUtils.getEventStreamRpcConnection()) { GreengrassCoreIPCClient ipcClient = new GreengrassCoreIPCClient(eventStreamRPCConnection); DeleteThingShadowResponseHandler responseHandler = DeleteThingShadow.deleteThingShadow(ipcClient, thingName, shadowName); CompletableFuture futureResponse = responseHandler.getResponse(); try { futureResponse.get(TIMEOUT_SECONDS, TimeUnit.SECONDS); System.out.printf("Successfully deleted shadow: %s/%s%n", thingName, shadowName); } catch (TimeoutException e) { System.err.printf("Timeout occurred while deleting shadow: %s/%s%n", thingName, shadowName); } catch (ExecutionException e) { if (e.getCause() instanceof UnauthorizedError) { System.err.printf("Unauthorized error while deleting shadow: %s/%s%n", thingName, shadowName); } else if (e.getCause() instanceof ResourceNotFoundError) { System.err.printf("Unable to find shadow to delete: %s/%s%n", thingName, shadowName); } else { throw e; } } } catch (InterruptedException e) { System.out.println("IPC interrupted."); } catch (ExecutionException e) { System.err.println("Exception occurred when using IPC."); e.printStackTrace(); System.exit(1); } } public static DeleteThingShadowResponseHandler deleteThingShadow(GreengrassCoreIPCClient greengrassCoreIPCClient, String thingName, String shadowName) { DeleteThingShadowRequest deleteThingShadowRequest = new DeleteThingShadowRequest(); deleteThingShadowRequest.setThingName(thingName); deleteThingShadowRequest.setShadowName(shadowName); return greengrassCoreIPCClient.deleteThingShadow(deleteThingShadowRequest, Optional.empty()); } } ``` ------ #### [ Python (IPC client V1) ] **Example 範例：刪除物件影子** ``` import awsiot.greengrasscoreipc import awsiot.greengrasscoreipc.client as client from awsiot.greengrasscoreipc.model import DeleteThingShadowRequest TIMEOUT = 10 def sample_delete_thing_shadow_request(thingName, shadowName): try: # set up IPC client to connect to the IPC server ipc_client = awsiot.greengrasscoreipc.connect() # create the DeleteThingShadow request delete_thing_shadow_request = DeleteThingShadowRequest() delete_thing_shadow_request.thing_name = thingName delete_thing_shadow_request.shadow_name = shadowName # retrieve the DeleteThingShadow response after sending the request to the IPC server op = ipc_client.new_delete_thing_shadow() op.activate(delete_thing_shadow_request) fut = op.get_response() result = fut.result(TIMEOUT) return result.payload except InvalidArgumentsError as e: # add error handling ... # except ResourceNotFoundError | UnauthorizedError | ServiceError ``` ------ #### [ JavaScript ] **Example 範例：刪除物件影子** ``` import { DeleteThingShadowRequest } from 'aws-iot-device-sdk-v2/dist/greengrasscoreipc/model'; import * as greengrasscoreipc from 'aws-iot-device-sdk-v2/dist/greengrasscoreipc'; class DeleteThingShadow { private ipcClient: greengrasscoreipc.Client; private thingName: string; private shadowName: string; constructor() { // Define args parameters here this.thingName = ""; this.shadowName = ""; this.bootstrap(); } async bootstrap() { try { this.ipcClient = await getIpcClient(); } catch (err) { // parse the error depending on your use cases throw err } try { await this.handleDeleteThingShadowOperation(this.thingName, this.shadowName) } catch (err) { // parse the error depending on your use cases throw err } } async handleDeleteThingShadowOperation(thingName: string, shadowName: string) { const request: DeleteThingShadowRequest = { thingName: thingName, shadowName: shadowName } // make the DeleteThingShadow request const response = await this.ipcClient.deleteThingShadow(request); } } export async function getIpcClient() { try { const ipcClient = greengrasscoreipc.createClient(); await ipcClient.connect() .catch(error => { // parse the error depending on your use cases throw error; }); return ipcClient } catch (err) { // parse the error depending on your use cases throw err } } const startScript = new DeleteThingShadow(); ``` ------ ## ListNamedShadowsForThing 列出指定物件的具名影子。 ### 請求此操作的請求具有下列參數： `thingName` (Python：`thing_name`) 物件名稱。類型：`string` `pageSize` (Python：`page_size`) （選用）每次呼叫中要傳回的陰影名稱數量。類型：`integer` 預設：25 上限：100 `nextToken` (Python：`next_token`) （選用）擷取下一組結果的字符。此值會在分頁結果上傳回，並在傳回下一頁的呼叫中使用。類型：`string` ### 回應此操作的回應包含下列資訊： `results` 陰影名稱清單。類型：`array` `timestamp` （選用）產生回應的日期和時間。類型：`integer` `nextToken` (Python：`next_token`) （選用）在分頁請求中使用的字符值，用於擷取序列中的下一頁。當不再傳回陰影名稱時，此字符不存在。類型：`string` 如果請求的頁面大小完全符合回應中的陰影名稱數量，則表示存在此字符；不過，使用時，它會傳回空清單。 ### 錯誤此操作可能會傳回下列錯誤。 `InvalidArgumentsError` 本機影子服務無法驗證請求參數。如果請求包含格式不正確的 JSON 或不支援的字元，就會發生這種情況。 `ResourceNotFoundError` 找不到請求的本機影子文件。 `ServiceError` 發生內部服務錯誤，或對 IPC 服務的請求數量超過影子管理員元件中 `maxLocalRequestsPerSecondPerThing`和 `maxTotalLocalRequestsRate`組態參數中指定的限制。 `UnauthorizedError` 元件的授權政策不包含此操作所需的許可。 ### 範例下列範例示範如何在自訂元件程式碼中呼叫此操作。 ------ #### [ Java (IPC client V1) ] **Example 範例：列出物件的命名影子** 此範例使用 `IPCUtils`類別來建立與 AWS IoT Greengrass Core IPC 服務的連線。如需詳細資訊，請參閱[連線至 AWS IoT Greengrass Core IPC 服務](interprocess-communication.md#ipc-service-connect)。 ``` package com.aws.greengrass.docs.samples.ipc; import com.aws.greengrass.docs.samples.ipc.util.IPCUtils; import software.amazon.awssdk.aws.greengrass.GreengrassCoreIPCClient; import software.amazon.awssdk.aws.greengrass.ListNamedShadowsForThingResponseHandler; import software.amazon.awssdk.aws.greengrass.model.ListNamedShadowsForThingRequest; import software.amazon.awssdk.aws.greengrass.model.ListNamedShadowsForThingResponse; import software.amazon.awssdk.aws.greengrass.model.ResourceNotFoundError; import software.amazon.awssdk.aws.greengrass.model.UnauthorizedError; import software.amazon.awssdk.eventstreamrpc.EventStreamRPCConnection; import java.util.ArrayList; import java.util.List; import java.util.Optional; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; public class ListNamedShadowsForThing { public static final int TIMEOUT_SECONDS = 10; public static void main(String[] args) { // Use the current core device's name if thing name isn't set. String thingName = args[0].isEmpty() ? System.getenv("AWS_IOT_THING_NAME") : args[0]; try (EventStreamRPCConnection eventStreamRPCConnection = IPCUtils.getEventStreamRpcConnection()) { GreengrassCoreIPCClient ipcClient = new GreengrassCoreIPCClient(eventStreamRPCConnection); List namedShadows = new ArrayList<>(); String nextToken = null; try { // Send additional requests until there's no pagination token in the response. do { ListNamedShadowsForThingResponseHandler responseHandler = ListNamedShadowsForThing.listNamedShadowsForThing(ipcClient, thingName, nextToken, 25); CompletableFuture futureResponse = responseHandler.getResponse(); ListNamedShadowsForThingResponse response = futureResponse.get(TIMEOUT_SECONDS, TimeUnit.SECONDS); List responseNamedShadows = response.getResults(); namedShadows.addAll(responseNamedShadows); nextToken = response.getNextToken(); } while (nextToken != null); System.out.printf("Successfully got named shadows for thing %s: %s%n", thingName, String.join(",", namedShadows)); } catch (TimeoutException e) { System.err.println("Timeout occurred while listing named shadows for thing: " + thingName); } catch (ExecutionException e) { if (e.getCause() instanceof UnauthorizedError) { System.err.println("Unauthorized error while listing named shadows for " + "thing: " + thingName); } else if (e.getCause() instanceof ResourceNotFoundError) { System.err.println("Unable to find thing to list named shadows: " + thingName); } else { throw e; } } } catch (InterruptedException e) { System.out.println("IPC interrupted."); } catch (ExecutionException e) { System.err.println("Exception occurred when using IPC."); e.printStackTrace(); System.exit(1); } } public static ListNamedShadowsForThingResponseHandler listNamedShadowsForThing(GreengrassCoreIPCClient greengrassCoreIPCClient, String thingName, String nextToken, int pageSize) { ListNamedShadowsForThingRequest listNamedShadowsForThingRequest = new ListNamedShadowsForThingRequest(); listNamedShadowsForThingRequest.setThingName(thingName); listNamedShadowsForThingRequest.setNextToken(nextToken); listNamedShadowsForThingRequest.setPageSize(pageSize); return greengrassCoreIPCClient.listNamedShadowsForThing(listNamedShadowsForThingRequest, Optional.empty()); } } ``` ------ #### [ Python (IPC client V1) ] **Example 範例：列出物件的命名影子** ``` import awsiot.greengrasscoreipc import awsiot.greengrasscoreipc.client as client from awsiot.greengrasscoreipc.model import ListNamedShadowsForThingRequest TIMEOUT = 10 def sample_list_named_shadows_for_thing_request(thingName, nextToken, pageSize): try: # set up IPC client to connect to the IPC server ipc_client = awsiot.greengrasscoreipc.connect() # create the ListNamedShadowsForThingRequest request list_named_shadows_for_thing_request = ListNamedShadowsForThingRequest() list_named_shadows_for_thing_request.thing_name = thingName list_named_shadows_for_thing_request.next_token = nextToken list_named_shadows_for_thing_request.page_size = pageSize # retrieve the ListNamedShadowsForThingRequest response after sending the request to the IPC server op = ipc_client.new_list_named_shadows_for_thing() op.activate(list_named_shadows_for_thing_request) fut = op.get_response() list_result = fut.result(TIMEOUT) # additional returned fields timestamp = list_result.timestamp next_token = result.next_token named_shadow_list = list_result.results return named_shadow_list, next_token, timestamp except InvalidArgumentsError as e: # add error handling ... # except ResourceNotFoundError | UnauthorizedError | ServiceError ``` ------ #### [ JavaScript ] **Example 範例：列出物件的命名影子** ``` import { ListNamedShadowsForThingRequest } from 'aws-iot-device-sdk-v2/dist/greengrasscoreipc/model'; import * as greengrasscoreipc from 'aws-iot-device-sdk-v2/dist/greengrasscoreipc'; class listNamedShadowsForThing { private ipcClient: greengrasscoreipc.Client; private thingName: string; private pageSizeStr: string; private nextToken: string; constructor() { // Define args parameters here this.thingName = ""; this.pageSizeStr = ""; this.nextToken = ""; this.bootstrap(); } async bootstrap() { try { this.ipcClient = await getIpcClient(); } catch (err) { // parse the error depending on your use cases throw err } try { await this.handleListNamedShadowsForThingOperation(this.thingName, this.nextToken, this.pageSizeStr); } catch (err) { // parse the error depending on your use cases throw err } } async handleListNamedShadowsForThingOperation( thingName: string, nextToken: string, pageSizeStr: string ) { let request: ListNamedShadowsForThingRequest = { thingName: thingName, nextToken: nextToken, }; if (pageSizeStr) { request.pageSize = parseInt(pageSizeStr); } // make the ListNamedShadowsForThing request const response = await this.ipcClient.listNamedShadowsForThing(request); const shadowNames = response.results; } } export async function getIpcClient(){ try { const ipcClient = greengrasscoreipc.createClient(); await ipcClient.connect() .catch(error => { // parse the error depending on your use cases throw error; }); return ipcClient } catch (err) { // parse the error depending on your use cases throw err } } const startScript = new listNamedShadowsForThing(); ``` ------ # 管理本機部署和元件 **注意** 此功能適用於 [Greengrass 核元件](greengrass-nucleus-component.md)的 v2.6.0 和更新版本。使用 Greengrass CLI IPC 服務來管理核心裝置上的本機部署和 Greengrass 元件。若要使用這些 IPC 操作，請在自訂[元件中包含 2.6.0 版或更新版本的 Greengrass CLI ](greengrass-cli-component.md)元件做為相依性。然後，您可以在自訂元件中使用 IPC 操作來執行下列動作： + 建立本機部署以修改和設定核心裝置上的 Greengrass 元件。 + 在核心裝置上重新啟動和停止 Greengrass 元件。 + 產生可用來登入[本機偵錯主控台](local-debug-console-component.md)的密碼。 **Topics** + [最低 SDK 版本](#ipc-local-deployments-components-sdk-versions) + [Authorization](#ipc-local-deployments-components-authorization) + [CreateLocalDeployment](#ipc-operation-createlocaldeployment) + [ListLocalDeployments](#ipc-operation-listlocaldeployments) + [GetLocalDeploymentStatus](#ipc-operation-getlocaldeploymentstatus) + [ListComponents](#ipc-operation-listcomponents) + [GetComponentDetails](#ipc-operation-getcomponentdetails) + [RestartComponent](#ipc-operation-restartcomponent) + [StopComponent](#ipc-operation-stopcomponent) + [CreateDebugPassword](#ipc-operation-createdebugpassword) ## 最低 SDK 版本下表列出 AWS IoT Device SDK 您必須用來與 Greengrass CLI IPC 服務互動的最低版本。 | SDK | 最低版本 | | --- | --- | | [AWS IoT Device SDK for Java v2](https://github.com/aws/aws-iot-device-sdk-java-v2) | v1.2.10 | | [AWS IoT Device SDK for Python v2](https://github.com/aws/aws-iot-device-sdk-python-v2) | 1.5.3 版 | | [AWS IoT Device SDK 適用於 C\$1\$1 v2](https://github.com/aws/aws-iot-device-sdk-cpp-v2) | 1.17.0 版 | | [AWS IoT Device SDK for JavaScript v2](https://github.com/aws/aws-iot-device-sdk-js-v2) | v1.12.0 | ## Authorization 若要在自訂元件中使用 Greengrass CLI IPC 服務，您必須定義授權政策，允許您的元件管理本機部署和元件。如需定義授權政策的資訊，請參閱 [授權元件執行 IPC 操作](interprocess-communication.md#ipc-authorization-policies)。 Greengrass CLI 的授權政策具有下列屬性。 **IPC 服務識別符：** `aws.greengrass.Cli` | 作業 | Description | Resources | | --- | --- | --- | | `aws.greengrass#CreateLocalDeployment` | 允許元件在核心裝置上建立本機部署。 | `*` | | `aws.greengrass#ListLocalDeployments` | 允許元件列出核心裝置上的本機部署。 | `*` | | `aws.greengrass#GetLocalDeploymentStatus` | 允許元件取得核心裝置上本機部署的狀態。 | 本機部署 ID，或 `*` 允許存取所有本機部署。 | | `aws.greengrass#ListComponents` | 允許元件列出核心裝置上的元件。 | `*` | | `aws.greengrass#GetComponentDetails` | 允許元件取得核心裝置上元件的詳細資訊。 | 元件名稱，例如 `com.example.HelloWorld`或 `*` 允許存取所有元件。 | | `aws.greengrass#RestartComponent` | 允許元件重新啟動核心裝置上的元件。 | 元件名稱，例如 `com.example.HelloWorld`或 `*` 允許存取所有元件。 | | `aws.greengrass#StopComponent` | 允許元件停止核心裝置上的元件。 | 元件名稱，例如 `com.example.HelloWorld`或 `*` 允許存取所有元件。 | | `aws.greengrass#CreateDebugPassword` | 允許元件產生密碼，以用來登入[本機偵錯主控台元件](local-debug-console-component.md)。 | `*` | **Example 範例授權政策** 下列範例授權政策允許元件建立本機部署、檢視所有本機部署和元件，以及重新啟動和停止名為的元件`com.example.HelloWorld`。 ``` { "accessControl": { "aws.greengrass.Cli": { "com.example.MyLocalManagerComponent:cli:1": { "policyDescription": "Allows access to create local deployments and view deployments and components.", "operations": [ "aws.greengrass#CreateLocalDeployment", "aws.greengrass#ListLocalDeployments", "aws.greengrass#GetLocalDeploymentStatus", "aws.greengrass#ListComponents", "aws.greengrass#GetComponentDetails" ], "resources": [ "*" ] } }, "aws.greengrass.Cli": { "com.example.MyLocalManagerComponent:cli:2": { "policyDescription": "Allows access to restart and stop the Hello World component.", "operations": [ "aws.greengrass#RestartComponent", "aws.greengrass#StopComponent" ], "resources": [ "com.example.HelloWorld" ] } } } } ``` ## CreateLocalDeployment 使用指定的元件配方、成品和執行時間引數建立或更新本機部署。此操作提供與 Greengrass CLI 中的[部署建立命令](gg-cli-deployment.md#deployment-create)相同的功能。 ### 請求此操作的請求具有下列參數： `recipeDirectoryPath` (Python：`recipe_directory_path`) （選用）包含元件配方檔案之資料夾的絕對路徑。 `artifactDirectoryPath` (Python：`artifact_directory_path`) （選用）資料夾的絕對路徑，其中包含要包含在部署中的成品檔案。成品資料夾必須包含下列資料夾結構： ``` /path/to/artifact/folder/component-name/component-version/artifacts ``` `rootComponentVersionsToAdd` (Python：`root_component_versions_to_add`) （選用）要在核心裝置上安裝的元件版本。此物件 `ComponentToVersionMap`是包含下列鍵值對的映射： `key` 元件的名稱。 `value` 元件的版本。 `rootComponentsToRemove` (Python：`root_components_to_remove`) （選用）從核心裝置解除安裝的元件。指定清單，其中每個項目都是元件的名稱。 `componentToConfiguration` (Python：`component_to_configuration`) （選用）部署中每個元件的組態更新。此物件 `ComponentToConfiguration`是包含下列鍵值對的映射： `key` 元件的名稱。 `value` 元件的組態更新 JSON 物件。JSON 物件必須具有下列格式。 ``` { "MERGE": { "config-key": "config-value" }, "RESET": [ "path/to/reset/" ] } ``` 如需組態更新的詳細資訊，請參閱 [更新元件組態](update-component-configurations.md)。 `componentToRunWithInfo` (Python：`component_to_run_with_info`) （選用）部署中每個元件的執行時間組態。此組態包含擁有每個元件程序的系統使用者，以及要套用至每個元件的系統限制。此物件 `ComponentToRunWithInfo`是包含下列鍵值對的映射： `key` 元件的名稱。 `value` 元件的執行時間組態。如果您省略執行時間組態參數，則 AWS IoT Greengrass Core 軟體會使用您在 [Greengrass 核](greengrass-nucleus-component.md)上設定的預設值。此物件 `RunWithInfo`包含下列資訊： `posixUser` (Python：`posix_user`) （選用） POSIX 系統使用者和選用的群組，用於在 Linux 核心裝置上執行此元件。如果指定，使用者和群組必須存在於每個 Linux 核心裝置上。以下列格式指定使用者和群組，並以冒號 (`:`) 分隔：`user:group`。群組為選用項目。如果您未指定群組，則 AWS IoT Greengrass Core 軟體會使用使用者的主要群組。如需詳細資訊，請參閱 [設定執行元件的使用者](configure-greengrass-core-v2.md#configure-component-user)。 `windowsUser` (Python：`windows_user`) （選用）用來在 Windows 核心裝置上執行此元件的 Windows 使用者。使用者必須存在於每個 Windows 核心裝置上，且其名稱和密碼必須存放在 LocalSystem 帳戶的 Credentials Manager 執行個體中。如需詳細資訊，請參閱 [設定執行元件的使用者](configure-greengrass-core-v2.md#configure-component-user)。 `systemResourceLimits` (Python：`system_resource_limits`) （選用）要套用至此元件程序的系統資源限制。您可以將系統資源限制套用至一般和非容器 Lambda 元件。如需詳細資訊，請參閱 [設定元件的系統資源限制](configure-greengrass-core-v2.md#configure-component-system-resource-limits)。 AWS IoT Greengrass 目前不支援 Windows 核心裝置上的此功能。此物件 `SystemResourceLimits`包含下列資訊： `cpus` （選用）此元件程序可在核心裝置上使用的 CPU 時間上限。核心裝置的 CPU 總時間等於裝置的 CPU 核心數。例如，在具有 4 個 CPU 核心的核心裝置上，您可以將此值設定為 `2`，將此元件的程序限制為每個 CPU 核心的 50% 用量。在具有 1 個 CPU 核心的裝置上，您可以將此值設定為 `0.25`，將此元件的程序限制為 CPU 的 25% 用量。如果您將此值設定為大於 CPU 核心數量的數字，則 AWS IoT Greengrass Core 軟體不會限制元件的 CPU 使用量。 `memory` （選用）此元件程序可在核心裝置上使用的 RAM 數量上限（以 KB 為單位）。 `groupName` (Python：`group_name`) （選用）此部署的目標物件群組名稱。 ### 回應此操作的回應包含下列資訊： `deploymentId` (Python：`deployment_id`) 請求建立的本機部署 ID。 ## ListLocalDeployments 取得過去 10 個本機部署的狀態。此操作提供與 Greengrass CLI 中的[部署清單命令](gg-cli-deployment.md#deployment-list)相同的功能。 ### 請求此操作的請求沒有任何參數。 ### 回應此操作的回應包含下列資訊： `localDeployments` (Python：`local_deployments`) 本機部署的清單。此清單中的每個物件都是 `LocalDeployment` 物件，其中包含下列資訊： `deploymentId` (Python：`deployment_id`) 本機部署的 ID。 `status` 本機部署的狀態。此列舉 `DeploymentStatus`具有下列值： + `QUEUED` + `IN_PROGRESS` + `SUCCEEDED` + `FAILED` ## GetLocalDeploymentStatus 取得本機部署的狀態。此操作提供與 Greengrass CLI 中的[部署狀態命令](gg-cli-deployment.md#deployment-status)相同的功能。 ### 請求此操作的請求具有下列參數： `deploymentId` (Python：`deployment_id`) 要取得的本機部署 ID。 ### 回應此操作的回應包含下列資訊： `deployment` 本機部署。此物件 `LocalDeployment`包含下列資訊： `deploymentId` (Python：`deployment_id`) 本機部署的 ID。 `status` 本機部署的狀態。此列舉 `DeploymentStatus`具有下列值： + `QUEUED` + `IN_PROGRESS` + `SUCCEEDED` + `FAILED` ## ListComponents 取得核心裝置上每個根元件的名稱、版本、狀態和組態。*根元件*是您在部署中指定的元件。此回應不包含安裝為其他元件相依性的元件。此操作提供與 Greengrass CLI 中的[元件清單命令](gg-cli-component.md#component-list)相同的功能。 ### 請求此操作的請求沒有任何參數。 ### 回應此操作的回應包含下列資訊： `components` 核心裝置上的根元件清單。此清單中的每個物件都是 `ComponentDetails` 物件，其中包含下列資訊： `componentName` (Python：`component_name`) 元件的名稱。 `version` 元件的版本。 `state` 元件的狀態。此狀態可以是下列其中一項： + `BROKEN` + `ERRORED` + `FINISHED` + `INSTALLED` + `NEW` + `RUNNING` + `STARTING` + `STOPPING` `configuration` 元件的組態做為 JSON 物件。 ## GetComponentDetails 取得核心裝置上元件的版本、狀態和組態。此操作提供與 Greengrass CLI 中的[元件詳細資訊命令](gg-cli-component.md#component-details)相同的功能。 ### 請求此操作的請求具有下列參數： `componentName` (Python：`component_name`) 要取得的元件名稱。 ### 回應此操作的回應包含下列資訊： `componentDetails` (Python：`component_details`) 元件的詳細資訊。此物件 `ComponentDetails`包含下列資訊： `componentName` (Python：`component_name`) 元件的名稱。 `version` 元件的版本。 `state` 元件的狀態。此狀態可以是下列其中一項： + `BROKEN` + `ERRORED` + `FINISHED` + `INSTALLED` + `NEW` + `RUNNING` + `STARTING` + `STOPPING` `configuration` 元件的組態做為 JSON 物件。 ## RestartComponent 重新啟動核心裝置上的元件。 **注意** 雖然您可以重新啟動任何元件，但我們建議您僅重新啟動[一般元件](develop-greengrass-components.md#component-types)。此操作提供與 Greengrass CLI 中的[元件重新啟動命令](gg-cli-component.md#component-restart)相同的功能。 ### 請求此操作的請求具有下列參數： `componentName` (Python：`component_name`) 元件的名稱。 ### 回應此操作的回應包含下列資訊： `restartStatus` (Python：`restart_status`) 重新啟動請求的狀態。請求狀態可以是下列其中一項： + `SUCCEEDED` + `FAILED` `message` 說明如果請求失敗，為何元件無法重新啟動的訊息。 ### 範例下列範例示範如何在自訂元件程式碼中呼叫此操作。 ------ #### [ Rust ] **Example 範例：重新啟動元件** ``` use gg_sdk::Sdk; fn main() { let sdk = Sdk::init(); sdk.connect().expect("Failed to establish IPC connection"); let component_name = "com.example.HelloWorld"; sdk.restart_component(component_name) .expect("Failed to restart component"); println!("Successfully requested restart for component: {component_name}"); } ``` ------ #### [ C ] **Example 範例：重新啟動元件** ``` #include #include #include #include #include int main(void) { gg_sdk_init(); GgError err = ggipc_connect(); if (err != GG_ERR_OK) { fprintf(stderr, "Failed to establish IPC connection.\n"); exit(-1); } GgBuffer component_name = GG_STR("com.example.HelloWorld"); err = ggipc_restart_component(component_name); if (err != GG_ERR_OK) { fprintf( stderr, "Failed to restart component: %.*s\n", (int) component_name.len, component_name.data ); exit(-1); } printf( "Successfully requested restart for component: %.*s\n", (int) component_name.len, component_name.data ); } ``` ------ #### [ C\$1\$1 (Component SDK) ] **Example 範例：重新啟動元件** ``` #include #include int main() { auto &client = gg::ipc::Client::get(); auto error = client.connect(); if (error) { std::cerr << "Failed to establish IPC connection.\n"; exit(-1); } std::string_view component_name = "com.example.HelloWorld"; error = client.restart_component(component_name); if (error) { std::cerr << "Failed to restart component: " << component_name << "\n"; exit(-1); } std::cout << "Successfully requested restart for component: " << component_name << "\n"; } ``` ------ ## StopComponent 在核心裝置上停止元件的程序。 **注意** 雖然您可以停止任何元件，但我們建議您僅停止[一般元件](develop-greengrass-components.md#component-types)。此操作提供與 Greengrass CLI 中的[元件停止命令](gg-cli-component.md#component-stop)相同的功能。 ### 請求此操作的請求具有下列參數： `componentName` (Python：`component_name`) 元件的名稱。 ### 回應此操作的回應包含下列資訊： `stopStatus` (Python：`stop_status`) 停止請求的狀態。請求狀態可以是下列其中一項： + `SUCCEEDED` + `FAILED` `message` 如果請求失敗，說明元件為何無法停止的訊息。 ## CreateDebugPassword 產生隨機密碼，您可以用來登入[本機偵錯主控台元件](local-debug-console-component.md)。密碼會在產生後 8 小時過期。此操作提供與 Greengrass CLI 中的 [get-debug-password 命令](gg-cli-get-debug-password.md)相同的功能。 ### 請求此操作的請求沒有任何參數。 ### 回應此操作的回應包含下列資訊： `username` 用於登入的使用者名稱。 `password` 用來登入的密碼。 `passwordExpiration` (Python：`password_expiration`) 密碼過期的時間。 `certificateSHA256Hash` (Python：`certificate_sha256_hash`) 啟用 HTTPS 時，本機偵錯主控台使用的自我簽署憑證的 SHA-256 指紋。當您開啟本機偵錯主控台時，請使用此指紋來驗證憑證是否合法且連線安全。 `certificateSHA1Hash` (Python：`certificate_sha1_hash`) 啟用 HTTPS 時，本機偵錯主控台使用的自我簽署憑證的 SHA-1 指紋。當您開啟本機偵錯主控台時，請使用此指紋來驗證憑證是否合法且連線安全。 # 驗證和授權用戶端裝置 **注意** 此功能適用於 [Greengrass 核元件](greengrass-nucleus-component.md)的 v2.6.0 和更新版本。使用用戶端裝置驗證 IPC 服務來開發自訂本機代理程式元件，例如用戶端裝置等本機 IoT 裝置可以連線。若要使用這些 IPC 操作，請在自訂[元件中包含 2.2.0 版或更新版本的用戶端裝置身分驗證](client-device-auth-component.md)元件做為相依性。然後，您可以在自訂元件中使用 IPC 操作來執行下列動作： + 驗證連接至核心裝置的用戶端裝置身分。 + 為用戶端裝置建立工作階段以連線至核心裝置。 + 驗證用戶端裝置是否具有執行動作的許可。 + 在核心裝置的伺服器憑證輪換時接收通知。 **Topics** + [最低 SDK 版本](#ipc-client-device-auth-sdk-versions) + [Authorization](#ipc-client-device-auth-authorization) + [VerifyClientDeviceIdentity](#ipc-operation-verifyclientdeviceidentity) + [GetClientDeviceAuthToken](#ipc-operation-getclientdeviceauthtoken) + [AuthorizeClientDeviceAction](#ipc-operation-authorizeclientdeviceaction) + [SubscribeToCertificateUpdates](#ipc-operation-subscribetocertificateupdates) ## 最低 SDK 版本下表列出 AWS IoT Device SDK 您必須用來與用戶端裝置驗證 IPC 服務互動的最低版本。 | SDK | 最低版本 | | --- | --- | | [AWS IoT Device SDK for Java v2](https://github.com/aws/aws-iot-device-sdk-java-v2) | 1.9.3 版 | | [AWS IoT Device SDK for Python v2](https://github.com/aws/aws-iot-device-sdk-python-v2) | 1.11.3 版 | | [AWS IoT Device SDK 適用於 C\$1\$1 v2](https://github.com/aws/aws-iot-device-sdk-cpp-v2) | 1.18.3 版 | | [AWS IoT Device SDK for JavaScript v2](https://github.com/aws/aws-iot-device-sdk-js-v2) | v1.12.0 | ## Authorization 若要在自訂元件中使用用戶端裝置驗證 IPC 服務，您必須定義授權政策，以允許您的元件執行這些操作。如需定義授權政策的資訊，請參閱 [授權元件執行 IPC 操作](interprocess-communication.md#ipc-authorization-policies)。用戶端裝置身分驗證和授權的授權政策具有下列屬性。 **IPC 服務識別符：** `aws.greengrass.clientdevices.Auth` | 作業 | Description | Resources | | --- | --- | --- | | `aws.greengrass#VerifyClientDeviceIdentity` | 允許元件驗證用戶端裝置的身分。 | `*` | | `aws.greengrass#GetClientDeviceAuthToken` | 允許元件驗證用戶端裝置的登入資料，並為該用戶端裝置建立工作階段。 | `*` | | `aws.greengrass#AuthorizeClientDeviceAction` | 允許元件驗證用戶端裝置是否具有執行動作的許可。 | `*` | | `aws.greengrass#SubscribeToCertificateUpdates` | 允許元件在核心裝置的伺服器憑證輪換時接收通知。 | `*` | | `*` | 允許元件執行所有用戶端裝置驗證 IPC 服務操作。 | `*` | ### 授權政策範例您可以參考下列授權政策範例，協助您設定元件的授權政策。 **Example 範例授權政策** 下列範例授權政策允許元件執行所有用戶端裝置驗證 IPC 操作。 ``` { "accessControl": { "aws.greengrass.clientdevices.Auth": { "com.example.MyLocalBrokerComponent:clientdevices:1": { "policyDescription": "Allows access to authenticate and authorize client devices.", "operations": [ "aws.greengrass#VerifyClientDeviceIdentity", "aws.greengrass#GetClientDeviceAuthToken", "aws.greengrass#AuthorizeClientDeviceAction", "aws.greengrass#SubscribeToCertificateUpdates" ], "resources": [ "*" ] } } } } ``` ## VerifyClientDeviceIdentity 驗證用戶端裝置的身分。此操作會驗證用戶端裝置是否為有效 AWS IoT 物件。 ### 請求此操作的請求具有下列參數： `credential` 用戶端裝置的登入資料。此物件 `ClientDeviceCredential`包含下列資訊： `clientDeviceCertificate` (Python：`client_device_certificate`) 用戶端裝置的 X.509 裝置憑證。 ### 回應此操作的回應包含下列資訊： `isValidClientDevice` (Python：`is_valid_client_device`) 用戶端裝置的身分是否有效。 ## GetClientDeviceAuthToken 驗證用戶端裝置的登入資料，並為用戶端裝置建立工作階段。此操作會傳回工作階段字符，您可以在後續請求中使用它來[授權用戶端裝置動作](#ipc-operation-authorizeclientdeviceaction)。若要成功連接用戶端裝置，[用戶端裝置身分驗證元件](client-device-auth-component.md#client-device-auth-component-configuration)必須授予用戶端裝置使用的用戶端 ID `mqtt:connect`許可。 ### 請求此操作的請求具有下列參數： `credential` 用戶端裝置的登入資料。此物件 `CredentialDocument`包含下列資訊： `mqttCredential` (Python：`mqtt_credential`) 用戶端裝置的 MQTT 登入資料。指定用戶端裝置用來連線的用戶端 ID 和憑證。此物件 `MQTTCredential`包含下列資訊： `clientId` (Python：`client_id`) 用於連線的用戶端 ID。 `certificatePem` (Python：`certificate_pem`) 用於連線的 X.509 裝置憑證。 `username` 此屬性目前未使用。 `password` 此屬性目前未使用。 ### 回應此操作的回應包含下列資訊： `clientDeviceAuthToken` (Python：`client_device_auth_token`) 用戶端裝置的工作階段字符。您可以在後續請求中使用此工作階段字符來授權此用戶端裝置的動作。 ## AuthorizeClientDeviceAction 驗證用戶端裝置是否具有對資源執行動作的許可。*用戶端裝置授權政策*指定用戶端裝置在連線至核心裝置時可執行的許可。您可以在設定用戶端裝置[身分驗證元件時定義用戶端裝置](client-device-auth-component.md#client-device-auth-component-configuration)授權政策。 ### 請求此操作的請求具有下列參數： `clientDeviceAuthToken` (Python：`client_device_auth_token`) 用戶端裝置的工作階段字符。 `operation` 要授權的操作。 `resource` 用戶端裝置執行操作的資源。 ### 回應此操作的回應包含下列資訊： `isAuthorized` (Python：`is_authorized`) 用戶端裝置是否獲授權在資源上執行操作。 ## SubscribeToCertificateUpdates 訂閱以在每次輪換時接收核心裝置的新伺服器憑證。當伺服器憑證輪換時，代理程式必須使用新的伺服器憑證重新載入。根據預設[，用戶端裝置身分驗證元件](client-device-auth-component.md)會每 7 天輪換一次伺服器憑證。您可以將輪換間隔設定為 2 到 10 天。此操作是您訂閱事件訊息串流的訂閱操作。若要使用此操作，請使用處理事件訊息、錯誤和串流關閉的函數來定義串流回應處理常式。如需詳細資訊，請參閱[訂閱 IPC 事件串流](interprocess-communication.md#ipc-subscribe-operations)。 **事件訊息類型：** `CertificateUpdateEvent` ### 請求此操作的請求具有下列參數： `certificateOptions` (Python：`certificate_options`) 要訂閱的憑證更新類型。此物件 `CertificateOptions`包含下列資訊： `certificateType` (Python：`certificate_type`) 要訂閱的憑證更新類型。選擇下列選項： + `SERVER` ### 回應此操作的回應包含下列資訊： `messages` 訊息串流。此物件 `CertificateUpdateEvent`包含下列資訊： `certificateUpdate` (Python：`certificate_update`) 新憑證的相關資訊。此物件 `CertificateUpdate`包含下列資訊： `certificate` 憑證。 `privateKey` (Python：`private_key`) 憑證的私有金鑰。 `publicKey` (Python：`public_key`) 憑證的公有金鑰。 `caCertificates` (Python：`ca_certificates`) 憑證的 CA 憑證鏈中的憑證授權機構 (CA) 憑證清單。