如何在 FUSE 中监听 AMQ 代理启动

Question

我想添加一个利用 JBoss FUSE 中内置的 AMQ 的功能，该功能将与容器一起启动 - 即，将添加到启动功能中，启动级别

此功能将立即尝试通过 AMQ 发送一些消息，因此我希望添加逻辑来“侦听”或只是等待代理启动事件。

我目前拥有的唯一句柄是一条记录的消息，例如：

... | ActiveMQServiceFactory           | 196 - io.fabric8.mq.mq-fabric
      - 1.2.0.redhat-143 | Broker amq has started.

在我的功能文件中，我尝试创建几个功能依赖项，例如：

<feature name="myFeature" ...
    <feature version="1.2.0.redhat-133">mq-fabric</feature>
    <feature version="5.11.0.redhat-620133">activemq</feature>

...但这没有用。我还尝试将起始级别设置为 90 年代的某个值，但这也不起作用。

我只是错过了这些方法中的某些东西吗？

我可以在我的蓝图中注入相关的 OSGi 服务吗？ -- 即，也许有一些 .isBrokerStarted() 方法？

我最近的想法是，也许我可以为咨询消息注册回调...如果可行，我会将其作为答案发布。

不幸的是，我不再认为尝试联系和倾听咨询主题是一种可行的方法。我已经部署了这个功能。如果在上述“代理启动”日志消息之后安装，它可以正常工作，否则会失败。

问题似乎是因为蓝图上下文初始化是在 AMQ 异步启动时进行的，因此任何生成的连接都是无效的。我原以为 AMQ 连接工厂或 Camel 组件会不透明地为这种情况提供一些保证。显然事实并非如此。调查他们的 API（特别是 org.apache.activemq.ActiveMQSslConnectionFactory 和 org.apache.camel.component.jms.JmsConfiguration），我也没有找到用于状态检查的公共方法......只有设置“testConnectionOnStartup”的能力Camel 组件的属性。据推测，这将按照记录完成，并引发 1+ 异常，但事实并非如此。非常混乱。

2016 年 8 月 24 日更新：

终于有了更多的时间来处理这个问题，并尝试了几件事...... 我在 camel-amq 上添加了一个功能依赖项，因为它位于功能依赖项链的“顶部”：

camel-amq -> 依赖于 mq-fabric -> 依赖于 activemq -> 依赖于 activemq-client

这不起作用。不过，在对日志进行了更多调整之后，我相信我已经确认了我上面暗示的内容。启动级别和功能依赖关系无法解决这个问题。所有较低启动级别的依赖功能或捆绑包都将在 amq 代理实际准备好（并记录上述消息）之前报告已安装/启动。

因此，我的功能将安装、启动和失败——除非我能找到一种方法来执行帖子标题所建议的操作：“监听 AMQ 代理启动”。有谁知道这样做的方法吗？

2016 年 8 月 25 日更新：

我看到 ActiveMQServiceFactory（它记录代理启动消息）本身在其上下文本地的 org.apache.activemq.broker.BrokerService 实例上有一个句柄。如果这也作为 OSGi 服务提供，那么我认为我可以等待“broker.waitUntilStarted();”然后以编程方式创建我的 JMS 连接。当我有时间尝试这个时会报告......

Answer 1

首先，我无法在ActiveMQServiceFactory 上找到可用的句柄...它是通过org.osgi.service.cm.ManagedServiceFactory 接口提供的，但也以某种方式代理，因此它不能转换为@ 987654323@ :-/

因此，我最终创建了一个功能，该功能不断循环尝试通过代理的最慢启动连接器（在我的情况下为 ssl 连接器）以编程方式创建连接/会话/临时队列，然后删除/关闭全部。之后，此功能以编程方式安装我的其他功能。

这行得通。

基本上：

// ... setup amq conn factory
while (true) {
  Thread.sleep(5000L);
  try { 
    amqConnFactory.createConnection()
                  .start()
                  .createSession(fals‌​e, Session.AUTO_ACKNOWLEDGE)
                  .createTemporaryQueue();
    break;
  } catch (Exception e) {
    // debug log 'not up yet'
  }
}