quarkusio / quarkus

Quarkus: Supersonic Subatomic Java.
https://quarkus.io
Apache License 2.0
13.75k stars 2.67k forks source link

Reactive Messaging: support dynamic emitters & publishers #40098

Open Serkan80 opened 6 months ago

Serkan80 commented 6 months ago

Description

Currently, afaik, it is not possible to send messages to topics/queues which have not been configured beforehand in applications.properties.

Especially in MQTT environment, it is common to create topics on the fly and subscribe to them.

Implementation ideas

An idea would be to do something like this:

@Inject 
//@Broadcast --> should work
Emitter emitter;

@Inject 
//@Channel("/house/+/temperature") --> should work: listen to temperatures in all rooms 
Publisher publisher;

public void sendMessage(String topic, String payload) {
  this.emitter.send(topic, payload);
}

public void subscribe(String topic) {
  this.publisher.subscribe(topic, this::handleMessage);
}

public void unsubscribe(String topic) {
  this.publisher.unsubscribe(topic);
}
quarkus-bot[bot] commented 6 months ago

/cc @cescoffier (reactive-messaging), @ozangunalp (reactive-messaging)

dometec commented 1 month ago

@Serkan80 you can, just emit an MqttMessage and leave the topic of the channel undefined:


import org.eclipse.microprofile.reactive.messaging.Channel;
import org.eclipse.microprofile.reactive.messaging.Emitter;
import org.eclipse.microprofile.reactive.messaging.Incoming;
import io.smallrye.reactive.messaging.mqtt.MqttMessage;
...

    @Inject
    @Channel(SEND_CHANNEL)
    Emitter<String> emitter;

        String topic = ...
        MqttMessage<String> message = MqttMessage.of(topic, payload);

        emitter.send(message);