Merge branch 'topic_tree_move'

This commit is contained in:
andsel
2012-08-03 16:33:04 +02:00
3 changed files with 47 additions and 25 deletions
@@ -16,7 +16,7 @@ public class SimpleMessaging implements IMessaging {
//TODO is poorly performace on concurrent load, use at least ReadWriteLock
private ReadWriteLock rwLock = new ReentrantReadWriteLock();
private List<Subscription> subscriptions = new ArrayList<Subscription>();
private SubscriptionsStore subscriptions = new SubscriptionsStore();
private INotifier m_notifier;
@@ -31,10 +31,8 @@ public class SimpleMessaging implements IMessaging {
rwLock.readLock().lock();
try {
for (Subscription sub : subscriptions) {
if (sub.match(topic)) {
m_notifier.notify(sub.clientId, topic, defQos, message);
}
for (Subscription sub : subscriptions.matches(topic)) {
m_notifier.notify(sub.clientId, topic, defQos, message);
}
} finally {
rwLock.readLock().unlock();
@@ -44,15 +42,11 @@ public class SimpleMessaging implements IMessaging {
public void subscribe(String clientId, String topic, QOSType qos) {
Subscription newSubscription = new Subscription(clientId, topic, qos);
rwLock.writeLock().lock();
if (subscriptions.contains(newSubscription)) {
rwLock.writeLock().unlock();
return;
}
subscriptions.add(newSubscription);
rwLock.writeLock().unlock();
}
protected List<Subscription> getSubscriptions() {
protected SubscriptionsStore getSubscriptions() {
return subscriptions;
}
}
@@ -0,0 +1,39 @@
package org.dna.mqtt.moquette.messaging.spi.impl;
import java.util.ArrayList;
import java.util.List;
/**
* Represents a tree of published topics
*
* @author andrea
*/
public class SubscriptionsStore {
private List<Subscription> subscriptions = new ArrayList<Subscription>();
public void add(Subscription newSubscription) {
if (subscriptions.contains(newSubscription)) {
return;
}
subscriptions.add(newSubscription);
}
public List<Subscription> matches(String topic) {
List<Subscription> m = new ArrayList<Subscription>();
for (Subscription sub : subscriptions) {
if (sub.match(topic)) {
m.add(sub);
}
}
return m;
}
public boolean contains(Subscription sub) {
return subscriptions.contains(sub);
}
public int size() {
return subscriptions.size();
}
}
@@ -1,12 +1,11 @@
package org.dna.mqtt.moquette.messaging.spi.impl;
import org.junit.Before;
import org.mockito.ArgumentCaptor;
import java.util.List;
import org.dna.mqtt.moquette.messaging.spi.INotifier;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
import org.junit.Test;
import static org.junit.Assert.*;
import org.junit.Before;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import static org.mockito.Mockito.*;
/**
@@ -43,17 +42,7 @@ public class SimpleMessagingTest {
//Verify
Subscription subscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE);
assertEquals(1, countMatchingSubscriptions(messaging.getSubscriptions(), subscription));
}
private int countMatchingSubscriptions(List<Subscription> l, Subscription matchingSub) {
int count = 0;
for (Subscription s : l) {
if (s.equals(matchingSub)) {
count++;
}
}
return count;
assertEquals(1, messaging.getSubscriptions().size());
}
@Test