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

Removed client active/not active state from persistence (issue #118)

Browse Source
This commit is contained in:
andsel
2015-12-19 18:01:54 +01:00
parent 4f65a59324
commit 8e891da49f
9 changed files with 67 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ChangeLog.txt
+1
View File
@@ -16,6 +16,7 @@ Version 0.8:
Added config property 'autosave_interval' (issue #97)
Removed leaking ref to parent's node (issue #119 and #126)
Added internal publish once the broker is embedded (issue #113)
Removed client active/not active state from persistence (issue #118)
Version 0.7:
Fix issue #38, added aligned to MQTT 3.1.1
broker/src/main/java/io/moquette/spi/ClientSession.java
-2
View File
@@ -148,12 +148,10 @@ public class ClientSession {
public void activate() {
this.active = true;
this.m_sessionsStore.activate(this.clientID);
}
public void deactivate() {
this.active = false;
this.m_sessionsStore.deactivate(this.clientID);
}
public boolean isActive() {
broker/src/main/java/io/moquette/spi/ISessionsStore.java
-4
View File
@@ -103,10 +103,6 @@ public interface ISessionsStore {
* */
ClientSession sessionForClient(String clientID);
void activate(String clientID);
void deactivate(String clientID);
void inFlightAck(String clientID, int messageID);
/**
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+15 -3
View File
@@ -92,7 +92,7 @@ public class ProtocolProcessor {
private static final Logger LOG = LoggerFactory.getLogger(ProtocolProcessor.class);
private Map<String, ConnectionDescriptor> m_clientIDs;
protected Map<String, ConnectionDescriptor> m_clientIDs;
private SubscriptionsStore subscriptions;
private boolean allowAnonymous;
private IAuthorizator m_authorizator;
@@ -264,9 +264,16 @@ public class ProtocolProcessor {
int messageID = msg.getMessageID();
//Remove the message from message store
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
verifyToActivate(clientID, targetSession);
targetSession.inFlightAcknowledged(messageID);
}
private void verifyToActivate(String clientID, ClientSession targetSession) {
if (m_clientIDs.containsKey(clientID)) {
targetSession.activate();
}
}
private static IMessagesStore.StoredMessage asStoredMessage(PublishMessage msg) {
IMessagesStore.StoredMessage stored = new IMessagesStore.StoredMessage(msg.getPayload().array(), msg.getQos(), msg.getTopicName());
stored.setRetained(msg.isRetainFlag());
@@ -452,6 +459,7 @@ public class ProtocolProcessor {
qos = sub.getRequestedQos();
}
ClientSession targetSession = m_sessionsStore.sessionForClient(sub.getClientId());
verifyToActivate(sub.getClientId(), targetSession);
LOG.debug("Broker republishing to client <{}> topic <{}> qos <{}>, active {}",
sub.getClientId(), sub.getTopicFilter(), qos, targetSession.isActive());
@@ -554,6 +562,7 @@ public class ProtocolProcessor {
int messageID = msg.getMessageID();
LOG.debug("PUB --PUBREL--> SRV processPubRel invoked for clientID {} ad messageID {}", clientID, messageID);
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
verifyToActivate(clientID, targetSession);
IMessagesStore.StoredMessage evt = targetSession.storedMessage(messageID);
route2Subscribers(evt);
@@ -581,6 +590,7 @@ public class ProtocolProcessor {
String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID);
int messageID = msg.getMessageID();
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
verifyToActivate(clientID, targetSession);
//remove from the inflight and move to the QoS2 second phase queue
targetSession.inFlightAcknowledged(messageID);
targetSession.secondPhaseAckWaiting(messageID);
@@ -599,6 +609,7 @@ public class ProtocolProcessor {
LOG.debug("\t\tSRV <--PUBCOMP-- SUB processPubComp invoked for clientID {} ad messageID {}", clientID, messageID);
//once received the PUBCOMP then remove the message from the temp memory
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
verifyToActivate(clientID, targetSession);
targetSession.secondPhaseAcknowledged(messageID);
}
@@ -646,7 +657,7 @@ public class ProtocolProcessor {
LOG.debug("UNSUBSCRIBE subscription on topics {} for clientID <{}>", topics, clientID);
ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);
verifyToActivate(clientID, clientSession);
for (String topic : topics) {
boolean validTopic = SubscriptionsStore.validate(topic);
if (!validTopic) {
@@ -674,6 +685,7 @@ public class ProtocolProcessor {
LOG.debug("SUBSCRIBE client <{}> packetID {}", clientID, msg.getMessageID());
ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);
verifyToActivate(clientID, clientSession);
//ack the client
SubAckMessage ackMessage = new SubAckMessage();
ackMessage.setMessageID(msg.getMessageID());
@@ -715,7 +727,7 @@ public class ProtocolProcessor {
});
ClientSession targetSession = m_sessionsStore.sessionForClient(newSubscription.getClientId());
verifyToActivate(newSubscription.getClientId(), targetSession);
for (IMessagesStore.StoredMessage storedMsg : messages) {
//fire the as retained the message
LOG.debug("send publish message for topic {}", newSubscription.getTopicFilter());
broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java
+1 -3
View File
@@ -46,11 +46,9 @@ public class MapDBPersistentStore {
* */
public static class PersistentSession implements Serializable {
public final boolean cleanSession;
public final boolean active;
public PersistentSession(boolean cleanSession, boolean active) {
public PersistentSession(boolean cleanSession) {
this.cleanSession = cleanSession;
this.active = active;
}
}
broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java
+2 -25
View File
@@ -129,7 +129,7 @@ class MapDBSessionsStore implements ISessionsStore {
throw new IllegalArgumentException("Can't create a session with the ID of an already existing" + clientID);
}
LOG.debug("clientID {} is a newcome, creating it's empty subscriptions set", clientID);
m_persistentSessions.putIfAbsent(clientID, new PersistentSession(cleanSession, false));
m_persistentSessions.putIfAbsent(clientID, new PersistentSession(cleanSession));
return new ClientSession(clientID, m_messagesStore, this, cleanSession);
}
@@ -140,30 +140,7 @@ class MapDBSessionsStore implements ISessionsStore {
}
PersistentSession storedSession = m_persistentSessions.get(clientID);
ClientSession clientSession = new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession);
if (storedSession.active) {
clientSession.activate();
}
return clientSession;
}
@Override
public void activate(String clientID) {
activationHelper(clientID, true);
}
@Override
public void deactivate(String clientID) {
activationHelper(clientID, false);
}
private void activationHelper(String clientID, boolean activation) {
PersistentSession storedSession = m_persistentSessions.get(clientID);
if (storedSession == null) {
throw new IllegalStateException((activation ? "activating" : "deactivating") + " a session never stored/created, clientID <"+ clientID + ">", null);
}
PersistentSession newStoredSession = new PersistentSession(storedSession.cleanSession, activation);
m_persistentSessions.put(clientID, newStoredSession);
return new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession);
}
@Override
broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java
+42
View File
@@ -401,4 +401,46 @@ public class ServerIntegrationPahoTest {
subscriberB.disconnect();
}
@Test
public void testSubcriptionDoesntStayActiveAfterARestart() throws Exception {
LOG.info("*** testSubcriptionDoesntStayActiveAfterARestart ***");
String tmpDir = System.getProperty("java.io.tmpdir");
//clientForSubscribe1 connect and subscribe to /topic QoS2
MqttClientPersistence dsSubscriberA = new MqttDefaultFilePersistence(tmpDir + File.separator + "clientForSubscribe1");
MqttClient clientForSubscribe1 = new MqttClient("tcp://localhost:1883", "clientForSubscribe1", dsSubscriberA);
TestCallback cbSubscriber1 = new TestCallback();
clientForSubscribe1.setCallback(cbSubscriber1);
clientForSubscribe1.connect();
clientForSubscribe1.subscribe("topic", 0);
//server stop
m_server.stopServer();
System.out.println("\n\n SEVER REBOOTING \n\n");
//server start
startServer();
//clientForSubscribe2 connect and subscribe to /topic QoS2
MqttClientPersistence dsSubscriberB = new MqttDefaultFilePersistence(tmpDir + File.separator + "clientForSubscribe2");
MqttClient clientForSubscribe2 = new MqttClient("tcp://localhost:1883", "clientForSubscribe2", dsSubscriberB);
TestCallback cbSubscriber2 = new TestCallback();
clientForSubscribe2.setCallback(cbSubscriber2);
clientForSubscribe2.connect();
clientForSubscribe2.subscribe("topic", 0);
//clientForPublish publish on /topic with QoS2 a message
MqttClientPersistence dsSubscriberPUB = new MqttDefaultFilePersistence(tmpDir + File.separator + "clientForPublish");
MqttClient clientForPublish = new MqttClient("tcp://localhost:1883", "clientForPublish", dsSubscriberPUB);
clientForPublish.connect();
clientForPublish.publish("topic", "Hello".getBytes(), 2, true);
//verify clientForSubscribe1 doesn't receive a notification but clientForSubscribe2 yes
System.out.println("Before waiting to receive 1 sec from " + clientForSubscribe1.getClientId());
assertFalse(clientForSubscribe1.isConnected());
assertTrue(clientForSubscribe2.isConnected());
System.out.println("Waiting to receive 1 sec from " + clientForSubscribe2.getClientId());
MqttMessage messageOnB = cbSubscriber2.getMessage(true);
assertEquals("Hello", new String(messageOnB.getPayload()));
}
}
broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java
+4
View File
@@ -18,6 +18,7 @@ package io.moquette.spi.impl;
import io.moquette.proto.messages.*;
import io.moquette.interception.InterceptHandler;
import io.moquette.proto.messages.AbstractMessage.QOSType;
import io.moquette.server.ConnectionDescriptor;
import io.moquette.server.netty.NettyChannel;
import io.moquette.spi.ClientSession;
import io.moquette.spi.IMatchingCondition;
@@ -504,6 +505,9 @@ public class ProtocolProcessorTest {
}
};
processor.init(subscriptions, memoryMessageStore, sessionsStore, null, true, null, NO_OBSERVERS_INTERCEPTOR);
//just to activate the two sessions
processor.m_clientIDs.put("Sub A", new ConnectionDescriptor("Sub A", null, true));
processor.m_clientIDs.put("Sub B", new ConnectionDescriptor("Sub B", null, true));
//Exercise
processor.route2Subscribers(forwardPublish);
broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java
+2 -25
View File
@@ -110,7 +110,7 @@ public class MemorySessionStore implements ISessionsStore {
}
LOG.debug("clientID {} is a newcome, creating it's empty subscriptions set", clientID);
m_persistentSubscriptions.put(clientID, new HashSet<Subscription>());
m_persistentSessions.put(clientID, new MapDBPersistentStore.PersistentSession(cleanSession, false));
m_persistentSessions.put(clientID, new MapDBPersistentStore.PersistentSession(cleanSession));
return new ClientSession(clientID, m_messagesStore, this, cleanSession);
}
@@ -121,11 +121,7 @@ public class MemorySessionStore implements ISessionsStore {
}
MapDBPersistentStore.PersistentSession storedSession = m_persistentSessions.get(clientID);
ClientSession clientSession = new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession);
if (storedSession.active) {
clientSession.activate();
}
return clientSession;
return new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession);
}
@Override
@@ -153,25 +149,6 @@ public class MemorySessionStore implements ISessionsStore {
return null;
}
@Override
public void activate(String clientID) {
activationHelper(clientID, true);
}
@Override
public void deactivate(String clientID) {
activationHelper(clientID, false);
}
private void activationHelper(String clientID, boolean activation) {
MapDBPersistentStore.PersistentSession storedSession = m_persistentSessions.get(clientID);
if (storedSession == null) {
throw new IllegalStateException((activation ? "activating" : "deactivating") + " a session never stored/created, clientID <"+ clientID + ">", null);
}
MapDBPersistentStore.PersistentSession newStoredSession = new MapDBPersistentStore.PersistentSession(storedSession.cleanSession, activation);
m_persistentSessions.put(clientID, newStoredSession);
}
@Override
public void inFlightAck(String clientID, int messageID) {
Map<Integer, String> m = this.m_inflightStore.get(clientID);