anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fixed ugly empty session storage, avoid fake NullObject empty subscription subscription

Browse Source
This commit is contained in:
andsel
2015-09-20 14:10:28 +02:00
parent f1e2aff331
commit 6e1524817e
6 changed files with 62 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ChangeLog.txt
+1
View File
@@ -8,6 +8,7 @@ Version 0.8:
Moved config to be not only filesystem based, but also classpath and memory (fix issue #73)
Switched from commit pesistence on every change to time based to fix delay in handling many msgs/sec (fix issue #79)
Changed willMessage and password in CONNECT message from String to byte[] (fix issue #94 and #95)
Fixed ugly empty session storage, avoid fake NullObject empty subscription subscription
Version 0.7:
Fix issue #38, added aligned to MQTT 3.1.1
broker/src/main/java/org/eclipse/moquette/spi/ISessionsStore.java
+2
View File
@@ -53,4 +53,6 @@ public interface ISessionsStore {
* @return true iff there are subscriptions persisted with clientID
*/
boolean contains(String clientID);
void createNewSession(String clientID);
}
broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java
+9
View File
@@ -208,6 +208,15 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore {
return m_persistentSubscriptions.containsKey(clientID);
}
@Override
public void createNewSession(String clientID) {
if (m_persistentSubscriptions.containsKey(clientID)) {
LOG.error("already exists a session for client <{}>", clientID);
return;
}
m_persistentSubscriptions.put(clientID, new HashSet<Subscription>());
}
@Override
public List<Subscription> listAllSubscriptions() {
List<Subscription> allSubscriptions = new ArrayList<Subscription>();
broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java
+6 -3
View File
@@ -243,14 +243,17 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
ConnAckMessage okResp = new ConnAckMessage();
okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED);
if (!msg.isCleanSession() && m_sessionsStore.contains(msg.getClientID())) {
boolean isSessionAlreadyStored = m_sessionsStore.contains(msg.getClientID());
if (!msg.isCleanSession() && isSessionAlreadyStored) {
okResp.setSessionPresent(true);
}
session.write(okResp);
m_interceptor.notifyClientConnected(msg);
LOG.info("Create persistent session for clientID <{}>", msg.getClientID());
m_sessionsStore.addNewSubscription(Subscription.createEmptySubscription(msg.getClientID(), true)); //null means EmptySubscription
if (!isSessionAlreadyStored) {
LOG.info("Create persistent session for clientID <{}>", msg.getClientID());
m_sessionsStore.createNewSession(msg.getClientID());
}
LOG.info("Connected client ID <{}> with clean session {}", msg.getClientID(), msg.isCleanSession());
if (!msg.isCleanSession()) {
//force the republish of stored QoS1 and QoS2
broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java
+12 -1
View File
@@ -285,7 +285,7 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
}
public List<Subscription> listAllSubscriptions() {
List<Subscription> allSubscriptions = new ArrayList<Subscription>();
List<Subscription> allSubscriptions = new ArrayList<>();
for (Map.Entry<String, Set<Subscription>> entry : m_persistentSubscriptions.entrySet()) {
allSubscriptions.addAll(entry.getValue());
}
@@ -298,6 +298,17 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
return m_persistentSubscriptions.containsKey(clientID);
}
@Override
public void createNewSession(String clientID) {
LOG.debug("createNewSession for client <{}>", clientID);
if (m_persistentSubscriptions.containsKey(clientID)) {
LOG.error("already exists a session for client <{}>", clientID);
return;
}
LOG.debug("clientID {} is a newcome, creating it's empty subscriptions set", clientID);
m_persistentSubscriptions.put(clientID, new HashSet<Subscription>());
}
public void close() {
this.m_db.commit();
LOG.debug("persisted subscriptions {}", m_persistentSubscriptions);
broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java
+32 -4
View File
@@ -15,14 +15,12 @@
*/
package org.eclipse.moquette.spi.impl;
import org.eclipse.moquette.proto.messages.AbstractMessage;
import org.eclipse.moquette.proto.messages.ConnAckMessage;
import org.eclipse.moquette.proto.messages.ConnectMessage;
import org.eclipse.moquette.proto.messages.DisconnectMessage;
import org.eclipse.moquette.proto.messages.*;
import org.eclipse.moquette.server.netty.NettyChannel;
import org.eclipse.moquette.spi.IMessagesStore;
import org.eclipse.moquette.spi.ISessionsStore;
import org.eclipse.moquette.spi.impl.security.PermitAllAuthorizator;
import org.eclipse.moquette.spi.impl.subscriptions.Subscription;
import org.eclipse.moquette.spi.impl.subscriptions.SubscriptionsStore;
import org.junit.Before;
import org.junit.Test;
@@ -258,4 +256,34 @@ public class ProtocolProcessor_CONNECT_Test {
assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, connAckMsg.getReturnCode());
}
@Test
public void testMultipleReconnection() throws InterruptedException {
//connect with clean a false and subscribe to a topic
connMsg = new ConnectMessage();
connMsg.setProtocolVersion(VERSION_3_1_1);
connMsg.setClientID("CliID");
connMsg.setCleanSession(false);
m_processor.processConnect(m_session, connMsg);
assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_session.getReturnCode());
//subscribe
SubscribeMessage subscribeMsg = new SubscribeMessage();
subscribeMsg.addSubscription(new SubscribeMessage.Couple((byte) AbstractMessage.QOSType.MOST_ONE.ordinal(), FAKE_TOPIC));
m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, "CliID");
m_session.setAttribute(NettyChannel.ATTR_KEY_CLEANSESSION, false);
m_processor.processSubscribe(m_session, subscribeMsg);
Subscription expectedSubscription = new Subscription("CliID", FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, false);
assertTrue(subscriptions.contains(expectedSubscription));
//disconnect
m_processor.processDisconnect(m_session, new DisconnectMessage());
//reconnect clean session a false
m_processor.processConnect(m_session, connMsg);
assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_session.getReturnCode());
//verify that the first subscription is still preserved
assertTrue(subscriptions.contains(expectedSubscription));
}
}