Spring Integration MQTT (EMQX): Сообщения, отправленные, когда клиент не принимается после повторного подключения ⇐ JAVA

[Anonymous]

Я использую весеннюю интеграцию с EMQX MQTT Broker и пытаюсь реализовать доставку сообщений для сценария, где:
Клиент находится в автономном режиме, когда брокер получает сообщения
после перезапуска клиента/reconnect, в Queued сообщества должны быть переданы и обрабатываются, однако это не происходит. /> Я использую mqttv5pahomessagedrivenchanneladapter и emqx со следующим: < /p>
Настройки: < /p>

[*] qos: 1 < /p>
< /li>
Qos: antabled < /p>
< /li>
< /p>
< /li>
/> очередь сообщений: видно в панели инструментов (например, 3/1000, когда клиент не в автономном режиме) < /p>
< /li>
пружина: < /p>
< /li>
< /ul>

Код: Выделить всё

@Bean
public ClientManager mqttClientManager() {
return new DefaultMqtt5ClientManager(() -> {
return new MqttConnectionOptionsBuilder()
.cleanStart(false)  // Attempt to enable persistent session
.sessionExpiryInterval(86400L) // 1 day session retention
.automaticReconnect(true)
.serverURI("tcp://localhost:1883")
.build();
});
}
< /code>
и IntegrationFlow: < /p>
@Bean
public IntegrationFlow mqttInputFlow(
ClientManager clientManager,
@Qualifier("mqttInboundTopics") List topics
) {
Mqttv5PahoMessageDrivenChannelAdapter adapter =
new Mqttv5PahoMessageDrivenChannelAdapter(clientManager, topics.toArray(new String[0]));

adapter.setQos(1);
adapter.setManualAcks(false); // Using automatic ack

return IntegrationFlow.from(adapter)
.handle(msg -> {
String payload = new String((byte[]) msg.getPayload(), StandardCharsets.UTF_8);
LOG.info("Received MQTT message: {}", payload);
})
.get();
}

Problem

When messages are sent while the client is offline, EMQX shows them as queued (e.g., 3/1000).

After the client restarts, EMQX shows the queue as empty (0/1000) — which indicates Доставка. < /p>
< /li>
Но метод .handle (...) весной не используется. Проверено: < /h2>

emqx показывает, что сеанс клиента сохраняется после отключения < /p>
< /li>
Qos установлен на 1 < /p>
< /li>
, оставленные, оставленные сообщения, включенные < /p>
< /li>
. /> CleanStart (false) и SessionExpiryInterval (...) установлены < /p>
< /li>
Используют оба. queeue () и. /> Когда клиент находится в Интернете, все работает отлично < /p>
< /li>
< /ul>
Вопрос < /h2>
Чего еще может быть отсутствует в Spring Integration + Setup Emqx, что предотвращает получение линейных сообщений при восстановлении? MQTTV5PahomessageDrivenchannelAdapter, или есть еще один механизм весны, который должен быть включен для получения этих сообщений?

Подробнее здесь: https://stackoverflow.com/questions/797 ... are-not-re