Récupérer un message sur un serveur MQTT
Bonjour a tous !
Je suis en train de développer un prototype qui récupérer des informations via le protocole MQTT. Je développe sous NetBeans.
J'ai donc commence par un exemple donne sur le site Paho avec leurs bibliothèques. Mon client envoie et reçoit bien les informations. Je les visualise sur MQTTLens et j'arrive bien a publier et récupérer.
Mon soucis est lorsque je souhaite diviser ces deux parties, je m'explique :
Je souhaiterais avoir une classe pour publier et une classe pour souscrire a un topic.
Pour la publication c'est tout bon, j'arrive bien a envoyer les données que je veux sur le serveur.
Mon problème survient quand je veux récupérer mes données avec mon subscripteur. J'utilise la méthode setcallback() qui appelle normalement la méthode messageArrived() et me retourne le message qu'elle a lu dans le serveur. Mais en fait non, je souscris au topic et après il se déconnecte (c'est la suite de mon code). Il ne passe jamais par la méthode messageArrived().
Dans la classe publier, j'utilise aussi la méthode setCallback() pour appeler la méthode deliveryComplete() et celle-ci est bien appelé, le message s'affiche après.
Bref, je ne comprends vraiment pas pourquoi il ne me l'appelle pas dans la classe qui souscrit, mon but étant de récupérer ce message et de le sauvegarder pour l'utiliser après, quelqu'un aurait une idée pour résoudre ce problème ?
Merci d'avance,
Je vous mets le code mes 2 classes et ce qui se trouve dans le main aussi :
Classe publier :
Code:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
| package publishclient;
import java.util.Random;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
/**
*
* @author
*/
public class PubClient implements MqttCallback{
MqttClient client; // Creation de l'objet client
MqttConnectOptions connOpt; // Creation de l'objet connOpt
Random rand = new Random(); //Permet de creer un nombre aleatoire
int nbRandom = 0;
int valMax =151, valMin = 40; //Delimite les valeurs max et min des pulsations cardiaques creees
public PubClient() throws MqttException{
String broker = "tcp://"; // Adresse du serveur MQTT Je ne sais pas si je peux la mettre ici sur le forum, ca ne craint pas ?
String clientId = "0bdd-4445-82f3-928f8ddb1887"; // Nom du clientID
String topic1f = "ApplicationRio/capteur"; // Topic simulateur
int QoSserveur = 2;
try{
String uuid = "ac8da3c6-0bdd-4445-82f3-928f8ddb1887";
MemoryPersistence persistence = new MemoryPersistence();
// Creation des 2 objets : client et connOpt
client = new MqttClient(broker, clientId, persistence);
connOpt = new MqttConnectOptions();
connOpt.setCleanSession(true); // Nettoyage de la session, remise a zero
connOpt.setKeepAliveInterval(30); // Temps maximum en secondes entre deux envois ou deux receptions
client.setCallback(this);
// Connexion au serveur MQTT
System.out.println("Connexion a : " + broker + " Publisher");
client.connect(connOpt);
nbRandom = valMin + rand.nextInt(valMax-valMin);
System.out.println("nb aleatoire = " + nbRandom);
String messageAEnvoyer = uuid + "//" + nbRandom;
System.out.println("Message a envoyer : " + messageAEnvoyer);
MqttMessage message = new MqttMessage();
message.setPayload(messageAEnvoyer.getBytes());
message.setQos(2);
client.publish(topic1f, message);
// Publication du message
// client.publish(topic1f, messageAEnvoyer.getBytes(),2,true);
System.out.println("Fin d'envoi des messages");
client.disconnect(); // Deconnexion du serveur MQTT
System.out.println("Deconnexion publieur");
}catch(MqttException e){
e.printStackTrace();
}
}
@Override
public void connectionLost(Throwable thrwbl) {}
@Override
public void messageArrived(String string, MqttMessage mm) throws Exception {}
@Override
public void deliveryComplete(IMqttDeliveryToken imdt) {
System.out.println("Message delivre au broker");
}
} |
Classe subscribe :
Code:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
| package publishclient;
import java.util.Arrays;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
/**
*
* @author
*/
public class SubClient implements MqttCallback {
MqttClient client; // Creation de l'objet client
MqttConnectOptions connOpt; // Creation de l'objet connOpt
public SubClient() throws MqttException{
String broker = ""; // Adresse du serveur MQTT Je ne sais pas si je peux la mettre ici sur le forum, ca ne craint pas ?
String clientId = "0bdd-4445-82f3-928f8ddb1887"; // Nom du clientID
String topic1f = "ApplicationRio/capteur"; // Topic simulateur
int QoSserveur = 2;
try{
// Creation des 2 objets : client et connOpt
client = new MqttClient(broker, clientId);
connOpt = new MqttConnectOptions();
connOpt.setCleanSession(false); // Nettoyage de la session, remise a zero
connOpt.setKeepAliveInterval(30); // Temps maximum en secondes entre deux envois ou deux receptions
client.setCallback(this);
// Connexion au serveur MQTT
System.out.println("Connexion a : " + broker + " Publisher");
client.connect(connOpt);
client.subscribe(topic1f,2);
MqttMessage mess = new MqttMessage();
String phrase = Arrays.toString(mess.getPayload());
System.out.println("La phrase est : " + phrase);
System.out.println("Fin de reception des messages");
client.disconnect(); // Deconnexion du serveur MQTT
System.out.println("Deconnexion subscriber");
}catch(MqttException e){
e.printStackTrace();
}
}
@Override
public void connectionLost(Throwable thrwbl) {
System.out.println("Connexion perdue");
}
@Override
public void messageArrived(String string, MqttMessage message) throws Exception {
System.out.println("Le message recu est : " + Arrays.toString(message.getPayload()));
}
@Override
public void deliveryComplete(IMqttDeliveryToken imdt) {
System.out.println("Message arrive");
}
} |
Mon main :
Code:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
| package publishclient;
import org.eclipse.paho.client.mqttv3.MqttException;
/**
*
* @author
*/
public class PublishClient {
/**
* @param args the command line arguments
* @throws org.eclipse.paho.client.mqttv3.MqttException
*/
public static void main(String[] args) throws MqttException {
PubClient publieur = new PubClient();
SubClient subscriber = new SubClient();
}
} |
