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

Big fix augment unit testing for retained cases

Browse Source
This commit is contained in:
selva.andre
2012-08-14 22:55:25 +02:00
parent af9fff2911
commit 6c8fa400d9
6 changed files with 107 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
parser/src/main/java/org/dna/mqtt/moquette/client/Client.java
+5
View File
@@ -215,7 +215,12 @@ public final class Client {
* It's admitted to publish a 0 -length payload.
*/
public void publish(String topic, byte[] payload) throws PublishException {
publish(topic, payload, false);
}
public void publish(String topic, byte[] payload, boolean retain) throws PublishException {
PublishMessage msg = new PublishMessage();
msg.setRetainFlag(retain);
msg.setTopicName(topic);
msg.setPayload(payload);
parser/src/main/java/org/dna/mqtt/moquette/client/IPublishCallback.java
+1 -1
View File
@@ -6,5 +6,5 @@ package org.dna.mqtt.moquette.client;
*/
public interface IPublishCallback {
void published(String topic, byte[] message);
void published(String topic, byte[] message/*, boolean retained*/);
}
parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java
+2 -1
View File
@@ -199,8 +199,9 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier {
m_authenticator = authenticator;
}
public void notify(String clientId, String topic, QOSType qOSType, byte[] payload) {
public void notify(String clientId, String topic, QOSType qOSType, byte[] payload, boolean retained) {
PublishMessage pubMessage = new PublishMessage();
pubMessage.setRetainFlag(retained);
pubMessage.setTopicName(topic);
pubMessage.setQos(qOSType);
pubMessage.setPayload(payload);
parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java
+27 -1
View File
@@ -56,7 +56,33 @@ public class SimpleMessagingTest {
//Verify
ArgumentCaptor<byte[]> argument = ArgumentCaptor.forClass(byte[].class);
verify(notifier).notify(eq(FAKE_CLIENT_ID), eq(FAKE_TOPIC), any(QOSType.class), argument.capture());
verify(notifier).notify(eq(FAKE_CLIENT_ID), eq(FAKE_TOPIC), any(QOSType.class), argument.capture(), eq(false));
assertEquals("Hello", new String(argument.getValue()));
}
@Test
public void testMatchTopics_simple() {
assertTrue(messaging.matchTopics("/", "/"));
assertTrue(messaging.matchTopics("/finance", "/finance"));
}
@Test
public void testMatchTopics_multi() {
assertTrue(messaging.matchTopics("finance", "#"));
assertTrue(messaging.matchTopics("finance", "finance/#"));
assertTrue(messaging.matchTopics("finance/stock", "finance/#"));
assertTrue(messaging.matchTopics("finance/stock/ibm", "finance/#"));
}
@Test
public void testMatchTopics_single() {
assertTrue(messaging.matchTopics("finance", "+"));
assertTrue(messaging.matchTopics("finance/stock", "finance/+"));
assertTrue(messaging.matchTopics("/finance", "/+"));
assertFalse(messaging.matchTopics("/finance", "+"));
assertTrue(messaging.matchTopics("/finance", "+/+"));
assertTrue(messaging.matchTopics("/finance/stock/ibm", "/finance/+/ibm"));
assertFalse(messaging.matchTopics("/finance/stock", "+"));
}
}
parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStoreTest.java
+43 -18
View File
@@ -1,10 +1,17 @@
package org.dna.mqtt.moquette.messaging.spi.impl;
import java.io.File;
import java.io.IOException;
import java.text.ParseException;
import java.util.Arrays;
import java.util.List;
import org.dna.mqtt.moquette.MQTTException;
import org.dna.mqtt.moquette.messaging.spi.impl.SubscriptionsStore.Token;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.fusesource.hawtdb.api.MultiIndexFactory;
import org.fusesource.hawtdb.api.PageFile;
import org.fusesource.hawtdb.api.PageFileFactory;
import org.junit.After;
import static org.junit.Assert.*;
import org.junit.Before;
import org.junit.Test;
@@ -14,15 +21,33 @@ import org.junit.Test;
* @author andrea
*/
public class SubscriptionsStoreTest {
private File hawtFile;
private SubscriptionsStore store;
public SubscriptionsStoreTest() {
}
@Before
public void setUp() {
public void setUp() throws IOException {
hawtFile = File.createTempFile("moquette_persistent_store", ".dat");
store = new SubscriptionsStore();
// store.init(hawtFile.getAbsolutePath());
PageFileFactory pageFactory = new PageFileFactory();
pageFactory.setFile(hawtFile);
pageFactory.open();
PageFile pageFile = pageFactory.getPageFile();
MultiIndexFactory multiIndexFactory = new MultiIndexFactory(pageFile);
store.init(multiIndexFactory);
}
@After
public void tearDown() {
if (hawtFile.exists()) {
hawtFile.delete();
}
}
@Test
@@ -80,11 +105,11 @@ public class SubscriptionsStoreTest {
@Test
public void testMatchSimple() {
Subscription slashSub = new Subscription(null, "/", AbstractMessage.QOSType.MOST_ONE);
Subscription slashSub = new Subscription("FAKE_CLI_ID_1", "/", AbstractMessage.QOSType.MOST_ONE);
store.add(slashSub);
assertTrue(store.matches("finance").isEmpty());
Subscription slashFinanceSub = new Subscription(null, "/finance", AbstractMessage.QOSType.MOST_ONE);
Subscription slashFinanceSub = new Subscription("FAKE_CLI_ID_1", "/finance", AbstractMessage.QOSType.MOST_ONE);
store.add(slashFinanceSub);
assertTrue(store.matches("finance").isEmpty());
@@ -94,19 +119,19 @@ public class SubscriptionsStoreTest {
@Test
public void testMatchSimpleMulti() {
Subscription anySub = new Subscription(null, "#", AbstractMessage.QOSType.MOST_ONE);
Subscription anySub = new Subscription("FAKE_CLI_ID_1", "#", AbstractMessage.QOSType.MOST_ONE);
store.add(anySub);
assertTrue(store.matches("finance").contains(anySub));
Subscription financeAnySub = new Subscription(null, "finance/#", AbstractMessage.QOSType.MOST_ONE);
Subscription financeAnySub = new Subscription("FAKE_CLI_ID_1", "finance/#", AbstractMessage.QOSType.MOST_ONE);
store.add(financeAnySub);
assertTrue(store.matches("finance").containsAll(Arrays.asList(financeAnySub, anySub)));
}
@Test
public void testMatchingDeepMulti_one_layer() {
Subscription anySub = new Subscription(null, "#", AbstractMessage.QOSType.MOST_ONE);
Subscription financeAnySub = new Subscription(null, "finance/#", AbstractMessage.QOSType.MOST_ONE);
Subscription anySub = new Subscription("FAKE_CLI_ID_1", "#", AbstractMessage.QOSType.MOST_ONE);
Subscription financeAnySub = new Subscription("FAKE_CLI_ID_1", "finance/#", AbstractMessage.QOSType.MOST_ONE);
store.add(anySub);
store.add(financeAnySub);
@@ -118,7 +143,7 @@ public class SubscriptionsStoreTest {
@Test
public void testMatchingDeepMulti_two_layer() {
Subscription financeAnySub = new Subscription(null, "finance/stock/#", AbstractMessage.QOSType.MOST_ONE);
Subscription financeAnySub = new Subscription("FAKE_CLI_ID_1", "finance/stock/#", AbstractMessage.QOSType.MOST_ONE);
store.add(financeAnySub);
//Verify
@@ -127,18 +152,18 @@ public class SubscriptionsStoreTest {
@Test
public void testMatchSimpleSingle() {
Subscription anySub = new Subscription(null, "+", AbstractMessage.QOSType.MOST_ONE);
Subscription anySub = new Subscription("FAKE_CLI_ID_1", "+", AbstractMessage.QOSType.MOST_ONE);
store.add(anySub);
assertTrue(store.matches("finance").contains(anySub));
Subscription financeOne = new Subscription(null, "finance/+", AbstractMessage.QOSType.MOST_ONE);
Subscription financeOne = new Subscription("FAKE_CLI_ID_1", "finance/+", AbstractMessage.QOSType.MOST_ONE);
store.add(financeOne);
assertTrue(store.matches("finance/stock").contains(financeOne));
}
@Test
public void testMatchManySingle() {
Subscription manySub = new Subscription(null, "+/+", AbstractMessage.QOSType.MOST_ONE);
Subscription manySub = new Subscription("FAKE_CLI_ID_1", "+/+", AbstractMessage.QOSType.MOST_ONE);
store.add(manySub);
//verify
@@ -148,9 +173,9 @@ public class SubscriptionsStoreTest {
@Test
public void testMatchSlashSingle() {
Subscription slashPlusSub = new Subscription(null, "/+", AbstractMessage.QOSType.MOST_ONE);
Subscription slashPlusSub = new Subscription("FAKE_CLI_ID_1", "/+", AbstractMessage.QOSType.MOST_ONE);
store.add(slashPlusSub);
Subscription anySub = new Subscription(null, "+", AbstractMessage.QOSType.MOST_ONE);
Subscription anySub = new Subscription("FAKE_CLI_ID_1", "+", AbstractMessage.QOSType.MOST_ONE);
store.add(anySub);
//Verify
@@ -162,10 +187,10 @@ public class SubscriptionsStoreTest {
@Test
public void testMatchManyDeepSingle() {
Subscription slashPlusSub = new Subscription(null, "/finance/+/ibm", AbstractMessage.QOSType.MOST_ONE);
Subscription slashPlusSub = new Subscription("FAKE_CLI_ID_1", "/finance/+/ibm", AbstractMessage.QOSType.MOST_ONE);
store.add(slashPlusSub);
Subscription slashPlusDeepSub = new Subscription(null, "/+/stock/+", AbstractMessage.QOSType.MOST_ONE);
Subscription slashPlusDeepSub = new Subscription("FAKE_CLI_ID_1", "/+/stock/+", AbstractMessage.QOSType.MOST_ONE);
store.add(slashPlusDeepSub);
//Verify
@@ -174,7 +199,7 @@ public class SubscriptionsStoreTest {
@Test
public void testMatchSimpleMulti_allTheTree() {
store.add(new Subscription(null, "#", AbstractMessage.QOSType.MOST_ONE));
store.add(new Subscription("FAKE_CLI_ID_1", "#", AbstractMessage.QOSType.MOST_ONE));
assertFalse(store.matches("finance").isEmpty());
assertFalse(store.matches("finance/ibm").isEmpty());
}
@@ -182,7 +207,7 @@ public class SubscriptionsStoreTest {
@Test
public void testMatchSimpleMulti_zeroLevel() {
//check MULTI in case of zero level match
store.add(new Subscription(null, "finance/#", AbstractMessage.QOSType.MOST_ONE));
store.add(new Subscription("FAKE_CLI_ID_1", "finance/#", AbstractMessage.QOSType.MOST_ONE));
assertFalse(store.matches("finance").isEmpty());
}
parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java
+29
View File
@@ -196,4 +196,33 @@ public class ServerIntegrationTest {
client.shutdown();
}
// @Test
// public void testRetain_maintainMessage_againstClientDestruction() {
// Client client = new Client("localhost", Server.PORT, "CLID_123");
// client.connect(false); //without session cleanup
//
// client.publish("/topic", "Test my payload".getBytes(), true);
// client.close();
// client.shutdown();
//
// //Exercise that the client maintain the subscriptions
// client = new Client("localhost", Server.PORT, "CLID_123");
// client.connect(false);
// client.subscribe("/topic", new IPublishCallback() {
//
// public void published(String topic, byte[] message) {
// received = true;
// }
// });
// client.close();
//
// //Verify
// assertTrue(received);
//
// //TearDown
// client.shutdown();
// }
}