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

Switched from plain synchronized to ReadWrite locking

Browse Source
This commit is contained in:
selva.andre
2012-08-01 23:51:25 +02:00
parent f0dd3f1f2e
commit 41f03b2d3e
1 changed files with 14 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+14 -6
View File
@@ -2,6 +2,8 @@ package org.dna.mqtt.moquette.messaging.spi.impl;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.List; import java.util.List;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.dna.mqtt.moquette.messaging.spi.IMessaging; import org.dna.mqtt.moquette.messaging.spi.IMessaging;
import org.dna.mqtt.moquette.messaging.spi.INotifier; import org.dna.mqtt.moquette.messaging.spi.INotifier;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
@@ -12,6 +14,8 @@ import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
*/ */
public class SimpleMessaging implements IMessaging { public class SimpleMessaging implements IMessaging {
//TODO is poorly performace on concurrent load, use at least ReadWriteLock //TODO is poorly performace on concurrent load, use at least ReadWriteLock
private ReadWriteLock rwLock = new ReentrantReadWriteLock();
private List<Subscription> subscriptions = new ArrayList<Subscription>(); private List<Subscription> subscriptions = new ArrayList<Subscription>();
private INotifier m_notifier; private INotifier m_notifier;
@@ -24,23 +28,27 @@ public class SimpleMessaging implements IMessaging {
//TODO for this moment no retain management and no qos > 0 //TODO for this moment no retain management and no qos > 0
//trivial implementation //trivial implementation
QOSType defQos = QOSType.MOST_ONE; QOSType defQos = QOSType.MOST_ONE;
synchronized(subscriptions) {
rwLock.readLock().lock();
try {
for (Subscription sub : subscriptions) { for (Subscription sub : subscriptions) {
if (sub.match(topic)) { if (sub.match(topic)) {
m_notifier.notify(sub.clientId, topic, defQos, message); m_notifier.notify(sub.clientId, topic, defQos, message);
} }
} }
} finally {
rwLock.readLock().unlock();
} }
} }
public void subscribe(String clientId, String topic, QOSType qos) { public void subscribe(String clientId, String topic, QOSType qos) {
Subscription newSubscription = new Subscription(clientId, topic, qos); Subscription newSubscription = new Subscription(clientId, topic, qos);
synchronized(subscriptions) { rwLock.writeLock().lock();
if (subscriptions.contains(newSubscription)) { if (subscriptions.contains(newSubscription)) {
return; return;
}
subscriptions.add(newSubscription);
} }
subscriptions.add(newSubscription);
rwLock.writeLock().unlock();
} }
protected List<Subscription> getSubscriptions() { protected List<Subscription> getSubscriptions() {