ActiveMQ:拦截发布到某个主题 [英] ActiveMQ: Intercept publishing to a certain topic

问题描述

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

I'm setting up an ActiveMQ broker with MQTT that uses an external service for user authentication.

我想出了如何编写BrokerFilter并将其插入Broker的方法，因此涵盖了所有基础知识.

I figured out how to write a BrokerFilter and plug it into a broker, so the basics are covered.

我什至可以限制用户使用addConsumer()替代订阅的主题.该方法看起来像这样，并且有效:

I could even limit users in the topics they are allowed to subscribe on using the addConsumer() override. That method looks like this, and works:

    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()进行覆盖时，限制发布的效果几乎相同，但是我偶然发现了一些问题.

So I thought that restricting publishing would work pretty much the same with an override of addProducer(), but I stumbled over a few problems.

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

The first problem was that subscriptions also called addProducer(), which was surprising to say the least. But the real surprise was that ConsumerInfo::destination is always null when this method is called. I've searched far and wide, but I'm unable to find a way to extract the topic being published to from the information passed to addProducer(). Without knowing which topic the producer wants to publish to, I obviously can't limit it.

因此，我尝试改写了messageDelivered()，以为我可以将消息发布到错误的主题时将其丢弃，从而获得大致相同的效果.根据文档，只要代理收到消息，就应调用此方法.但是，当我向经纪人发送消息时，它不会被调用，因此我可能误解了传递给经纪人的消息"是什么意思，或者是有些可疑. 我还尝试了addSession()，它也没有被调用.

So I tried overriding messageDelivered() instead, figuring I could just discard the message when it is published to the wrong topic, achieving more or less the same effect. According to the documentation, this method should be called whenever the broker receives a message. But it doesn't get called when I send a message to the broker, so either I'm misunderstanding what "a message delivered to the broker" means or something is fishy. I also tried addSession(), which also doesn't get called.

那么...当客户发布到某个主题时，我该如何截取?

So... How can I intercept when a client publishes to a certain topic?

推荐答案

在仔细研究了源代码和许多替代方法之后，看到了它们何时被调用以及它们得到了什么，我学到了两点有助于我做些什么的事情我想要:

After some hunting through the source code and a lot of overriding methods and seeing when they get called and what they receive, I learned two things that helped me do what I wanted:

1. 要发布到的主题封装在消息中.生产者无需注册即可发布到某个主题，而只是注册即可发布到任何主题.代理本身不知道哪个主题，该主题是逐个消息地处理的.因此，我第一次尝试限制addProducer()中的主题是徒劳的，因为没人还不知道该主题.

发布链中可以同时使用用户名和主题的最早点是addDestination().首先可以在send()中知道该主题，但是您没有该用户.可以通过保留上下文的用户名来进行授权，但是我不喜欢状态.因此，我将授权放入addDestination()中，它可以正常工作.

The earliest point in the publish chain where both username and topic are available is addDestination(). The topic can first be known in send(), but you don't have the user. It might be possible to do the authorisation there by persisting the username for the context, but I don't like state. So I put the authorisation into addDestination(), and it works.

不过，这里有一个潜在的警告.根据文档，addDestination()仅在目标不存在时才被调用.我可以在有限的时间内进行的所有测试都证实，即使有人订阅了该主题，发布中也总是如此.但是，如果生产者保持持久的连接，则可能会有所不同.因此，请谨慎使用此解决方案.

There's a potential caveat here, though. According to the documentation, addDestination() only gets called if a destination doesn't exist yet. All tests I could do in the limited time confirmed that this is always the case on a publish, even if somebody is subscribed to the topic. But it might be different if the producer maintains a persistent connection, So use this solution with care.

这篇关于ActiveMQ:拦截发布到某个主题的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！