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

Moved processPublish from SimpleMessaging

Browse Source
This commit is contained in:
andrea
2013-08-30 16:14:50 +02:00
parent 90e2277490
commit dd4ca33ed0
3 changed files with 167 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java
+102 -7
View File
@@ -1,6 +1,9 @@
package org.dna.mqtt.moquette.messaging.spi.impl;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.mina.core.filterchain.IoFilter;
import org.apache.mina.core.filterchain.IoFilterAdapter;
import org.apache.mina.core.session.DummySession;
@@ -8,12 +11,19 @@ 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.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 static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import org.junit.Before;
import org.junit.Test;
import static org.mockito.Mockito.*;
@@ -23,11 +33,18 @@ import static org.mockito.Mockito.*;
* @author andrea
*/
public class ProtocolProcessorTest {
final static String FAKE_TOPIC = "/news";
IoSession m_session;
byte m_returnCode;
ConnectMessage connMsg;
ProtocolProcessor m_processor;
IStorageService m_storageService;
AbstractMessage m_receivedMessage;
@Before
public void setUp() throws InterruptedException {
connMsg = new ConnectMessage();
@@ -41,9 +58,9 @@ public class ProtocolProcessorTest {
public void filterWrite(IoFilter.NextFilter nextFilter, IoSession session,
WriteRequest writeRequest) throws Exception {
try {
AbstractMessage receivedMessage = (AbstractMessage) writeRequest.getMessage();
if (receivedMessage instanceof ConnAckMessage) {
ConnAckMessage buf = (ConnAckMessage) receivedMessage;
m_receivedMessage = (AbstractMessage) writeRequest.getMessage();
if (m_receivedMessage instanceof ConnAckMessage) {
ConnAckMessage buf = (ConnAckMessage) m_receivedMessage;
m_returnCode = buf.getReturnCode();
}
} catch (Exception ex) {
@@ -55,14 +72,14 @@ public class ProtocolProcessorTest {
//sleep to let the messaging batch processor to process the initEvent
Thread.sleep(300);
IStorageService storageService = new HawtDBStorageService();
storageService.initStore();
m_storageService = new HawtDBStorageService();
m_storageService.initStore();
SubscriptionsStore subscriptions = new SubscriptionsStore();
subscriptions.init(storageService);
subscriptions.init(m_storageService);
m_processor = new ProtocolProcessor();
IMessaging mockedMessaging = mock(IMessaging.class);
m_processor.init(new HashMap<String, ConnectionDescriptor>(), subscriptions, storageService, mockedMessaging);
m_processor.init(new HashMap<String, ConnectionDescriptor>(), subscriptions, m_storageService, mockedMessaging);
}
@Test
@@ -105,4 +122,82 @@ public class ProtocolProcessorTest {
/*verify(mockedMessaging).publish(eq("topic"), eq("Topic message".getBytes()),
any(AbstractMessage.QOSType.class), anyBoolean(), eq("123"), any(IoSession.class));*/
}
@Test
public void testPublish() throws InterruptedException {
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<Subscription> matches(String topic) {
if (topic.equals(FAKE_TOPIC)) {
return Arrays.asList(subscription);
} else {
throw new IllegalArgumentException("Expected " + FAKE_TOPIC + " buf found " + topic);
}
}
};
//simulate a connect that register a clientID to an IoSession
IMessaging mockedMessaging = mock(IMessaging.class);
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, mockedMessaging);
//Exercise
PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, "Hello".getBytes(), false, "FakeCLI", null);
m_processor.processPublish(pubEvt);
//Verify
assertNotNull(m_receivedMessage);
//TODO check received message attributes
}
// @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());
// }
}
broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java
+63 -63
View File
@@ -120,69 +120,69 @@ public class SimpleMessagingTest {
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());
}
// @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() {