Я пытаюсь заставить некоторых клиентов на основе Paho работать с сервером Vert.x MQTT. Я пытаюсь опубликовать в тестовой теме, на которую подписан мой принимающий клиент. Мне трудно получить сообщение от моего клиентского издателя моему клиентскому подписчику.
Используя верные примеры, которые я видел в Интернете, я собрал брокера MQTT. Код для кода брокера MQTT Vert.x приведен ниже:
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);
});
}
}
У меня есть клиент Paho, который должен получать сообщения от темы, на которую он подписан. Код для этого ниже:
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();
}
}
}
Подписчик и издатель являются локальными, хотя они могут (и будут) удаленными от брокера, как только я разверну свои приложения. Код на основе Paho, который я использую для публикации, приведен ниже:
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();
}
}
Обратите внимание, что я публикую жестко закодированную строку (для целей тестирования) от брокера (как показано в методе doPublish), хотя следует отметить, что я улавливаю попытку клиента издателя опубликовать в моем обработчике конечной точки. К сожалению, хотя у меня нет проблем с подключением подписчика и издателя к брокеру, по какой-то причине сообщение не доходит до подписчика. Я пробовал разные вещи, но, хотя клиент издателя фактически публикует для брокера, по какой-то причине публикация брокера не может отправить строку подписчику.
Я почти уверен, что что-то здесь не хватает, но не могу представить, что это могло быть. Может ли кто-нибудь помочь мне заставить это работать ???
Заранее благодарим за любую помощь или идеи.