Подписчик клиента блокировки Java HiveMQ не потребляет никаких сообщений - PullRequest
Новая
       36

Подписчик клиента блокировки Java HiveMQ не потребляет никаких сообщений

2 голосов
/ 01 апреля 2019

Я использую Java-клиент HiveMQ для подключения к брокеру HiveMQ. Подписчик блокирующего клиента не потребляет никаких сообщений. С MQTTBox публикация и подписка работают нормально. Вот код Я слежу за клиентской документацией HiveMQ Java 

public class MQTTMain {
public static void main(String[] args) {
Mqtt3BlockingClient pubClient = MqttClient.builder()
                                          .useMqttVersion3()
                                          .identifier("pub")
                                          .serverHost("hostaddress")
                                          .serverPort(1883)
                                          .buildBlocking();
 Mqtt3BlockingClient subClient = MqttClient.builder()
                                           .useMqttVersion3()
                                           .identifier("sub")
                                           .serverHost("hostaddress")
                                           .serverPort(1883)
                                           .buildBlocking();

    pubClient.connectWith().keepAlive(10000).send();
    publish(pubClient, "test/topic", "test");
    subClient.connectWith().keepAlive(10000).send();
    subscribe(subClient, "test/topic");
    while (true) {
    }
}

public static void subscribe(Mqtt3BlockingClient client, String topic) {
    try (final Mqtt3Publishes publishes = 
              client.publishes(MqttGlobalPublishFilter.ALL)) {
        try {
             publishes.receive(1, TimeUnit.SECONDS)
            .ifPresent(System.out::println);
            publishes.receive(10000, TimeUnit.MILLISECONDS)
            .ifPresent(System.out::println);
            } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    } catch (Exception e) {
        // TODO: handle exception
    }

   client
  .subscribeWith()
  .topicFilter(topic)
  .qos(MqttQos.AT_LEAST_ONCE)
  .send();
}

public static void publish(Mqtt3BlockingClient client, String topic, 
String payload) {
   client
  `enter code here`.publishWith()
  .topic(topic)
  .qos(MqttQos.AT_LEAST_ONCE)
  .payload(payload.getBytes())
  .send();
}
}

Зависимость Maven: 

<!-- MQTT Client -->      
<dependency>
        <groupId>com.hivemq</groupId>
        <artifactId>hivemq-mqtt-client</artifactId>
        <version>1.0.0</version>
</dependency>

Я что-то упустил? Любой указатель будет очень полезен.

1 Ответ

7 голосов
/ 02 апреля 2019

Вы должны подписаться, прежде чем сможете получать сообщения.

Publishes.receive является операцией блокировки, поэтому она ожидает получения сообщения.Вы подписываетесь после операции получения.Вы должны сделать что-то вроде этого: 

Mqtt3BlockingClient client = Mqtt3Client.builder()
        .identifier("sub")
        .serverHost("hostaddress")
        .serverPort(1883)
        .buildBlocking();

client.connect();

try (final Mqtt3BlockingClient.Mqtt3Publishes publishes = 
             client.publishes(MqttGlobalPublishFilter.ALL)) {

    client.subscribeWith().topicFilter("test/topic").qos(MqttQos.AT_LEAST_ONCE).send();

    publishes.receive(1, TimeUnit.SECONDS).ifPresent(System.out::println);
}
...