AMQ JavaScrip は、以下の業界標準およびネットワークプロトコルをサポートします。

AMQ JavaScript でサポートされている設定については、Red Hat カスタマーポータルの「Red Hat AMQ 7 でサポートされる構成」を参照してください。

本セクションでは、コア API エンティティーを紹介し、それらが一緒に操作する方法を説明します。

AMQ JavaScript は メッセージを送受信します。メッセージは、送信側と受信側を介して接続されたピア間で転送されます。送信側およびレシーバーは セッション 上で確立されます。セッションはコネクションを介して確立されます。接続は、一意に識別された 2 つのコンテナー間で確立されます。コネクションには複数のセッションを含めることができますが、多くの場合、これは必要ありません。API を使用すると、セッションが必要でない限り、セッションを無視できます。

送信ピアは、メッセージを送信するために送信者を作成します。送信側には、リモートピアでキューまたはトピックを識別する ターゲット があります。受信ピアは、メッセージを受信するための受信側を作成します。受信側には、リモートピアでキューまたはトピックを識別する ソース があります。

メッセージの送信は、配信 と呼ばれます。メッセージは送信される内容で、ヘッダーやアノテーションなどのすべてのメタデータが含まれます。配信は、そのコンテンツの移動に関連するプロトコルエクスチェンジです。

配信が完了したことを示すには、送信側または受信側セットのいずれかです。これが設定されていることを知らせると、その配信に関する通信はなくなります。受信側は、メッセージを受諾または拒否するかどうかを指定することもできます。

sudo コマンド

本書では、root 権限を必要とするコマンドには sudo が使用されています。何らかの変更がシステム全体に影響する可能性があるため、sudo を使用する場合は注意が必要です。sudo の詳細は、「sudo コマンドの使用」を参照してください。

ファイルパス

本書では、すべてのファイルパスが Linux、UNIX、および同様のオペレーティングシステムで有効です (例: /home/andrea)。Microsoft Windows では、同等の Windows パスを使用する必要があります (例: C:\Users\andrea)。

変数テキスト

本書には、実際の環境に固有の値に置き換える必要がある変数を含むコードブロックが含まれています。変数テキストは中括弧で囲まれ、斜体の等幅フォントとしてスタイル設定されます。たとえば、以下の例では、<project-dir> を実際の環境の値に置き換えます。

$ cd <project-dir>

.zip ファイルの内容を展開すると、amq-clients-2.10.0-javascript という名前のディレクトリーが作成されます。これはインストールの最上位ディレクトリーであり、本書全体で <install-dir> と呼ばれます。

インストールされたライブラリーを使用するように環境を設定するには、node_modules ディレクトリーを NODE_PATH 環境変数に追加します。

$ cd amq-clients-2.10.0-javascript
$ export NODE_PATH=$PWD/node_modules:$NODE_PATH

新しいコンソールセッションすべてでこの設定を有効にするには、$HOME/.bashrc ファイルに NODE_PATH を設定します。

インストールをテストするには、次のコマンドを使用します。インストールされたライブラリーを正常にインポートした場合は、OK をコンソールに出力します。

$ node -e 'require("rhea")' && echo OK
OK

.zip ファイルの内容を展開すると、amq-clients-2.10.0-javascript という名前のディレクトリーが作成されます。これはインストールの最上位ディレクトリーであり、本書全体で <install-dir> と呼ばれます。

インストールされたライブラリーを使用するように環境を設定するには、node_modules ディレクトリーを NODE_PATH 環境変数に追加します。

$ cd amq-clients-2.10.0-javascript
$ set NODE_PATH=%cd%\node_modules;%NODE_PATH%

AMQ JavaScript は Web ブラウザー内で実行できます。ブラウザーと互換性のあるライブラリーのバージョンを作成するには、npm run browserify コマンドを使用します。

$ cd amq-clients-2.10.0-javascript/node_modules/rhea
$ npm install
$ npm run browserify

これにより、rhea.js という名前のファイルが作成され、ブラウザーベースのアプリケーションで使用できます。

Hello World の例では、ブローカーへの接続を作成し、グリーティングが含まれるメッセージを examples キューに送信し、それを受け取ります。成功すると、受け取ったメッセージをコンソールに出力します。

examples ディレクトリーに移動し、helloworld.js の例を実行します。

$ cd <install-dir>/node_modules/rhea/examples
$ node helloworld.js
Hello World!

Hello World の例では、ブローカーへの接続を作成し、グリーティングが含まれるメッセージを examples キューに送信し、それを受け取ります。成功すると、受け取ったメッセージをコンソールに出力します。

examples ディレクトリーに移動し、helloworld.js の例を実行します。

> cd <install-dir>/node_modules/rhea/examples
> node helloworld.js
Hello World!

このクライアントプログラムは、<connection-url> を使用してサーバーに接続します。ターゲット <address> の送信側は <message-body> が含まれるメッセージを送信し、接続を閉じて終了します。

"use strict";

var rhea = require("rhea");
var url = require("url");

if (process.argv.length !== 5) {
    console.error("Usage: send.js <connection-url> <address> <message-body>");
    process.exit(1);
}

var conn_url = url.parse(process.argv[2]);
var address = process.argv[3];
var message_body = process.argv[4];

var container = rhea.create_container();

container.on("sender_open", function (event) {
    console.log("SEND: Opened sender for target address '" +
                event.sender.target.address + "'");
});

container.on("sendable", function (event) {
    var message = {
        body: message_body
    };

    event.sender.send(message);

    console.log("SEND: Sent message '" + message.body + "'");

    event.sender.close();
    event.connection.close();
});

var opts = {
    host: conn_url.hostname,
    port: conn_url.port || 5672,
    // To connect with a user and password:
    // username: "<username>",
    // password: "<password>",
};

var conn = container.connect(opts);
conn.open_sender(address);

サンプルの実行

サンプルプログラムを実行するには、これをローカルファイルにコピーし、node コマンドを使用してこれを呼び出します。詳細は、3章スタートガイド を参照してください。

$ node send.js amqp://localhost queue1 hello

このクライアントプログラムは <connection-url> を使用してサーバーに接続し、ソース <address> のレシーバーを作成し、終了するか <count> メッセージに到達するまでメッセージを受信します。

"use strict";

var rhea = require("rhea");
var url = require("url");

if (process.argv.length !== 4 && process.argv.length !== 5) {
    console.error("Usage: receive.js <connection-url> <address> [<message-count>]");
    process.exit(1);
}

var conn_url = url.parse(process.argv[2]);
var address = process.argv[3];
var desired = 0;
var received = 0;

if (process.argv.length === 5) {
    desired = parseInt(process.argv[4]);
}

var container = rhea.create_container();

container.on("receiver_open", function (event) {
    console.log("RECEIVE: Opened receiver for source address '" +
                event.receiver.source.address + "'");
});

container.on("message", function (event) {
    var message = event.message;

    console.log("RECEIVE: Received message '" + message.body + "'");

    received++;

    if (received == desired) {
        event.receiver.close();
        event.connection.close();
    }
});

var opts = {
    host: conn_url.hostname,
    port: conn_url.port || 5672,
    // To connect with a user and password:
    // username: "<username>",
    // password: "<password>",
};

var conn = container.connect(opts);
conn.open_receiver(address);

サンプルの実行

サンプルプログラムを実行するには、これをローカルファイルにコピーし、python コマンドを使用してこれを呼び出します。詳細は、3章スタートガイド を参照してください。

$ node receive.js amqp://localhost queue1

AMQ JavaScript は非同期イベント駆動型 API です。アプリケーションがイベントを処理する方法を定義するには、ユーザーが container オブジェクトのイベント処理機能を登録します。これらの機能は、ネットワークアクティビティーとして呼び出され、タイマーが新規イベントをトリガーします。

var rhea = require("rhea");
var container = rhea.create_container();

container.on("sendable", function (event) {
    console.log("A message can be sent");
});

container.on("message", function (event) {
    console.log("A message is received");
});

これらはいくつかの一般的なケースイベントのみです。完全セットは AMQ JavaScript API リファレンス に文書化されています。

event 引数には、イベントが関連するオブジェクトにアクセスするための属性があります。たとえば、connection_open イベントはイベント connection 属性を設定します。

イベントのプライマリーオブジェクトに加えて、イベントのコンテキストを形成するすべてのオブジェクトも設定されます。特定のイベントに対する関連性のない属性は null です。

event.container
event.connection
event.session
event.sender
event.receiver
event.delivery
event.message

コンテナーはトップレベルの API オブジェクトです。これは、接続を作成するエントリーポイントであり、メインのイベントループを実行します。多くの場合、これはグローバルイベントハンドラーで構築されます。

var rhea = require("rhea");
var container = rhea.create_container();

各コンテナーインスタンスには、コンテナー ID と呼ばれる一意のアイデンティティーがあります。AMQ JavaScript がネットワーク接続を作成したら、コンテナー ID をリモートピアに送信します。コンテナー ID を設定するには、id オプションを create_container メソッドに渡します。

var container = rhea.create_container({id: "job-processor-3"});

ユーザーが ID を設定しないと、コンテナーが構成されると、ライブラリーは UUID を生成します。

リモートサーバーに接続するには、ホストとポートを含む接続オプションを container.connect() メソッドに渡します。

container.on("connection_open", function (event) {
    console.log("Connection " + event.connection + " is open");
});

var opts = {
    host: "example.com",
    port: 5672
};

container.connect(opts);

デフォルトのホストは localhost です。デフォルトのポートは 5672 です。

セキュアな接続の作成に関する詳細は、「7章セキュリティー」を参照してください。

再接続すると、クライアントが失われた接続から回復できます。これは、一時的なネットワークまたはコンポーネントの障害後に、分散システムのコンポーネントが再確立された状態にするために使用されます。

AMQ JavaScript はデフォルトで再接続を有効にします。接続試行に失敗すると、クライアントは若干時間が経ってから再度試行します。遅延は、デフォルトの最大値 60 秒まで、新しい試行ごとに指数関数的に増加します。

再接続を無効にするには、reconnect 接続オプションを false に設定します。

var opts = {
    host: "example.com",
    reconnect: false
};

container.connect(opts);

接続試行間の遅延を制御するには、initial_reconnect_delay および max_reconnect_delay 接続オプションを設定します。遅延オプションはミリ秒単位で指定します。

再接続試行回数を制限するには、reconnect_limit オプションを設定します。

var opts = {
    host: "example.com",
    initial_reconnect_delay: 100,
    max_reconnect_delay: 60 * 1000,
    reconnect_limit: 10
};

container.connect(opts);

AMQ JavaScript を使用すると、代替の接続エンドポイントをプログラム的に設定できます。

複数の接続エンドポイントを指定するには、新しい接続オプションを返す関数を定義し、connection_details オプションで関数を渡します。この関数は、接続試行ごとに 1 回呼び出されます。

var hosts = ["alpha.example.com", "beta.example.com"];
var index = -1;

function failover_fn() {
    index += 1;

    if (index == hosts.length) index = 0;

    return {host: hosts[index].hostname};
};

var opts = {
    host: "example.com",
    connection_details: failover_fn
}

container.connect(opts);

この例では、ホストの一覧に対して、ラウンドロビンフェイルオーバーを繰り返すことを実装します。このインターフェースを使用すると、独自のフェイルオーバー動作を実装できます。

AMQ JavaScript はインバウンドネットワーク接続を受け入れ、カスタムメッセージングサーバーを構築できます。

接続のリッスンを開始するには、ローカルホストアドレスおよびリッスンするポートが含まれるオプションで container.listen() メソッドを使用します。

container.on("connection_open", function (event) {
    console.log("New incoming connection " + event.connection);
});

var opts = {
    host: "0.0.0.0",
    port: 5672
};

container.listen(opts);

特別な IP アドレス 0.0.0.0 は、利用可能なすべての IPv4 インターフェースでリッスンします。すべての IPv6 インターフェースをリッスンするには、[::0] を使用します。

詳細は、サーバー receive.js の例 を参照してください。

AMQ JavaScript は SSL/TLS を使用して、クライアントとサーバー間の通信を暗号化します。

SSL/TLS を使用してリモートサーバーに接続するには、transport 接続オプションを tls に設定します。

var opts = {
    host: "example.com",
    port: 5671,
    transport: "tls"
};

container.connect(opts);

AMQ JavaScript は、ユーザーとパスワードによる接続を認証できます。

認証に使用するクレデンシャルを指定するには、username および password 接続オプションを設定します。

var opts = {
    host: "example.com",
    username: "alice",
    password: "secret"
};

container.connect(opts);

AMQ JavaScript は SASL プロトコルを使用して認証を実行します。SASL は多くの異なる 認証 メカニズム を使用できます。2 つのネットワークピアに接続すると、許可されるメカニズムを交換し、両方で許可される最も強力なメカニズムが選択されます。

AMQ JavaScript は、ユーザーとパスワード情報の有無に基づいて SASL メカニズムを有効にします。ユーザーとパスワードの両方が指定されている場合は、PLAIN が使用されます。ユーザーのみが指定されている場合には、ANONYMOUS が使用されます。いずれも指定されていない場合、SASL は無効になります。

一部のメッセージサーバーは、キューとトピックのオンデマンド作成をサポートします。送信側またはレシーバーが割り当てられている場合、サーバーは送信側のターゲットアドレスまたは受信側ソースアドレスを使用して、アドレスに一致する名前を持つキューまたはトピックを作成します。

メッセージサーバーは通常、キュー (1 対 1 のメッセージ配信用) またはトピック （1 対多のメッセージ配信の場合) を作成します。クライアントは、ソースまたはターゲットに queue または topic 機能を設定することで、希望のものを指定できます。

キューまたはトピックセマンティクスを選択するには、以下の手順に従います。

var conn = container.connect({host: "example.com"});

var sender_opts = {
    target: {
        address: "jobs",
        capabilities: ["queue"]
    }
}

conn.open_sender(sender_opts);

var conn = container.connect({host: "example.com"});

var receiver_opts = {
    source: {
        address: "notifications",
        capabilities: ["topic"]
    }
}

conn.open_receiver(receiver_opts);

詳細は、以下の例を参照してください。

永続サブスクリプションは、メッセージの受信側を表すリモートサーバーの状態です。通常、クライアントが閉じられると、メッセージ受信側は破棄されます。ただし、永続サブスクリプションは永続的であるため、クライアントはこれらのサブスクリプションの割り当てを解除してから、後で再度アタッチすることができます。デタッチ中に受信したすべてのメッセージは、クライアントの再割り当て時に利用できます。

永続サブスクリプションは、クライアントコンテナー ID とレシーバー名を組み合わせてサブスクリプション ID を形成することで一意に識別されます。サブスクリプションが回復できるようにするには、これらの値に安定した値が必要です。

サブスクリプションからデタッチするには、receiver.detach() メソッドを使用します。サブスクリプションを終了するには、receiver.close() メソッドを使用します。

詳細は、durable-subscribe.js の例を参照してください。

共有サブスクリプションとは、1 つ以上のメッセージレシーバーを表すリモートサーバーの状態のことです。共有されているので、複数のクライアントは同じメッセージのストリームから消費できます。

クライアントは、レシーバーソースに shared 機能を設定して、共有サブスクリプションを設定します。

共有サブスクリプションは、クライアントコンテナー ID とレシーバー名を組み合わせてサブスクリプション ID を形成することで一意に識別されます。複数のクライアントプロセスで同じサブスクリプションを見つけることができるように、これらの値が安定している必要があります。shared に加えて global 機能が設定されている場合、レシーバー名のみを使用してサブスクリプションを特定します。

永続サブスクリプションを作成するには、以下の手順に従います。

サブスクリプションからデタッチするには、receiver.detach() メソッドを使用します。サブスクリプションを終了するには、receiver.close() メソッドを使用します。

詳細は、shared-subscribe.js の例を参照してください。

以下のイベントをインターセプトして、プロトコルレベルのエラーを処理できます。

これらのイベントは、イベントにある特定のオブジェクトにエラー状態が生じるたびに呼び出されます。エラーハンドラーを呼び出すと、対応する <object>_close ハンドラーも呼び出されます。

event 引数は、エラーオブジェクトにアクセスするための error 属性を持ちます。

container.on("error", function (event) {
    console.log("An error!", event.error);
});

AMQ JavaScript は JavaScript デバッグモジュール を使用してロギングを実装します。

たとえば、詳細なクライアントロギングを有効にするには、DEBUG 環境変数を rhea* に設定します。

$ export DEBUG=rhea*
$ <your-client-program>

クライアントは AMQP プロトコルフレームをコンソールに記録できます。通常、このデータは問題を診断する際に重要です。

プロトコルロギングを有効にするには、DEBUG 環境変数を rhea:frames に設定します。

$ export DEBUG=rhea:frames
$ <your-client-program>

設定された場合、AMQ JavaScript は MESSAGING_CONNECT_FILE 環境変数の値を使用して設定ファイルを見つけます。

MESSAGING_CONNECT_FILE が設定されていない場合、AMQ JavaScript は以下の場所で connect.json という名前のファイルを検索します。最初の一致で停止します。

connect.json ファイルが見つからない場合、ライブラリーはすべてのオプションにデフォルト値を使用します。

connect.json ファイルには JSON データが含まれ、JavaScript コメントの追加サポートが提供されます。

すべての設定属性は任意で、またはデフォルト値を持っているため、簡単な例にはいくつかの詳細のみが必要になります。

{
    "host": "example.com",
    "user": "alice",
    "password": "secret"
}

SASL および SSL/TLS オプションは、"sasl" および "tls" namespace で入れ子になっています。

{
    "host": "example.com",
    "user": "ortega",
    "password": "secret",
    "sasl": {
        "mechanisms": ["SCRAM-SHA-1", "SCRAM-SHA-256"]
    },
    "tls": {
        "cert": "/home/ortega/cert.pem",
        "key": "/home/ortega/key.pem"
    }
}

オプションキーは、ドット (.) を含む属性は namespace 内でネストされた属性を表します。

AMQP メッセージは AMQP タイプシステムを使用して構成されます。この一般的な形式を使用するのは、異なる言語の AMQP クライアントが、相互運用できることが理由です。

メッセージを送信する場合、AMQ JavaScript は自動的に言語ネイティブの型を AMQP エンコードデータに変換します。メッセージの受信時に、リバース変換が行われます。

JavaScript がエンコードできるネイティブタイプは、AMQP よりも少ないです。特定の AMQP タイプを含むメッセージを送信するには、rhea/types.js モジュールの wrap_ 関数を使用します。

AMQP は、JMS メッセージングモデルへの標準的なマッピングを定義します。本項では、そのマッピングのさまざまな側面について説明します。詳細は、「AMQ JMS 相互運用性」を参照してください。

JMS メッセージタイプ

AMQ JavaScript は、本文タイプが異なる、単一のメッセージを提供します。一方、JMS API は異なるメッセージタイプを使用して、さまざまな種類のデータを表します。以下の表は、特定のボディ型が JMS メッセージタイプにマッピングする方法を示しています。

結果として生成される JMS メッセージタイプの明示的な制御を行うために、x-opt-jms-msg-type メッセージアノテーションを設定できます。詳細は、「AMQ JMS 相互運用性」の章を参照してください。

AMQ Broker は AMQP 1.0 クライアントと相互運用するために設計されています。以下をチェックして、ブローカーが AMQP メッセージング用に設定されていることを確認します。

AMQ Interconnect は AMQP 1.0 クライアントと動作します。以下をチェックして、コンポーネントが正しく設定されていることを確認します。