第3章 AMQ への接続

インテグレーションでは、Red Hat AMQ ブローカーからメッセージを取得したり、Red Hat AMQ ブローカーへメッセージを公開することができます。Red Hat AMQ は、クライアントとメッセージブローカーとの間の通信に OpenWire プロトコルを使用します。以下のブローカータイプと通信するには、Red Hat AMQ コネクターを使用して、対象のブローカーへのコネクションを作成します。

  • AMQP をサポートしない Apache ActiveMQ ブローカー
  • AMQ 6 ブローカー

以下のブローカータイプの 1 つと通信するには、AMQP コネクターを使用 して、対象のブローカーへのコネクションを作成します。

  • AMQP をサポートする Apache ActiveMQ ブローカー
  • Apache ActiveMQ Artemis
  • AMQ 7 ブローカー
  • EnMasse (オープンソースのメッセージングプラットフォーム)

Red Hat AMQ コネクターを使用するには、以下を参照してください。

3.1. AMQ コネクションの作成

インテグレーションでは、以下からメッセージを取得したり、以下にメッセージを公開することがあります。

  • AMQP をサポートしない Apache ActiveMQ ブローカー
  • AMQ 6 ブローカー

このような場合、インテグレーションに追加できる Red Hat AMQ コネクションを作成します。

前提条件

接続する Red Hat AMQ ブローカーには以下が必要となります。

  • ブローカーの URL
  • ユーザーアカウントのクレデンシャル
  • ブローカーの PEM 形式の証明書

手順

  1. Fuse Online の左パネルで Connections をクリックし、利用可能なコネクションを表示します。
  2. Create Connection をクリックしてコネクターを表示します。
  3. Red Hat AMQ コネクターを表示します。
  4. コネクションを設定します。

    1. Broker URL フィールドにデータの送信先またはデータの取得元の場所を入力します (例: tcp://localhost:61616)。
    2. User name フィールドに、このブローカーへアクセスするために使用するアカウントのユーザー名を入力します。
    3. Password フィールドに、このブローカーへアクセスするために使用するアカウントのパスワードを入力します。
    4. Client ID フィールドに、メッセージを見逃さずにコネクションを開閉できるようにする ID を入力します。宛先のタイプはトピックにする必要があります。
    5. このコネクションが開発環境で使用される場合は、Check certificates を無効にすると時間を一部節約することができます。証明書確認の無効化は開発環境で便利です。セキュアな本番環境では、常に Check certificates を有効にするようにしてください。
    6. Broker certificate フィールドに Red Hat AMQ ブローカーの PEM 形式の証明書テキストを貼り付けます。これは、証明書確認を無効にした場合以外は必須です。
    7. Client certificate フィールドに Red Hat クライアントの PEM 形式の証明書テキストを貼り付けます。このフィールドは常に任意です。
  5. Validate をクリックします。Fuse Online は即座にコネクションを検証しようとし、検証の成功または失敗を示すメッセージを表示します。検証に失敗した場合は、必要に応じて設定の詳細を修正し、再度検証を行います。
  6. 検証に成功した場合は Next をクリックします。
  7. Name フィールドに、このコネクションを別のコネクションと区別するために使用する名前を入力します。たとえば、Red Hat AMQ 1 を入力します。
  8. Description は任意のフィールドで、このコネクションに関する便利な情報を入力します。たとえば、Sample Red Hat AMQ connection that uses a provided broker. を入力します。
  9. Save をクリックし、作成したコネクションが利用可能になったことを確認します。上記の例の名前を入力した場合は、インテグレーションに追加できるコネクションとして Red Hat AMQ 1 が表示されます。

3.2. AMQ コネクションを追加してメッセージ受信時にインテグレーションの実行をトリガー

Red Hat AMQ ブローカーからメッセージを受信したときにインテグレーションの実行をトリガーするには、Red Hat AMQ コネクションを最初のコネクションとして追加します。

前提条件

メッセージの取得元となる Red Hat AMQ ブローカーへの Red Hat AMQ コネクションが作成されている必要があります。

手順

  1. Fuse Online パネルの左側にある Integrations をクリックします。
  2. Create Integration をクリックします。
  3. Choose a connection ページで、インテグレーションを開始するために使用する Red Hat AMQ コネクションをクリックします。
  4. Choose an action ページで、Subscribe for messages アクションを選択し、指定したキューまたはトピックからメッセージを受信します。
  5. アクションを設定するには以下を行います。

    1. Destination name フィールドに、データの受信元となるキューまたはトピックの名前を入力します。
    2. Destination type では Queue を許可するか、Topic を選択します。
    3. メッセージを見逃さずにコネクションを開閉できるようにするため、Durable subscription ID フィールドに永続サブスクリプション ID を入力します。宛先のタイプはトピックにする必要があります。
    4. 特定の条件を満たすデータのみを受信する場合は、Message selector フィールドにフィルター式を入力します。

      メッセージセレクターは式が含まれる文字列です。式の構文は、SQL92 条件式構文のサブセットを基にしています。以下の例のメッセージセレクターは、値が Sports または Opinion に設定された NewsType プロパティーがあるメッセージをすべて選択します。

      NewsType = ’Sports’ OR NewsType = ’Opinion’

      メッセージコンシューマーは、ヘッダーとプロパティーがメッセージセレクターの式と一致するメッセージのみを受信します。メッセージセレクターは、メッセージのボディー部の内容を基にしてメッセージを選択することはできません。

  6. Next をクリックして、アクションの出力タイプを指定します。
  7. データタイプの認識が必要がない場合は、Select Type フィールドに Type specification not required を使用し、Next をクリックします。残りの手順を行う必要はありません。

    ただし、構造化データタイプが推奨されます。たとえば、コネクション出力をデータマッパーステップでマップする場合、データタイプを指定する必要があります。データマッパーは、非構造化データのフィールドを表示できません。

    データタイプを指定するには、Select Type フィールドをクリックし、以下の 1 つをスキーマタイプとして選択します。

    • JSON schema は JSON データの構造を記述するドキュメントです。ドキュメントのメディアタイプは application/schema+json です。
    • JSON instance は JSON データが含まれるドキュメントです。ドキュメントのメディアタイプは application/json です。
    • XML schema は XML データの構造を記述するドキュメントです。このドキュメントのファイル拡張子は .xsd です。
    • XML instance は XML データが含まれるドキュメントです。このドキュメントのファイル拡張子は .xml です。
  8. Definition 入力ボックスに、選択したスキーマタイプに準拠する定義を貼り付けます。たとえば、JSON schema を選択した場合は、メディアタイプが application/schema+json の JSON スキーマファイルの内容を貼り付けます。
  9. Data Type Name フィールドにはデータタイプに選択した名前を入力します。たとえば、ベンダーの JSON スキーマを指定する場合は、Vendor をデータタイプ名として指定することができます。

    このタイプを指定するコネクションを使用するインテグレーションを作成または編集するときにこのデータタイプ名が表示されます。Fuse Online では、インテグレーションビジュアライゼーションとデータマッパーにタイプ名が表示されます。

  10. Data Type Description フィールドには、このタイプを区別するのに便利な情報を提供します。データマッパーでこのタイプを処理するステップにカーソルを合わせると、この説明が表示されます。
  11. Next をクリックします。

結果

インテグレーションフローの最初にコネクションが表示されます。

3.3. AMQ メッセージを公開してインテグレーションを終了

メッセージを Red Hat AMQ ブローカーに公開してシンプルなインテグレーションを終了するには、Red Hat AMQ コネクションをシンプルなインテグレーションの最後のコネクションとして追加します。

前提条件

  • メッセージの公開先となる Red Hat AMQ ブローカーへのコネクションが作成されている必要があります。
  • インテグレーションを作成することになり、Fuse Online は最後のコネクションの選択を要求します。または、インテグレーションを変更して最後のコネクションを変更することになります。

手順

  1. Choose a connection ページで、インテグレーションの終了に使用する Red Hat AMQ コネクションをクリックします。
  2. Choose an action ページで、Publish messages アクションを選択し、指定したキューまたはトピックにメッセージを公開します。
  3. Destination name フィールドに、メッセージの送信先となるキューまたはトピックの名前を入力します。
  4. Destination type では Queue を許可するか、Topic を選択します。
  5. Persistent を選択し、コネクションが失敗した場合でもメッセージが確実に送信されるようにします。
  6. Next をクリックし、アクションの入力および出力タイプを指定します。
  7. データタイプの認識が必要がない場合は、Select Type フィールドに Type specification not required を使用し、Next をクリックします。残りの手順を行う必要はありません。

    ただし、構造化データタイプが推奨されます。たとえば、コネクションの入出力をデータマッパーステップでマップする場合、データタイプを指定する必要があります。データマッパーは、非構造化データのフィールドを表示できません。

    データタイプを指定するには、Select Type フィールドをクリックし、以下の 1 つをスキーマタイプとして選択します。

    • JSON schema は JSON データの構造を記述するドキュメントです。ドキュメントのメディアタイプは application/schema+json です。
    • JSON instance は JSON データが含まれるドキュメントです。ドキュメントのメディアタイプは application/json です。
    • XML schema は XML データの構造を記述するドキュメントです。このドキュメントのファイル拡張子は .xsd です。
    • XML instance は XML データが含まれるドキュメントです。このドキュメントのファイル拡張子は .xml です。
  8. Definition 入力ボックスに、選択したスキーマタイプに準拠する定義を貼り付けます。たとえば、JSON schema を選択した場合は、メディアタイプが application/schema+json の JSON スキーマファイルの内容を貼り付けます。
  9. Data Type Name フィールドにはデータタイプに選択した名前を入力します。たとえば、ベンダーの JSON スキーマを指定する場合は、Vendor をデータタイプ名として指定することができます。

    このタイプを指定するコネクションを使用するインテグレーションを作成または編集するときにこのデータタイプ名が表示されます。Fuse Online では、インテグレーションビジュアライゼーションとデータマッパーにタイプ名が表示されます。

  10. Data Type Description フィールドには、このタイプを区別するのに便利な情報を提供します。データマッパーでこのタイプを処理するステップにカーソルを合わせると、この説明が表示されます。
  11. Next をクリックします。

結果

インテグレーションビジュアライゼーションの最後にコネクションが表示されます。

3.4. インテグレーションの途中に AMQ メッセージを公開

インテグレーションの途中に Red Hat AMQ ブローカーにメッセージを公開するには、Red Hat AMQ コネクションをフローの途中に追加します。

前提条件

  • メッセージの公開先となる Red Hat AMQ ブローカーへのコネクションが作成されている必要があります。
  • フローを作成または編集することになり、Fuse Online でインテグレーションに追加するよう要求されます。または、Fuse Online で最後のコネクションを選択するよう要求されます。

手順

  1. Add to Integration ページで、コネクションの追加先のプラス記号をクリックします。Fuse Online が最後のコネクションを選択するよう要求する場合は、このステップをスキップします。
  2. フローの途中に追加する Red Hat AMQ コネクションをクリックします。
  3. Choose an action ページで、以下のアクションの 1 つを選択します。

    • Publish messages は指定するキューまたはトピックにメッセージを公開します。このアクションを設定するには、以下を行います。

      1. Destination name フィールドに、メッセージの送信先となるキューまたはトピックの名前を入力します。
      2. Destination type では Queue を許可するか、Topic を選択します。
      3. Persistent を選択し、コネクションが失敗した場合でもメッセージが確実に送信されるようにします。
    • Request response using messages で、応答の受信に指定した JMS 宛先にメッセージを送ります。このアクションを設定するには、以下を行います。

      1. Destination name フィールドに、メッセージの送信先となるキューまたはトピックの名前を入力します。
      2. Destination type では Queue を許可するか、Topic を選択します。
      3. 特定の条件を満たす応答のみを受信する場合は、Message selector フィールドにフィルター式を入力します。

        メッセージセレクターは式が含まれる文字列です。式の構文は、SQL92 条件式構文のサブセットを基にしています。以下の例のメッセージセレクターは、値が Sports または Opinion に設定された NewsType プロパティーがあるメッセージをすべて選択します。

        NewsType = ’Sports’ OR NewsType = ’Opinion’

        メッセージコンシューマーは、ヘッダーとプロパティーがメッセージセレクターの式と一致するメッセージのみを受信します。メッセージセレクターは、メッセージのボディー部の内容を基にしてメッセージを選択することはできません。

      4. Named reply to フィールドには、キューまたはトピックの名前を入力します。宛先によって、応答がこのキューまたはトピックに送信されます。
      5. Persistent を選択し、コネクションが失敗した場合でもメッセージが確実に送信されるようにします。
      6. Response time out フィールドには、ランタイム例外をスローする前にこのコネクションが応答メッセージを待つ期間 (ミリ秒単位) を指定します。デフォルトは 5000 ミリ秒 (5 秒)です。
  4. Next をクリックし、アクションの入力タイプを指定してから、アクションの出力タイプを指定します。
  5. データタイプの認識が必要がない場合は、Select Type フィールドに Type specification not required を使用し、Next をクリックします。残りの手順を行う必要はありません。

    ただし、構造化データタイプが推奨されます。たとえば、コネクションの入出力をデータマッパーステップでマップする場合、データタイプを指定する必要があります。データマッパーは、非構造化データのフィールドを表示できません。

    データタイプを指定するには、Select Type フィールドをクリックし、以下の 1 つをスキーマタイプとして選択します。

    • JSON schema は JSON データの構造を記述するドキュメントです。ドキュメントのメディアタイプは application/schema+json です。
    • JSON instance は JSON データが含まれるドキュメントです。ドキュメントのメディアタイプは application/json です。
    • XML schema は XML データの構造を記述するドキュメントです。このドキュメントのファイル拡張子は .xsd です。
    • XML instance は XML データが含まれるドキュメントです。このドキュメントのファイル拡張子は .xml です。
  6. Definition 入力ボックスに、選択したスキーマタイプに準拠する定義を貼り付けます。たとえば、JSON schema を選択した場合は、メディアタイプが application/schema+json の JSON スキーマファイルの内容を貼り付けます。
  7. Data Type Name フィールドにはデータタイプに選択した名前を入力します。たとえば、ベンダーの JSON スキーマを指定する場合は、Vendor をデータタイプ名として指定することができます。

    このタイプを指定するコネクションを使用するインテグレーションを作成または編集するときにこのデータタイプ名が表示されます。Fuse Online では、インテグレーションビジュアライゼーションとデータマッパーにタイプ名が表示されます。

  8. Data Type Description フィールドには、このタイプを区別するのに便利な情報を提供します。データマッパーでこのタイプを処理するステップにカーソルを合わせると、この説明が表示されます。
  9. Next をクリックします。

結果

インテグレーションビジュアライゼーションで追加した場所にコネクションが表示されます。