Re-inserted missed CTrie unit tests
This commit is contained in:
andsel
+2
-2
@@ -28,7 +28,7 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory {
|
||||
private static final Token ROOT = new Token("root");
|
||||
private static final INode NO_PARENT = null;
|
||||
|
||||
private INode root;
|
||||
INode root;
|
||||
private volatile ISubscriptionsRepository subscriptionsRepository;
|
||||
|
||||
interface IVisitor<T> {
|
||||
@@ -108,7 +108,7 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory {
|
||||
return new HashSet<>(subsGroupedByClient.values());
|
||||
}
|
||||
|
||||
private Set<Subscription> recursiveMatch(Topic topic, INode inode) {
|
||||
Set<Subscription> recursiveMatch(Topic topic, INode inode) {
|
||||
CNode cnode = inode.mainNode();
|
||||
if (Token.MULTI.equals(cnode.token)) {
|
||||
return cnode.subscriptions;
|
||||
|
||||
+271
@@ -0,0 +1,271 @@
|
||||
/*
|
||||
* Copyright (c) 2012-2018 The original author or authors
|
||||
* ------------------------------------------------------
|
||||
* All rights reserved. This program and the accompanying materials
|
||||
* are made available under the terms of the Eclipse Public License v1.0
|
||||
* and Apache License v2.0 which accompanies this distribution.
|
||||
*
|
||||
* The Eclipse Public License is available at
|
||||
* http://www.eclipse.org/legal/epl-v10.html
|
||||
*
|
||||
* The Apache License v2.0 is available at
|
||||
* http://www.opensource.org/licenses/apache2.0.php
|
||||
*
|
||||
* You may elect to redistribute this code under either of these licenses.
|
||||
*/
|
||||
package io.moquette.broker.subscriptions;
|
||||
|
||||
|
||||
import io.moquette.broker.ISubscriptionsRepository;
|
||||
import io.moquette.persistence.MemorySubscriptionsRepository;
|
||||
import io.netty.handler.codec.mqtt.MqttQoS;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
|
||||
import java.util.Optional;
|
||||
import java.util.Set;
|
||||
|
||||
import static io.moquette.broker.subscriptions.CTrieSubscriptionDirectoryTest.clientSubOnTopic;
|
||||
import static io.moquette.broker.subscriptions.Topic.asTopic;
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
|
||||
public class CTrieSubscriptionDirectoryMatchingTest {
|
||||
|
||||
private CTrieSubscriptionDirectory sut;
|
||||
private ISubscriptionsRepository sessionsRepository;
|
||||
|
||||
@Before
|
||||
public void setUp() {
|
||||
sut = new CTrieSubscriptionDirectory();
|
||||
|
||||
this.sessionsRepository = new MemorySubscriptionsRepository();
|
||||
sut.init(this.sessionsRepository);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMatchSimple() {
|
||||
Subscription slashSub = clientSubOnTopic("TempSensor1", "/");
|
||||
sut.add(slashSub);
|
||||
assertThat(sut.recursiveMatch(asTopic("finance"), sut.root)).isEmpty();
|
||||
|
||||
Subscription slashFinanceSub = clientSubOnTopic("TempSensor1", "/finance");
|
||||
sut.add(slashFinanceSub);
|
||||
assertThat(sut.recursiveMatch(asTopic("finance"), sut.root)).isEmpty();
|
||||
|
||||
assertThat(sut.recursiveMatch(asTopic("/finance"), sut.root)).contains(slashFinanceSub);
|
||||
assertThat(sut.recursiveMatch(asTopic("/"), sut.root)).contains(slashSub);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMatchSimpleMulti() {
|
||||
Subscription anySub = clientSubOnTopic("TempSensor1", "#");
|
||||
sut.add(anySub);
|
||||
assertThat(sut.recursiveMatch(asTopic("finance"), sut.root)).contains(anySub);
|
||||
|
||||
Subscription financeAnySub = clientSubOnTopic("TempSensor1", "finance/#");
|
||||
sut.add(financeAnySub);
|
||||
assertThat(sut.recursiveMatch(asTopic("finance"), sut.root)).containsExactlyInAnyOrder(financeAnySub, anySub);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMatchingDeepMulti_one_layer() {
|
||||
Subscription anySub = clientSubOnTopic("AllSensor1", "#");
|
||||
Subscription financeAnySub = clientSubOnTopic("FinanceSensor", "finance/#");
|
||||
sut.add(anySub);
|
||||
sut.add(financeAnySub);
|
||||
|
||||
// Verify
|
||||
assertThat(sut.recursiveMatch(asTopic("finance/stock"), sut.root))
|
||||
.containsExactlyInAnyOrder(financeAnySub, anySub);
|
||||
assertThat(sut.recursiveMatch(asTopic("finance/stock/ibm"), sut.root))
|
||||
.containsExactlyInAnyOrder(financeAnySub, anySub);
|
||||
// System.out.println(sut.dumpTree());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMatchingDeepMulti_two_layer() {
|
||||
Subscription financeAnySub = clientSubOnTopic("FinanceSensor", "finance/stock/#");
|
||||
sut.add(financeAnySub);
|
||||
|
||||
// Verify
|
||||
assertThat(sut.recursiveMatch(asTopic("finance/stock/ibm"), sut.root)).containsExactly(financeAnySub);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMatchSimpleSingle() {
|
||||
Subscription anySub = clientSubOnTopic("AnySensor", "+");
|
||||
sut.add(anySub);
|
||||
assertThat(sut.recursiveMatch(asTopic("finance"), sut.root)).containsExactly(anySub);
|
||||
|
||||
Subscription financeOne = clientSubOnTopic("AnySensor", "finance/+");
|
||||
sut.add(financeOne);
|
||||
assertThat(sut.recursiveMatch(asTopic("finance/stock"), sut.root)).containsExactly(financeOne);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMatchManySingle() {
|
||||
Subscription manySub = clientSubOnTopic("AnySensor", "+/+");
|
||||
sut.add(manySub);
|
||||
|
||||
// verify
|
||||
assertThat(sut.recursiveMatch(asTopic("/finance"), sut.root)).contains(manySub);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMatchSlashSingle() {
|
||||
Subscription slashPlusSub = clientSubOnTopic("AnySensor", "/+");
|
||||
sut.add(slashPlusSub);
|
||||
Subscription anySub = clientSubOnTopic("AnySensor", "+");
|
||||
sut.add(anySub);
|
||||
|
||||
// Verify
|
||||
assertThat(sut.recursiveMatch(asTopic("/finance"), sut.root)).containsOnly(slashPlusSub);
|
||||
assertThat(sut.recursiveMatch(asTopic("/finance"), sut.root)).doesNotContain(anySub);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMatchManyDeepSingle() {
|
||||
Subscription slashPlusSub = clientSubOnTopic("FinanceSensor1", "/finance/+/ibm");
|
||||
sut.add(slashPlusSub);
|
||||
Subscription slashPlusDeepSub = clientSubOnTopic("FinanceSensor2", "/+/stock/+");
|
||||
sut.add(slashPlusDeepSub);
|
||||
|
||||
// Verify
|
||||
assertThat(sut.recursiveMatch(asTopic("/finance/stock/ibm"), sut.root))
|
||||
.containsExactlyInAnyOrder(slashPlusSub, slashPlusDeepSub);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMatchSimpleMulti_allTheTree() {
|
||||
Subscription sub = clientSubOnTopic("AnySensor1", "#");
|
||||
sut.add(sub);
|
||||
|
||||
assertThat(sut.recursiveMatch(asTopic("finance"), sut.root)).isNotEmpty();
|
||||
assertThat(sut.recursiveMatch(asTopic("finance/ibm"), sut.root)).isNotEmpty();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void rogerLightTopicMatches() {
|
||||
assertMatch("foo/bar", "foo/bar");
|
||||
assertMatch("foo/bar", "foo/bar");
|
||||
assertMatch("foo/+", "foo/bar");
|
||||
assertMatch("foo/+/baz", "foo/bar/baz");
|
||||
assertMatch("foo/+/#", "foo/bar/baz");
|
||||
assertMatch("#", "foo/bar/baz");
|
||||
|
||||
assertNotMatch("foo/bar", "foo");
|
||||
assertNotMatch("foo/+", "foo/bar/baz");
|
||||
assertNotMatch("foo/+/baz", "foo/bar/bar");
|
||||
assertNotMatch("foo/+/#", "fo2/bar/baz");
|
||||
|
||||
assertMatch("#", "/foo/bar");
|
||||
assertMatch("/#", "/foo/bar");
|
||||
assertNotMatch("/#", "foo/bar");
|
||||
|
||||
assertMatch("foo//bar", "foo//bar");
|
||||
assertMatch("foo//+", "foo//bar");
|
||||
assertMatch("foo/+/+/baz", "foo///baz");
|
||||
assertMatch("foo/bar/+", "foo/bar/");
|
||||
}
|
||||
|
||||
private void assertMatch(String s, String t) {
|
||||
sut = new CTrieSubscriptionDirectory();
|
||||
ISubscriptionsRepository sessionsRepository = new MemorySubscriptionsRepository();
|
||||
sut.init(sessionsRepository);
|
||||
|
||||
Subscription sub = clientSubOnTopic("AnySensor1", s);
|
||||
sut.add(sub);
|
||||
|
||||
assertThat(sut.recursiveMatch(asTopic(t), sut.root)).isNotEmpty();
|
||||
}
|
||||
|
||||
private void assertNotMatch(String subscription, String topic) {
|
||||
sut = new CTrieSubscriptionDirectory();
|
||||
ISubscriptionsRepository sessionsRepository = new MemorySubscriptionsRepository();
|
||||
sut.init(sessionsRepository);
|
||||
|
||||
Subscription sub = clientSubOnTopic("AnySensor1", subscription);
|
||||
sut.add(sub);
|
||||
|
||||
assertThat(sut.recursiveMatch(asTopic(topic), sut.root)).isEmpty();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testOverlappingSubscriptions() {
|
||||
Subscription genericSub = new Subscription("Sensor1", asTopic("a/+"), MqttQoS.AT_MOST_ONCE);
|
||||
this.sessionsRepository.addNewSubscription(genericSub);
|
||||
sut.add(genericSub);
|
||||
|
||||
Subscription specificSub = new Subscription("Sensor1", asTopic("a/b"), MqttQoS.AT_MOST_ONCE);
|
||||
this.sessionsRepository.addNewSubscription(specificSub);
|
||||
sut.add(specificSub);
|
||||
|
||||
//Exercise
|
||||
final Set<Subscription> matchingForSpecific = sut.matchQosSharpening(asTopic("a/b"));
|
||||
|
||||
// Verify
|
||||
assertThat(matchingForSpecific.size()).isEqualTo(1);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void removeSubscription_withDifferentClients_subscribedSameTopic() {
|
||||
Subscription slashSub = clientSubOnTopic("Sensor1", "/topic");
|
||||
sut.add(slashSub);
|
||||
Subscription slashSub2 = clientSubOnTopic("Sensor2", "/topic");
|
||||
sut.add(slashSub2);
|
||||
|
||||
// Exercise
|
||||
sut.removeSubscription(asTopic("/topic"), slashSub2.clientId);
|
||||
|
||||
// Verify
|
||||
Subscription remainedSubscription = sut.recursiveMatch(asTopic("/topic"), sut.root).iterator().next();
|
||||
assertThat(remainedSubscription.clientId).isEqualTo(slashSub.clientId);
|
||||
assertEquals(slashSub.clientId, remainedSubscription.clientId);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void removeSubscription_sameClients_subscribedSameTopic() {
|
||||
Subscription slashSub = clientSubOnTopic("Sensor1", "/topic");
|
||||
sut.add(slashSub);
|
||||
|
||||
// Exercise
|
||||
sut.removeSubscription(asTopic("/topic"), slashSub.clientId);
|
||||
|
||||
// Verify
|
||||
final Set<Subscription> matchingSubscriptions = sut.recursiveMatch(asTopic("/topic"), sut.root);
|
||||
assertThat(matchingSubscriptions).isEmpty();
|
||||
}
|
||||
|
||||
/*
|
||||
* Test for Issue #49
|
||||
*/
|
||||
@Test
|
||||
public void duplicatedSubscriptionsWithDifferentQos() {
|
||||
Subscription client2Sub = new Subscription("client2", asTopic("client/test/b"), MqttQoS.AT_MOST_ONCE);
|
||||
this.sut.add(client2Sub);
|
||||
Subscription client1SubQoS0 = new Subscription("client1", asTopic("client/test/b"), MqttQoS.AT_MOST_ONCE);
|
||||
this.sut.add(client1SubQoS0);
|
||||
|
||||
Subscription client1SubQoS2 = new Subscription("client1", asTopic("client/test/b"), MqttQoS.EXACTLY_ONCE);
|
||||
this.sut.add(client1SubQoS2);
|
||||
|
||||
// Verify
|
||||
Set<Subscription> subscriptions = this.sut.matchQosSharpening(asTopic("client/test/b"));
|
||||
assertThat(subscriptions).contains(client1SubQoS2);
|
||||
assertThat(subscriptions).contains(client2Sub);
|
||||
|
||||
final Optional<Subscription> matchingClient1Sub = subscriptions
|
||||
.stream()
|
||||
.filter(s -> s.equals(client1SubQoS0))
|
||||
.findFirst();
|
||||
assertTrue(matchingClient1Sub.isPresent());
|
||||
Subscription client1Sub = matchingClient1Sub.get();
|
||||
|
||||
assertThat(client1SubQoS0.getRequestedQos()).isNotEqualTo(client1Sub.getRequestedQos());
|
||||
|
||||
// client1SubQoS2 should override client1SubQoS0
|
||||
assertThat(client1Sub.getRequestedQos()).isEqualTo(client1SubQoS2.getRequestedQos());
|
||||
}
|
||||
}
|
||||
+308
@@ -0,0 +1,308 @@
|
||||
/*
|
||||
* Copyright (c) 2012-2018 The original author or authors
|
||||
* ------------------------------------------------------
|
||||
* All rights reserved. This program and the accompanying materials
|
||||
* are made available under the terms of the Eclipse Public License v1.0
|
||||
* and Apache License v2.0 which accompanies this distribution.
|
||||
*
|
||||
* The Eclipse Public License is available at
|
||||
* http://www.eclipse.org/legal/epl-v10.html
|
||||
*
|
||||
* The Apache License v2.0 is available at
|
||||
* http://www.opensource.org/licenses/apache2.0.php
|
||||
*
|
||||
* You may elect to redistribute this code under either of these licenses.
|
||||
*/
|
||||
package io.moquette.broker.subscriptions;
|
||||
|
||||
|
||||
import io.moquette.broker.ISubscriptionsRepository;
|
||||
import io.moquette.persistence.MemorySubscriptionsRepository;
|
||||
import io.netty.handler.codec.mqtt.MqttQoS;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
|
||||
import java.util.Optional;
|
||||
import java.util.Set;
|
||||
|
||||
import static io.moquette.broker.subscriptions.Topic.asTopic;
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.junit.Assert.assertFalse;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
|
||||
public class CTrieSubscriptionDirectoryTest {
|
||||
|
||||
private CTrieSubscriptionDirectory sut;
|
||||
|
||||
@Before
|
||||
public void setUp() {
|
||||
sut = new CTrieSubscriptionDirectory();
|
||||
ISubscriptionsRepository sessionsRepository = new MemorySubscriptionsRepository();
|
||||
sut.init(sessionsRepository);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testAddOnSecondLayerWithEmptyTokenOnEmptyTree() {
|
||||
//Exercise
|
||||
sut.add(clientSubOnTopic("TempSensor1", "/"));
|
||||
|
||||
//Verify
|
||||
final Optional<CNode> matchedNode = sut.lookup(asTopic("/"));
|
||||
assertTrue("Node on path / must be present", matchedNode.isPresent());
|
||||
//verify structure, only root INode and the first CNode should be present
|
||||
assertThat(this.sut.root.mainNode().subscriptions).isEmpty();
|
||||
assertThat(this.sut.root.mainNode().allChildren()).isNotEmpty();
|
||||
|
||||
INode firstLayer = this.sut.root.mainNode().allChildren().get(0);
|
||||
assertThat(firstLayer.mainNode().subscriptions).isEmpty();
|
||||
assertThat(firstLayer.mainNode().allChildren()).isNotEmpty();
|
||||
|
||||
INode secondLayer = firstLayer.mainNode().allChildren().get(0);
|
||||
assertThat(secondLayer.mainNode().subscriptions).isNotEmpty();
|
||||
assertThat(secondLayer.mainNode().allChildren()).isEmpty();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testAddFirstLayerNodeOnEmptyTree() {
|
||||
//Exercise
|
||||
sut.add(clientSubOnTopic("TempSensor1", "/temp"));
|
||||
|
||||
//Verify
|
||||
final Optional<CNode> matchedNode = sut.lookup(asTopic("/temp"));
|
||||
assertTrue("Node on path /temp must be present", matchedNode.isPresent());
|
||||
assertFalse(matchedNode.get().subscriptions.isEmpty());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testLookup() {
|
||||
final Subscription existingSubscription = clientSubOnTopic("TempSensor1", "/temp");
|
||||
sut.add(existingSubscription);
|
||||
|
||||
//Exercise
|
||||
final Optional<CNode> matchedNode = sut.lookup(asTopic("/humidity"));
|
||||
|
||||
//Verify
|
||||
assertFalse("Node on path /humidity can't be present", matchedNode.isPresent());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testAddNewSubscriptionOnExistingNode() {
|
||||
final Subscription existingSubscription = clientSubOnTopic("TempSensor1", "/temp");
|
||||
sut.add(existingSubscription);
|
||||
|
||||
//Exercise
|
||||
final Subscription newSubscription = clientSubOnTopic("TempSensor2", "/temp");
|
||||
sut.add(newSubscription);
|
||||
|
||||
//Verify
|
||||
final Optional<CNode> matchedNode = sut.lookup(asTopic("/temp"));
|
||||
assertTrue("Node on path /temp must be present", matchedNode.isPresent());
|
||||
final Set<Subscription> subscriptions = matchedNode.get().subscriptions;
|
||||
assertTrue(subscriptions.contains(newSubscription));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testAddNewDeepNodes() {
|
||||
sut.add(clientSubOnTopic("TempSensorRM", "/italy/roma/temp"));
|
||||
sut.add(clientSubOnTopic("TempSensorFI", "/italy/firenze/temp"));
|
||||
sut.add(clientSubOnTopic("HumSensorFI", "/italy/roma/humidity"));
|
||||
final Subscription happinessSensor = clientSubOnTopic("HappinessSensor", "/italy/happiness");
|
||||
sut.add(happinessSensor);
|
||||
|
||||
//Verify
|
||||
final Optional<CNode> matchedNode = sut.lookup(asTopic("/italy/happiness"));
|
||||
assertTrue("Node on path /italy/happiness must be present", matchedNode.isPresent());
|
||||
final Set<Subscription> subscriptions = matchedNode.get().subscriptions;
|
||||
assertTrue(subscriptions.contains(happinessSensor));
|
||||
}
|
||||
|
||||
static Subscription clientSubOnTopic(String clientID, String topicName) {
|
||||
return new Subscription(clientID, asTopic(topicName), null);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void givenTreeWithSomeNodeWhenRemoveContainedSubscriptionThenNodeIsUpdated() {
|
||||
sut.add(clientSubOnTopic("TempSensor1", "/temp"));
|
||||
|
||||
//Exercise
|
||||
sut.removeSubscription(asTopic("/temp"), "TempSensor1");
|
||||
|
||||
//Verify
|
||||
final Optional<CNode> matchedNode = sut.lookup(asTopic("/temp"));
|
||||
assertFalse("Node on path /temp can't be present", matchedNode.isPresent());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void givenTreeWithSomeNodeUnsubscribeAndResubscribeCleanTomb() {
|
||||
sut.add(clientSubOnTopic("TempSensor1", "test"));
|
||||
sut.removeSubscription(asTopic("test"), "TempSensor1");
|
||||
|
||||
sut.add(clientSubOnTopic("TempSensor1", "test"));
|
||||
assertTrue(sut.root.mainNode().allChildren().size() == 1); // looking to see if TNode is cleaned up
|
||||
}
|
||||
|
||||
@Test
|
||||
public void givenTreeWithSomeNodeWhenRemoveMultipleTimes() {
|
||||
sut.add(clientSubOnTopic("TempSensor1", "test"));
|
||||
|
||||
// make sure no TNode exceptions
|
||||
sut.removeSubscription(asTopic("test"), "TempSensor1");
|
||||
sut.removeSubscription(asTopic("test"), "TempSensor1");
|
||||
sut.removeSubscription(asTopic("test"), "TempSensor1");
|
||||
sut.removeSubscription(asTopic("test"), "TempSensor1");
|
||||
|
||||
//Verify
|
||||
final Optional<CNode> matchedNode = sut.lookup(asTopic("/temp"));
|
||||
assertFalse("Node on path /temp can't be present", matchedNode.isPresent());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void givenTreeWithSomeDeepNodeWhenRemoveMultipleTimes() {
|
||||
sut.add(clientSubOnTopic("TempSensor1", "/test/me/1/2/3"));
|
||||
|
||||
// make sure no TNode exceptions
|
||||
sut.removeSubscription(asTopic("/test/me/1/2/3"), "TempSensor1");
|
||||
sut.removeSubscription(asTopic("/test/me/1/2/3"), "TempSensor1");
|
||||
sut.removeSubscription(asTopic("/test/me/1/2/3"), "TempSensor1");
|
||||
|
||||
//Verify
|
||||
final Optional<CNode> matchedNode = sut.lookup(asTopic("/temp"));
|
||||
assertFalse("Node on path /temp can't be present", matchedNode.isPresent());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void givenTreeWithSomeNodeHierarchWhenRemoveContainedSubscriptionThenNodeIsUpdated() {
|
||||
sut.add(clientSubOnTopic("TempSensor1", "/temp/1"));
|
||||
sut.add(clientSubOnTopic("TempSensor1", "/temp/2"));
|
||||
|
||||
//Exercise
|
||||
sut.removeSubscription(asTopic("/temp/1"), "TempSensor1");
|
||||
|
||||
sut.removeSubscription(asTopic("/temp/1"), "TempSensor1");
|
||||
final Set<Subscription> matchingSubs = sut.recursiveMatch(asTopic("/temp/2"), sut.root);
|
||||
|
||||
//Verify
|
||||
final Subscription expectedMatchingsub = new Subscription("TempSensor1", asTopic("/temp/2"), MqttQoS.AT_MOST_ONCE);
|
||||
assertThat(matchingSubs).contains(expectedMatchingsub);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void givenTreeWithSomeNodeHierarchWhenRemoveContainedSubscriptionSmallerThenNodeIsNotUpdated() {
|
||||
sut.add(clientSubOnTopic("TempSensor1", "/temp/1"));
|
||||
sut.add(clientSubOnTopic("TempSensor1", "/temp/2"));
|
||||
|
||||
//Exercise
|
||||
sut.removeSubscription(asTopic("/temp"), "TempSensor1");
|
||||
|
||||
final Set<Subscription> matchingSubs1 = sut.recursiveMatch(asTopic("/temp/1"), sut.root);
|
||||
final Set<Subscription> matchingSubs2 = sut.recursiveMatch(asTopic("/temp/2"), sut.root);
|
||||
|
||||
//Verify
|
||||
// not clear to me, but I believe /temp unsubscribe should not unsub you from downstream /temp/1 or /temp/2
|
||||
final Subscription expectedMatchingsub1 = new Subscription("TempSensor1", asTopic("/temp/1"), MqttQoS.AT_MOST_ONCE);
|
||||
assertThat(matchingSubs1).contains(expectedMatchingsub1);
|
||||
final Subscription expectedMatchingsub2 = new Subscription("TempSensor1", asTopic("/temp/2"), MqttQoS.AT_MOST_ONCE);
|
||||
assertThat(matchingSubs2).contains(expectedMatchingsub2);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void givenTreeWithDeepNodeWhenRemoveContainedSubscriptionThenNodeIsUpdated() {
|
||||
sut.add(clientSubOnTopic("TempSensor1", "/bah/bin/bash"));
|
||||
|
||||
sut.removeSubscription(asTopic("/bah/bin/bash"), "TempSensor1");
|
||||
|
||||
//Verify
|
||||
final Optional<CNode> matchedNode = sut.lookup(asTopic("/bah/bin/bash"));
|
||||
assertFalse("Node on path /temp can't be present", matchedNode.isPresent());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMatchSubscriptionNoWildcards() {
|
||||
sut.add(clientSubOnTopic("TempSensor1", "/temp"));
|
||||
|
||||
//Exercise
|
||||
final Set<Subscription> matchingSubs = sut.recursiveMatch(asTopic("/temp"), sut.root);
|
||||
|
||||
//Verify
|
||||
final Subscription expectedMatchingsub = new Subscription("TempSensor1", asTopic("/temp"), MqttQoS.AT_MOST_ONCE);
|
||||
assertThat(matchingSubs).contains(expectedMatchingsub);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testRemovalInnerTopicOffRootSameClient() {
|
||||
sut.add(clientSubOnTopic("TempSensor1", "temp"));
|
||||
sut.add(clientSubOnTopic("TempSensor1", "temp/1"));
|
||||
|
||||
//Exercise
|
||||
final Set<Subscription> matchingSubs1 = sut.recursiveMatch(asTopic("temp"), sut.root);
|
||||
final Set<Subscription> matchingSubs2 = sut.recursiveMatch(asTopic("temp/1"), sut.root);
|
||||
|
||||
//Verify
|
||||
final Subscription expectedMatchingsub1 = new Subscription("TempSensor1", asTopic("temp"), MqttQoS.AT_MOST_ONCE);
|
||||
final Subscription expectedMatchingsub2 = new Subscription("TempSensor1", asTopic("temp/1"), MqttQoS.AT_MOST_ONCE);
|
||||
|
||||
assertThat(matchingSubs1).contains(expectedMatchingsub1);
|
||||
assertThat(matchingSubs2).contains(expectedMatchingsub2);
|
||||
|
||||
sut.removeSubscription(asTopic("temp"), "TempSensor1");
|
||||
|
||||
//Exercise
|
||||
final Set<Subscription> matchingSubs3 = sut.recursiveMatch(asTopic("temp"), sut.root);
|
||||
final Set<Subscription> matchingSubs4 = sut.recursiveMatch(asTopic("temp/1"), sut.root);
|
||||
|
||||
assertThat(matchingSubs3).doesNotContain(expectedMatchingsub1);
|
||||
assertThat(matchingSubs4).contains(expectedMatchingsub2);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testRemovalInnerTopicOffRootDiffClient() {
|
||||
sut.add(clientSubOnTopic("TempSensor1", "temp"));
|
||||
sut.add(clientSubOnTopic("TempSensor2", "temp/1"));
|
||||
|
||||
//Exercise
|
||||
final Set<Subscription> matchingSubs1 = sut.recursiveMatch(asTopic("temp"), sut.root);
|
||||
final Set<Subscription> matchingSubs2 = sut.recursiveMatch(asTopic("temp/1"), sut.root);
|
||||
|
||||
//Verify
|
||||
final Subscription expectedMatchingsub1 = new Subscription("TempSensor1", asTopic("temp"), MqttQoS.AT_MOST_ONCE);
|
||||
final Subscription expectedMatchingsub2 = new Subscription("TempSensor2", asTopic("temp/1"), MqttQoS.AT_MOST_ONCE);
|
||||
|
||||
assertThat(matchingSubs1).contains(expectedMatchingsub1);
|
||||
assertThat(matchingSubs2).contains(expectedMatchingsub2);
|
||||
|
||||
sut.removeSubscription(asTopic("temp"), "TempSensor1");
|
||||
|
||||
//Exercise
|
||||
final Set<Subscription> matchingSubs3 = sut.recursiveMatch(asTopic("temp"), sut.root);
|
||||
final Set<Subscription> matchingSubs4 = sut.recursiveMatch(asTopic("temp/1"), sut.root);
|
||||
|
||||
assertThat(matchingSubs3).doesNotContain(expectedMatchingsub1);
|
||||
assertThat(matchingSubs4).contains(expectedMatchingsub2);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testRemovalOuterTopicOffRootDiffClient() {
|
||||
sut.add(clientSubOnTopic("TempSensor1", "temp"));
|
||||
sut.add(clientSubOnTopic("TempSensor2", "temp/1"));
|
||||
|
||||
//Exercise
|
||||
final Set<Subscription> matchingSubs1 = sut.recursiveMatch(asTopic("temp"), sut.root);
|
||||
final Set<Subscription> matchingSubs2 = sut.recursiveMatch(asTopic("temp/1"), sut.root);
|
||||
|
||||
//Verify
|
||||
final Subscription expectedMatchingsub1 = new Subscription("TempSensor1", asTopic("temp"), MqttQoS.AT_MOST_ONCE);
|
||||
final Subscription expectedMatchingsub2 = new Subscription("TempSensor2", asTopic("temp/1"), MqttQoS.AT_MOST_ONCE);
|
||||
|
||||
assertThat(matchingSubs1).contains(expectedMatchingsub1);
|
||||
assertThat(matchingSubs2).contains(expectedMatchingsub2);
|
||||
|
||||
sut.removeSubscription(asTopic("temp/1"), "TempSensor2");
|
||||
|
||||
//Exercise
|
||||
final Set<Subscription> matchingSubs3 = sut.recursiveMatch(asTopic("temp"), sut.root);
|
||||
final Set<Subscription> matchingSubs4 = sut.recursiveMatch(asTopic("temp/1"), sut.root);
|
||||
|
||||
assertThat(matchingSubs3).contains(expectedMatchingsub1);
|
||||
assertThat(matchingSubs4).doesNotContain(expectedMatchingsub2);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,179 @@
|
||||
/*
|
||||
* Copyright (c) 2012-2018 The original author or authors
|
||||
* ------------------------------------------------------
|
||||
* All rights reserved. This program and the accompanying materials
|
||||
* are made available under the terms of the Eclipse Public License v1.0
|
||||
* and Apache License v2.0 which accompanies this distribution.
|
||||
*
|
||||
* The Eclipse Public License is available at
|
||||
* http://www.eclipse.org/legal/epl-v10.html
|
||||
*
|
||||
* The Apache License v2.0 is available at
|
||||
* http://www.opensource.org/licenses/apache2.0.php
|
||||
*
|
||||
* You may elect to redistribute this code under either of these licenses.
|
||||
*/
|
||||
package io.moquette.broker.subscriptions;
|
||||
|
||||
import org.assertj.core.api.AbstractAssert;
|
||||
import org.assertj.core.api.Assertions;
|
||||
import org.junit.Test;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
|
||||
public class TopicTest {
|
||||
|
||||
@Test
|
||||
public void testParseTopic() {
|
||||
assertThat(new Topic("finance/stock/ibm")).containsToken("finance", "stock", "ibm");
|
||||
|
||||
assertThat(new Topic("/finance/stock/ibm")).containsToken(Token.EMPTY, "finance", "stock", "ibm");
|
||||
|
||||
assertThat(new Topic("/")).containsToken(Token.EMPTY, Token.EMPTY);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testParseTopicMultiValid() {
|
||||
assertThat(new Topic("finance/stock/#")).containsToken("finance", "stock", Token.MULTI);
|
||||
|
||||
assertThat(new Topic("#")).containsToken(Token.MULTI);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testValidationProcess() {
|
||||
// TopicMultiInTheMiddle
|
||||
assertThat(new Topic("finance/#/closingprice")).isInValid();
|
||||
|
||||
// MultiNotAfterSeparator
|
||||
assertThat(new Topic("finance#")).isInValid();
|
||||
|
||||
// TopicMultiNotAlone
|
||||
assertThat(new Topic("/finance/#closingprice")).isInValid();
|
||||
|
||||
// SingleNotAferSeparator
|
||||
assertThat(new Topic("finance+")).isInValid();
|
||||
|
||||
assertThat(new Topic("finance/+")).isValid();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testParseTopicSingleValid() {
|
||||
assertThat(new Topic("finance/stock/+")).containsToken("finance", "stock", Token.SINGLE);
|
||||
|
||||
assertThat(new Topic("+")).containsToken(Token.SINGLE);
|
||||
|
||||
assertThat(new Topic("finance/+/ibm")).containsToken("finance", Token.SINGLE, "ibm");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMatchTopics_simple() {
|
||||
assertThat(new Topic("/")).matches("/");
|
||||
assertThat(new Topic("/finance")).matches("/finance");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMatchTopics_multi() {
|
||||
assertThat(new Topic("finance")).matches("#");
|
||||
assertThat(new Topic("finance")).matches("finance/#");
|
||||
assertThat(new Topic("finance/stock")).matches("finance/#");
|
||||
assertThat(new Topic("finance/stock/ibm")).matches("finance/#");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMatchTopics_single() {
|
||||
assertThat(new Topic("finance")).matches("+");
|
||||
assertThat(new Topic("finance/stock")).matches("finance/+");
|
||||
assertThat(new Topic("finance")).doesNotMatch("finance/+");
|
||||
assertThat(new Topic("/finance")).matches("/+");
|
||||
assertThat(new Topic("/finance")).doesNotMatch("+");
|
||||
assertThat(new Topic("/finance")).matches("+/+");
|
||||
assertThat(new Topic("/finance/stock/ibm")).matches("/finance/+/ibm");
|
||||
assertThat(new Topic("/")).matches("+/+");
|
||||
assertThat(new Topic("sport/")).matches("sport/+");
|
||||
assertThat(new Topic("/finance/stock")).doesNotMatch("+");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void rogerLightMatchTopics() {
|
||||
assertThat(new Topic("foo/bar")).matches("foo/bar");
|
||||
assertThat(new Topic("foo/bar")).matches("foo/+");
|
||||
assertThat(new Topic("foo/bar/baz")).matches("foo/+/baz");
|
||||
assertThat(new Topic("foo/bar/baz")).matches("foo/+/#");
|
||||
assertThat(new Topic("foo/bar/baz")).matches("#");
|
||||
|
||||
assertThat(new Topic("foo")).doesNotMatch("foo/bar");
|
||||
assertThat(new Topic("foo/bar/baz")).doesNotMatch("foo/+");
|
||||
assertThat(new Topic("foo/bar/bar")).doesNotMatch("foo/+/baz");
|
||||
assertThat(new Topic("fo2/bar/baz")).doesNotMatch("foo/+/#");
|
||||
|
||||
assertThat(new Topic("/foo/bar")).matches("#");
|
||||
assertThat(new Topic("/foo/bar")).matches("/#");
|
||||
assertThat(new Topic("foo/bar")).doesNotMatch("/#");
|
||||
|
||||
assertThat(new Topic("foo//bar")).matches("foo//bar");
|
||||
assertThat(new Topic("foo//bar")).matches("foo//+");
|
||||
assertThat(new Topic("foo///baz")).matches("foo/+/+/baz");
|
||||
assertThat(new Topic("foo/bar/")).matches("foo/bar/+");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void exceptHeadToken() {
|
||||
assertEquals(Topic.asTopic("token"), Topic.asTopic("/token").exceptHeadToken());
|
||||
assertEquals(Topic.asTopic("a/b"), Topic.asTopic("/a/b").exceptHeadToken());
|
||||
}
|
||||
|
||||
public static TopicAssert assertThat(Topic topic) {
|
||||
return new TopicAssert(topic);
|
||||
}
|
||||
|
||||
static class TopicAssert extends AbstractAssert<TopicAssert, Topic> {
|
||||
|
||||
TopicAssert(Topic actual) {
|
||||
super(actual, TopicAssert.class);
|
||||
}
|
||||
|
||||
public TopicAssert matches(String topic) {
|
||||
Assertions.assertThat(actual.match(new Topic(topic))).isTrue();
|
||||
|
||||
return myself;
|
||||
}
|
||||
|
||||
public TopicAssert doesNotMatch(String topic) {
|
||||
Assertions.assertThat(actual.match(new Topic(topic))).isFalse();
|
||||
|
||||
return myself;
|
||||
}
|
||||
|
||||
public TopicAssert containsToken(Object... tokens) {
|
||||
Assertions.assertThat(actual.getTokens()).containsExactly(asArray(tokens));
|
||||
|
||||
return myself;
|
||||
}
|
||||
|
||||
private Token[] asArray(Object... l) {
|
||||
Token[] tokens = new Token[l.length];
|
||||
for (int i = 0; i < l.length; i++) {
|
||||
Object o = l[i];
|
||||
if (o instanceof Token) {
|
||||
tokens[i] = (Token) o;
|
||||
} else {
|
||||
tokens[i] = new Token(o.toString());
|
||||
}
|
||||
}
|
||||
|
||||
return tokens;
|
||||
}
|
||||
|
||||
public TopicAssert isValid() {
|
||||
Assertions.assertThat(actual.isValid()).isTrue();
|
||||
|
||||
return myself;
|
||||
}
|
||||
|
||||
public TopicAssert isInValid() {
|
||||
Assertions.assertThat(actual.isValid()).isFalse();
|
||||
|
||||
return myself;
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user