I am trying to get some Paho- based clients to work with the Vert.x MQTT Server. I am trying to publish to a test Topic which my receiving client is subscribed to. I am having difficulty getting a message from my client publisher to my client subscriber.
Using verious examples I have seen on the Internet, I have put together an MQTT Broker. The code for the Vert.x MQTT Broker code is below:
public class MQTTBroker
{
public MQTTBroker()
{
MqttServerOptions opts = new MqttServerOptions();
opts.setHost("localhost");
opts.setPort(1883);
MqttServer server = MqttServer.create(Vertx.vertx(),opts);
server.endpointHandler(endpoint -> {
System.out.println("MQTT client [" + endpoint.clientIdentifier() + "] request to connect, clean session = " + endpoint.isCleanSession());
endpoint.accept(false);
if("Test_Send".equals(endpoint.clientIdentifier()))
{
doPublish(endpoint);
handleClientDisconnect(endpoint);
}
else
{
handleSubscription(endpoint);
handleUnsubscription(endpoint);
handleClientDisconnect(endpoint);
}
}).listen(ar -> {
if (ar.succeeded()) {
System.out.println("MQTT server is listening on port " + ar.result().actualPort());
} else {
System.out.println("Error on starting the server");
ar.cause().printStackTrace();
}
});
}
protected void handleSubscription(MqttEndpoint endpoint) {
endpoint.subscribeHandler(subscribe -> {
List grantedQosLevels = new ArrayList < > ();
for (MqttTopicSubscription s: subscribe.topicSubscriptions()) {
System.out.println("Subscription for " + s.topicName() + " with QoS " + s.qualityOfService());
grantedQosLevels.add(s.qualityOfService());
}
endpoint.subscribeAcknowledge(subscribe.messageId(), grantedQosLevels);
});
}
protected void handleUnsubscription(MqttEndpoint endpoint) {
endpoint.unsubscribeHandler(unsubscribe -> {
for (String t: unsubscribe.topics()) {
System.out.println("Unsubscription for " + t);
}
endpoint.unsubscribeAcknowledge(unsubscribe.messageId());
});
}
protected void publishHandler(MqttEndpoint endpoint) {
endpoint.publishHandler(message -> {
endpoint.publishAcknowledge(message.messageId());
}).publishReleaseHandler(messageId -> {
endpoint.publishComplete(messageId);
});
}
protected void handleClientDisconnect(MqttEndpoint endpoint) {
endpoint.disconnectHandler(h -> {
System.out.println("The remote client has closed the connection.");
});
}
protected void doPublish(MqttEndpoint endpoint) {
// just as example, publish a message with QoS level 2
endpoint.publish("Test_Topic",
Buffer.buffer("Hello from the Vert.x MQTT server"),
MqttQoS.EXACTLY_ONCE,
false,
false);
publishHandler(endpoint);
// specifing handlers for handling QoS 1 and 2
endpoint.publishAcknowledgeHandler(messageId -> {
System.out.println("Received ack for message = " + messageId);
}).publishReceivedHandler(messageId -> {
endpoint.publishRelease(messageId);
}).publishCompleteHandler(messageId -> {
System.out.println("Received ack for message = " + messageId);
});
}
}
I have a Paho client which is supposed to receive messages from the Topic it subscribes to. The code for that is below:
public class Receiver implements MqttCallback
{
public Receiver()
{
MqttClient client = null;
try
{
MemoryPersistence persist = new MemoryPersistence();
client = new MqttClient("tcp://localhost:1883",persist);
client.setCallback(this);
client.connect();
client.subscribe("Test_Topic");
System.out.println("The receiver is initialized.");
}
catch(Exception exe)
{
exe.printStackTrace();
}
}
@Override
public void connectionLost(Throwable arg0)
{
System.out.println("Connection is lost!");
}
@Override
public void deliveryComplete(IMqttDeliveryToken arg0)
{
System.out.println("Delivered!");
}
@Override
public void messageArrived(String theStr, MqttMessage theMsg)
{
System.out.println("Message from: "+theStr);
try
{
String str = new String(theMsg.getPayload());
System.out.println("Message is: "+str);
}
catch(Exception exe)
{
exe.printStackTrace();
}
}
}
The subscriber and the publisher are local, though they can (and will be) remote from the Broker once I deploy my applications. The Paho- based code that I use to publish is below:
public void publish(String theMsg)
{
MqttClient nwclient = null;
try
{
ServConfigurator serv = ServConfigurator.getInstance();
MemoryPersistence persist = new MemoryPersistence();
String url = "tcp://localhost:1883";
nwclient = new MqttClient(url,"Test_Send",persist);
MqttConnectOptions option = new MqttConnectOptions();
option.setCleanSession(true);
nwclient.connect(option);
MqttMessage message = new MqttMessage(theMsg.getBytes());
message.setQos(2);
nwclient.publish("Test_Topic",message);
nwclient.disconnect();
nwclient.close();
}
catch(Exception exe)
{
exe.printStackTrace();
}
}
Note that I am publishing a hard- coded string (for testing purposes) from the Broker (as shown in the doPublish method) though it should be noted that I am catching the publisher client's attempt to publish in my endpoint handler. Unfortunately, while I have no problem connecting the Subscriber and the Publisher to the Broker, for some reason the message is not getting to the Subscriber. I have tried various things, but while the publisher client actually publishes to the Broker, for some reason the Broker's publish fails to send the string to the subscriber.
I am pretty sure I am missing something here, but I cannot think what it could be. Can anyone help me to get this to work???
Thanks in advance for any help or insights.
When the MQTT server receives a message from the publisher so that the endpoint.publishHandler is called passing it the message, I don't see any logic in your code to get the topic and search for the subscribers (endpoints) registered for that topic and sending them the message. At same time I don't see that you code save in some way references between subscriber endpoints and subscribed topics for doing the above research. Remember that the MQTT server is not a broker, it doesn't take care of a list of subscribed clients on topics for you; you can build a broker on top of it which should be something you are trying to do ?