Stompプロトコル
StompプロトコルはApache ActiveMQなどのメッセージキュー管理サーバで利用されている、メッセージをやりとりするためのプロトコルです。ここで、Stompプロトコルで通信するためのスクリプトを書く方法について、少し触れておきます。
前回述べたとおり、ActivemqアダプタはStompプロトコルでメッセージキューサーバと通信していますが、この処理は内部ではZend_Queue_Stomp関係のクラスを利用して実装されています。
Zend_Queue_Stomp関係のクラスにはZend_Queue_Stomp_ClientクラスとZend_Queue_Stomp_Frameクラスが存在します。メッセージをやりとりするサーバとクライアントの接続の単位を管理するのがZend_Queue_Stomp_Clientクラスで、Stompプロトコルでの通信の単位であるフレームに対応するのがZend_Queue_Stomp_Frameクラスです。
Zend_Queue_Stomp関係のクラスを利用する手順は
- （1）クライアントの初期化
- （2）Stompプロトコルによるサーバとの通信
となります。ここではローカルホストのポート「61613」に対してStompプロトコルでメッセージを登録する例を見てみます。
<?php
require_once 'Zend/Queue/Stomp/Frame.php';
require_once 'Zend/Queue/Stomp/Client.php';
/* （1）クライアントの初期化 */
$client = new Zend_Queue_Stomp_Client('tcp', 'localhost', '61613');
/* （2）サーバとの通信 */
/* （2-1）接続 */
$connect = $client->createFrame();
$connect->setCommand('CONNECT');
$response = $client->send($connect)->receive();
/* （2-2）メッセージの登録 */
$frame = $client->createFrame();
$message = "Hello World!!";
$frame->setCommand('SEND');
$frame->setHeader('destination', '/queue/codezine_queue');
$frame->setHeader('content-length', strlen($message));
$client->send($frame);
このように（1）Zend_Queue_Stomp_Clientクラスのコンストラクタでサーバへの接続方法について指定した後（2）サーバに送りたい情報を順番にStompプロトコルで送信して行きます。ここでは（2-1）サーバとの接続を確立した後（2-2）サーバに「Hello World!!」のメッセージを登録しています。
アダプタの自作
既存のアダプタが準備されていないメッセージキューを利用するためには、自分でアダプタを作成する必要があります。似ているアダプタがある場合には、そのアダプタを継承／拡張するのが良いでしょう。
全く新しいアダプタを作成する場合には、Zend_Queue_Adapter_AdapterAbstractクラスを継承して、必要な機能を実装していきます。ここではファイルにテキストでメッセージを格納するメッセージキューを例に説明します。なお、話を単純にするため単一のキューのみを扱えるものとし、またメッセージが取得された段階でキューから取り除かれるとします。
まず、実装する必要のあるメソッドの一覧を見てみます。
| 分類 | メソッド | 機能 |
| 基本 | __construct | コンストラクタ |
| キュー | create | キューを作成 |
| delete | キューを削除 | |
| isExists | キューが存在するかを取得 | |
| getQueues | キューの一覧を取得 | |
| count | キュー内にあるメッセージを取得 | |
| メッセージ | send | メッセージを追加 |
| receive | メッセージを取得 | |
| deleteMessage | メッセージを削除 |
今回は単一のキューのみを扱うため、上の表の「キューの管理」にあるメソッドは実装しません。また、deleteMessageメソッド／countメソッドも実装しません。また、キューを記録しておくファイルの名前は、オプションの「driverOptions.file」で与えることにします。
では、まずコンストラクタから見ていきます。
class Zend_Queue_Adapter_File extends Zend_Queue_Adapter_AdapterAbstract
{
public function __construct($options, Zend_Queue $queue = null)
{
/* （1）AdapterAbstract のコンストラクタを呼ぶ
* この中で $options の解析が行われる
*/
parent::__construct($options, $queue);
/* （2）ここでキューの初期化 */
}
まず（1）コンストラクタでは親クラスのコンストラクタを呼び出しています。この中では、引数の$option変数を解析し、アダプタのオプションを$this->_options['adapterOptions']に、メッセージキュー管理システムへの引数を$this->_options['driverOptions']へ設定しています。
また、キューの初期化等の作業が必要な場合には（2）で行います。今回のサンプルでは特に何も行いませんが、例えばDbアダプタではデータベースへの接続はここで行なっています。
次にメッセージの登録と取得です。
public function send($message, Zend_Queue $queue=null)
{
/* ファイルを追記用に開く */
$fp = fopen($this->_options['driverOptions']['file'], 'a');
/* ファイルをロック */
flock($fp, LOCK_EX);
/* メッセージの追記 */
fputs($fp, $message."\n");
/* ロックの開放 */
flock($fp, LOCK_UN);
/* ファイルを閉じる */
fclose($fp);
}
メッセージの登録は簡単で、ファイルにメッセージ追記するだけです。他のプロセスがファイルへ同時にアクセスしないようにするため、排他的なロックをかけています。
メッセージの取得も基本的に同じような処理になりますが
- 取得したメッセージを削除したい
- 取得したメッセージを正しい型で返さないといけない
ため、処理がやや面倒になっています。
public function receive($maxMessages = null, $timeout = null, Zend_Queue $queue = null)
{
if ($queue === null) {
$queue = $this->_queue;
}
/* （1）ファイルを開く */
$count = 0;
$fp = fopen($this->_options['driverOptions']['file'], 'r');
$fp_new = fopen('queue_new', 'w');
flock($fp, LOCK_EX);
flock($fp_new, LOCK_EX);
/* （2）ファイルの各行を取得 */
while (!feof($fp)) {
$line = fgets($fp);
if ($count < $maxMessages) {
/* （2-1）取得上限に達していないならメッセージを取得 */
$count++;
$message = rtrim($line, "\n");
/* （2-2）メッセージの準備 */
$msg['body'] = $message;
$data[] = $msg;
} else {
/* （2-3）取得上限に達しているならそのまま残す */
fputs($fp_new, $line);
}
}
/* （3）ロック等の開放 */
/* （3-1）一時的なロック */
$fp_tmp = fopen('queue_tmp', 'w');
flock($fp_tmp, LOCK_EX);
/* （3-2）ロック等の開放 */
flock($fp, LOCK_UN);
flock($fp_new, LOCK_UN);
fclose($fp);
fclose($fp_new);
/* （4）残すメッセージを含むファイルに旧ファイルを置き換える */
rename('queue_new', $this->_options['driverOptions']['file']);
flock($fp_tmp, LOCK_UN);
fclose($fp_tmp);
/* （5）返り値の準備 */
$options = array(
'queue' => $queue,
'data' => $data,
'messageClass' => $queue->getMessageClass(),
);
$classname = $queue->getMessageSetClass();
if (!class_exists($classname)) {
require_once 'Zend/Loader.php';
Zend_Loader::loadClass($classname);
}
return new $classname($options);
}
このうち（1）～（4）までがメッセージの取得と取得済メッセージの削除、（5）が返り値の準備になっています（図1_1、図1_2）。
（1）では2つのファイルを開いていますが、これはそれぞれメッセージを読み込むためのファイル（$fp）と、キューに残しておくためのメッセージを記録するためのファイル（$fp_new）です。（2）では、メッセージの記録されているファイルを一行一行見て、メッセージ取得数の上限に達していない場合には（2-1）メッセージを取得して（2-2）で配列$msgに必要な情報（'body'）を記録しています。
一方で、メッセージ取得数の上限に達している場合には（2-3）でキューに残すメッセージとして$fp_newに記録しています。
（4）ではキューに残すメッセージが記録されたファイルでメッセージキューが記録されているファイルを置き換えています。このことで、実質、取得されたメッセージが削除されます。
（5）では、得られたメッセージをreceiveメソッドの返り値としてふさわしいように、Zend_Queue_Message_Iteratorクラスのオブジェクトに格納しています（このreceiveメソッドの返り値については後から少し説明します）。
