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

MOved the method nextClientID from IMessagesStore to the ISessionsStore because it's session related

Browse Source
This commit is contained in:
andsel
2016-01-17 13:00:36 +01:00
parent 5daf1362cc
commit 4c1f36cf67
9 changed files with 73 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/spi/ClientSession.java
+1 -1
View File
@@ -159,7 +159,7 @@ public class ClientSession {
}
public int nextPacketId() {
return this.messagesStore.nextPacketID(this.clientID);
return this.m_sessionsStore.nextPacketID(this.clientID);
}
public void inFlightAcknowledged(int messageID) {
broker/src/main/java/io/moquette/spi/IMessagesStore.java
-5
View File
@@ -136,10 +136,5 @@ public interface IMessagesStore {
StoredMessage getMessageByGuid(String guid);
/**
* Return the next valid packetIdentifier for the given client session.
* */
int nextPacketID(String clientID);
void cleanRetained(String topic);
}
broker/src/main/java/io/moquette/spi/ISessionsStore.java
+5
View File
@@ -110,6 +110,11 @@ public interface ISessionsStore {
* */
void inFlight(String clientID, int messageID, String guid);
/**
* Return the next valid packetIdentifier for the given client session.
* */
int nextPacketID(String clientID);
/**
* Store the guid to be later published.
* */
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+2 -2
View File
@@ -427,7 +427,7 @@ public class ProtocolProcessor {
//NB it's a will publish, it needs a PacketIdentifier for this conn, default to 1
Integer messageId = null;
if (will.getQos() != AbstractMessage.QOSType.MOST_ONE) {
messageId = m_messagesStore.nextPacketID(clientID);
messageId = m_sessionsStore.nextPacketID(clientID);
}
IMessagesStore.StoredMessage tobeStored = asStoredMessage(will);
@@ -737,7 +737,7 @@ public class ProtocolProcessor {
LOG.debug("send publish message for topic {}", newSubscription.getTopicFilter());
//forwardPublishQoS0(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true);
Integer packetID = storedMsg.getQos() == QOSType.MOST_ONE ? null :
m_messagesStore.nextPacketID(newSubscription.getClientId());
targetSession.nextPacketId();
directSend(targetSession, storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true, packetID);
}
broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java
+20 -20
View File
@@ -110,26 +110,26 @@ class MapDBMessagesStore implements IMessagesStore {
return m_persistentMessageStore.get(guid);
}
/**
* Return the next valid packetIdentifier for the given client session.
* */
@Override
public int nextPacketID(String clientID) {
Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID);
if (inFlightForClient == null) {
int nextPacketId = 1;
inFlightForClient = new HashSet<>();
inFlightForClient.add(nextPacketId);
this.m_inFlightIds.put(clientID, inFlightForClient);
return nextPacketId;
}
int maxId = inFlightForClient.isEmpty() ? 0 : Collections.max(inFlightForClient);
int nextPacketId = (maxId + 1) % 0xFFFF;
inFlightForClient.add(nextPacketId);
return nextPacketId;
}
// /**
// * Return the next valid packetIdentifier for the given client session.
// * */
// @Override
// public int nextPacketID(String clientID) {
// Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID);
// if (inFlightForClient == null) {
// int nextPacketId = 1;
// inFlightForClient = new HashSet<>();
// inFlightForClient.add(nextPacketId);
// this.m_inFlightIds.put(clientID, inFlightForClient);
// return nextPacketId;
//
// }
//
// int maxId = inFlightForClient.isEmpty() ? 0 : Collections.max(inFlightForClient);
// int nextPacketId = (maxId + 1) % 0xFFFF;
// inFlightForClient.add(nextPacketId);
// return nextPacketId;
// }
@Override
public void cleanRetained(String topic) {
broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java
+21
View File
@@ -143,6 +143,27 @@ class MapDBSessionsStore implements ISessionsStore {
return new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession);
}
/**
* Return the next valid packetIdentifier for the given client session.
* */
@Override
public int nextPacketID(String clientID) {
Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID);
if (inFlightForClient == null) {
int nextPacketId = 1;
inFlightForClient = new HashSet<>();
inFlightForClient.add(nextPacketId);
this.m_inFlightIds.put(clientID, inFlightForClient);
return nextPacketId;
}
int maxId = inFlightForClient.isEmpty() ? 0 : Collections.max(inFlightForClient);
int nextPacketId = (maxId + 1) % 0xFFFF;
inFlightForClient.add(nextPacketId);
return nextPacketId;
}
@Override
public void inFlightAck(String clientID, int messageID) {
Map<Integer, String> m = this.m_inflightStore.get(clientID);
broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java
-20
View File
@@ -34,7 +34,6 @@ public class MemoryMessagesStore implements IMessagesStore {
private Map<String, String> m_retainedStore = new HashMap<>();
private Map<String, StoredMessage> m_persistentMessageStore = new HashMap<>();
private Map<String, Map<Integer, String>> m_messageToGuids;
private Map<String, Set<Integer>> m_inflightIDs = new HashMap<>();
MemoryMessagesStore(Map<String, Map<Integer, String>> messageToGuids) {
m_messageToGuids = messageToGuids;
@@ -97,25 +96,6 @@ public class MemoryMessagesStore implements IMessagesStore {
return m_persistentMessageStore.get(guid);
}
/**
* Return the next valid packetIdentifier for the given client session.
* */
@Override
public int nextPacketID(String clientID) {
Set<Integer> inFlightForClient = m_inflightIDs.get(clientID);
if (inFlightForClient == null) {
int nextPacketId = 1;
inFlightForClient = new HashSet<>();
inFlightForClient.add(nextPacketId);
m_inflightIDs.put(clientID, inFlightForClient);
return nextPacketId;
}
int maxId = Collections.max(inFlightForClient);
int nextPacketId = (maxId + 1) % 0xFFFF;
inFlightForClient.add(nextPacketId);
return nextPacketId;
}
@Override
public void cleanRetained(String topic) {
m_retainedStore.remove(topic);
broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java
+5 -5
View File
@@ -83,18 +83,18 @@ public class MapDBPersistentStoreTest {
@Test
public void testNextPacketID_notExistingClientSession() {
int packetId = m_messagesStore.nextPacketID("NOT_EXISTING_CLI");
int packetId = m_sessionsStore.nextPacketID("NOT_EXISTING_CLI");
assertEquals(1, packetId);
}
@Test
public void testNextPacketID_existingClientSession() {
//Force creation of inflight map for the CLIENT session
int packetId = m_messagesStore.nextPacketID("CLIENT");
int packetId = m_sessionsStore.nextPacketID("CLIENT");
assertEquals(1, packetId);
//request a second packetID
packetId = m_messagesStore.nextPacketID("CLIENT");
packetId = m_sessionsStore.nextPacketID("CLIENT");
assertEquals(2, packetId);
}
@@ -102,7 +102,7 @@ public class MapDBPersistentStoreTest {
public void testNextPacketID() {
//request a first ID
int packetId = m_messagesStore.nextPacketID("CLIENT");
int packetId = m_sessionsStore.nextPacketID("CLIENT");
m_sessionsStore.inFlight("CLIENT", packetId, "ABCDE"); //simulate an inflight
assertEquals(1, packetId);
@@ -110,7 +110,7 @@ public class MapDBPersistentStoreTest {
m_sessionsStore.inFlightAck("CLIENT", packetId);
//request a second packetID, counter restarts from 0
packetId = m_messagesStore.nextPacketID("CLIENT");
packetId = m_sessionsStore.nextPacketID("CLIENT");
assertEquals(1, packetId);
}
broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java
+19
View File
@@ -39,6 +39,7 @@ public class MemorySessionStore implements ISessionsStore {
//maps clientID->[MessageId -> guid]
private Map<String, Map<Integer, String>> m_inflightStore = new HashMap<>();
// private Map<String, Set<Integer>> m_inflightIDs = new HashMap<>();
//maps clientID->[guid*]
private Map<String, Set<String>> m_enqueuedStore = new HashMap<>();
//maps clientID->[messageID*]
@@ -169,6 +170,24 @@ public class MemorySessionStore implements ISessionsStore {
this.m_inflightStore.put(clientID, m);
}
/**
* Return the next valid packetIdentifier for the given client session.
* */
@Override
public int nextPacketID(String clientID) {
Map<Integer, String> m = this.m_inflightStore.get(clientID);
if (m == null) {
m = new HashMap<>();
int nextPacketId = 1;
m.put(nextPacketId, null);
return nextPacketId;
}
int maxId = Collections.max(m.keySet());
int nextPacketId = (maxId + 1) % 0xFFFF;
m.put(nextPacketId, null);
return nextPacketId;
}
@Override
public void bindToDeliver(String guid, String clientID) {
Set<String> guids = Utils.defaultGet(m_enqueuedStore, clientID, new HashSet<String>());