From 4c1f36cf675b3a49913058b8a431d5bfad70280e Mon Sep 17 00:00:00 2001 From: andsel Date: Sun, 17 Jan 2016 13:00:36 +0100 Subject: [PATCH] MOved the method nextClientID from IMessagesStore to the ISessionsStore because it's session related --- .../java/io/moquette/spi/ClientSession.java | 2 +- .../java/io/moquette/spi/IMessagesStore.java | 5 --- .../java/io/moquette/spi/ISessionsStore.java | 5 +++ .../moquette/spi/impl/ProtocolProcessor.java | 4 +- .../spi/persistence/MapDBMessagesStore.java | 40 +++++++++---------- .../spi/persistence/MapDBSessionsStore.java | 21 ++++++++++ .../spi/impl/MemoryMessagesStore.java | 20 ---------- .../persistence/MapDBPersistentStoreTest.java | 10 ++--- .../spi/persistence/MemorySessionStore.java | 19 +++++++++ 9 files changed, 73 insertions(+), 53 deletions(-) diff --git a/broker/src/main/java/io/moquette/spi/ClientSession.java b/broker/src/main/java/io/moquette/spi/ClientSession.java index 9ebd6f75..f76d04f3 100644 --- a/broker/src/main/java/io/moquette/spi/ClientSession.java +++ b/broker/src/main/java/io/moquette/spi/ClientSession.java @@ -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) { diff --git a/broker/src/main/java/io/moquette/spi/IMessagesStore.java b/broker/src/main/java/io/moquette/spi/IMessagesStore.java index 8b2d1188..34ce5ecd 100644 --- a/broker/src/main/java/io/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/io/moquette/spi/IMessagesStore.java @@ -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); } diff --git a/broker/src/main/java/io/moquette/spi/ISessionsStore.java b/broker/src/main/java/io/moquette/spi/ISessionsStore.java index 99af0908..ed894266 100644 --- a/broker/src/main/java/io/moquette/spi/ISessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/ISessionsStore.java @@ -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. * */ diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index 70f946c6..e2b9500c 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -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); } diff --git a/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java b/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java index 8d4d0886..8e744dcc 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java +++ b/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java @@ -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 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 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) { diff --git a/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java b/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java index 1d019446..50848f0c 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java @@ -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 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 m = this.m_inflightStore.get(clientID); diff --git a/broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java b/broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java index e60277d7..5c4eed04 100644 --- a/broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java +++ b/broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java @@ -34,7 +34,6 @@ public class MemoryMessagesStore implements IMessagesStore { private Map m_retainedStore = new HashMap<>(); private Map m_persistentMessageStore = new HashMap<>(); private Map> m_messageToGuids; - private Map> m_inflightIDs = new HashMap<>(); MemoryMessagesStore(Map> 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 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); diff --git a/broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java b/broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java index ea0cab6a..58aed9c5 100644 --- a/broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java +++ b/broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java @@ -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); } diff --git a/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java b/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java index 0d21f1a6..731cbb4b 100644 --- a/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java +++ b/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java @@ -39,6 +39,7 @@ public class MemorySessionStore implements ISessionsStore { //maps clientID->[MessageId -> guid] private Map> m_inflightStore = new HashMap<>(); +// private Map> m_inflightIDs = new HashMap<>(); //maps clientID->[guid*] private Map> 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 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 guids = Utils.defaultGet(m_enqueuedStore, clientID, new HashSet());