第6章 Camel コンポーネント

@Startup
@ApplicationScoped
@ContextName("activemq-camel-context")
public class ActiveMQRouteBuilder extends RouteBuilder {

  @Override
  public void configure() throws Exception {
    from("timer://sendJMSMessage?fixedRate=true&period=10000")
    .transform(constant("<?xml version='1.0><message><greeting>hello world</greeting></message>"))
    .to("activemq:queue:WildFlyCamelQueue?brokerURL=vm://localhost")
    .log("JMS Message sent");
  }
}

メッセージが WildFlyCamelQueue 宛先に追加されるたびに、ログメッセージがコンソールに出力されます。メッセージが実際にキューに配置されていることを確認するには、Camel on EAP サブシステムによって提供される ../features/hawtio.md[Hawtio console,window=_blank] を使用できます。

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

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

@Override
public void configure() throws Exception {
  from("activemq:queue:WildFlyCamelQueue?brokerURL=vm://localhost")
  .to("log:jms?showAll=true");
}

camel-activemq コンポーネントには、タイプ 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 ActiveMQ コンポーネントに必要な依存関係を注入できるようになりました。リソースアダプター設定に設定した ActiveMQ 接続ファクトリーは、以前に設定したトランザクションマネージャーと共に注入されます。

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

@Startup
@ApplicationScoped
@ContextName("activemq-camel-context")
public class ActiveMQRouteBuilder extends RouteBuilder {

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

  @Inject
  private CdiTransactionManager transactionManager;

  @Override
  public void configure() throws Exception {
    ActiveMQComponent activeMQComponent = ActiveMQComponent.activeMQComponent();
    activeMQComponent.setTransacted(false);
    activeMQComponent.setConnectionFactory(connectionFactory);
    activeMQComponent.setTransactionManager(transactionManager);

    getContext().addComponent("activemq", activeMQComponent);

      errorHandler(deadLetterChannel("activemq:queue:ActiveMQ.DLQ")
      .useOriginalMessage()
      .maximumRedeliveries(0)
      .redeliveryDelay(1000));

    from("activemq:queue:queue1F")
      .transacted("PROPAGATION_REQUIRED")
      .to("activemq:queue:queue2");

    from("activemq:queue:queue2")
      .to("log:end")
      .rollback();
  }
}