# メッセージ変換

メッセージ変換は、ユーザー定義のルールに基づいてメッセージを変更およびフォーマットし、その後の処理やサブスクライバーへの配信前に適用する機能です。この機能は非常にカスタマイズ可能で、複数のエンコーディングや高度な変換をサポートしています。

## ワークフロー

メッセージがパブリッシュされると、以下のワークフローが実行されます。

1. **スキーマ検証**：メッセージがパブリッシュされ認可を通過すると、まず[スキーマ検証](./schema-validation.md)が行われます。メッセージが検証に合格すると、次のステップに進みます。

2. **メッセージ変換パイプライン**：

   - **変換マッチング**：メッセージは、そのトピックに基づいてユーザー定義の変換リストと照合されます。異なるトピックやトピックフィルターに対して複数の変換を設定可能です。
   - **変換実行**：マッチした変換は設定された順序で実行されます。パイプラインはJSON、Protobuf、Avroなどの各種エンコーダー・デコーダーをサポートし、[Variform式](../../guides/configuration/configuration.md#variform-expressions)を使ってメッセージの拡張や修正が可能です。
   - **変換後処理**：メッセージが変換パイプラインを正常に通過すると、ルールエンジンのトリガーやサブスクライバーへのメッセージ配信など、次の処理に進みます。

3. **失敗時の処理**：変換が失敗した場合、ユーザー設定のアクションが実行されます。

   - **メッセージ破棄**：パブリッシュを終了しメッセージを破棄します。QoS 1およびQoS 2のメッセージにはPUBACKで特定の理由コード（131 - 実装固有のエラー）が返されます。
   - **切断してメッセージ破棄**：メッセージを破棄し、パブリッシュしたクライアントを切断します。
   - **無視**：追加の処理は行いません。

   変換失敗時には設定に関わらずログエントリを生成可能です。ログの出力レベルはユーザーが設定でき、デフォルトは `warning` です。さらに、変換失敗はルールエンジンイベント（`$events/message_transformation/failed`）をトリガーでき、誤ったメッセージを別トピックに再パブリッシュしたり、Kafkaに送信して解析するなどのカスタム処理を実装可能です。

## ユーザーガイド

このセクションでは、メッセージ変換機能の設定方法とテスト方法を説明します。

### ダッシュボードでのメッセージ変換設定

ダッシュボードでメッセージ変換を作成・設定する手順を示します。

1. ダッシュボードにアクセスし、左側のナビゲーションメニューから **Smart Data Hub** -> **Message Transform** をクリックします。

2. **Message Transform** ページ右上の **Create** をクリックします。

3. 「Create Message Transform」ページで以下の情報を設定します。

   - **Name**：変換の名前を入力します。

   - **Message Source Topic**：変換対象のメッセージが属するトピックを設定します。複数のトピックやトピックフィルターを設定可能です。

   - **Note**（任意）：任意のメモを入力します。

   - **Message Format Transformation**：
     - **Source Format**：変換パイプラインに入るメッセージのペイロードに適用するデコーダーを指定します。選択肢は以下の通りです。

       - `None`（デコードなし）
       - `JSON`
       - `Avro`
       - `Protobuf`
       - `Custom (External HTTP)`

       これらのデコーダーはバイナリの入力ペイロードを構造化マップに変換します。`Avro`、`Protobuf`、`Custom (External HTTP)`を選択する場合は、あらかじめ[スキーマレジストリ](./schema-registry.md)で作成済みである必要があります。

       複数の変換が連なるパイプラインでは、各ステップでのデコードは必須ではありません。例えば、変換 `T1` でペイロードが既にデコードされていれば、後続の変換 `T2` はデコードをスキップし、正しい形式のペイロードを利用できます。

     - **Target Format**：変換パイプラインの最後にメッセージペイロードをバイナリ値としてエンコードするエンコーダーを指定します。エンコーダーの選択肢は **Source Format** と同じです。

       パイプラインの最後の変換のみがペイロードをバイナリ値にエンコードする必要があり、中間の変換はバイナリエンコードを処理する必要はありません。

   - **Message Properties Transformation**：

     - **Properties**：式の結果として得られた変換後の値を書き込む先を指定します。指定可能な宛先は `payload`、`topic`、`qos`、`retain`（対応するフラグを設定）、および `user_property`（MQTTのUser-Property）です。`user_property` を使用する場合は、このフィールドの下に正確に1つのキーを指定する必要があります（例：`user_property.my_custom_prop`）。`payload` はそのままメッセージペイロード全体を上書きするか、あるいはネストされたキーのパスを指定してJSONオブジェクトのように扱うことができます（例：`payload.x.y`）。
     - **Target Value**：設定したプロパティに書き込む値を定義します。この値は `qos`、`retain`、`topic`、`payload`、`payload.x.y` など他のフィールドからコピーするか、[Variform式](../../guides/configuration/configuration.md#variform-expressions)を使って生成できます。

   - **Transformation Failure Operation**：
     - **Action After Failure**：変換失敗時に実行するアクションを選択します。
       - **Drop Message**：パブリッシュを終了しメッセージを破棄、QoS 1およびQoS 2メッセージにはPUBACKで特定の理由コードを返します。
       - **Disconnect and Drop Message**：メッセージを破棄し、パブリッシュしたクライアントを切断します。
       - **Ignore**：追加の処理は行いません。

   - **Output Logs**：変換失敗時にログを生成するか選択します。ログはデフォルトで有効です。

   - **Logs Level**：ログの出力レベルを設定します。デフォルトは `warning` です。

4. **Create** をクリックして設定を完了します。

作成前に **Preview** をクリックすると、変換のテストが可能です。新しいペインが開き、QoS、ペイロード、リテインフラグの有無、パブリッシャーのユーザー名やクライアントIDなど、入力メッセージのコンテキストを指定できます。必要な情報を入力後、**Execute Transformation** をクリックすると指定したコンテキストで変換が実行され、結果を確認できます。

作成後は、メッセージ変換ページのリストに変換が表示され、デフォルトで有効になっています。必要に応じて無効化したり、**Actions** 列の **Settings** をクリックして設定を更新できます。削除や順序変更は **More** から行えます。

### 設定ファイルでのメッセージ変換設定

Avro形式でエンコードされたメッセージを受信し、JSONにデコードしたいとします。デコード後、パブリッシュクライアントのクライアント属性から取得した `tenant` 属性をトピックの先頭に付加し、ルールエンジンで処理する前にトピックを変更したい場合、以下の設定で実現可能です。

```hocon
message_transformation {
  transformations = [
    {
      name = mytransformation
      topics = ["t"]
      failure_action = drop
      payload_decoder = {type = avro, schema = myschema}
      payload_encoder = {type = json}
      operations = [
        {key = "topic", value = "concat([client_attrs.tenant, '/', topic])"}
      ]
    }
  ]
}
```

この設定は、`mytransformation` という名前の変換を指定し、

- 指定したスキーマを使ってAvro形式のペイロードをデコード、
- ペイロードをJSON形式にエンコード、
- クライアント属性の `tenant` と元のトピックを連結してトピックを変更、

という処理を行います。

詳細な設定方法は[設定マニュアル](https://docs.emqx.com/en/enterprise/v6.3.1/hocon/)を参照してください。

### REST API

REST APIを使ったメッセージ変換の詳細な利用方法は[EMQX Enterprise API](https://docs.emqx.com/en/enterprise/v6.3/admin/api-docs.html)をご覧ください。

### デコード／エンコードスキーマの作成

デコーダーおよびエンコーダースキーマの作成方法については、[スキーマレジストリ](./schema-registry.md)のセクションを参照してください。

## 統計と指標

有効化すると、メッセージ変換はダッシュボード上で統計と指標を公開します。メッセージ変換ページで変換名をクリックすると、以下を確認できます。

**統計情報**：

- **Total**：システム起動以来のトリガー総数
- **Success**：成功したデータ変換の数
- **Failed**：失敗したデータ変換の数

**レート指標**：

- 現在の変換速度
- 過去5分間の速度
- 過去の最大速度

統計はリセット可能で、Prometheusの `/prometheus/message_transformation` からも取得可能です。