diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java index dd4053a2..ecfa8d7f 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java @@ -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 m_clientIDs; + private Map m_clientIDs = new HashMap(); private SubscriptionsStore subscriptions; private IStorageService m_storageService; private IAuthenticator m_authenticator; @@ -48,9 +49,9 @@ class ProtocolProcessor { } - void init(Map clientIDs, SubscriptionsStore subscriptions, + void init(/*Map 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); diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java index 7d42c95c..bcedb5ea 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java @@ -39,8 +39,6 @@ public class SimpleMessaging implements IMessaging, EventHandler { private IStorageService m_storageService; - Map m_clientIDs = new HashMap(); - private ExecutorService m_executor; BatchEventProcessor m_eventProcessor; @@ -166,7 +164,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { m_storageService.initStore(); subscriptions.init(m_storageService); - m_processor.init(m_clientIDs, subscriptions, m_storageService); + m_processor.init(subscriptions, m_storageService); } diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java index 9021579a..5eff2944 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java @@ -84,7 +84,7 @@ public class ProtocolProcessorTest { subscriptions = new SubscriptionsStore(); subscriptions.init(m_storageService); m_processor = new ProtocolProcessor(); - m_processor.init(new HashMap(), 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 connectionsMap = new HashMap(); - 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 connectionsMap = new HashMap(); - 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);