diff --git a/broker/src/main/java/org/eclipse/moquette/server/ConnectionDescriptor.java b/broker/src/main/java/org/eclipse/moquette/server/ConnectionDescriptor.java index 9cdacb0f..56a6a7e7 100644 --- a/broker/src/main/java/org/eclipse/moquette/server/ConnectionDescriptor.java +++ b/broker/src/main/java/org/eclipse/moquette/server/ConnectionDescriptor.java @@ -16,38 +16,26 @@ package org.eclipse.moquette.server; /** - * Maintains the information of single connection, like ClientID, IoSession, - * and other connection related flags. + * Value object to maintain the information of single connection, like ClientID, IoSession, + * and other clean session fla. * * * @author andrea */ public class ConnectionDescriptor { - private String m_clientID; - private ServerChannel m_session; - private boolean m_cleanSession; + public final String clientID; + public final ServerChannel session; + public final boolean cleanSession; public ConnectionDescriptor(String clientID, ServerChannel session, boolean cleanSession) { - this.m_clientID = clientID; - this.m_session = session; - this.m_cleanSession = cleanSession; - } - - public boolean isCleanSession() { - return m_cleanSession; - } - - public String getClientID() { - return m_clientID; - } - - public ServerChannel getSession() { - return m_session; + this.clientID = clientID; + this.session = session; + this.cleanSession = cleanSession; } @Override public String toString() { - return "ConnectionDescriptor{" + "m_clientID=" + m_clientID + ", m_cleanSession=" + m_cleanSession + '}'; + return "ConnectionDescriptor{" + "clientID=" + clientID + ", cleanSession=" + cleanSession + '}'; } } diff --git a/broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java b/broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java index 5c8f277a..508a929e 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java @@ -67,7 +67,7 @@ public class ClientSession { * Remove a message previously stored for delivery. * */ public void removeDelivered(int messageID) { - messagesStore.removeMessageInSession(clientID, messageID); + messagesStore.removeMessage(clientID, messageID); } @Override 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 c3a89687..abf0a561 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java @@ -77,7 +77,7 @@ public interface IMessagesStore { */ List listMessagesInSession(String clientID); - void removeMessageInSession(String clientID, Integer packetID); + void removeMessage(String clientID, Integer packetID); void dropMessagesInSession(String clientID); 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 19f9dc2c..96e2a009 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 @@ -176,7 +176,7 @@ public class ProtocolProcessor { if (m_clientIDs.containsKey(msg.getClientID())) { LOG.info("Found an existing connection with same client ID <{}>, forcing to close", msg.getClientID()); //clean the subscriptions if the old used a cleanSession = true - ServerChannel oldSession = m_clientIDs.get(msg.getClientID()).getSession(); + ServerChannel oldSession = m_clientIDs.get(msg.getClientID()).session; ClientSession oldClientSession = m_sessionsStore.sessionForClient(msg.getClientID()); oldClientSession.disconnect(); oldSession.setAttribute(NettyChannel.ATTR_KEY_SESSION_STOLEN, true); @@ -263,7 +263,7 @@ public class ProtocolProcessor { String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID); int messageID = msg.getMessageID(); //Remove the message from message store - m_messagesStore.removeMessageInSession(clientID, messageID); + m_messagesStore.removeMessage(clientID, messageID); } public void processPublish(ServerChannel session, PublishMessage msg) { @@ -416,7 +416,7 @@ public class ProtocolProcessor { // could happen is not an error HANDLE IT throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client <%s> in cache <%s>", clientId, m_clientIDs)); } - ServerChannel session = m_clientIDs.get(clientId).getSession(); + ServerChannel session = m_clientIDs.get(clientId).session; LOG.debug("Session for clientId {} is {}", clientId, session); String user = (String) session.getAttribute(NettyChannel.ATTR_KEY_USERNAME); @@ -431,7 +431,7 @@ public class ProtocolProcessor { LOG.trace("PUB <--PUBREC-- SRV sendPubRec invoked for clientID {} with messageID {}", clientID, messageID); PubRecMessage pubRecMessage = new PubRecMessage(); pubRecMessage.setMessageID(messageID); - m_clientIDs.get(clientID).getSession().write(pubRecMessage); + m_clientIDs.get(clientID).session.write(pubRecMessage); } private void sendPubAck(PubAckEvent evt) { @@ -448,7 +448,7 @@ public class ProtocolProcessor { if (m_clientIDs.get(clientId) == null) { throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client %s in cache %s", clientId, m_clientIDs)); } - m_clientIDs.get(clientId).getSession().write(pubAckMessage); + m_clientIDs.get(clientId).session.write(pubAckMessage); } catch(Throwable t) { LOG.error(null, t); } @@ -481,7 +481,7 @@ public class ProtocolProcessor { PubCompMessage pubCompMessage = new PubCompMessage(); pubCompMessage.setMessageID(messageID); - m_clientIDs.get(clientID).getSession().write(pubCompMessage); + m_clientIDs.get(clientID).session.write(pubCompMessage); } public void processPubRec(ServerChannel session, PubRecMessage msg) { diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/events/PublishEvent.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/events/PublishEvent.java index 60e8be09..48007897 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/events/PublishEvent.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/events/PublishEvent.java @@ -85,7 +85,7 @@ public class PublishEvent extends MessagingEvent { public String toString() { return "PublishEvent{" + "m_msgID=" + m_msgID + - ", m_clientID='" + m_clientID + '\'' + + ", clientID='" + m_clientID + '\'' + ", m_retain=" + m_retain + ", m_qos=" + m_qos + ", m_topic='" + m_topic + '\'' + 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 beac521e..f5ff72ef 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 @@ -181,7 +181,7 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { } @Override - public void removeMessageInSession(String clientID, Integer messageID) { + public void removeMessage(String clientID, Integer messageID) { List events = m_persistentMessageStore.get(clientID); if (events == null) { return; 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 bca02ecb..28711f67 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 @@ -99,7 +99,7 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { } @Override - public void removeMessageInSession(String clientID, Integer messageID) { + public void removeMessage(String clientID, Integer messageID) { List events = m_persistentMessageStore.get(clientID); if (events == null) { return;