java - ActiveMQ:拦截发布到某个主题

Question

我正在使用 MQTT 设置 ActiveMQ 代理，该代理使用外部服务进行用户身份验证。

我弄清楚了如何编写 BrokerFilter 并将其插入代理，因此基础知识已涵盖。

我什至可以使用 addConsumer() 覆盖来限制用户可以订阅的主题。该方法如下所示，并且有效:

    override fun addConsumer(context: ConnectionContext?, info: ConsumerInfo?): Subscription {
    // let connections from the local vm through
    return if (isLocalConnection(context!!)) {
        super.addConsumer(context, info)
    } else {
        val username = context?.userName ?: ""
        val cameraCode = getTopicElementsFromDestination(info!!.destination).first()
        assertUserHasPermissionForCamera(username, cameraCode)
        super.addConsumer(context, info)
    }
}

因此，我认为限制发布与重写 addProducer() 的效果几乎相同，但我偶然发现了一些问题。

第一个问题是订阅也调用了 addProducer()，这至少可以说是令人惊讶的。但真正令人惊讶的是，调用此方法时 ConsumerInfo::destination 始终为 null。我进行了广泛的搜索，但无法找到一种方法来从传递给 addProducer() 的信息中提取要发布的主题。在不知道制作者要发布到哪个主题的情况下，我显然无法限制它。

所以我尝试重写 messageDelivered() ，认为当消息发布到错误的主题时我可以丢弃该消息，从而达到或多或少相同的效果。根据文档，只要代理收到消息，就应该调用此方法。但是当我向经纪人发送消息时它不会被调用，所以要么我误解了“消息传递给经纪人”的含义，要么有些可疑。 我还尝试了 addSession()，它也没有被调用。

那么...当客户端发布到某个主题时如何拦截？

Answer 1

在浏览了源代码和许多重写方法并查看它们何时被调用以及它们收到什么之后，我学到了两件事帮助我做我想做的事情:

1. 要发布的主题封装在消息中。 Producer 并不注册发布到某个主题，它只是注册发布到任何主题。代理本身不知道哪个主题是逐条消息处理的。因此，我第一次尝试限制 addProducer() 中的主题是徒劳的，因为没有人知道该主题。

2. 发布链中用户名和主题均可用的最早点是 addDestination()。主题首先可以在 send() 中获知，但你没有用户。可以通过保留上下文的用户名来进行授权，但我不喜欢状态。所以我将授权放入 addDestination() 中，并且它起作用了。

不过，这里有一个潜在的警告。根据文档，仅当目的地尚不存在时才会调用 addDestination() 。我在有限的时间内所做的所有测试都证实，即使有人订阅了该主题，发布时也总是出现这种情况。但如果生产者维护持久连接，情况可能会有所不同，因此请谨慎使用此解决方案。