5.6. JMS

DefaultJMSConnectionFactory 接続ファクトリーは、JNDI から RouteBuilder に注入されます。メッセージングサブシステム内の JBoss EAP XML 設定に、接続ファクトリーがあります。

次に、タイマーエンドポイントが 10 秒ごとに実行され、以前に設定した WildFlyCamelQueue 宛先に XML ペイロードを送信します。

@Startup
@ApplicationScoped
@ContextName("jms-camel-context")
public class JmsRouteBuilder extends RouteBuilder {

  @Resource(mappedName = "java:jboss/DefaultJMSConnectionFactory")
  private ConnectionFactory connectionFactory;

  @Override
  public void configure() throws Exception {
    JmsComponent component = new JmsComponent();
    component.setConnectionFactory(connectionFactory);

    getContext().addComponent("jms", component);

    from("timer://sendJMSMessage?fixedRate=true&period=10000")
    .transform(constant("<?xml version='1.0><message><greeting>hello world</greeting></message>"))
    .to("jms:queue:WildFlyCamelQueue")
    .log("JMS Message sent");
  }
}

JMS メッセージが WildFlyCamelQueue 宛先に追加されるたびに、ログメッセージがコンソールに出力されます。メッセージが実際にキューに配置されていることを確認するには、JBoss EAP 管理コンソールを使用できます。

JMS メッセージを消費するため、Camel RouteBuilder 実装はプロデューサーの例と似ています。

前述のように、接続ファクトリーは JNDI から検出され、JMSComponent インスタンスに注入され設定されます。

JMS エンドポイントが WildFlyCamelQueue 宛先からのメッセージを消費すると、内容がコンソールに記録されます。

@Override
public void configure() throws Exception {
  JmsComponent component = new JmsComponent();
  component.setConnectionFactory(connectionFactory);

  getContext().addComponent("jms", component);

  from("jms:queue:WildFlyCamelQueue")
  .to("log:jms?showAll=true");
}

Camel JMS ルートが JMS トランザクションに参加できるようにするには、追加の設定が必要になります。camel-jms は spring-jms を中心に構築されているため、JBoss EAP のトランザクションマネージャーおよび接続ファクトリーと連携できるようにいくつかの Spring クラスを設定する必要があります。以下のコード例は、CDI を使用してトランザクション JMS Camel ルートを設定する方法を示しています。

camel-jms コンポーネントには、タイプ org.springframework.transaction.PlatformTransactionManager トランザクションマネージャーが必要です。そのため、JtaTransactionManager を拡張する Bean の作成から開始します。Bean には @Named アノテーションが付けられており、Camel Bean レジストリー内に Bean を登録できることに注意してください。また、JBoss EAP トランザクションマネージャーとユーザートランザクションインスタンスは CDI を使用して注入されることにも注意してください。

@Named("transactionManager")
public class CdiTransactionManager extends JtaTransactionManager {

  @Resource(mappedName = "java:/TransactionManager")
  private TransactionManager transactionManager;

  @Resource
  private UserTransaction userTransaction;

  @PostConstruct
  public void initTransactionManager() {
    setTransactionManager(transactionManager);
    setUserTransaction(userTransaction);
  }
}

次に、使用するトランザクションポリシーを宣言する必要があります。ここでも、@Named アノテーションを使用して Bean を Camel で利用可能にします。また、トランザクションマネージャーも、必要なトランザクションポリシーで TransactionTemplate を作成できるように注入されます(ここでは PROPAGATION_REQUIRED)。

@Named("PROPAGATION_REQUIRED")
public class CdiRequiredPolicy extends SpringTransactionPolicy {
  @Inject
  public CdiRequiredPolicy(CdiTransactionManager cdiTransactionManager) {
    super(new TransactionTemplate(cdiTransactionManager,
      new DefaultTransactionDefinition(TransactionDefinition.PROPAGATION_REQUIRED)));
  }
}

これで、Camel RouteBuilder クラスを設定し、Camel JMS コンポーネントに必要な依存関係を注入できるようになりました。JBoss EAP XA 接続ファクトリーは、以前に設定されたトランザクションマネージャーとともに注入されます。

この例の RouteBuilder では、queue1 から任意のメッセージが消費されるたびに、メッセージは queue2 という名前の別の JMS キューにルーティングされます。queue2 から消費されるメッセージにより、rollback() DSL メソッドを使用して JMS トランザクションがロールバックされます。これにより、元のメッセージがデッドレターキュー (DLQ) に配置されます。

@Startup
@ApplicationScoped
@ContextName("jms-camel-context")
public class JMSRouteBuilder extends RouteBuilder {

  @Resource(mappedName = "java:/JmsXA")
  private ConnectionFactory connectionFactory;

  @Inject
  CdiTransactionManager transactionManager;

  @Override
  public void configure() throws Exception {
    // Creates a JMS component which supports transactions
    JmsComponent jmsComponent = JmsComponent.jmsComponentTransacted(connectionFactory, transactionManager);
    getContext().addComponent("jms", jmsComponent);

    from("jms:queue:queue1")
      .transacted("PROPAGATION_REQUIRED")
      .to("jms:queue:queue2");

    // Force the transaction to roll back. The message will end up on the Wildfly 'DLQ' message queue
    from("jms:queue:queue2")
      .to("log:end")
      .rollback();
  }

ある JBoss EAP インスタンスが リモート JNDI を使用して別の JBoss EAP インスタンスに設定された ActiveMQ Artemis 宛先にメッセージを送信することができます。

これには、追加の JBoss EAP 設定が必要です。最初に、エクスポートされた JMS キューが設定されます。

java:jboss/exported namespace でバインドされた JNDI 名のみがリモートクライアントの候補として考慮されるため、キューには適切な名前が付けられます。

<jms-queue name="RemoteQueue">
  <entry name="java:jboss/exported/jms/queues/RemoteQueue"/>
</jms-queue>

クライアントがリモートサーバーに接続できるようにするには、ユーザーアクセスクレデンシャルを設定する必要があります。リモートサーバーで add user utility を実行し、guest グループ内に新しいアプリケーションユーザーを作成します。この例には、名前が admin でパスワードが secret のユーザーがいます。

RouteBuilder の実装は、前述の例とは異なります。接続ファクトリーを注入する代わりに、InitialContext を設定し、JNDI 自体から取得する必要があります。

configureInitialContext メソッドはこの InitialContext を作成します。リモート JBoss EAP インスタンスのホスト名とポート番号を参照するプロバイダー URL を設定する必要があります。この例では JBoss EAP JMS http-connector を使用していますが、ここ に記載されている代替手段があります。

最後に、ルートは、前項で定義されたリモート宛先 ('RemoteQueue') に 10 秒ごとに XML ペイロードを送信するように設定されます。

@Override
public void configure() throws Exception {
  Context initialContext = configureInitialContext();
  ConnectionFactory connectionFactory = (ConnectionFactory) initialContext.lookup("java:jms/RemoteConnectionFactory");

  JmsComponent component = new JmsComponent();
  component.setConnectionFactory(connectionFactory);

  getContext().addComponent("jms", component);

  from("timer://foo?fixedRate=true&period=10000")
  .transform(constant("<?xml version='1.0><message><greeting>hello world</greeting></message>"))
  .to("jms:queue:RemoteQueue?username=admin&password=secret")
  .to("log:jms?showAll=true");
}

private Context configureInitialContext() throws NamingException {
  final Properties env = new Properties();
  env.put(Context.INITIAL_CONTEXT_FACTORY, "org.jboss.naming.remote.client.InitialContextFactory");
  env.put(Context.PROVIDER_URL, System.getProperty(Context.PROVIDER_URL, "http-remoting://my-remote-host:8080"));
  env.put(Context.SECURITY_PRINCIPAL, System.getProperty("username", "admin"));
  env.put(Context.SECURITY_CREDENTIALS, System.getProperty("password", "secret"));
  return new InitialContext(env);
}