1.3. コネクション内のデータへの操作の適用

Kamelet とイベントチャネルの間に渡すデータで操作を実行する場合は、Kamelet Binding 内の中間ステップとしてアクション Kamelets を使用します。たとえば、アクション Kamelet を使用して、データをシリアライズまたはデシリアライズしたり、データをフィルターしたり、フィールドやメッセージヘッダーを挿入したりできます。

フィールドのフィルターリングや追加などの操作操作は、JSON データ (Content-Type ヘッダーが application/json に設定されている場合) でのみ動作します。イベントデータが JSON 以外の形式 (Avro または Protocol Buffers など) を使用する場合は、操作アクションおよびその後のシリアライズステップ (例: protobuf-deserialize-action または avro-deserialize-action Kamelet を参照する) の前にデシリアライズステップ (例: protobuf-serialize-action または avro-serialize-action Kamelet を参照する) を追加して、データの形式を変換する必要があります。接続のデータ形式を変換する方法は、Data conversion Kamelets を参照してください。

アクション Kamelets には以下が含まれます。

1.3.1. Kamelet Binding への操作の追加

アクション Kamelet を実装するには、Kamelet Binding ファイルの spec セクションで、ソースセクションとシンクセクションの間に steps セクションを追加します。

前提条件

  • Kamelet Binding でのソースおよびシンクコンポーネントの接続 で説明されているように、Kamelet Binding を作成している。
  • Kamelet Binding に追加するアクション Kamelet とアクション Kamelet に必要なパラメーターを知っている必要があります。

    この手順の例では、predicate-filter-action Kamelet のパラメーターは string タイプの式で、コーヒーデータをフィルターして "deep" taste intensity を持つコーヒーだけをログに記録するための JSON パス式を提供します。predicate-filter-action Kamelet では、Kamelet Binding に Builder トレイト設定プロパティーを設定する必要があります。

    この例には、イベントデータフォーマットが JSON であるため、この場合はオプションのデシリアライズおよびシリアライズアクションも含まれます。

手順

  1. エディターで KameletBinding ファイルを開きます。

    たとえば、以下は coffee-to-log.yaml ファイルの内容になります。

    apiVersion: camel.apache.org/v1alpha1
    kind: KameletBinding
    metadata:
      name: coffee-to-log
    spec:
      source:
        ref:
          kind: Kamelet
          apiVersion: camel.apache.org/v1alpha1
          name: coffee-source
        properties:
          period: 5000
      sink:
        ref:
          kind: Kamelet
          apiVersion: camel.apache.org/v1alpha1
          name: log-sink
  2. source セクションの上に integration セクションを追加し、(predicate-filter-action Kamelet で必要とされるように) 以下の Builder トレイト設定プロパティーを指定します。

    apiVersion: camel.apache.org/v1alpha1
    kind: KameletBinding
    metadata:
      name: coffee-to-log
    spec:
      integration:
             traits:
               builder:
                 configuration:
                   properties:
                     - "quarkus.arc.unremovable- types=com.fasterxml.jackson.databind.ObjectMapper"
      source:
        ref:
          kind: Kamelet
          apiVersion: camel.apache.org/v1alpha1
          name: coffee-source
        properties:
          period: 5000
      sink:
        ref:
          kind: Kamelet
          apiVersion: camel.apache.org/v1alpha1
          name: log-sink
  3. source セクションと sink セクションの間に steps セクションを追加し、アクション Kamelet を定義します。以下に例を示します。

    apiVersion: camel.apache.org/v1alpha1
    kind: KameletBinding
    metadata:
      name: coffee-to-log
      spec:
        integration:
               traits:
                 builder:
                   configuration:
                     properties:
                       - "quarkus.arc.unremovable-types=com.fasterxml.jackson.databind.ObjectMapper"
      source:
        ref:
          kind: Kamelet
          apiVersion: camel.apache.org/v1alpha1
          name: coffee-source
        properties:
          period: 5000
      steps:
      - ref:
          kind: Kamelet
          apiVersion: camel.apache.org/v1alpha1
          name: json-deserialize-action
      - ref:
          kind: Kamelet
          apiVersion: camel.apache.org/v1alpha1
          name: predicate-filter-action
        properties:
          expression: "@.intensifier =~ /.*deep/"
      - ref:
          kind: Kamelet
          apiVersion: camel.apache.org/v1alpha1
          name: json-serialize-action
      sink:
        ref:
          kind: Kamelet
          apiVersion: camel.apache.org/v1alpha1
          name: log-sink
  4. 変更を保存します。
  5. 以下のように、oc apply コマンドを使用して KameletBinding リソースを更新します。

    oc apply -f coffee-to-log.yaml

    Camel K Operator は再生成し、更新された KameletBinding リソースに基づいて生成する CamelK インテグレーションを実行します。

  6. Kamelet Binding のステータスを表示するには、以下を実行します。

    oc get kameletbindings

  7. 対応するインテグレーションのステータスを表示するには、次のコマンドを実行します。

    oc get integrations

  8. インテグレーションのログファイルの出力を表示するには、以下を実行します。

    kamel logs <integration-name>

    たとえば、インテグレーション名が coffee-to-log の場合は、以下のコマンドを使用します。

    kamel logs coffee-to-log

  9. インテグレーションを停止するには、Kamelet Binding を削除します。

    oc delete kameletbindings/<kameletbinding-name>

    以下に例を示します。

    oc delete kameletbindings/coffee-to-log