RailsのActiveMessaging入門

はじめに

Railsはすでに、データベース駆動型のWebアプリケーションの領域でしっかりと旗を掲げています。RailsプラグインのフレームワークであるActiveMessagingは、簡易化したメッセージング統合を追加することでRails領域の境界を拡大します。ActiveMessagingとRailsを使用することで、MQメッセージを送信するメインフレームやJ2EE Webアプリケーションのように異種のシステムと簡易統合することや、長期タスクの処理をオフロードすることや、イベントまたはメッセージ駆動型のアーキテクチャでアプリケーションを作成することができます。以前に他の人が言ったように、ActiveMessagingは、ActiveRecordがデータベースに対して実行することを、メッセージングに対して実行しようとします。この記事では、ActiveMessagingと関連技術について紹介し、それをRailsアプリケーションで利用する方法を説明します。

メッセージング標準とブローカー

メッセージングを開始するために必要な3つのものは、通信プロトコル、このプロトコルを話すブローカー、プロトコルを話すアプリケーションのクライアントライブラリです。技術的には、4つ目として、メッセージを送信するシステムがありますが、この例では、説明だけのための少し不十分なシステムを構築します。プロトコルに関して、ActiveMessagingはStomp(リンク)（Streaming Text-Oriented Messaging Protocol）をサポートしています。Stompは、ワイヤ接続（通常はtcp/ip）を介してテキストベースのメッセージを交換するメッセージング標準です。その目的は、ほぼすべての言語で容易にクライアントおよびサーバーを作成できるようにプロトコルをシンプルにすることであったため（たとえば、Ruby Stompクライアントは約400行で、そのほとんどをコメントと空白行が占める）、ほぼどんなシステムでも統合することが可能です。

他のブローカーオプション(リンク)も利用可能で、常に増え続けていますが、Apache ActiveMQ(リンク)メッセージブローカーはStompをネイティブにサポートしています(リンク)。事実、最近発表されたStompConnect(リンク)プロジェクトはJMSブローカーをStompブローカーにします。これは、Stomp、およびActiveMessagingを、実質的にすべてのオープンソースと独占のメッセージング製品に解放します。クライアント側で、ActiveMessagingはアダプタパターンを使用してプロトコルをサポートし、Stompに関しては、そのアダプタになるようにStomp RubyGemを拡張します。さらなるアダプタのサポートが見込まれており計画中ですが、現在のところStompが唯一サポートされています。

この記事では、ActiveMQをブローカーとして使用し、まだリリースされていない4.2ブランチを扱います。ActiveMQには、すべてのプラットフォームに対するインストールガイド(リンク)が用意されていますが、ここでは、Java 1.4以降を利用していると仮定して単純な手順を用います。

cd /usr/local/src
 #unix and cygwin
 wget http://people.apache.org/repo/m2-snapshot-repository/org/apache/activemq/apache-activemq/4.2-SNAPSHOT/apache-activemq-4.2-20070221.081507-10-src.tar.gz
#os x
curl -O http://people.apache.org/repo/m2-snapshot-repository/org/apache/activemq/apache-activemq/4.2-SNAPSHOT/apache-activemq-4.2-20070221.081507-10-src.tar.gz
 
cd ..
tar xvfz apache-activemq-4.2-20070221.081507-10-src.tar.gz
cd apache-activemq-4.2-incubator-SNAPSHOT
 ./bin/activemq 

以上です! デフォルトで、ActiveMQのStompサポートは自動的に起動するように設定されるはずですが、設定されない場合は次の行をconf/activemq.xmlに追加する必要があります。

<transportConnectors> 
    ...
    <!-- Add this line -->
    <transportConnector name="stomp" uri="stomp://localhost:61613"/>
</transportConnectors> 

これは最小の設定であるため、ActiveMQをご自分の望みどおりに設定するには、ActiveMQのActiveMessaging wikiページ(リンク)、およびActiveMQ使用法のガイド(リンク)を参照してください。

Railsの設定

Rails 1.1以降とMySqlをすでに利用していると仮定して、さらに2つのRubyGems: daemons(リンク)およびStomp(リンク)をインストールする必要があります。

sudo gem install daemons
sudo gem install stomp 

新たに作られたRails「MessageMonster」というデータベースから始めて、ActiveMessagingプラグインをインストールします。

cd /usr/local/dev
rails MessageMonster
mysqladmin create messagemonster_development -u root
cd MessageMonster
script/plugin install http://activemessaging.googlecode.com/svn/trunk/plugins/activemessaging 

すべての前提条件が整ったので、アプリケーションを記述する段階に進みます。

メッセージの処理

「プロセッサ」を作成することから始めましょう。プロセッサは、ActiveMessagingの用語で、メッセージを受信するとそのメッセージに対して何らかの処理をするために宛先にサブスクライブ（subscribe）するクラスです。これは、メッセージ駆動型のコントローラに相当し、コードの大半を配置することになるモデルが伴います。

Railsからあなたは期待することを教わったように、プロセッサを作成するジェネレータがあります。ActiveRecord統合を示すための受信メッセージを保存するものを作成しましょう。

script/generate processor PersistMessage 

これを実行すると、次のような最初の出力が生成されます。

create  app/processors
create  app/processors/persist_message_processor.rb
create  config/messaging.rb
create  config/broker.yml
create  app/processors/application.rb
create  script/poller 

• app/processors は、すべてのプロセッサのディレクトリです。
• app/processors/persist_message_processor.rb は、作業用のデフォルトの実装による新規プロセッサです。
• app/processors/application.rb は、プロセッサの派生元の共通のスーパークラスです。これを使用して、エラー処理などの共有コードを配置します。
• config/broker.yml は、各環境におけるメッセージブローカーへの接続方法を設定します。これはdatabase.ymlと同等ですが、メッセージブローカー用です。
• config/messaging.rb は、宛先やメッセージヘッダーなど、残りのメッセージング設定に使用されます。
• script/poller は、メッセージを聴取するためにActiveMessagingによって使用されるデーモンです。このスクリプトについては後ほど詳しく説明します。

まずはbroker.yml lファイルを調べ、オプションを見て開発ブローカーの設定が正しいことを確認します。

development:
    adapter: stomp
    login: ""
    passcode: ""
    host: localhost
    port: 61613
    reliable: true
    reconnectDelay: 5
 ... 

これらの値はすべて、まさにデフォルトのActiveMQインストールに必要なものです。ただし本番環境では、ユーザーとパスワードのオプションも必要になるでしょうし、ブローカーをリモートで実行する場合は、違うホストとポートが必要になります。

では次に、persist_message_processor.rbに注目しましょう。

class PersistMessageProcessor < ApplicationProcessor
  subscribes_to :persist_message
  def on_message(message)
    logger.debug "PersistMessageProcessor received: " + message
  end
 end

デフォルトでこれは:persist_message にサブスクライブされ、on_messageという1つのメソッドを持つことが分かります。on_messageは、:persist_message サブスクリプションに対するメッセージが到着したときにActiveMessagingによって呼び出されます。メッセージの本文は文字列引数messageとして渡されますが、Stompメッセージオブジェクト自体に到達する必要がある場合は、@message プロセッサ属性でそれが可能です。プロセッサは、複数のキューにサブスクライブできます。単に別の呼び出しをsubscribes_toに追加するだけです。

:persist_message シンボルは、プロセッサのサブスクライブ先の論理名です。このサブスクライブ先の定義は、config/messaging.rbで作成されました。

ActiveMessaging::Gateway.define do |s|
  s.queue :persist_message, '/queue/PersistMessage'
 end 

Railsとの比較を用いると、messaging.rb はroutes.rbに類似します。論理名(:persist_message)をブローカーの宛先(/queue/PersistMessage)にマップすることができます。ここで設定できるものはさらに多くありますが、生成されたデフォルトはこの例に十分適しています。

次に、まさかの時に備えてメッセージを保存するようプロセッサに教え込むため、メッセージ用のモデルを作成することから取り掛かります。

script/generate model message 

ここで、次の2つのカラムを追加して「db/migrate/001_create_messages.rb」を更新します。

create_table :messages do |t|
  t.column :body, :text
  t.column :received_date, :datetime
 end 

migrateを呼び出してテーブルを作成します...

rake db:migrate 

...続いて、プロセッサのon_messageメソッドを更新し、新しいMessageモデルを使用します。

def on_message(message)
  logger.debug "PersistMessageProcessor received: " + message
  my_message = Message.create(:body=>message, :received_date=>DateTime.new)
 end 

これを見て、何も目新しいことではないと思うでしょう。Railsコントローラで同じものを記述する方法とたいして違いはありません。ポイントは、ActiveMessagingを使用する場合、Railsアプリケーション環境の利便性がすべて利用可能であることです。

メッセージをキャッチするプロセッサを準備したら、メッセージを（個別に瓶詰めして売るかのように）スローする段階に進みます。

メッセージの送信

メッセージを送信するには、同じアプリケーションのRailsビューとコントローラを使用します。

script/generate controller SendMessage index

メッセージテキストをサブミットするように、そして、これまでに永続化されたメッセージの更新可能なリストを表示するように、ビューを編集します。

<p style="color: green"><%= flash[:notice] %></p> <h1>Send Message</h1> <%= start_form_tag :action => 'new' %> <label>Message</label> <%=text_field_tag :message, @message %>  <%= submit_tag "submit" %> <%= end_form_tag %> <%= link_to "Refresh List", :action => "index" %> <table> <% @messages.each do |m| -%>   <tr><td><%= m.received_date %></td><td><%= m.body %></td></tr> <% end %> </table>

require 'activemessaging/processor'
 
 class SendMessageController < ApplicationController
 
   include ActiveMessaging::MessageSender
  publishes_to :persist_message
 
   def index
    @messages = Message.find :all
  end
 
   def new
    @message = params[:message]
    publish :persist_message, @message
    flash[:notice] = "'#{@message}' sent"
    @messages = Message.find :all
    render :action => 'index'
  end
 end
 

というような感じです。コントローラはサブミット時にメッセージを送信し、あなたは次回にページを表示または更新するとき、プロセッサによって保存されたメッセージを目にするでしょう。

例題の実行

ActiveMessagingは独自のプロセスで実行し、script/pollerによって制御されます。そのため、メッセージを受信したい場合、Railsサーバーを開始するだけでは不十分であり、メッセージはRailsと異なる実行プロセスの中で受信されます。この例を実行するには、3つのプロセスが必要です。ActiveMQブローカー、メッセージの送信を実行するRailsサーバー、それを受信するためのscript/poller の3つで、それぞれ独自のターミナルで実行します。

ActiveMQを開始します。

cd /usr/local/apache-activemq-4.2-incubator-SNAPSHOT
 ./bin/activemq

Railsを開始します。

cd /usr/local/dev/MessageMonster
 script/server

ActiveMessagingを実行します。

cd /usr/local/dev/MessageMonster
script/poller run

The script/poller は通常のデーモンの意味に従います。「run」はフォアグラウンドでその処理を維持する上、あなたはそれに「start」、「stop」、「restart」、「status」を渡すこともできます。また、同時に複数のインスタンスを実行できるため、大量のメッセージを処理するのに有用です。

次に、テストページ（http://localhost:3000/send_message) に進み、一般的な「Hello World」などのメッセージを入力します。メッセージが送信されると、pollerウィンドウに次のように表示されます。

Loading /usr/local/dev/MessageMonster/app/processors/application.rb
Loading /usr/local/dev/MessageMonster/app/processors/persist_message_processor.rb
=> Subscribing to /queue/PersistMessage (processed by PersistMessageProcessor)
=> All subscribed, now polling
PersistMessageProcessor received: Hello World 

pollerは常に、起動時にロードするプロセッサと、ブローカーにどんなサブスクリプションの要求があるかを示します。ブローカーが利用不能か見つからない場合は、broker.yml設定に基づいて再接続しようとします。pollerによる出力の開始後、あなたはRailsから送信されたメッセージをActiveMessagingプロセッサが受信したことを確認できます。同様に、「Refresh List」ボタンを使用する場合、ActiveMessagingプロセッサによって永続化された時点でRailsビュー内にメッセージを確認できます。

今後の展開

このActiveMessaging入門をお楽しみいただけたでしょうか。そのポテンシャルと使いやすさの両方を感じていただけることを願っております。ActiveMessagingには、この記事では取り上げていない多数の高度な機能があります。たとえば、改善された例外処理、複数のpollerによって実行するプロセッサのグループ化、同期メッセージングやJMS統合や選択的サブスクリプションのようなトリックのためのStompおよびActiveMQメッセージヘッダーの使用法などです。あなたのニーズに対しActiveMessagingはあまりに制限されているという考えがベースにあった上で検討する場合、うれしい驚きを経験するだろうと思います。