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

Introduced 'fat' ClientSession class responsible to model the concept of an MQTT client session

Browse Source
This commit is contained in:
andsel
2015-11-08 18:42:03 +01:00
parent 0b5c95ee57
commit c793ded63e
11 changed files with 184 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java
+59
View File
@@ -0,0 +1,59 @@
/*
* Copyright (c) 2012-2015 The original author or authors
* ------------------------------------------------------
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Apache License v2.0 which accompanies this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* The Apache License v2.0 is available at
* http://www.opensource.org/licenses/apache2.0.php
*
* You may elect to redistribute this code under either of these licenses.
*/
package org.eclipse.moquette.spi;
import org.eclipse.moquette.spi.impl.events.PublishEvent;
import java.util.List;
/**
* @author andrea
*/
public class ClientSession {
public final String clientID;
private final IMessagesStore messagesStore;
public ClientSession(String clientID, IMessagesStore messagesStore) {
this.clientID = clientID;
this.messagesStore = messagesStore;
}
//List of client's subscriptions
//list of messages not acknowledged by client
//list of in-flight messages
/**
* @return the list of messages to be delivered for client related to the session.
* */
public List<PublishEvent> storedMessages() {
return messagesStore.listMessagesInSession(clientID);
}
/**
* Remove a message previously stored for delivery.
* */
public void removeDelivered(int messageID) {
messagesStore.removeMessageInSession(clientID, messageID);
}
@Override
public String toString() {
return "ClientSession{" +
"clientID='" + clientID + '\'' +
'}';
}
}
broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java
+1 -1
View File
@@ -28,7 +28,7 @@ import java.util.List;
*/
public interface IMessagesStore {
public static class StoredMessage implements Serializable {
class StoredMessage implements Serializable {
final AbstractMessage.QOSType m_qos;
final byte[] m_payload;
final String m_topic;
broker/src/main/java/org/eclipse/moquette/spi/ISessionsStore.java
+6 -1
View File
@@ -54,5 +54,10 @@ public interface ISessionsStore {
*/
boolean contains(String clientID);
void createNewSession(String clientID);
ClientSession createNewSession(String clientID);
/**
* @return the session for the given clientID, null if not found.
* */
ClientSession sessionForClient(String clientID);
}
broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java
+12 -9
View File
@@ -20,6 +20,7 @@ import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import org.eclipse.moquette.server.netty.NettyChannel;
import org.eclipse.moquette.spi.ClientSession;
import org.eclipse.moquette.spi.IMatchingCondition;
import org.eclipse.moquette.spi.IMessagesStore;
import org.eclipse.moquette.spi.ISessionsStore;
@@ -220,7 +221,9 @@ public class ProtocolProcessor {
ConnAckMessage okResp = new ConnAckMessage();
okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED);
boolean isSessionAlreadyStored = m_sessionsStore.contains(msg.getClientID());
ClientSession clientSession = m_sessionsStore.sessionForClient(msg.getClientID());
boolean isSessionAlreadyStored = clientSession != null;
if (!msg.isCleanSession() && isSessionAlreadyStored) {
okResp.setSessionPresent(true);
}
@@ -229,12 +232,12 @@ public class ProtocolProcessor {
if (!isSessionAlreadyStored) {
LOG.info("Create persistent session for clientID <{}>", msg.getClientID());
m_sessionsStore.createNewSession(msg.getClientID());
clientSession = 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
republishStoredInSession(msg.getClientID());
republishStoredInSession(clientSession);
}
}
@@ -248,19 +251,19 @@ public class ProtocolProcessor {
/**
* Republish QoS1 and QoS2 messages stored into the session for the clientID.
* */
private void republishStoredInSession(String clientID) {
LOG.trace("republishStoredInSession for client <{}>", clientID);
List<PublishEvent> publishedEvents = m_messagesStore.listMessagesInSession(clientID);
private void republishStoredInSession(ClientSession clientSession) {
LOG.trace("republishStoredInSession for client <{}>", clientSession);
List<PublishEvent> publishedEvents = clientSession.storedMessages();
if (publishedEvents.isEmpty()) {
LOG.info("No stored messages for client <{}>", clientID);
LOG.info("No stored messages for client <{}>", clientSession.clientID);
return;
}
LOG.info("republishing stored messages to client <{}>", clientID);
LOG.info("republishing stored messages to client <{}>", clientSession.clientID);
for (PublishEvent pubEvt : publishedEvents) {
sendPublish(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(),
pubEvt.getMessage(), false, pubEvt.getMessageID());
m_messagesStore.removeMessageInSession(clientID, pubEvt.getMessageID());
clientSession.removeDelivered(pubEvt.getMessageID());
}
}
broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java
+15 -4
View File
@@ -17,6 +17,7 @@
package org.eclipse.moquette.spi.persistence;
import org.eclipse.moquette.proto.MQTTException;
import org.eclipse.moquette.spi.ClientSession;
import org.eclipse.moquette.spi.IMatchingCondition;
import org.eclipse.moquette.spi.IMessagesStore;
import org.eclipse.moquette.spi.ISessionsStore;
@@ -257,7 +258,7 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
Set<Subscription> subs = m_persistentSubscriptions.get(clientID);
if (!subs.contains(newSubscription)) {
LOG.debug("updating clientID {} subscriptions set with new subscription", clientID);
LOG.debug("updating <{}> subscriptions set with new subscription", clientID);
//TODO check the subs doesn't contain another subscription to the same topic with different
Subscription existingSubscription = null;
for (Subscription scanSub : subs) {
@@ -300,14 +301,24 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
}
@Override
public void createNewSession(String clientID) {
public ClientSession 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.error("already exists a session for client <{}>, bad condition", clientID);
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_persistentSubscriptions.put(clientID, new HashSet<Subscription>());
return new ClientSession(clientID, this);
}
@Override
public ClientSession sessionForClient(String clientID) {
if (!m_persistentSubscriptions.containsKey(clientID)) {
return null;
}
return new ClientSession(clientID, this);
}
@Override
broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationFuseTest.java
+2 -2
View File
@@ -96,8 +96,8 @@ public class ServerIntegrationFuseTest {
}
@Test
public void checkWillTestmaentIsPublishedOnConnectionKill_noRetain() throws Exception {
LOG.info("checkWillTestmaentIsPublishedOnConnectionKill");
public void checkWillTestamentIsPublishedOnConnectionKill_noRetain() throws Exception {
LOG.info("checkWillTestamentIsPublishedOnConnectionKill");
String willTestamentTopic = "/will/test";
String willTestamentMsg = "Bye bye";
broker/src/test/java/org/eclipse/moquette/spi/impl/MemorySessionStore.java
+24 -4
View File
@@ -15,6 +15,8 @@
*/
package org.eclipse.moquette.spi.impl;
import org.eclipse.moquette.spi.ClientSession;
import org.eclipse.moquette.spi.IMessagesStore;
import org.eclipse.moquette.spi.ISessionsStore;
import org.eclipse.moquette.spi.impl.subscriptions.Subscription;
import org.slf4j.Logger;
@@ -24,13 +26,19 @@ import java.util.*;
/**
*
* @author andrea
* @author andream_messagesStore
*/
public class MemorySessionStore implements ISessionsStore {
private static final Logger LOG = LoggerFactory.getLogger(MemorySessionStore.class);
private Map<String, Set<Subscription>> m_persistentSubscriptions = new HashMap<>();
private final IMessagesStore m_messagesStore;
public MemorySessionStore(IMessagesStore messagesStore) {
this.m_messagesStore = messagesStore;
}
@Override
public void removeSubscription(String topic, String clientID) {
LOG.debug("removeSubscription topic filter: {} for clientID: {}", topic, clientID);
@@ -82,12 +90,24 @@ public class MemorySessionStore implements ISessionsStore {
}
@Override
public void createNewSession(String clientID) {
public ClientSession 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.error("already exists a session for client <{}>, bad condition", clientID);
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_persistentSubscriptions.put(clientID, new HashSet<Subscription>());
return new ClientSession(clientID, m_messagesStore);
}
@Override
public ClientSession sessionForClient(String clientID) {
if (!m_persistentSubscriptions.containsKey(clientID)) {
return null;
}
return new ClientSession(clientID, m_messagesStore);
}
@Override
broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java → broker/src/test/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java
+47 -77
View File
@@ -18,6 +18,8 @@ package org.eclipse.moquette.spi.impl;
import java.nio.ByteBuffer;
import java.util.*;
import jdk.nashorn.internal.objects.NativeJava;
import org.eclipse.moquette.spi.ClientSession;
import org.eclipse.moquette.spi.IMatchingCondition;
import org.eclipse.moquette.spi.IMessagesStore;
import org.eclipse.moquette.spi.impl.events.PublishEvent;
@@ -34,17 +36,18 @@ import static org.eclipse.moquette.spi.impl.Utils.defaultGet;
*/
public class MemoryStorageService implements IMessagesStore, ISessionsStore {
private Map<String, Set<Subscription>> m_persistentSubscriptions = new HashMap<>();
private Map<String, StoredMessage> m_retainedStore = new HashMap<>();
//TODO move in a multimap because only Qos1 and QoS2 are stored here and they have messageID(key of secondary map)
private Map<String, List<PublishEvent>> m_persistentMessageStore = new HashMap<>();
private Map<String, PublishEvent> m_inflightStore = new HashMap<>();
private Map<String, Set<Integer>> m_inflightIDs = new HashMap<>();
private Map<String, PublishEvent> m_qos2Store = new HashMap<>();
private MemorySessionStore m_sessionsStore;
private static final Logger LOG = LoggerFactory.getLogger(MemoryStorageService.class);
public void initStore() {
m_sessionsStore = new MemorySessionStore(this);
}
@Override
@@ -69,7 +72,7 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore {
public Collection<StoredMessage> searchMatching(IMatchingCondition condition) {
LOG.debug("searchMatching scanning all retained messages, presents are {}", m_retainedStore.size());
List<StoredMessage> results = new ArrayList<StoredMessage>();
List<StoredMessage> results = new ArrayList<>();
for (Map.Entry<String, StoredMessage> entry : m_retainedStore.entrySet()) {
StoredMessage storedMsg = entry.getValue();
@@ -84,13 +87,7 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore {
@Override
public void storePublishForFuture(PublishEvent evt) {
LOG.debug("storePublishForFuture store evt {}", evt);
// List<PublishEvent> storedEvents;
String clientID = evt.getClientID();
// if (!m_persistentMessageStore.containsKey(clientID)) {
// storedEvents = new ArrayList<PublishEvent>();
// } else {
// storedEvents = m_persistentMessageStore.get(clientID);
// }
List<PublishEvent> storedEvents = defaultGet(m_persistentMessageStore, clientID, new ArrayList<PublishEvent>());
storedEvents.add(evt);
m_persistentMessageStore.put(clientID, storedEvents);
@@ -143,7 +140,7 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore {
}
/**
* Return the next valid packetIdentifer for the given client session.
* Return the next valid packetIdentifier for the given client session.
* */
@Override
public int nextPacketID(String clientID) {
@@ -161,73 +158,6 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore {
return nextPacketId;
}
@Override
public void removeSubscription(String topic, String clientID) {
LOG.debug("removeSubscription topic filter: {} for clientID: {}", topic, clientID);
if (!m_persistentSubscriptions.containsKey(clientID)) {
return;
}
Set<Subscription> clientSubscriptions = m_persistentSubscriptions.get(clientID);
//search for the subscription to remove
Subscription toBeRemoved = null;
for (Subscription sub : clientSubscriptions) {
if (sub.getTopicFilter().equals(topic)) {
toBeRemoved = sub;
break;
}
}
if (toBeRemoved != null) {
clientSubscriptions.remove(toBeRemoved);
}
}
@Override
public void addNewSubscription(Subscription newSubscription) {
final String clientID = newSubscription.getClientId();
if (!m_persistentSubscriptions.containsKey(clientID)) {
m_persistentSubscriptions.put(clientID, new HashSet<Subscription>());
}
Set<Subscription> subs = m_persistentSubscriptions.get(clientID);
if (!subs.contains(newSubscription)) {
subs.add(newSubscription);
m_persistentSubscriptions.put(clientID, subs);
}
}
@Override
public void wipeSubscriptions(String clientID) {
m_persistentSubscriptions.remove(clientID);
}
@Override
public void updateSubscriptions(String clientID, Set<Subscription> subscriptions) {
m_persistentSubscriptions.put(clientID, subscriptions);
}
@Override
public boolean contains(String clientID) {
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>();
for (Map.Entry<String, Set<Subscription>> entry : m_persistentSubscriptions.entrySet()) {
allSubscriptions.addAll(entry.getValue());
}
return allSubscriptions;
}
@Override
public void close() {
@@ -249,4 +179,44 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore {
public PublishEvent retrieveQoS2Message(String publishKey) {
return m_qos2Store.get(publishKey);
}
@Override
public void addNewSubscription(Subscription newSubscription) {
m_sessionsStore.addNewSubscription(newSubscription);
}
@Override
public void removeSubscription(String topic, String clientID) {
m_sessionsStore.removeSubscription(topic, clientID);
}
@Override
public void wipeSubscriptions(String sessionID) {
m_sessionsStore.wipeSubscriptions(sessionID);
}
@Override
public void updateSubscriptions(String clientID, Set<Subscription> subscriptions) {
m_sessionsStore.updateSubscriptions(clientID, subscriptions);
}
@Override
public List<Subscription> listAllSubscriptions() {
return m_sessionsStore.listAllSubscriptions();
}
@Override
public boolean contains(String clientID) {
return m_sessionsStore.contains(clientID);
}
@Override
public ClientSession createNewSession(String clientID) {
return m_sessionsStore.createNewSession(clientID);
}
@Override
public ClientSession sessionForClient(String clientID) {
return m_sessionsStore.sessionForClient(clientID);
}
}
broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java
+13 -6
View File
@@ -75,6 +75,7 @@ public class ProtocolProcessorTest {
//sleep to let the messaging batch processor to process the initEvent
Thread.sleep(300);
MemoryStorageService memStorage = new MemoryStorageService();
memStorage.initStore();
m_storageService = memStorage;
m_sessionStore = memStorage;
//m_storageService.initStore();
@@ -84,7 +85,7 @@ public class ProtocolProcessorTest {
m_mockAuthenticator = new MockAuthenticator(users);
subscriptions = new SubscriptionsStore();
subscriptions.init(new MemoryStorageService());
subscriptions.init(memStorage);
m_processor = new ProtocolProcessor();
m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, true,
new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
@@ -109,7 +110,9 @@ public class ProtocolProcessorTest {
};
//simulate a connect that register a clientID to an IoSession
subs.init(new MemoryStorageService());
MemoryStorageService storageService = new MemoryStorageService();
storageService.initStore();
subs.init(storageService);
m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
ConnectMessage connectMessage = new ConnectMessage();
connectMessage.setProtocolVersion((byte) 3);
@@ -153,7 +156,9 @@ public class ProtocolProcessorTest {
};
//simulate a connect that register a clientID to an IoSession
subs.init(new MemoryStorageService());
MemoryStorageService storageService = new MemoryStorageService();
storageService.initStore();
subs.init(storageService);
m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
MockReceiverChannel firstReceiverSession = new MockReceiverChannel();
@@ -306,8 +311,10 @@ public class ProtocolProcessorTest {
}
}
};
subs.init(new MemoryStorageService());
MemoryStorageService storageService = new MemoryStorageService();
storageService.initStore();
subs.init(storageService);
//simulate a connect that register a clientID to an IoSession
m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
ConnectMessage connectMessage = new ConnectMessage();
@@ -457,7 +464,7 @@ public class ProtocolProcessorTest {
Subscription subQos1 = new Subscription("Sub A", "a/b", QOSType.LEAST_ONE, false);
Subscription subQos2 = new Subscription("Sub B", "a/+", QOSType.EXACTLY_ONCE, false);
SubscriptionsStore subscriptions = new SubscriptionsStore();
subscriptions.init(new MemorySessionStore());
subscriptions.init(new MemorySessionStore(memoryMessageStore));
subscriptions.add(subQos1);
subscriptions.add(subQos2);
broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java
+2 -1
View File
@@ -59,6 +59,7 @@ public class ProtocolProcessor_CONNECT_Test {
//sleep to let the messaging batch processor to process the initEvent
Thread.sleep(300);
MemoryStorageService memStorage = new MemoryStorageService();
memStorage.initStore();
m_storageService = memStorage;
m_sessionStore = memStorage;
//m_storageService.initStore();
@@ -68,7 +69,7 @@ public class ProtocolProcessor_CONNECT_Test {
m_mockAuthenticator = new MockAuthenticator(users);
subscriptions = new SubscriptionsStore();
subscriptions.init(new MemoryStorageService());
subscriptions.init(memStorage);
m_processor = new ProtocolProcessor();
m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, true,
new PermitAllAuthorizator(), ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR);
broker/src/test/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java
+3 -1
View File
@@ -41,7 +41,9 @@ public class SubscriptionsStoreTest {
@Before
public void setUp() throws IOException {
store = new SubscriptionsStore();
store.init(new MemoryStorageService());
MemoryStorageService storageService = new MemoryStorageService();
storageService.initStore();
store.init(storageService);
}
@Test