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

Moved the clientIDs map (ConnectionDescriptors cache) from SimpleMessaging to ProtocolProcessor

Browse Source
This commit is contained in:
selva.andre
2013-09-17 11:08:05 +02:00
parent 3d3432762e
commit 8f10f3ed15
3 changed files with 22 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java
+5 -4
View File
@@ -1,6 +1,7 @@
package org.dna.mqtt.moquette.messaging.spi.impl;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition;
@@ -39,7 +40,7 @@ class ProtocolProcessor {
private static final Logger LOG = LoggerFactory.getLogger(ProtocolProcessor.class);
private Map<String, ConnectionDescriptor> m_clientIDs;
private Map<String, ConnectionDescriptor> m_clientIDs = new HashMap<String, ConnectionDescriptor>();
private SubscriptionsStore subscriptions;
private IStorageService m_storageService;
private IAuthenticator m_authenticator;
@@ -48,9 +49,9 @@ class ProtocolProcessor {
}
void init(Map<String, ConnectionDescriptor> clientIDs, SubscriptionsStore subscriptions,
void init(/*Map<String, ConnectionDescriptor> clientIDs, */SubscriptionsStore subscriptions,
IStorageService storageService) {
m_clientIDs = clientIDs;
//m_clientIDs = clientIDs;
this.subscriptions = subscriptions;
m_storageService = storageService;
}
@@ -244,7 +245,7 @@ class ProtocolProcessor {
}
LOG.debug("clientIDs are " + m_clientIDs);
if (m_clientIDs.get(clientId) == null) {
throw new RuntimeException(String.format("Can't find a ConnectionDEwcriptor for client %s in cache %s", clientId, m_clientIDs));
throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client %s in cache %s", clientId, m_clientIDs));
}
LOG.debug("Session for clientId " + clientId + " is " + m_clientIDs.get(clientId).getSession());
m_clientIDs.get(clientId).getSession().write(pubMessage);
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+1 -3
View File
@@ -39,8 +39,6 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
private IStorageService m_storageService;
Map<String, ConnectionDescriptor> m_clientIDs = new HashMap<String, ConnectionDescriptor>();
private ExecutorService m_executor;
BatchEventProcessor<ValueEvent> m_eventProcessor;
@@ -166,7 +164,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
m_storageService.initStore();
subscriptions.init(m_storageService);
m_processor.init(m_clientIDs, subscriptions, m_storageService);
m_processor.init(subscriptions, m_storageService);
}
broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java
+16 -9
View File
@@ -84,7 +84,7 @@ public class ProtocolProcessorTest {
subscriptions = new SubscriptionsStore();
subscriptions.init(m_storageService);
m_processor = new ProtocolProcessor();
m_processor.init(new HashMap<String, ConnectionDescriptor>(), subscriptions, m_storageService);
m_processor.init(subscriptions, m_storageService);
}
@Test
@@ -143,13 +143,18 @@ public class ProtocolProcessorTest {
throw new IllegalArgumentException("Expected " + FAKE_TOPIC + " buf found " + topic);
}
}
};
//simulate a connect that register a clientID to an IoSession
Map<String, ConnectionDescriptor> connectionsMap = new HashMap<String, ConnectionDescriptor>();
ConnectionDescriptor connDescr = new ConnectionDescriptor(FAKE_CLIENT_ID, m_session, subscription.isCleanSession());
connectionsMap.put(FAKE_CLIENT_ID, connDescr);
m_processor.init(connectionsMap, subs, m_storageService);
subs.init(m_storageService);
m_processor.init(subs, m_storageService);
ConnectMessage connectMessage = new ConnectMessage();
connectMessage.setProcotolVersion((byte)3);
connectMessage.setClientID(FAKE_CLIENT_ID);
connectMessage.setCleanSession(subscription.isCleanSession());
m_processor.processConnect(m_session, connectMessage);
//Exercise
@@ -236,10 +241,12 @@ public class ProtocolProcessorTest {
subs.init(new MemoryStorageService());
//simulate a connect that register a clientID to an IoSession
Map<String, ConnectionDescriptor> connectionsMap = new HashMap<String, ConnectionDescriptor>();
ConnectionDescriptor connDescr = new ConnectionDescriptor(FAKE_CLIENT_ID, m_session, subscription.isCleanSession());
connectionsMap.put(FAKE_CLIENT_ID, connDescr);
m_processor.init(connectionsMap, subs, m_storageService);
m_processor.init(subs, m_storageService);
ConnectMessage connectMessage = new ConnectMessage();
connectMessage.setClientID(FAKE_CLIENT_ID);
connectMessage.setProcotolVersion((byte)3);
connectMessage.setCleanSession(subscription.isCleanSession());
m_processor.processConnect(m_session, connectMessage);
PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, "Hello".getBytes(), true, "FakeCLI", null);
m_processor.processPublish(pubEvt);