Minor changes, refactored ConnectionDescriptor to be an immutable object

This commit is contained in:
andsel
2015-11-10 22:24:35 +01:00
parent 608a72d3ac
commit ce1618cb52
7 changed files with 20 additions and 32 deletions
@@ -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 + '}';
}
}
@@ -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
@@ -77,7 +77,7 @@ public interface IMessagesStore {
*/
List<PublishEvent> listMessagesInSession(String clientID);
void removeMessageInSession(String clientID, Integer packetID);
void removeMessage(String clientID, Integer packetID);
void dropMessagesInSession(String clientID);
@@ -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) {
@@ -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 + '\'' +
@@ -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<StoredPublishEvent> events = m_persistentMessageStore.get(clientID);
if (events == null) {
return;