使用SmallRye反应式消息动态发布/订阅MQTT

Question

我们尝试使用小型反应式消息传递发布和订阅MQTT协议。我们设法通过以下简单代码将消息实际发布到特定主题/频道

import io.smallrye.mutiny.Multi;
import org.eclipse.microprofile.reactive.messaging.Outgoing;
import javax.enterprise.context.ApplicationScoped;
import java.time.Duration;

@ApplicationScoped
public class Publish {
    
    @Outgoing("pao")
    public Multi<String> generate() {
        return Multi.createFrom().ticks().every(Duration.ofSeconds(1))
                .map(x -> "A Message in here");
    }
}

我们想要做的是在我们需要生成（）方法时以某种方式调用动态主题，用户将在其中定义它。那是我们的问题，但后来我们在github中从该存储库中找到了这些类。包名称io.smallrye.reactive.messaging.mqtt

例如，我们发现有一个类说它对MQTT代理进行了发布调用（Mosquito server up）。

在该语句中SendingMqttMessage

UPDATE（发布完成）最后向mqtt代理（蚊子服务器）发出发布请求，所有这些都使用用户配置的动态主题。正如我们发现之前的ClassSendingMqttMessage根本不应该使用。我们发现我们还需要和发射器来实际发出带有动态主题的发布请求。

    @Inject
    @Channel("panatha")
    Emitter<String> emitter;

    @POST
    @Consumes(MediaType.APPLICATION_JSON)
    @Produces(MediaType.APPLICATION_JSON)
    public Response createUser(Device device) {
        System.out.println("New Publish request: message->"+device.getMessage()+" & topic->"+device.getTopic());
        emitter.send(MqttMessage.of(device.getTopic(), device.getMessage()));
        return Response.ok().status(Response.Status.CREATED).build();
    }

现在我们需要了解如何动态订阅主题。




Answer 1

首先让我们回到同一个页面：
反应式消息不适用于主题，而是适用于频道。这一点很重要，因为您可以专门读取或写入某个频道。因此，如果您想同时提供这两个频道，您需要配置指向同一主题的两个频道，一个传入频道和一个传出频道

你用发射器做了一个很好的开始，但是你仍然缺乏你想要的动态特性。在您的示例中，您通过CDI获得了该发射器
这就是我们所需要的，使其动态化，因为我们可以在运行时使用如下CDI动态注入bean：

private Emitter<byte[]> dynamicEmitter(String topic){
        return CDI.current().select(new TypeLiteral<Emitter<byte[]>>() {}, new ChannelAnnotation(topic)).get();
    }

另请注意，我正在创建一个类型为byte[]的发射器，因为这是根据其留档的小Rey-mqtt连接器（版本3.4.0）目前唯一支持的类型。

要从反应式消息传递通道读取消息，您可以使用发射器的对应物，即发布者。
它可以用于模拟：

private Publisher<byte[]> dynamicReceiver(String topic){
        return CDI.current().select(new TypeLiteral<Publisher<byte[]>>() {}, new ChannelAnnotation(topic)).get();
    }

然后，您可以以任何您喜欢的方式处理这些Date。作为演示，它将其挂在一个简单的RESTendpoint上

@GET
    @Produces(MediaType.SERVER_SENT_EVENTS) 
    public Multi<String> stream(@QueryParam("topic") String topic) {
        return Multi.createFrom().publisher(dynamicReceiver(topic)).onItem().transform(String::new); 
    }
    
    @GET
    @Path("/publish")
    public boolean publish(@QueryParam("msg") String msg, @QueryParam("topic") String topic) {
        dynamicEmitter(topic).send(msg.getBytes());
        return true; 
    }

创建此解决方案时，我遇到了一些您应该知道的陷阱：

1. Quarkus会删除任何“未使用”的CDI-Bean。因此，如果您想动态注入它们，您需要排除它们，或者关闭该功能。
2. 必须配置以这种方式注入的所有通道。否则注入将失败。
3. 由于某些原因，（即使完全禁用删除）我无法动态注入发射器，除非它们曾经被注入其他地方。