Add 'getSubscriptions' API on Server.

Add a test for sessions without clean session flag.
This commit is contained in:
teranisi
2016-10-20 14:23:57 +09:00
parent cf67d5ac52
commit 81a1d8d2b2
6 changed files with 69 additions and 3 deletions
@@ -97,6 +97,17 @@ public class ServerIntegrationEmbeddedPublishTest {
m_server.internalPublish(message);
}
private void connectNoCleanSession() throws Exception {
MqttConnectOptions opts = new MqttConnectOptions();
opts.setCleanSession(false);
m_subscriber.connect(opts);
}
private void subscribeToWithQosAndNoCleanSession(String topic, int qos) throws Exception {
connectNoCleanSession();
m_subscriber.subscribe(topic, qos);
}
private void verifyNoMessageIsReceived() throws Exception {
MqttMessage msg = m_callback.getMessage(false);
assertNull(msg);
@@ -276,4 +287,16 @@ public class ServerIntegrationEmbeddedPublishTest {
MqttMessage message = m_callback.getMessage(true);
assertTrue(message == null);
}
@Test
public void testClientSubscribeWithoutCleanSession() throws Exception {
LOG.info("*** testClientSubscribeWithoutCleanSession ***");
subscribeToWithQosAndNoCleanSession("foo", 1);
m_subscriber.disconnect();
assertTrue(m_server.getSubscriptions().size() == 1);
connectNoCleanSession();
assertTrue(m_server.getSubscriptions().size() == 1);
internalPublishToWithQosAndRetained("foo", QOSType.MOST_ONE, false);
verifyMessageIsReceivedSuccessfully();
}
}
@@ -157,6 +157,15 @@ public class MemorySessionStore implements ISessionsStore {
return null;
}
@Override
public List<Subscription> getSubscriptions() {
List<Subscription> subscriptions = new ArrayList<>();
for (Map.Entry<String, Set<Subscription>> entry : m_persistentSubscriptions.entrySet()) {
subscriptions.addAll(entry.getValue());
}
return subscriptions;
}
@Override
public void inFlightAck(String clientID, int messageID) {
Map<Integer, MessageGUID> m = this.m_inflightStore.get(clientID);