From fdf5b039834311162b058a1778c7356518596771 Mon Sep 17 00:00:00 2001 From: andsel Date: Mon, 16 Nov 2015 23:53:10 +0100 Subject: [PATCH] Removed no more used methods --- .../eclipse/moquette/spi/IMessagesStore.java | 6 ------ .../moquette/spi/impl/ProtocolProcessor.java | 8 ------- .../spi/persistence/MapDBPersistentStore.java | 21 ------------------- .../ServerIntegrationWebSocketTest.java | 4 ++-- .../spi/impl/MemoryStorageService.java | 18 ---------------- 5 files changed, 2 insertions(+), 55 deletions(-) diff --git a/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java b/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java index 0c3e530b..2f32c433 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java @@ -92,11 +92,5 @@ public interface IMessagesStore { void close(); - void persistQoS2Message(String publishKey, PublishEvent evt); - - void removeQoS2Message(String publishKey); - - PublishEvent retrieveQoS2Message(String publishKey); - void cleanRetained(String topic); } diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java index d548c418..8b9c1c77 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java @@ -291,10 +291,6 @@ public class ProtocolProcessor { sendPubAck(new PubAckEvent(messageID, clientID)); LOG.debug("replying with PubAck to MSG ID {}", messageID); } else if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) { //QoS2 -// String publishKey = String.format("%s%d", clientID, messageID); - //store the message in temp store -//TODO use the global message store -// m_messagesStore.persistQoS2Message(publishKey, publishEvt); m_messagesStore.storePublishForFuture(publishEvt); sendPubRec(clientID, messageID); //Next the client will send us a pub rel @@ -456,13 +452,9 @@ public class ProtocolProcessor { String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID); int messageID = msg.getMessageID(); LOG.debug("PUB --PUBREL--> SRV processPubRel invoked for clientID {} ad messageID {}", clientID, messageID); -// String publishKey = String.format("%s%d", clientID, messageID); -// PublishEvent evt = m_messagesStore.retrieveQoS2Message(publishKey); - //get guid, get message ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); PublishEvent evt = targetSession.storedMessage(messageID); forward2Subscribers(evt); -// m_messagesStore.removeQoS2Message(publishKey); if (evt.isRetain()) { final String topic = evt.getTopic(); diff --git a/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java b/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java index ad02e916..b38c3d3e 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java @@ -70,8 +70,6 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { private ConcurrentMap> m_inflightStore; //map clientID <-> set of currently in flight packet identifiers Map> m_inFlightIds; - //bind clientID+MsgID -> evt message published - private ConcurrentMap m_qos2Store; //persistent Map of clientID, set of Subscriptions private ConcurrentMap> m_persistentSubscriptions; private ConcurrentMap m_persistentSessions; @@ -119,7 +117,6 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { m_inflightStore = m_db.getHashMap("inflight"); m_inFlightIds = m_db.getHashMap("inflightPacketIDs"); m_persistentSubscriptions = m_db.getHashMap("subscriptions"); - m_qos2Store = m_db.getHashMap("qos2Store"); m_persistentSessions = m_db.getHashMap("sessions"); m_enqueuedStore = m_db.getHashMap("sessionQueue"); m_secondPhaseStore = m_db.getHashMap("secondPhase"); @@ -417,24 +414,6 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { LOG.debug("Persistence commit scheduler is shutdown"); } - /*-------- QoS 2 storage management --------------*/ - @Override - public void persistQoS2Message(String publishKey, PublishEvent evt) { - LOG.debug("persistQoS2Message store pubKey: {}, evt: {}", publishKey, evt); - m_qos2Store.put(publishKey, convertToStored(evt)); - } - - @Override - public void removeQoS2Message(String publishKey) { - LOG.debug("Removing stored Q0S2 message <{}>", publishKey); - m_qos2Store.remove(publishKey); - } - - public PublishEvent retrieveQoS2Message(String publishKey) { - StoredPublishEvent storedEvt = m_qos2Store.get(publishKey); - return convertFromStored(storedEvt); - } - private StoredPublishEvent convertToStored(PublishEvent evt) { return new StoredPublishEvent(evt); } diff --git a/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationWebSocketTest.java b/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationWebSocketTest.java index c740d8db..5482ae76 100644 --- a/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationWebSocketTest.java +++ b/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationWebSocketTest.java @@ -42,7 +42,7 @@ import static org.junit.Assert.assertTrue; * @author andreaorg.eclipse.moquette.commons.Constants. */ public class ServerIntegrationWebSocketTest { - private static final Logger LOG = LoggerFactory.getLogger(ServerIntegrationPahoTest.class); + private static final Logger LOG = LoggerFactory.getLogger(ServerIntegrationWebSocketTest.class); Server m_server; WebSocketClient client; @@ -79,7 +79,7 @@ public class ServerIntegrationWebSocketTest { URI echoUri = new URI(destUri); ClientUpgradeRequest request = new ClientUpgradeRequest(); client.connect(socket, echoUri, request); - LOG.info("Connecting to : %s%n", echoUri); + LOG.info("Connecting to : %s", echoUri); boolean connected = socket.awaitConnected(4, TimeUnit.SECONDS); assertTrue(connected); diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java index 70e7c35a..63a552f2 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java @@ -46,7 +46,6 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { //maps clientID->[messageID*] private Map> m_secondPhaseStore = new HashMap<>(); private Map> m_inflightIDs = new HashMap<>(); - private Map m_qos2Store = new HashMap<>(); private Map> m_messageToGuids = new HashMap<>(); private MemorySessionStore m_sessionsStore; @@ -147,28 +146,11 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { return nextPacketId; } - @Override public void close() { //To change body of implemented methods use File | Settings | File Templates. } - @Override - public void persistQoS2Message(String publishKey, PublishEvent evt) { - LOG.debug("persistQoS2Message store pubKey {}, evt {}", publishKey, evt); - m_qos2Store.put(publishKey, evt); - } - - @Override - public void removeQoS2Message(String publishKey) { - m_qos2Store.remove(publishKey); - } - - @Override - public PublishEvent retrieveQoS2Message(String publishKey) { - return m_qos2Store.get(publishKey); - } - @Override public void addNewSubscription(Subscription newSubscription) { m_sessionsStore.addNewSubscription(newSubscription);