Topic implements com.hazelcast.nio.serialization.IdentifiedDataSerializable { public void publish(final Message message) { assert message != null; assert name.equals(message.topicName()); if (message.qos() != MqttQoS.AT_MOST_ONCE) { Message.NEXUS.put(message.key(), message); } TopicSubscriber.NEXUS.clientIdsOf(name).forEach(id -> { Session session = Session.NEXUS.get(id); logger.debug("Before publish to [clientId={}, sessionsSize={}]", id, Session.NEXUS.keySet().size()); assert session != null; publish(session, message); }); } Topic();  Topic(String name); String name(); Message retainedMessage(); void setRetainedMessage(Message message); void publish(final Message message); void publish(final Session session, final Message message); void publish(Session session, String messageKey); @JsonIgnore @Override int getFactoryId(); @JsonIgnore @Override int getId(); @Override void writeData(ObjectDataOutput out); @Override void readData(ObjectDataInput in); static final Topics NEXUS; static final int ID; }
@Test public void persistMessagesForDisconnectedPersistantSession() throws Exception { String clientId = TestUtil.newClientId(); ConnectOptions options = new ConnectOptions(); options.clientId(clientId); options.cleanSession(false); String topicName = "testTopic"; String message = "test message"; MqttClient client = new MqttClient("mqtt: client.connectOptions(options).connect(); client.subscribe(new MqttTopicSubscription(topicName, MqttQoS.EXACTLY_ONCE)); client.disconnect(true); Thread.sleep(100); Assert.assertNotNull(Session.NEXUS.get(clientId)); String publisherId = TestUtil.newClientId(); MqttClient publisher = new MqttClient("mqtt: ConnectOptions pubOptions = new ConnectOptions(); pubOptions.clientId(publisherId); pubOptions.cleanSession(true); int messageId = 1; publisher.connectOptions(pubOptions).connect(); publisher.publish(new Message(messageId, topicName, publisherId, message.getBytes(CharsetUtil.UTF_8), MqttQoS.EXACTLY_ONCE, false)); Thread.sleep(100); publisher.disconnect(true); Thread.sleep(1000); Assert.assertNull(Session.NEXUS.get(publisherId)); Topic topic = Topic.NEXUS.get(topicName); Assert.assertNotNull(topic); OutboundMessageStatus status = OutboundMessageStatus.NEXUS.getBy(messageId, clientId); Assert.assertNotNull(status); Assert.assertTrue(Messages.key(publisherId, messageId).equals(status.messageKey())); }