第41章 Consumer インターフェイス

org.apache.camel.Consumer タイプのインスタンスは、ルートのソースエンドポイントを表します。コンシューマーの実装方法がいくつかあります (「コンシューマーパターンおよびスレッド」 を参照)。このレベルの柔軟性は継承階層 (図41.1「コンシューマー継承階層」 を参照) に反映されます。これには、コンシューマーを実装するための複数の異なるベースクラスが含まれます。

図41.1 コンシューマー継承階層

Apache Camel は、スケジュールされたポーリングパターン (「スケジュールされたポーリングパターン」 を参照) を使用しているコンシューマーの場合、パラメーターをコンシューマーインスタンスに注入するサポートを提供します。たとえば、custom 接頭辞で識別されるコンポーネントの以下のエンドポイント URI について考えてみましょう。

custom:destination?consumer.myConsumerParam

Apache Camel は、フォームのクエリーオプション consumer.\* を自動的に注入するためのサポートを提供します。consumer.myConsumerParam パラメーターには、以下のように Consumer 実装クラスで対応する setter メソッドと getter メソッドを定義する必要があります。

public class CustomConsumer extends ScheduledPollConsumer {
    ...
    String getMyConsumerParam() { ... }
    void setMyConsumerParam(String s) { ... }
    ...
}

getter および setter メソッドは、通常の Java Bean の命名規則に従います (プロパティー名の最初の文字を大文字にすることも含む)。

Consumer 実装で Bean メソッドを定義する他に、必ず Endpoint.createConsumer() の実装で configureConsumer() メソッドを呼び出す必要があります (「スケジュールされたポーリングエンドポイントの実装」 を参照してください)。

例41.1「FileEndpoint createConsumer () 実装」 は、ファイルコンポーネントの FileEndpoint クラスから取得した createConsumer() メソッド実装の例を示しています。

...
public class FileEndpoint extends ScheduledPollEndpoint {
    ...
    public Consumer createConsumer(Processor processor) throws Exception {
        Consumer result = new FileConsumer(this, processor);
        configureConsumer(result);
        return result;
    }
    ...
    }

ランタイム時に、コンシューマーパラメーターの注入は以下のようになります。

スケジュールされたポーリングパターンに続くコンシューマーは、表41.1「スケジュールされたポーリングパラメーター」 に記載されているコンシューマーパラメーターを自動的にサポートします (エンドポイント URI のクエリーオプションとして表示されます)。

Apache Camel は、イベント駆動型のコンシューマーとポーリングコンシューマー間の変換に使用できる 2 つの特殊なコンシューマー実装を提供します。以下の変換クラスが提供されます。

実際には、これらのクラスを使用して Endpoint タイプの実装タスクを単純化します。Endpoint インターフェイスは、コンシューマーインスタンスを作成するための以下の 2 つのメソッドを定義します。

package org.apache.camel;

public interface Endpoint {
    ...
    Consumer createConsumer(Processor processor) throws Exception;
    PollingConsumer createPollingConsumer() throws Exception;
}

createConsumer() はイベント駆動型のコンシューマーを返し、createPollingConsumer() はポーリングコンシューマーを返します。これらのメソッドは 1 つのみ実装します。たとえば、コンシューマーのイベント駆動パターンに従っている場合は、単に例外を発生させる createConsumer() のメソッド実装を提供するために createPollingConsumer() メソッドを実装します。ただし、変換クラスを利用して Apache Camel はより有用なデフォルト実装を提供できます。

たとえば、イベント駆動のパターンに従ってコンシューマーを実装する場合は、DefaultEndpoint を拡張し、createConsumer() メソッドを実装してエンドポイントを実装します。createPollingConsumer() の実装は、以下のように定義される DefaultEndpoint から継承されます。

public PollingConsumer<E> createPollingConsumer() throws Exception {
    return new EventDrivenPollingConsumer<E>(this);
}

EventDrivenPollingConsumer コンストラクターはイベント駆動のコンシューマーへの参照 this を取得し、効果的にラップし、ポーリングコンシューマーに変換します。変換を実装するには、EventDrivenPollingConsumer インスタンスは受信イベントをバッファーし、receive()、receive(long timeout)、および receiveNoWait() メソッドを介してオンデマンドで利用できるようにします。

同様に、ポーリングパターンに従ってコンシューマーを実装する場合は、DefaultPollingEndpoint を拡張し、createPollingConsumer() メソッドを実装してエンドポイントを実装します。この場合、createConsumer() メソッドの実装は DefaultPollingEndpoint から継承され、デフォルトの実装は DefaultScheduledPollConsumer インスタンス (ポーリングコンシューマーをイベント駆動型のコンシューマーに変換) を返します。

コンシューマークラスはオプションで org.apache.camel.spi.ShutdownPrepared インターフェイスを実装できます。これにより、カスタムコンシューマーエンドポイントがシャットダウン通知を受け取ることができます。

例41.2「ShutdownPrepared インターフェイス」 は、ShutdownPrepared インターフェイスの定義を示しています。

package org.apache.camel.spi;

public interface ShutdownPrepared {

    void prepareShutdown(boolean forced);

}

ShutdownPrepared インターフェイスは以下のメソッドを定義します。

コンシューマークラスはオプションで org.apache.camel.spi.ShutdownAware インターフェイスを実装できます。このインターフェイスは、正常なシャットダウンメカニズムと対話し、コンシューマーがシャットダウンするための追加の時間を要求できるようにします。これは通常、内部キューに保留中のエクスチェンジを保存できる SEDA などのコンポーネントに必要です。通常、SEDA コンシューマーをシャットダウンする前にキューのすべてのエクスチェンジを処理します。

例41.3「ShutdownAware インターフェイス」 は、ShutdownAware インターフェイスの定義を示しています。

// Java
package org.apache.camel.spi;

import org.apache.camel.ShutdownRunningTask;

public interface ShutdownAware extends ShutdownPrepared {

    boolean deferShutdown(ShutdownRunningTask shutdownRunningTask);

    int getPendingExchangesSize();
}

ShutdownAware インターフェイスは以下のメソッドを定義します。

ShutdownAware メソッドを定義する方法の例は、例41.7「カスタムスレッド実装」 を参照してください。