diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java index 9bb69773..7503c933 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java @@ -2,6 +2,16 @@ package org.dna.mqtt.moquette.messaging.spi.impl; import com.lmax.disruptor.BatchEventProcessor; import com.lmax.disruptor.EventHandler; +import com.lmax.disruptor.RingBuffer; +import com.lmax.disruptor.SequenceBarrier; +import java.io.Serializable; +import java.text.ParseException; +import java.util.Collection; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import org.apache.mina.core.session.IdleStatus; import org.apache.mina.core.session.IoSession; import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition; @@ -21,17 +31,6 @@ import org.dna.mqtt.moquette.server.IAuthenticator; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.Serializable; -import java.text.ParseException; -import java.util.Collection; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import com.lmax.disruptor.RingBuffer; -import com.lmax.disruptor.SequenceBarrier; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; - /** * * @author andrea @@ -254,7 +253,7 @@ public class SimpleMessaging implements IMessaging, EventHandler/*, // } // } - private void processConnect(ConnectEvent evt) { + protected void processConnect(ConnectEvent evt) { ConnectMessage msg = evt.getMessage(); IoSession session = evt.getSession(); diff --git a/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java index 924427c7..2c940bdb 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java @@ -2,16 +2,29 @@ package org.dna.mqtt.moquette.messaging.spi.impl; import java.util.concurrent.BlockingQueue; +import org.apache.mina.core.filterchain.IoFilterAdapter; +import org.apache.mina.core.session.DummySession; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.core.write.WriteRequest; +import org.dna.mqtt.moquette.messaging.spi.IMessaging; import org.dna.mqtt.moquette.messaging.spi.INotifier; -import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent; -import org.dna.mqtt.moquette.messaging.spi.impl.events.NotifyEvent; -import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; -import org.dna.mqtt.moquette.messaging.spi.impl.events.SubscribeEvent; +import org.dna.mqtt.moquette.messaging.spi.impl.events.*; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; import static org.junit.Assert.*; + +import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; +import org.dna.mqtt.moquette.proto.messages.ConnectMessage; +import org.dna.mqtt.moquette.proto.messages.SubscribeMessage; +import org.dna.mqtt.moquette.server.ConnectionDescriptor; +import org.dna.mqtt.moquette.server.Constants; +import org.dna.mqtt.moquette.server.MQTTHandler; +import org.junit.After; import org.junit.Before; import org.junit.Test; import org.mockito.ArgumentCaptor; + +import static org.junit.Assert.assertEquals; import static org.mockito.Mockito.*; /** @@ -24,9 +37,42 @@ public class SimpleMessagingTest { final static String FAKE_TOPIC = "/news"; SimpleMessaging messaging; + byte m_returnCode; + IoSession m_session; +// MQTTHandler m_handler; + ConnectMessage connMsg; + + AbstractMessage m_receivedMessage; + @Before public void setUp() { messaging = new SimpleMessaging(); + connMsg = new ConnectMessage(); + connMsg.setProcotolVersion((byte) 0x03); + + m_session = new DummySession(); + + m_session.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() { + + @Override + public void filterWrite(NextFilter nextFilter, IoSession session, + WriteRequest writeRequest) throws Exception { + try { + m_receivedMessage = (AbstractMessage) writeRequest.getMessage(); + if (m_receivedMessage instanceof ConnAckMessage) { + ConnAckMessage buf = (ConnAckMessage) m_receivedMessage; + m_returnCode = buf.getReturnCode(); + } + } catch (Exception ex) { + throw new AssertionError("Wrong return code"); + } + } + }); + } + + @After + public void tearDown() { + m_receivedMessage = null; } @Test @@ -55,18 +101,20 @@ public class SimpleMessagingTest { @Test public void testPublish() throws InterruptedException { + //simulate a connect that register a clientID to an IoSession + ConnectionDescriptor connDescr = new ConnectionDescriptor(FAKE_CLIENT_ID, m_session, true); + messaging.m_clientIDs.put(FAKE_CLIENT_ID, connDescr); + SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false)); messaging.processSubscribe(evt); - INotifier notifier = mock(INotifier.class); - messaging.setNotifier(notifier); //Exercise PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, QOSType.MOST_ONE, "Hello".getBytes(), false, "FakeCLI", null); messaging.processPublish(pubEvt); //Verify - ArgumentCaptor argument = ArgumentCaptor.forClass(NotifyEvent.class); - verify(notifier).notify(argument.capture()); + assertNotNull(m_receivedMessage); + //TODO check received message attributes } @Test @@ -94,4 +142,74 @@ public class SimpleMessagingTest { assertTrue(messaging.matchTopics("/finance/stock/ibm", "/finance/+/ibm")); assertFalse(messaging.matchTopics("/finance/stock", "+")); } + + @Test + public void testHandleConnect_BadProtocol() { + connMsg.setProcotolVersion((byte) 0x02); + + //Exercise + ConnectEvent pubEvt = new ConnectEvent(m_session, connMsg); + messaging.processConnect(pubEvt); + + //Verify + assertEquals(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION, m_returnCode); + } + + @Test + public void testConnect_badClientID() { + connMsg.setClientID("extremely_long_clientID_greater_than_23"); + + //Exercise + ConnectEvent pubEvt = new ConnectEvent(m_session, connMsg); + messaging.processConnect(pubEvt); + + //Verify + assertEquals(ConnAckMessage.IDENTIFIER_REJECTED, m_returnCode); + } + + @Test + public void testWill() { + connMsg.setClientID("123"); + connMsg.setWillFlag(true); + connMsg.setWillTopic("topic"); + connMsg.setWillMessage("Topic message"); + //IMessaging mockedMessaging = mock(IMessaging.class); + + //Exercise + //m_handler.setMessaging(mockedMessaging); + ConnectEvent pubEvt = new ConnectEvent(m_session, connMsg); + messaging.processConnect(pubEvt); + + //Verify + assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_returnCode); + //TODO verify the call + /*verify(mockedMessaging).publish(eq("topic"), eq("Topic message".getBytes()), + any(AbstractMessage.QOSType.class), anyBoolean(), eq("123"), any(IoSession.class));*/ + } + + + @Test + public void testHandleSubscribe() { + String topicName = "/news"; + SubscribeMessage subscribeMsg = new SubscribeMessage(); + subscribeMsg.setMessageID(0x555); + subscribeMsg.addSubscription(new SubscribeMessage.Couple( + (byte)AbstractMessage.QOSType.EXACTLY_ONCE.ordinal(), topicName)); + //IMessaging mockedMessaging = mock(IMessaging.class); + + m_session.setAttribute(Constants.ATTR_CLIENTID, "fakeID"); + m_session.setAttribute(Constants.CLEAN_SESSION, false); + Subscription subscription = new Subscription("fakeID", "/news", QOSType.EXACTLY_ONCE, true); + + //Exercise + //m_handler.setMessaging(mockedMessaging); + SubscribeEvent pubEvt = new SubscribeEvent(subscription); + messaging.processSubscribe(pubEvt); + + //Verify + /*ArgumentCaptor argument = ArgumentCaptor.forClass(AbstractMessage.QOSType.class); + verify(mockedMessaging).subscribe(eq("fakeID"), eq(topicName), argument.capture(), eq(false)); + assertEquals(AbstractMessage.QOSType.EXACTLY_ONCE, argument.getValue());*/ + //TODO verify the message ACK and corresponding QoS!! + } } diff --git a/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java b/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java deleted file mode 100644 index dd227bf3..00000000 --- a/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java +++ /dev/null @@ -1,120 +0,0 @@ -package org.dna.mqtt.moquette.server; - -import org.mockito.ArgumentCaptor; -import org.dna.mqtt.moquette.proto.messages.AbstractMessage; -import org.apache.mina.core.write.WriteRequest; -import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; -import org.apache.mina.core.filterchain.IoFilterAdapter; -import org.apache.mina.core.session.DummySession; -import org.apache.mina.core.session.IoSession; -import org.dna.mqtt.moquette.messaging.spi.IMessaging; -import org.dna.mqtt.moquette.proto.messages.ConnectMessage; -import org.dna.mqtt.moquette.proto.messages.SubscribeMessage; -import org.junit.Before; -import org.junit.Test; -import static org.junit.Assert.*; -import static org.mockito.Mockito.*; - -/** - * - * @author andrea - */ -public class MQTTHandlerTest { - - byte m_returnCode; - IoSession m_session; - MQTTHandler m_handler; - ConnectMessage connMsg; - - @Before - public void setUp() { - m_handler = new MQTTHandler(); - connMsg = new ConnectMessage(); - connMsg.setProcotolVersion((byte) 0x03); - - m_session = new DummySession(); - m_session.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() { - - @Override - public void filterWrite(NextFilter nextFilter, IoSession session, - WriteRequest writeRequest) throws Exception { - try { - ConnAckMessage buf = (ConnAckMessage) writeRequest.getMessage(); - m_returnCode = buf.getReturnCode(); - } catch (Exception ex) { - throw new AssertionError("Wrong return code"); - } - } - }); - } - - @Test - public void testHandleConnect_BadProtocol() { - connMsg.setProcotolVersion((byte) 0x02); - - //Exercise - m_handler.handleConnect(m_session, connMsg); - - //Verify - assertEquals(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION, m_returnCode); - } - - - @Test - public void testConnect_badClientID() { - connMsg.setClientID("extremely_long_clientID_greater_than_23"); - - //Exercise - m_handler.handleConnect(m_session, connMsg); - - //Verify - assertEquals(ConnAckMessage.IDENTIFIER_REJECTED, m_returnCode); - } - - @Test - public void testWill() { - connMsg.setClientID("123"); - connMsg.setWillFlag(true); - connMsg.setWillTopic("topic"); - connMsg.setWillMessage("Topic message"); - IMessaging mockedMessaging = mock(IMessaging.class); - - //Exercise - m_handler.setMessaging(mockedMessaging); - m_handler.handleConnect(m_session, connMsg); - - //Verify - assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_returnCode); - verify(mockedMessaging).publish(eq("topic"), eq("Topic message".getBytes()), - any(AbstractMessage.QOSType.class), anyBoolean(), eq("123"), any(IoSession.class)); - } - - - @Test - public void testHandleSubscribe() { - String topicName = "/news"; - SubscribeMessage subscribeMsg = new SubscribeMessage(); - subscribeMsg.setMessageID(0x555); - subscribeMsg.addSubscription(new SubscribeMessage.Couple( - (byte)AbstractMessage.QOSType.EXACTLY_ONCE.ordinal(), topicName)); - IMessaging mockedMessaging = mock(IMessaging.class); - - m_session.setAttribute(Constants.ATTR_CLIENTID, "fakeID"); - m_session.setAttribute(Constants.CLEAN_SESSION, false); - - //Exercise - m_handler.setMessaging(mockedMessaging); - m_handler.handleSubscribe(m_session, subscribeMsg); - - //Verify - ArgumentCaptor argument = ArgumentCaptor.forClass(AbstractMessage.QOSType.class); - verify(mockedMessaging).subscribe(eq("fakeID"), eq(topicName), argument.capture(), eq(false)); - assertEquals(AbstractMessage.QOSType.EXACTLY_ONCE, argument.getValue()); - //TODO verify the message ACK and corresponding QoS!! - } - -// @Test -// public void testHandleSubscribeMultipleTopics() { -// fail("TOBE implemented"); -// } -}