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

Removed never used and unsufull no-op interceptor, forced usage of BrokerInterceptor, changed BrokerInterceptor to work on a list of InterceptHandler

Browse Source
This commit is contained in:
andrea
2015-08-08 18:37:32 +02:00
parent 9daff2b738
commit dc55ff4078
7 changed files with 94 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/eclipse/moquette/interception/Interceptor.java
+1 -5
View File
@@ -23,8 +23,7 @@ import org.eclipse.moquette.spi.impl.subscriptions.Subscription;
* This interface is to be used internally by the broker components.
* <p>
* An interface is used instead of a class to allow more flexibility in changing
* an implementation. For instance, this interface provide a lightweight
* no-handler interceptor.
* an implementation.
* <p>
* Interceptor implementations forward notifications to a <code>InterceptHandler</code>,
* that is normally a field. So, the implementations should act as a proxy to a custom
@@ -34,9 +33,6 @@ import org.eclipse.moquette.spi.impl.subscriptions.Subscription;
* @author Wagner Macedo
*/
public interface Interceptor {
/**
* When no custom handler is provided, this interceptor should be used.
*/
void notifyClientConnected(ConnectMessage msg);
broker/src/main/java/org/eclipse/moquette/spi/impl/BrokerInterceptor.java
+44 -33
View File
@@ -21,6 +21,7 @@ import org.eclipse.moquette.proto.messages.ConnectMessage;
import org.eclipse.moquette.proto.messages.PublishMessage;
import org.eclipse.moquette.spi.impl.subscriptions.Subscription;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -30,11 +31,11 @@ import java.util.concurrent.Executors;
* @author Wagner Macedo
*/
final class BrokerInterceptor implements Interceptor {
private final InterceptHandler handler;
private final List<InterceptHandler> handlers;
private final ExecutorService executor;
BrokerInterceptor(InterceptHandler handler) {
this.handler = handler;
BrokerInterceptor(List<InterceptHandler> handlers) {
this.handlers = handlers;
executor = Executors.newFixedThreadPool(1);
}
@@ -47,51 +48,61 @@ final class BrokerInterceptor implements Interceptor {
@Override
public void notifyClientConnected(final ConnectMessage msg) {
executor.execute(new Runnable() {
@Override
public void run() {
handler.onConnect(msg.readOnlyClone());
}
});
for (final InterceptHandler handler : this.handlers) {
executor.execute(new Runnable() {
@Override
public void run() {
handler.onConnect(msg.readOnlyClone());
}
});
}
}
@Override
public void notifyClientDisconnected(final String clientID) {
executor.execute(new Runnable() {
@Override
public void run() {
handler.onDisconnect(clientID);
}
});
for (final InterceptHandler handler : this.handlers) {
executor.execute(new Runnable() {
@Override
public void run() {
handler.onDisconnect(clientID);
}
});
}
}
@Override
public void notifyTopicPublished(final PublishMessage msg) {
executor.execute(new Runnable() {
@Override
public void run() {
handler.onPublish(msg.readOnlyClone());
}
});
for (final InterceptHandler handler : this.handlers) {
executor.execute(new Runnable() {
@Override
public void run() {
handler.onPublish(msg.readOnlyClone());
}
});
}
}
@Override
public void notifyTopicSubscribed(final Subscription sub) {
executor.execute(new Runnable() {
@Override
public void run() {
handler.onSubscribe(sub.clone());
}
});
for (final InterceptHandler handler : this.handlers) {
executor.execute(new Runnable() {
@Override
public void run() {
handler.onSubscribe(sub);
}
});
}
}
@Override
public void notifyTopicUnsubscribed(final String sub) {
executor.execute(new Runnable() {
@Override
public void run() {
handler.onUnsubscribe(sub);
}
});
for (final InterceptHandler handler : this.handlers) {
executor.execute(new Runnable() {
@Override
public void run() {
handler.onUnsubscribe(sub);
}
});
}
}
}
broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java
+10 -22
View File
@@ -23,9 +23,6 @@ import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import io.netty.util.internal.StringUtil;
import org.apache.commons.codec.binary.StringUtils;
import org.eclipse.moquette.interception.Interceptor;
import org.eclipse.moquette.server.netty.NettyChannel;
import org.eclipse.moquette.spi.IMatchingCondition;
import org.eclipse.moquette.spi.IMessagesStore;
@@ -106,7 +103,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
private IMessagesStore m_messagesStore;
private ISessionsStore m_sessionsStore;
private IAuthenticator m_authenticator;
private Interceptor interceptor = null;
private BrokerInterceptor m_interceptor;
//maps clientID to Will testament, if specified on CONNECT
private Map<String, WillMessage> m_willStore = new HashMap<>();
@@ -116,13 +113,6 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
ProtocolProcessor() {}
void init(SubscriptionsStore subscriptions, IMessagesStore storageService,
ISessionsStore sessionsStore,
IAuthenticator authenticator,
boolean allowAnonymous, IAuthorizator authorizator) {
init(subscriptions, storageService, sessionsStore, authenticator, allowAnonymous, authorizator, null);
}
/**
* @param subscriptions the subscription store where are stored all the existing
* clients subscriptions.
@@ -137,11 +127,9 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
void init(SubscriptionsStore subscriptions, IMessagesStore storageService,
ISessionsStore sessionsStore,
IAuthenticator authenticator,
boolean allowAnonymous, IAuthorizator authorizator, Interceptor interceptor) {
boolean allowAnonymous, IAuthorizator authorizator, BrokerInterceptor interceptor) {
//m_clientIDs = clientIDs;
if (interceptor != null) {
this.interceptor = interceptor;
}
this.m_interceptor = interceptor;
this.subscriptions = subscriptions;
this.allowAnonymous = allowAnonymous;
m_authorizator = authorizator;
@@ -177,7 +165,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
ConnAckMessage okResp = new ConnAckMessage();
okResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED);
session.write(okResp);
interceptor.notifyClientConnected(msg);
m_interceptor.notifyClientConnected(msg);
return;
}
@@ -233,7 +221,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
AbstractMessage.QOSType willQos = AbstractMessage.QOSType.values()[msg.getWillQos()];
byte[] willPayload = msg.getWillMessage().getBytes();
ByteBuffer bb = (ByteBuffer) ByteBuffer.allocate(willPayload.length).put(willPayload).flip();
//save the will testment in the clientID store
//save the will testament in the clientID store
WillMessage will = new WillMessage(msg.getWillTopic(), bb, msg.isWillRetain(),willQos );
m_willStore.put(msg.getClientID(), will);
}
@@ -253,7 +241,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
okResp.setSessionPresent(true);
}
session.write(okResp);
interceptor.notifyClientConnected(msg);
m_interceptor.notifyClientConnected(msg);
LOG.info("Create persistent session for clientID <{}>", msg.getClientID());
m_sessionsStore.addNewSubscription(Subscription.createEmptySubscription(msg.getClientID(), true)); //null means EmptySubscription
@@ -317,7 +305,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
String user = (String) session.getAttribute(NettyChannel.ATTR_KEY_USERNAME);
if (m_authorizator.canWrite(topic, user, clientID)) {
executePublish(clientID, msg);
interceptor.notifyTopicPublished(msg);
m_interceptor.notifyTopicPublished(msg);
} else {
LOG.debug("topic {} doesn't have write credentials", topic);
}
@@ -576,7 +564,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
m_willStore.remove(clientID);
LOG.info("DISCONNECT client <{}> with clean session {}", clientID, cleanSession);
interceptor.notifyClientDisconnected(clientID);
m_interceptor.notifyClientDisconnected(clientID);
}
void processConnectionLost(LostConnectionEvent evt) {
@@ -610,7 +598,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
for (String topic : topics) {
subscriptions.removeSubscription(topic, clientID);
m_sessionsStore.removeSubscription(topic, clientID);
interceptor.notifyTopicUnsubscribed(topic);
m_interceptor.notifyTopicUnsubscribed(topic);
}
//ack the client
@@ -655,7 +643,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
subscriptions.add(newSubscription);
//notify the Observables
interceptor.notifyTopicSubscribed(newSubscription);
m_interceptor.notifyTopicSubscribed(newSubscription);
//scans retained messages to be published to the new subscription
Collection<IMessagesStore.StoredMessage> messages = m_messagesStore.searchMatching(new IMatchingCondition() {
broker/src/main/java/org/eclipse/moquette/spi/impl/SimpleMessaging.java
+18 -12
View File
@@ -18,9 +18,9 @@ package org.eclipse.moquette.spi.impl;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import io.netty.util.internal.StringUtil;
import org.HdrHistogram.Histogram;
import org.eclipse.moquette.interception.InterceptHandler;
import org.eclipse.moquette.interception.Interceptor;
import org.eclipse.moquette.proto.messages.AbstractMessage;
import org.eclipse.moquette.spi.impl.security.*;
import org.eclipse.moquette.server.ServerChannel;
@@ -40,6 +40,8 @@ import java.io.File;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
@@ -75,7 +77,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
private ExecutorService m_executor;
private Disruptor<ValueEvent> m_disruptor;
private Interceptor interceptor;
private BrokerInterceptor m_interceptor;
private static SimpleMessaging INSTANCE;
@@ -194,14 +196,18 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
m_sessionsStore = mapStorage;
m_storageService.initStore();
try {
final InterceptHandler handler = Class.forName(props.getProperty("intercept.handler"))
.asSubclass(InterceptHandler.class).newInstance();
interceptor = new BrokerInterceptor(handler);
} catch (Throwable ex) {
LOG.error("Can't load the intercept handler {}", ex);
List<InterceptHandler> observers = new ArrayList<>();
String interceptorClassName = props.getProperty("intercept.handler");
if (interceptorClassName != null && !interceptorClassName.isEmpty()) {
try {
InterceptHandler handler = Class.forName(interceptorClassName).asSubclass(InterceptHandler.class).newInstance();
observers.add(handler);
} catch (Throwable ex) {
LOG.error("Can't load the intercept handler {}", ex);
}
}
m_interceptor = new BrokerInterceptor(observers);
//List<Subscription> storedSubscriptions = m_sessionsStore.listAllSubscriptions();
//subscriptions.init(storedSubscriptions);
@@ -251,7 +257,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
}
boolean allowAnonymous = Boolean.parseBoolean(props.getProperty(ALLOW_ANONYMOUS_PROPERTY_NAME, "true"));
m_processor.init(subscriptions, m_storageService, m_sessionsStore, authenticator, allowAnonymous, authorizator, interceptor);
m_processor.init(subscriptions, m_storageService, m_sessionsStore, authenticator, allowAnonymous, authorizator, m_interceptor);
}
private Object loadClass(String className, Class<?> cls) {
@@ -291,8 +297,8 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
private void processStop() {
LOG.debug("processStop invoked");
if (interceptor != null) {
((BrokerInterceptor) interceptor).stop();
if (m_interceptor != null) {
this.m_interceptor.stop();
}
m_storageService.close();
LOG.debug("subscription tree {}", subscriptions.dumpTree());
broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java
-1
View File
@@ -19,7 +19,6 @@ import java.text.ParseException;
import java.util.*;
import java.util.concurrent.LinkedBlockingDeque;
import org.eclipse.moquette.interception.Interceptor;
import org.eclipse.moquette.spi.ISessionsStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
broker/src/test/java/org/eclipse/moquette/spi/impl/BrokerInterceptorTest.java
+5 -2
View File
@@ -24,6 +24,7 @@ import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import java.util.Arrays;
import java.util.concurrent.atomic.AtomicInteger;
import static org.junit.Assert.*;
@@ -36,7 +37,7 @@ public class BrokerInterceptorTest {
private static final AtomicInteger n = new AtomicInteger(0);
// Interceptor loaded with a custom InterceptHandler special for the tests
private static final BrokerInterceptor interceptor = new BrokerInterceptor(new InterceptHandler() {
private static final class MockObserver implements InterceptHandler {
@Override
public void onConnect(ConnectMessage msg) {
n.set(40);
@@ -61,7 +62,9 @@ public class BrokerInterceptorTest {
public void onUnsubscribe(String topic) {
n.set(80);
}
});
}
private static final BrokerInterceptor interceptor = new BrokerInterceptor(Arrays.<InterceptHandler>asList(new MockObserver()));
@BeforeClass
public static void beforeAllTests() {
broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java
+16 -11
View File
@@ -21,6 +21,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import io.netty.util.AttributeKey;
import org.eclipse.moquette.interception.InterceptHandler;
import org.eclipse.moquette.server.netty.NettyChannel;
import org.eclipse.moquette.spi.IMatchingCondition;
import org.eclipse.moquette.spi.IMessagesStore;
@@ -59,6 +60,9 @@ public class ProtocolProcessorTest {
final static String TEST_PWD = "fakepwd";
final static String EVIL_TEST_USER = "eviluser";
final static String EVIL_TEST_PWD = "unsecret";
final static List<InterceptHandler> EMPTY_OBSERVERS = Collections.emptyList();
final static BrokerInterceptor NO_OBSERVERS_INTERCEPTOR = new BrokerInterceptor(EMPTY_OBSERVERS);
DummyChannel m_session;
byte m_returnCode;
@@ -172,7 +176,8 @@ public class ProtocolProcessorTest {
subscriptions = new SubscriptionsStore();
subscriptions.init(new MemoryStorageService());
m_processor = new ProtocolProcessor();
m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, true, new PermitAllAuthorizator());
m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, true,
new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
}
@Test
@@ -263,7 +268,7 @@ public class ProtocolProcessorTest {
@Test
public void prohibitAnonymousClient() {
connMsg.setClientID("123");
m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, false, new PermitAllAuthorizator());
m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, false, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
//Exercise
m_processor.processConnect(m_session, connMsg);
@@ -277,7 +282,7 @@ public class ProtocolProcessorTest {
connMsg.setClientID("123");
connMsg.setUserFlag(true);
connMsg.setUsername(TEST_USER + "_fake");
m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, false, new PermitAllAuthorizator());
m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, false, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
//Exercise
m_processor.processConnect(m_session, connMsg);
@@ -289,7 +294,7 @@ public class ProtocolProcessorTest {
@Test
public void acceptAnonymousClient() {
connMsg.setClientID("123");
m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, true, new PermitAllAuthorizator());
m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
//Exercise
m_processor.processConnect(m_session, connMsg);
@@ -376,7 +381,7 @@ public class ProtocolProcessorTest {
//simulate a connect that register a clientID to an IoSession
subs.init(new MemoryStorageService());
m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator());
m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
ConnectMessage connectMessage = new ConnectMessage();
connectMessage.setProcotolVersion((byte)3);
connectMessage.setClientID(FAKE_CLIENT_ID);
@@ -420,7 +425,7 @@ public class ProtocolProcessorTest {
//simulate a connect that register a clientID to an IoSession
subs.init(new MemoryStorageService());
m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator());
m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
MockReceiverChannel firstReceiverSession = new MockReceiverChannel();
ConnectMessage connectMessage = new ConnectMessage();
@@ -559,7 +564,7 @@ public class ProtocolProcessorTest {
subs.init(new MemoryStorageService());
//simulate a connect that register a clientID to an IoSession
m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator());
m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
ConnectMessage connectMessage = new ConnectMessage();
connectMessage.setClientID(FAKE_PUBLISHER_ID);
connectMessage.setProcotolVersion((byte)3);
@@ -598,7 +603,7 @@ public class ProtocolProcessorTest {
ByteBuffer.wrap("Hello".getBytes()), true, FAKE_PUBLISHER_ID, 120);
m_storageService.storePublishForFuture(retainedMessage);
m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator());
m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
ConnectMessage connectMessage = new ConnectMessage();
connectMessage.setClientID(FAKE_PUBLISHER_ID);
connectMessage.setProcotolVersion((byte)3);
@@ -617,7 +622,7 @@ public class ProtocolProcessorTest {
List<Subscription> inactiveSubscriptions = Arrays.asList(inactiveSub);
when(mockedSubscriptions.matches(eq("/topic"))).thenReturn(inactiveSubscriptions);
m_processor = new ProtocolProcessor();
m_processor.init(mockedSubscriptions, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator());
m_processor.init(mockedSubscriptions, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
//Exercise
ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes());
@@ -642,7 +647,7 @@ public class ProtocolProcessorTest {
List<Subscription> inactiveSubscriptions = Arrays.asList(inactiveSub);
when(mockedSubscriptions.matches(eq("/topic"))).thenReturn(inactiveSubscriptions);
m_processor = new ProtocolProcessor();
m_processor.init(mockedSubscriptions, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator());
m_processor.init(mockedSubscriptions, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
//Exercise
ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes());
@@ -718,7 +723,7 @@ public class ProtocolProcessorTest {
publishedForwarded.add(new PublishEvent(topic, qos, message, retained, clientId, messageID));
}
};
processor.init(subscriptions, memoryMessageStore, null, null, true, null);
processor.init(subscriptions, memoryMessageStore, null, null, true, null, NO_OBSERVERS_INTERCEPTOR);
//Exercise
processor.forward2Subscribers(forwardPublish);