Vert.x MQTTサーバーからのメッセージを受信できません

Question

一部のPaho-ベースクライアントをVert.x MQTTサーバーと連携させようとしています。私は受信クライアントが購読しているテストトピックに公開しようとしています。クライアントのパブリッシャからクライアントの購読者にメッセージを受け取ることができません。

私はインターネット上で見た例を使って、MQTT Brokerをまとめました。 Vert.x MQTTブローカーコードのためのコードは以下の通りです：

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-ベースのコードは、以下である：（doPublish方法に示されるように）、それがなければならないけど、ブローカから（テスト目的のために）ハードコード化された文字列を公開してい

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(); 
    } 
} 

注エンドポイントハンドラでパブリッシャクライアントがパブリッシュしようとしていることに注目しています。残念ながら、私はブローカにサブスクライバとパブリッシャを接続する際に問題はありませんが、何らかの理由でメッセージがサブスクライバに届いていません。私は様々なことを試みましたが、パブリッシャークライアントが実際にブローカーにパブリッシュする間、何らかの理由でブローカーのパブリッシュがストリングをサブスクライバーに送信できませんでした。

私はここに何かが欠けていると確信していますが、それが何であるかは考えられません。誰でも私にこれを働かせる手助けができますか？

ご協力いただきありがとうございます。

Answer 1

MQTTサーバーがパブリッシャーからメッセージを受け取り、endpoint.publishHandlerがメッセージを渡すようにメッセージを受け取ると、コードにトピックがあり、登録されたサブスクライバー（エンドポイント）を検索するロジックが表示されませんそのトピックのメッセージを送信します。 同時に、上記の調査を行うためのサブスクライバエンドポイントとサブスクリプションされたトピックの間の参照を何らかの方法で保存するコードはありません。 MQTTサーバーはブローカーではなく、トピックに関するサブスクライブされたクライアントのリストを処理しません。あなたはそれの上にあなたがやろうとしているものでなければならないブローカーを作ることができますか？