From fb7c137a11effc0e754afaecc2755b4d2b049a63 Mon Sep 17 00:00:00 2001 From: andrea Date: Fri, 30 Aug 2013 17:06:15 +0200 Subject: [PATCH] Moved processSubscribe into ProtocolProcessor --- .../messaging/spi/impl/ProtocolProcessor.java | 43 ++++++ .../messaging/spi/impl/SimpleMessaging.java | 66 +------- .../spi/impl/ProtocolProcessorTest.java | 143 ++++++++++++------ .../spi/impl/SimpleMessagingTest.java | 135 ----------------- 4 files changed, 142 insertions(+), 245 deletions(-) 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 4aef4ab2..8c82a789 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,9 +1,11 @@ package org.dna.mqtt.moquette.messaging.spi.impl; +import java.util.Collection; import java.util.List; import java.util.Map; import org.apache.mina.core.session.IdleStatus; import org.apache.mina.core.session.IoSession; +import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition; import org.dna.mqtt.moquette.messaging.spi.IMessaging; import org.dna.mqtt.moquette.messaging.spi.IStorageService; import org.dna.mqtt.moquette.messaging.spi.impl.events.NotifyEvent; @@ -19,6 +21,8 @@ import org.dna.mqtt.moquette.proto.messages.PubAckMessage; import org.dna.mqtt.moquette.proto.messages.PubRecMessage; import org.dna.mqtt.moquette.proto.messages.PubRelMessage; import org.dna.mqtt.moquette.proto.messages.PublishMessage; +import org.dna.mqtt.moquette.proto.messages.SubAckMessage; +import org.dna.mqtt.moquette.proto.messages.SubscribeMessage; import org.dna.mqtt.moquette.proto.messages.UnsubAckMessage; import org.dna.mqtt.moquette.server.ConnectionDescriptor; import org.dna.mqtt.moquette.server.Constants; @@ -339,5 +343,44 @@ class ProtocolProcessor { LOG.info("replying with UnsubAck to MSG ID {0}", messageID); session.write(ackMessage); } + + + void processSubscribe(IoSession session, SubscribeMessage msg, String clientID, boolean cleanSession) { + LOG.debug("processSubscribe invoked"); + + for (SubscribeMessage.Couple req : msg.subscriptions()) { + AbstractMessage.QOSType qos = AbstractMessage.QOSType.values()[req.getQos()]; + Subscription newSubscription = new Subscription(clientID, req.getTopic(), qos, cleanSession); + subscribeSingleTopic(newSubscription, req.getTopic()); + } + + //ack the client + SubAckMessage ackMessage = new SubAckMessage(); + ackMessage.setMessageID(msg.getMessageID()); + + //TODO by now it handles only QoS 0 messages + for (int i = 0; i < msg.subscriptions().size(); i++) { + ackMessage.addType(AbstractMessage.QOSType.MOST_ONE); + } + LOG.info("replying with SubAct to MSG ID " + msg.getMessageID()); + session.write(ackMessage); + } + + private void subscribeSingleTopic(Subscription newSubscription, final String topic) { + subscriptions.add(newSubscription); + + //scans retained messages to be published to the new subscription + Collection messages = m_storageService.searchMatching(new IMatchingCondition() { + public boolean match(String key) { + return SubscriptionsStore.matchTopics(key, topic); + } + }); + + for (HawtDBStorageService.StoredMessage storedMsg : messages) { + //fire the as retained the message + LOG.debug("Inserting NotifyEvent into outbound for topic " + topic); + notify(new NotifyEvent(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true)); + } + } } 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 4f57fe93..9dca827f 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 @@ -4,27 +4,21 @@ import com.lmax.disruptor.BatchEventProcessor; import com.lmax.disruptor.EventHandler; import com.lmax.disruptor.RingBuffer; import com.lmax.disruptor.SequenceBarrier; -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; import org.dna.mqtt.moquette.messaging.spi.IMessaging; import org.dna.mqtt.moquette.messaging.spi.IStorageService; -import org.dna.mqtt.moquette.messaging.spi.impl.HawtDBStorageService.StoredMessage; import org.dna.mqtt.moquette.messaging.spi.impl.events.*; -import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore; import org.dna.mqtt.moquette.proto.PubCompMessage; import org.dna.mqtt.moquette.proto.messages.*; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; import org.dna.mqtt.moquette.server.ConnectionDescriptor; import org.dna.mqtt.moquette.server.Constants; -import org.dna.mqtt.moquette.server.IAuthenticator; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -44,8 +38,6 @@ public class SimpleMessaging implements IMessaging, EventHandler { private RingBuffer m_ringBuffer; -// private INotifier m_notifier; - private IStorageService m_storageService; Map m_clientIDs = new HashMap(); @@ -172,7 +164,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { } else if (message instanceof SubscribeMessage) { String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION); - processSubscribe(session, (SubscribeMessage) message, clientID, cleanSession); + m_processor.processSubscribe(session, (SubscribeMessage) message, clientID, cleanSession); } else if (message instanceof PubRelMessage) { String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); int messageID = ((PubRelMessage) message).getMessageID(); @@ -203,62 +195,6 @@ public class SimpleMessaging implements IMessaging, EventHandler { } - private void subscribeSingleTopic(Subscription newSubscription, final String topic) { - subscriptions.add(newSubscription); - - //scans retained messages to be published to the new subscription - Collection messages = m_storageService.searchMatching(new IMatchingCondition() { - public boolean match(String key) { - return SubscriptionsStore.matchTopics(key, topic); - } - }); - - for (StoredMessage storedMsg : messages) { - //fire the as retained the message - LOG.debug("Inserting NotifyEvent into outbound for topic " + topic); - m_processor.notify(new NotifyEvent(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true)); - } - } - - protected void processSubscribe(IoSession session, SubscribeMessage msg, String clientID, boolean cleanSession) { - LOG.debug("processSubscribe invoked"); - - for (SubscribeMessage.Couple req : msg.subscriptions()) { - QOSType qos = AbstractMessage.QOSType.values()[req.getQos()]; - Subscription newSubscription = new Subscription(clientID, req.getTopic(), qos, cleanSession); - subscribeSingleTopic(newSubscription, req.getTopic()); - } - - //ack the client - SubAckMessage ackMessage = new SubAckMessage(); - ackMessage.setMessageID(msg.getMessageID()); - - //TODO by now it handles only QoS 0 messages - for (int i = 0; i < msg.subscriptions().size(); i++) { - ackMessage.addType(QOSType.MOST_ONE); - } - LOG.info("replying with SubAct to MSG ID " + msg.getMessageID()); - session.write(ackMessage); - } - -// /** -// * Remove the clientID from topic subscription, if not previously subscribed, -// * doesn't reply any error -// */ -// protected void processUnsubscribe(IoSession session, String clientID, List topics, int messageID) { -// LOG.debug("processSubscribe invoked"); -// -// for (String topic : topics) { -// subscriptions.removeSubscription(topic, clientID); -// } -// //ack the client -// UnsubAckMessage ackMessage = new UnsubAckMessage(); -// ackMessage.setMessageID(messageID); -// -// LOG.info("replying with UnsubAck to MSG ID {0}", messageID); -// session.write(ackMessage); -// } - private void processStop() { LOG.debug("processStop invoked"); m_storageService.close(); 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 1f5cb5eb..091fbd05 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 @@ -11,14 +11,15 @@ 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.IStorageService; -import static org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessagingTest.FAKE_CLIENT_ID; import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; import org.dna.mqtt.moquette.proto.messages.ConnectMessage; import org.dna.mqtt.moquette.proto.messages.PublishMessage; +import org.dna.mqtt.moquette.proto.messages.SubAckMessage; import org.dna.mqtt.moquette.proto.messages.SubscribeMessage; import org.dna.mqtt.moquette.server.ConnectionDescriptor; import static org.junit.Assert.assertEquals; @@ -33,7 +34,7 @@ import static org.mockito.Mockito.*; * @author andrea */ public class ProtocolProcessorTest { - + final static String FAKE_CLIENT_ID = "FAKE_123"; final static String FAKE_TOPIC = "/news"; IoSession m_session; @@ -42,9 +43,10 @@ public class ProtocolProcessorTest { ProtocolProcessor m_processor; IStorageService m_storageService; - + SubscriptionsStore subscriptions; AbstractMessage m_receivedMessage; + @Before public void setUp() throws InterruptedException { connMsg = new ConnectMessage(); @@ -75,7 +77,7 @@ public class ProtocolProcessorTest { m_storageService = new HawtDBStorageService(); m_storageService.initStore(); - SubscriptionsStore subscriptions = new SubscriptionsStore(); + subscriptions = new SubscriptionsStore(); subscriptions.init(m_storageService); m_processor = new ProtocolProcessor(); IMessaging mockedMessaging = mock(IMessaging.class); @@ -156,48 +158,99 @@ public class ProtocolProcessorTest { assertNotNull(m_receivedMessage); //TODO check received message attributes } + + @Test + public void testSubscribe() { + //Exercise + SubscribeMessage msg = new SubscribeMessage(); + msg.addSubscription(new SubscribeMessage.Couple((byte)AbstractMessage.QOSType.MOST_ONE.ordinal(), FAKE_TOPIC)); + m_processor.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false); + + //Verify + assertTrue(m_receivedMessage instanceof SubAckMessage); + Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, false); + assertTrue(subscriptions.contains(expectedSubscription)); + } + + @Test + public void testDoubleSubscribe() { + SubscribeMessage msg = new SubscribeMessage(); + msg.addSubscription(new SubscribeMessage.Couple((byte)AbstractMessage.QOSType.MOST_ONE.ordinal(), FAKE_TOPIC)); + + subscriptions.clearAllSubscriptions(); + assertEquals(0, subscriptions.size()); + + m_processor.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false); + + //Exercise + m_processor.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false); + + //Verify + assertEquals(1, subscriptions.size()); + Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, false); + assertTrue(subscriptions.contains(expectedSubscription)); + } - // @Test -// public void testPublishOfRetainedMessage_afterNewSubscription() throws Exception { -// m_session.getFilterChain().remove("MessageCatcher"); -// m_session.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() { -// -// @Override -// public void filterWrite(IoFilter.NextFilter nextFilter, IoSession session, -// WriteRequest writeRequest) throws Exception { -// try { -// System.out.println("filterReceived class " + writeRequest.getMessage().getClass().getName()); -// if (writeRequest.getMessage() instanceof PublishMessage) { -// 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"); -// } -// } -// }); -// + @Test + public void testPublishOfRetainedMessage_afterNewSubscription() throws Exception { + m_session.getFilterChain().remove("MessageCatcher"); + m_session.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() { + + @Override + public void filterWrite(IoFilter.NextFilter nextFilter, IoSession session, + WriteRequest writeRequest) throws Exception { + try { + System.out.println("filterReceived class " + writeRequest.getMessage().getClass().getName()); + if (writeRequest.getMessage() instanceof PublishMessage) { + 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"); + } + } + }); + // //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); -// -// PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, QOSType.MOST_ONE, "Hello".getBytes(), true, "FakeCLI", null); -// messaging.processPublish(pubEvt); -// -// //Exercise -// SubscribeMessage msg = new SubscribeMessage(); -// msg.addSubscription(new SubscribeMessage.Couple((byte)QOSType.MOST_ONE.ordinal(), "#")); -// messaging.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false); -// -// //Verify -// assertNotNull(m_receivedMessage); -// assertTrue(m_receivedMessage instanceof PublishMessage); -// PublishMessage pubMessage = (PublishMessage) m_receivedMessage; -// assertEquals(FAKE_TOPIC, pubMessage.getTopicName()); -// } + final Subscription subscription = new Subscription(FAKE_CLIENT_ID, + FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, true); + + //subscriptions.matches(topic) redefine the method to return true + SubscriptionsStore subs = new SubscriptionsStore() { + @Override + public List matches(String topic) { + if (topic.equals(FAKE_TOPIC)) { + return Arrays.asList(subscription); + } else { + throw new IllegalArgumentException("Expected " + FAKE_TOPIC + " buf found " + topic); + } + } + }; + subs.init(m_storageService); + + //simulate a connect that register a clientID to an IoSession + IMessaging mockedMessaging = mock(IMessaging.class); + 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, mockedMessaging); + PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, "Hello".getBytes(), true, "FakeCLI", null); + m_processor.processPublish(pubEvt); + + //Exercise + SubscribeMessage msg = new SubscribeMessage(); + msg.addSubscription(new SubscribeMessage.Couple((byte)QOSType.MOST_ONE.ordinal(), "#")); + m_processor.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false); + + //Verify + assertNotNull(m_receivedMessage); + assertTrue(m_receivedMessage instanceof PublishMessage); + PublishMessage pubMessage = (PublishMessage) m_receivedMessage; + assertEquals(FAKE_TOPIC, pubMessage.getTopicName()); + } } diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java index c9ac64ad..d27fa915 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java @@ -1,32 +1,18 @@ package org.dna.mqtt.moquette.messaging.spi.impl; import java.io.File; -import java.util.List; -import org.apache.mina.core.filterchain.IoFilter; 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.impl.events.*; -import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; -import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore; 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.PublishMessage; -import org.dna.mqtt.moquette.proto.messages.SubscribeMessage; -import org.dna.mqtt.moquette.server.ConnectionDescriptor; import org.dna.mqtt.moquette.server.Server; import org.junit.After; import org.junit.Before; -import org.junit.Test; - -import static org.junit.Assert.assertEquals; import org.junit.BeforeClass; -import org.junit.Ignore; /** * @@ -89,125 +75,4 @@ public class SimpleMessagingTest { messaging.stop(); } - @Test - public void testSubscribe() { - //Exercise - SubscribeMessage msg = new SubscribeMessage(); - msg.addSubscription(new SubscribeMessage.Couple((byte)QOSType.MOST_ONE.ordinal(), FAKE_TOPIC)); - messaging.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false); - - //Verify - Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false); - assertTrue(messaging.getSubscriptions().contains(expectedSubscription)); - } - - @Test - public void testDoubleSubscribe() { -// SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false), 0); - SubscribeMessage msg = new SubscribeMessage(); - msg.addSubscription(new SubscribeMessage.Couple((byte)QOSType.MOST_ONE.ordinal(), FAKE_TOPIC)); - - messaging.getSubscriptions().clearAllSubscriptions(); - assertEquals(0, messaging.getSubscriptions().size()); - - messaging.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false); - - //Exercise - messaging.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false); - - //Verify - Subscription subscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false); - assertEquals(1, messaging.getSubscriptions().size()); - } - -// @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), 0); -// SubscribeMessage msg = new SubscribeMessage(); -// msg.addSubscription(new SubscribeMessage.Couple((byte)QOSType.MOST_ONE.ordinal(), FAKE_TOPIC)); -// -// messaging.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false); -// -// //Exercise -// PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, QOSType.MOST_ONE, "Hello".getBytes(), false, "FakeCLI", null); -// messaging.processPublish(pubEvt); -// -// //Verify -// assertNotNull(m_receivedMessage); -// //TODO check received message attributes -// } -// -// @Test -// public void testPublishOfRetainedMessage() throws Exception { -// m_session.getFilterChain().remove("MessageCatcher"); -// m_session.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() { -// -// @Override -// public void filterWrite(IoFilter.NextFilter nextFilter, IoSession session, -// WriteRequest writeRequest) throws Exception { -// try { -// System.out.println("filterReceived class " + writeRequest.getMessage().getClass().getName()); -// if (writeRequest.getMessage() instanceof PublishMessage) { -// 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"); -// } -// } -// }); -// -// //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); -// -// PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, QOSType.MOST_ONE, "Hello".getBytes(), true, "FakeCLI", null); -// messaging.processPublish(pubEvt); -// -// //Exercise -// SubscribeMessage msg = new SubscribeMessage(); -// msg.addSubscription(new SubscribeMessage.Couple((byte)QOSType.MOST_ONE.ordinal(), "#")); -// messaging.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false); -// -// //Verify -// assertNotNull(m_receivedMessage); -// assertTrue(m_receivedMessage instanceof PublishMessage); -// PublishMessage pubMessage = (PublishMessage) m_receivedMessage; -// assertEquals(FAKE_TOPIC, pubMessage.getTopicName()); -// } - - @Ignore - 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, 0); - SubscribeMessage msg = new SubscribeMessage(); - msg.addSubscription(new SubscribeMessage.Couple((byte)QOSType.EXACTLY_ONCE.ordinal(), "/news")); - messaging.processSubscribe(m_session, msg, "fakeID", false); - - //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!! - } }