- * @param
+ * @param The type of message this bus consumes
+ * @param The publication commands this bus supports depend on P
*/
public abstract class AbstractSyncAsyncMessageBus
extends AbstractPubSubSupport implements IMessageBus {
@@ -36,14 +37,20 @@ protected AbstractSyncAsyncMessageBus(IBusConfiguration configuration) {
// configure asynchronous message dispatch
Feature.AsynchronousMessageDispatch asyncDispatch = configuration.getFeature(Feature.AsynchronousMessageDispatch.class);
+ if(asyncDispatch == null){
+ configuration.handleError(ConfigurationError.Missing(Feature.AsynchronousMessageDispatch.class));
+ }
pendingMessages = asyncDispatch.getPendingMessages();
dispatchers = new ArrayList(asyncDispatch.getNumberOfMessageDispatchers());
initDispatcherThreads(asyncDispatch);
// configure asynchronous handler invocation
Feature.AsynchronousHandlerInvocation asyncInvocation = configuration.getFeature(Feature.AsynchronousHandlerInvocation.class);
+ if(asyncInvocation == null){
+ configuration.handleError(ConfigurationError.Missing(Feature.AsynchronousHandlerInvocation.class));
+ }
this.executor = asyncInvocation.getExecutor();
- getRuntime().add(BusRuntime.Properties.AsynchronousHandlerExecutor, executor);
+ getRuntime().add(Properties.Handler.AsynchronousHandlerExecutor, executor);
}
@@ -117,9 +124,4 @@ public boolean hasPendingMessages() {
return pendingMessages.size() > 0;
}
- @Override
- public Executor getExecutor() {
- return executor;
- }
-
}
diff --git a/src/main/java/net/engio/mbassy/bus/BusFactory.java b/src/main/java/net/engio/mbassy/bus/BusFactory.java
index 683a84cd..bd014644 100644
--- a/src/main/java/net/engio/mbassy/bus/BusFactory.java
+++ b/src/main/java/net/engio/mbassy/bus/BusFactory.java
@@ -16,8 +16,6 @@ public class BusFactory {
* Create a message bus supporting only synchronous message publication.
* All message publications will run in the calling thread, no bus internal
* multi-threading will occur.
- *
- * @return
*/
public static SyncMessageBus SynchronousOnly(){
BusConfiguration syncPubSubCfg = new BusConfiguration();
@@ -26,12 +24,9 @@ public static SyncMessageBus SynchronousOnly(){
}
/**
- * Create a message bus supporting synchronous and asynchronous message publication.
+ * Create a message bus with support for synchronous and asynchronous message publication.
* Asynchronous message publication will be handled by a single thread such that FIFO
* order of message processing is guaranteed.
- *
- *
- * @return
*/
public static IMessageBus AsynchronousSequentialFIFO(){
BusConfiguration asyncFIFOConfig = new BusConfiguration();
diff --git a/src/main/java/net/engio/mbassy/bus/BusRuntime.java b/src/main/java/net/engio/mbassy/bus/BusRuntime.java
index 047edb09..17c2f43a 100644
--- a/src/main/java/net/engio/mbassy/bus/BusRuntime.java
+++ b/src/main/java/net/engio/mbassy/bus/BusRuntime.java
@@ -18,13 +18,6 @@
*/
public class BusRuntime {
- public static class Properties{
-
- public static final String ErrorHandlers = "error.handlers";
- public static final String AsynchronousHandlerExecutor = "handler.async.executor";
-
- }
-
private PubSubSupport provider;
private Map properties = new HashMap();
diff --git a/src/main/java/net/engio/mbassy/bus/MBassador.java b/src/main/java/net/engio/mbassy/bus/MBassador.java
index ec249f42..aa140a7b 100644
--- a/src/main/java/net/engio/mbassy/bus/MBassador.java
+++ b/src/main/java/net/engio/mbassy/bus/MBassador.java
@@ -12,15 +12,16 @@
public class MBassador extends AbstractSyncAsyncMessageBus> implements IMessageBus> {
+
public MBassador(IBusConfiguration configuration) {
super(configuration);
}
public MBassador(){
- super(new BusConfiguration()
- .addFeature(Feature.SyncPubSub.Default())
- .addFeature(Feature.AsynchronousHandlerInvocation.Default())
- .addFeature(Feature.AsynchronousMessageDispatch.Default()));
+ this(new BusConfiguration()
+ .addFeature(Feature.SyncPubSub.Default())
+ .addFeature(Feature.AsynchronousHandlerInvocation.Default())
+ .addFeature(Feature.AsynchronousMessageDispatch.Default()));
}
@@ -47,7 +48,7 @@ public void publish(T message) {
handlePublicationError(new PublicationError()
.setMessage("Error during publication of message")
.setCause(e)
- .setPublishedObject(message));
+ .setPublishedMessage(message));
}
}
diff --git a/src/main/java/net/engio/mbassy/bus/SyncMessageBus.java b/src/main/java/net/engio/mbassy/bus/SyncMessageBus.java
index 5cd4892f..9d56cf16 100644
--- a/src/main/java/net/engio/mbassy/bus/SyncMessageBus.java
+++ b/src/main/java/net/engio/mbassy/bus/SyncMessageBus.java
@@ -28,7 +28,7 @@ public void publish(T message) {
handlePublicationError(new PublicationError()
.setMessage("Error during publication of message")
.setCause(e)
- .setPublishedObject(message));
+ .setPublishedMessage(message));
}
}
diff --git a/src/main/java/net/engio/mbassy/bus/common/GenericMessagePublicationSupport.java b/src/main/java/net/engio/mbassy/bus/common/GenericMessagePublicationSupport.java
index d55a8e15..e074cd3b 100644
--- a/src/main/java/net/engio/mbassy/bus/common/GenericMessagePublicationSupport.java
+++ b/src/main/java/net/engio/mbassy/bus/common/GenericMessagePublicationSupport.java
@@ -4,9 +4,10 @@
/**
* This interface is meant to be implemented by different bus implementations to offer a consistent way
- * to plugin different flavors of message publication.
+ * to plugin different methods of message publication.
*
- * The parametrization of the IPostCommand influences which publication flavours are available.
+ * The parametrization of the IPostCommand influences which publication methods (asynchronous, synchronous or
+ * conditional etc.) are available.
*
*/
public interface GenericMessagePublicationSupport extends PubSubSupport, ErrorHandlingSupport{
@@ -15,10 +16,10 @@ public interface GenericMessagePublicationSupport
@Override
P post(T message);
- /**
- * Get the executor service that is used for asynchronous message publications.
- * The executor is passed to the message bus at creation time.
- *
- * Note: The executor can be obtained from the run time. See
- * @return
- */
- @Deprecated
- Executor getExecutor();
-
/**
* Check whether any asynchronous message publications are pending to be processed
*
diff --git a/src/main/java/net/engio/mbassy/bus/common/Properties.java b/src/main/java/net/engio/mbassy/bus/common/Properties.java
new file mode 100644
index 00000000..e152d8fa
--- /dev/null
+++ b/src/main/java/net/engio/mbassy/bus/common/Properties.java
@@ -0,0 +1,25 @@
+package net.engio.mbassy.bus.common;
+
+/**
+ * A collection of properties commonly used by different parts of the library.
+ *
+ * @author bennidi
+ * Date: 22.02.15
+ */
+public final class Properties {
+
+ public static final class Handler {
+
+ public static final String PublicationError = "bus.handlers.error";
+ public static final String AsynchronousHandlerExecutor = "bus.handlers.async-executor";
+ }
+
+ public static final class Common {
+
+ public static final String Id = "bus.id";
+ }
+
+
+
+
+}
diff --git a/src/main/java/net/engio/mbassy/bus/config/BusConfiguration.java b/src/main/java/net/engio/mbassy/bus/config/BusConfiguration.java
index 561ad1fe..93523cef 100644
--- a/src/main/java/net/engio/mbassy/bus/config/BusConfiguration.java
+++ b/src/main/java/net/engio/mbassy/bus/config/BusConfiguration.java
@@ -1,6 +1,8 @@
package net.engio.mbassy.bus.config;
import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
import java.util.Map;
/**
@@ -8,26 +10,51 @@
*/
public class BusConfiguration implements IBusConfiguration {
- // the registered features
- private Map, Feature> features = new HashMap, Feature>();
+ // the registered properties
+ private final Map properties = new HashMap();
+ private final List errorHandlerList = new LinkedList();
public BusConfiguration() {
super();
}
+ @Override
+ public IBusConfiguration setProperty(String name, Object value) {
+ properties.put(name, value);
+ return this;
+ }
+
+ @Override
+ public T getProperty(String name, T defaultValue) {
+ return properties.containsKey(name) ? (T) properties.get(name) : defaultValue;
+ }
+
+ @Override
+ public boolean hasProperty(String name) {
+ return properties.containsKey(name);
+ }
+
@Override
public T getFeature(Class feature) {
- return (T)features.get(feature);
+ return (T) properties.get(feature);
}
@Override
public IBusConfiguration addFeature(Feature feature) {
- features.put(feature.getClass(), feature);
+ properties.put(feature.getClass(), feature);
+ return this;
+ }
+
+ @Override
+ public IBusConfiguration addConfigurationErrorHandler(ConfigurationErrorHandler handler) {
+ errorHandlerList.add(handler);
return this;
}
@Override
- public IBusConfiguration addErrorHandler(ConfigurationErrorHandler handler) {
- return null; // TODO: implement configuration validation
+ public void handleError(ConfigurationError error) {
+ for(ConfigurationErrorHandler errorHandler : errorHandlerList){
+ errorHandler.handle(error);
+ }
}
}
diff --git a/src/main/java/net/engio/mbassy/bus/config/ConfigurationError.java b/src/main/java/net/engio/mbassy/bus/config/ConfigurationError.java
index 2e4ed420..33092c36 100644
--- a/src/main/java/net/engio/mbassy/bus/config/ConfigurationError.java
+++ b/src/main/java/net/engio/mbassy/bus/config/ConfigurationError.java
@@ -1,10 +1,30 @@
package net.engio.mbassy.bus.config;
/**
- * Todo: Add javadoc
+ * Configuration errors represent specific misconfigurations of features in a {@link net.engio.mbassy.bus.config.IBusConfiguration}
*
* @author bennidi
* Date: 8/29/14
*/
public class ConfigurationError {
+
+ private Class featureType;
+ private Feature feature;
+ private String message;
+
+ public ConfigurationError(Class featureType, Feature feature, String message) {
+ this.featureType = featureType;
+ this.feature = feature;
+ this.message = message;
+ }
+
+ public static ConfigurationError Missing(Class featureType){
+ return new ConfigurationError(featureType, null, "An expected feature was missing. Use addFeature() in IBusConfiguration to add features.");
+ }
+
+ @Override
+ public String toString() {
+ return "Error for " + featureType + ":" + message +
+ ", (" + feature + ")";
+ }
}
diff --git a/src/main/java/net/engio/mbassy/bus/config/ConfigurationErrorHandler.java b/src/main/java/net/engio/mbassy/bus/config/ConfigurationErrorHandler.java
index 7554bdd9..8952b622 100644
--- a/src/main/java/net/engio/mbassy/bus/config/ConfigurationErrorHandler.java
+++ b/src/main/java/net/engio/mbassy/bus/config/ConfigurationErrorHandler.java
@@ -1,10 +1,16 @@
package net.engio.mbassy.bus.config;
/**
- * Todo: Add javadoc
+ * Respond to a {@link net.engio.mbassy.bus.config.ConfigurationError} with any kind of action.
*
* @author bennidi
* Date: 8/29/14
*/
public interface ConfigurationErrorHandler {
+
+ /**
+ * Called when a misconfiguration is detected on a {@link net.engio.mbassy.bus.config.IBusConfiguration}
+ * @param error The error that represents the detected misconfiguration.
+ */
+ void handle(ConfigurationError error);
}
diff --git a/src/main/java/net/engio/mbassy/bus/config/Feature.java b/src/main/java/net/engio/mbassy/bus/config/Feature.java
index 5d0cec5c..95bfb5b7 100644
--- a/src/main/java/net/engio/mbassy/bus/config/Feature.java
+++ b/src/main/java/net/engio/mbassy/bus/config/Feature.java
@@ -96,9 +96,8 @@ public static final AsynchronousHandlerInvocation Default(){
return Default(numberOfCores, numberOfCores * 2);
}
- public static final AsynchronousHandlerInvocation Default(int initialCoreThreads, int maximumCoreThreads){
- int numberOfCores = Runtime.getRuntime().availableProcessors();
- return new AsynchronousHandlerInvocation().setExecutor(new ThreadPoolExecutor(initialCoreThreads, maximumCoreThreads, 1,
+ public static final AsynchronousHandlerInvocation Default(int minThreadCount, int maxThreadCount){
+ return new AsynchronousHandlerInvocation().setExecutor(new ThreadPoolExecutor(minThreadCount, maxThreadCount, 1,
TimeUnit.MINUTES, new LinkedBlockingQueue(), MessageHandlerThreadFactory));
}
diff --git a/src/main/java/net/engio/mbassy/bus/config/IBusConfiguration.java b/src/main/java/net/engio/mbassy/bus/config/IBusConfiguration.java
index 51581c24..e47d6b0d 100644
--- a/src/main/java/net/engio/mbassy/bus/config/IBusConfiguration.java
+++ b/src/main/java/net/engio/mbassy/bus/config/IBusConfiguration.java
@@ -12,19 +12,57 @@
*/
public interface IBusConfiguration{
+ /**
+ * Set a property which will be read by the message bus constructor. Existing value will be overwritten.
+ * Null values are supported (checking for existence of property will return true even if set to null).
+ *
+ * @param name The name of the property. Note: Each implementation may support different properties.
+ * @param value The value of the property.
+ * @return A reference to this bus configuration.
+ */
+ IBusConfiguration setProperty(String name, Object value);
+
+ /**
+ * Read a property from this configuration.
+ *
+ * @param name The name of the property to be read.
+ * @param defaultValue The value to be returned if property was not found
+ * @param The type of property
+ * @return The value associated with the given property name or defaultValue if not present
+ */
+ T getProperty(String name, T defaultValue);
+
+ /**
+ * Check whether a property has been set.
+ *
+ * @return true if property was set (even if set to null)
+ * false otherwise
+ */
+ boolean hasProperty(String name);
+
+
/**
* Get a registered feature by its type (class).
*
- * @param feature
- * @param
- * @return
*/
T getFeature(Class feature);
+ /**
+ * Add a feature to the given configuration, replacing any existing feature of the same type.
+ *
+ * @param feature The feature to add
+ * @return A reference to this bus configuration.
+ */
IBusConfiguration addFeature(Feature feature);
- IBusConfiguration addErrorHandler(ConfigurationErrorHandler handler);
-
+ /**
+ * Add a handler that is called when a misconfiguration is detected.
+ */
+ IBusConfiguration addConfigurationErrorHandler(ConfigurationErrorHandler handler);
+ /**
+ * Calls all ConfigurationErrorHandlers
+ */
+ void handleError(ConfigurationError error);
}
diff --git a/src/main/java/net/engio/mbassy/bus/error/IPublicationErrorHandler.java b/src/main/java/net/engio/mbassy/bus/error/IPublicationErrorHandler.java
index 1bbfd31a..c5153914 100644
--- a/src/main/java/net/engio/mbassy/bus/error/IPublicationErrorHandler.java
+++ b/src/main/java/net/engio/mbassy/bus/error/IPublicationErrorHandler.java
@@ -26,6 +26,16 @@ public interface IPublicationErrorHandler {
*/
static final class ConsoleLogger implements IPublicationErrorHandler {
+ private final boolean printStackTrace;
+
+ public ConsoleLogger() {
+ this(false);
+ }
+
+ public ConsoleLogger(boolean printStackTrace) {
+ this.printStackTrace = printStackTrace;
+ }
+
/**
* {@inheritDoc}
*/
@@ -36,7 +46,7 @@ public void handleError(final PublicationError error) {
System.out.println(error);
// Printout the stacktrace from the cause.
- if (error.getCause() != null) {
+ if (printStackTrace && error.getCause() != null) {
error.getCause().printStackTrace();
}
}
diff --git a/src/main/java/net/engio/mbassy/bus/error/PublicationError.java b/src/main/java/net/engio/mbassy/bus/error/PublicationError.java
index dc985555..72bc3400 100644
--- a/src/main/java/net/engio/mbassy/bus/error/PublicationError.java
+++ b/src/main/java/net/engio/mbassy/bus/error/PublicationError.java
@@ -1,6 +1,7 @@
package net.engio.mbassy.bus.error;
import net.engio.mbassy.bus.IMessagePublication;
+import net.engio.mbassy.subscription.SubscriptionContext;
import java.lang.reflect.Method;
@@ -21,7 +22,7 @@ public class PublicationError{
private String message;
private Method handler;
private Object listener;
- private Object publishedObject;
+ private Object publishedMessage;
/**
@@ -43,7 +44,7 @@ public PublicationError(final Throwable cause,
this.message = message;
this.handler = handler;
this.listener = listener;
- this.publishedObject = publishedObject;
+ this.publishedMessage = publishedObject;
}
public PublicationError(final Throwable cause,
@@ -51,10 +52,16 @@ public PublicationError(final Throwable cause,
final IMessagePublication publication) {
this.cause = cause;
this.message = message;
- this.publishedObject = publication != null ? publication.getMessage() : null;
+ this.publishedMessage = publication != null ? publication.getMessage() : null;
}
-
+ public PublicationError(final Throwable cause,
+ final String message,
+ final SubscriptionContext context) {
+ this.cause = cause;
+ this.message = message;
+ this.handler = context.getHandler().getMethod();
+ }
/**
@@ -109,12 +116,12 @@ public PublicationError setListener(Object listener) {
return this;
}
- public Object getPublishedObject() {
- return publishedObject;
+ public Object getPublishedMessage() {
+ return publishedMessage;
}
- public PublicationError setPublishedObject(Object publishedObject) {
- this.publishedObject = publishedObject;
+ public PublicationError setPublishedMessage(Object publishedMessage) {
+ this.publishedMessage = publishedMessage;
return this;
}
@@ -134,7 +141,7 @@ public String toString() {
newLine +
"\tlistener=" + listener +
newLine +
- "\tpublishedObject=" + publishedObject +
+ "\tpublishedMessage=" + publishedMessage +
'}';
}
}
diff --git a/src/main/java/net/engio/mbassy/common/ReflectionUtils.java b/src/main/java/net/engio/mbassy/common/ReflectionUtils.java
index dc5515c7..cf623672 100644
--- a/src/main/java/net/engio/mbassy/common/ReflectionUtils.java
+++ b/src/main/java/net/engio/mbassy/common/ReflectionUtils.java
@@ -41,7 +41,6 @@ public static List getMethods( IPredicate condition, Class ta
*
* @param overridingMethod
* @param subclass
- * @return
*/
public static Method getOverridingMethod( final Method overridingMethod, final Class subclass ) {
Class current = subclass;
diff --git a/src/main/java/net/engio/mbassy/dispatch/AsynchronousHandlerInvocation.java b/src/main/java/net/engio/mbassy/dispatch/AsynchronousHandlerInvocation.java
index 7b80830c..8ad6c1a0 100644
--- a/src/main/java/net/engio/mbassy/dispatch/AsynchronousHandlerInvocation.java
+++ b/src/main/java/net/engio/mbassy/dispatch/AsynchronousHandlerInvocation.java
@@ -1,6 +1,6 @@
package net.engio.mbassy.dispatch;
-import net.engio.mbassy.bus.BusRuntime;
+import net.engio.mbassy.bus.common.Properties;
import net.engio.mbassy.subscription.AbstractSubscriptionContextAware;
import java.util.concurrent.ExecutorService;
@@ -20,7 +20,7 @@ public class AsynchronousHandlerInvocation extends AbstractSubscriptionContextAw
public AsynchronousHandlerInvocation(IHandlerInvocation delegate) {
super(delegate.getContext());
this.delegate = delegate;
- this.executor = delegate.getContext().getRuntime().get(BusRuntime.Properties.AsynchronousHandlerExecutor);
+ this.executor = delegate.getContext().getRuntime().get(Properties.Handler.AsynchronousHandlerExecutor);
}
/**
diff --git a/src/main/java/net/engio/mbassy/dispatch/FilteredMessageDispatcher.java b/src/main/java/net/engio/mbassy/dispatch/FilteredMessageDispatcher.java
index d46bdda5..fe53c6a4 100644
--- a/src/main/java/net/engio/mbassy/dispatch/FilteredMessageDispatcher.java
+++ b/src/main/java/net/engio/mbassy/dispatch/FilteredMessageDispatcher.java
@@ -17,7 +17,7 @@ public final class FilteredMessageDispatcher extends DelegatingMessageDispatcher
public FilteredMessageDispatcher(IMessageDispatcher dispatcher) {
super(dispatcher);
- this.filter = dispatcher.getContext().getHandlerMetadata().getFilter();
+ this.filter = dispatcher.getContext().getHandler().getFilter();
}
private boolean passesFilter(Object message) {
@@ -26,7 +26,7 @@ private boolean passesFilter(Object message) {
return true;
} else {
for (IMessageFilter aFilter : filter) {
- if (!aFilter.accepts(message, getContext().getHandlerMetadata())) {
+ if (!aFilter.accepts(message, getContext())) {
return false;
}
}
diff --git a/src/main/java/net/engio/mbassy/dispatch/ReflectiveHandlerInvocation.java b/src/main/java/net/engio/mbassy/dispatch/ReflectiveHandlerInvocation.java
index 796c3d6d..b4430a26 100644
--- a/src/main/java/net/engio/mbassy/dispatch/ReflectiveHandlerInvocation.java
+++ b/src/main/java/net/engio/mbassy/dispatch/ReflectiveHandlerInvocation.java
@@ -46,6 +46,6 @@ protected void invokeHandler(final Object message, final Object listener, Method
*/
@Override
public void invoke(final Object listener, final Object message){
- invokeHandler(message, listener, getContext().getHandlerMetadata().getHandler());
+ invokeHandler(message, listener, getContext().getHandler().getMethod());
}
}
diff --git a/src/main/java/net/engio/mbassy/dispatch/el/ElFilter.java b/src/main/java/net/engio/mbassy/dispatch/el/ElFilter.java
index d15b0139..90425ba1 100644
--- a/src/main/java/net/engio/mbassy/dispatch/el/ElFilter.java
+++ b/src/main/java/net/engio/mbassy/dispatch/el/ElFilter.java
@@ -1,30 +1,36 @@
package net.engio.mbassy.dispatch.el;
+import net.engio.mbassy.bus.error.IPublicationErrorHandler;
+import net.engio.mbassy.bus.error.PublicationError;
import net.engio.mbassy.listener.IMessageFilter;
import net.engio.mbassy.listener.MessageHandler;
+import net.engio.mbassy.subscription.SubscriptionContext;
import javax.el.ExpressionFactory;
import javax.el.ValueExpression;
-/*****************************************************************************
- * A filter that will use a expression from the handler annotation and
+/**
+ * A filter that will use a expression from the handler annotation and
* parse it as EL.
- ****************************************************************************/
-
+ *
+ * Accepts a message if the associated EL expression evaluates to true
+ */
public class ElFilter implements IMessageFilter {
// thread-safe initialization of EL factory singleton
- public static final class ExpressionFactoryHolder{
+ public static final class ExpressionFactoryHolder {
// if runtime exception is thrown, this will
public static final ExpressionFactory ELFactory = getELFactory();
- /*************************************************************************
+ /**
+ * **********************************************************************
* Get an implementation of the ExpressionFactory. This uses the
* Java service lookup mechanism to find a proper implementation.
- * If none if available we do not support EL filters.
- ************************************************************************/
- private static final ExpressionFactory getELFactory(){
+ * If none is available we do not support EL filters.
+ * **********************************************************************
+ */
+ private static final ExpressionFactory getELFactory() {
try {
return ExpressionFactory.newInstance();
} catch (RuntimeException e) {
@@ -33,40 +39,39 @@ private static final ExpressionFactory getELFactory(){
}
}
- public static final boolean isELAvailable(){
+ public static final boolean isELAvailable() {
return ExpressionFactoryHolder.ELFactory != null;
}
- public static final ExpressionFactory ELFactory(){
+ public static final ExpressionFactory ELFactory() {
return ExpressionFactoryHolder.ELFactory;
}
- /**
- * Accepts a message if the associated EL expression of the message handler resolves to 'true'
- *
- * @param message the message to be handled by the handler
- * @param metadata the metadata object which describes the message handler
- * @return
- */
- @Override
- public boolean accepts(Object message, MessageHandler metadata) {
- String expression = metadata.getCondition();
- StandardELResolutionContext context = new StandardELResolutionContext(message);
- return evalExpression(expression, context);
- }
- private boolean evalExpression(String expression, StandardELResolutionContext context) {
- ValueExpression ve = ELFactory().createValueExpression(context, expression, Boolean.class);
- try{
- Object result = ve.getValue(context);
- return (Boolean)result;
- }
- catch(Throwable exception){
- // TODO: BusRuntime should be available in this filter to propagate resolution errors
- // -> this is generally a good feature for filters
+ @Override
+ public boolean accepts(Object message, final SubscriptionContext context) {
+ final MessageHandler metadata = context.getHandler();
+ String expression = metadata.getCondition();
+ StandardELResolutionContext resolutionContext = new StandardELResolutionContext(message);
+ return evalExpression(expression, resolutionContext, context, message);
+ }
+
+ private boolean evalExpression(final String expression,
+ final StandardELResolutionContext resolutionContext,
+ final SubscriptionContext context,
+ final Object message) {
+ ValueExpression ve = ELFactory().createValueExpression(resolutionContext, expression, Boolean.class);
+ try {
+ Object result = ve.getValue(resolutionContext);
+ return (Boolean) result;
+ } catch (Throwable exception) {
+ PublicationError publicationError = new PublicationError(exception, "Error while evaluating EL expression on message", context)
+ .setPublishedMessage(message);
+ for (IPublicationErrorHandler errorHandler : context.getErrorHandlers()) {
+ errorHandler.handleError(publicationError);
+ }
return false;
- //throw new IllegalStateException("A handler uses an EL filter but the output is not \"true\" or \"false\".");
}
- }
+ }
}
diff --git a/src/main/java/net/engio/mbassy/listener/Filter.java b/src/main/java/net/engio/mbassy/listener/Filter.java
index 0251aaf8..101bda01 100644
--- a/src/main/java/net/engio/mbassy/listener/Filter.java
+++ b/src/main/java/net/engio/mbassy/listener/Filter.java
@@ -10,8 +10,6 @@
* It references a class that implements the IMessageFilter interface.
* The filter will be used to check whether a message should be delivered
* to the listener or not.
- *
- *
*
* @author bennidi
* Date: 2/14/12
@@ -22,9 +20,7 @@
/**
* The class that implements the filter.
- * Note: A filter always needs to provide a non-arg constructor
- *
- * @return
+ * IMPORTANT: A filter always needs to provide a non-arg constructor
*/
Class value();
}
diff --git a/src/main/java/net/engio/mbassy/listener/Filters.java b/src/main/java/net/engio/mbassy/listener/Filters.java
index db47bd10..2f69264a 100644
--- a/src/main/java/net/engio/mbassy/listener/Filters.java
+++ b/src/main/java/net/engio/mbassy/listener/Filters.java
@@ -1,5 +1,7 @@
package net.engio.mbassy.listener;
+import net.engio.mbassy.subscription.SubscriptionContext;
+
/**
* A set of standard filters for common use cases.
*
@@ -21,7 +23,8 @@ public class Filters {
public static final class RejectSubtypes implements IMessageFilter {
@Override
- public boolean accepts(Object event, MessageHandler metadata) {
+ public boolean accepts(final Object event, final SubscriptionContext context) {
+ final MessageHandler metadata = context.getHandler();
for (Class handledMessage : metadata.getHandledMessages()) {
if (handledMessage.equals(event.getClass())) {
return true;
@@ -40,7 +43,8 @@ public boolean accepts(Object event, MessageHandler metadata) {
public static final class SubtypesOnly implements IMessageFilter{
@Override
- public boolean accepts(Object message, MessageHandler metadata) {
+ public boolean accepts(final Object message, final SubscriptionContext context) {
+ final MessageHandler metadata = context.getHandler();
for(Class acceptedClasses : metadata.getHandledMessages()){
if(acceptedClasses.isAssignableFrom(message.getClass())
&& ! acceptedClasses.equals(message.getClass()))
diff --git a/src/main/java/net/engio/mbassy/listener/IMessageFilter.java b/src/main/java/net/engio/mbassy/listener/IMessageFilter.java
index 12efcfa7..6aa3f10e 100644
--- a/src/main/java/net/engio/mbassy/listener/IMessageFilter.java
+++ b/src/main/java/net/engio/mbassy/listener/IMessageFilter.java
@@ -1,5 +1,7 @@
package net.engio.mbassy.listener;
+import net.engio.mbassy.subscription.SubscriptionContext;
+
/**
* Message filters can be used to control what messages are delivered to a specific message handler.
* Filters are attached to message handler using the @Listener annotation.
@@ -8,6 +10,7 @@
*
* Example:
*
+ * {@code
* @Lister
* @Filters(Urlfilter.class)
* public void someHandler(String message){...}
@@ -22,6 +25,7 @@
* bus.post("www.stackoverflow.com"); // will not be delivered
*
* NOTE: A message filter must provide a no-arg constructor!!!
+ * }
*
* @author bennidi
* Date: 2/8/12
@@ -29,12 +33,12 @@
public interface IMessageFilter {
/**
- * Check the message for whatever criteria
+ * Check whether the message matches some criteria
*
- * @param message the message to be handled by the handler
- * @param metadata the metadata object which describes the message handler
+ * @param message The message to be handled by the handler
+ * @param context The context object containing a description of the message handler and the bus environment
* @return true: if the message matches the criteria and should be delivered to the handler
* false: otherwise
*/
- boolean accepts(M message, MessageHandler metadata);
+ boolean accepts(M message, SubscriptionContext context);
}
diff --git a/src/main/java/net/engio/mbassy/listener/Listener.java b/src/main/java/net/engio/mbassy/listener/Listener.java
index 3b039516..eb3655d1 100644
--- a/src/main/java/net/engio/mbassy/listener/Listener.java
+++ b/src/main/java/net/engio/mbassy/listener/Listener.java
@@ -18,10 +18,9 @@
public @interface Listener {
/**
- * By default, references to message listeners are weak to eliminate risks of memory leaks.
+ * BY DEFAULT, REFERENCES to message listeners ARE WEAK to eliminate risks of memory leaks.
* It is possible to use strong references instead.
*
- * @return
*/
References references() default References.Weak;
diff --git a/src/main/java/net/engio/mbassy/listener/MessageHandler.java b/src/main/java/net/engio/mbassy/listener/MessageHandler.java
index eb22bb40..bb533797 100644
--- a/src/main/java/net/engio/mbassy/listener/MessageHandler.java
+++ b/src/main/java/net/engio/mbassy/listener/MessageHandler.java
@@ -41,9 +41,12 @@ public static final class Properties{
* @param filter The set of preconfigured filters if any
* @param listenerConfig The listener metadata
* @return A map of properties initialized from the given parameters that will conform to the requirements of the
- * {@link MessageHandler} constructor. See {@see MessageHandler.validate()} for more details.
+ * {@link MessageHandler} constructor.
*/
- public static final Map Create(Method handler, Handler handlerConfig, IMessageFilter[] filter, MessageListener listenerConfig){
+ public static final Map Create(Method handler,
+ Handler handlerConfig,
+ IMessageFilter[] filter,
+ MessageListener listenerConfig){
if(handler == null){
throw new IllegalArgumentException("The message handler configuration may not be null");
}
@@ -133,6 +136,7 @@ public MessageHandler(Map properties){
}
private void validate(Map properties){
+ // define expected types of known properties
Object[][] expectedProperties = new Object[][]{
new Object[]{Properties.HandlerMethod, Method.class },
new Object[]{Properties.Priority, Integer.class },
@@ -145,13 +149,12 @@ private void validate(Map properties){
new Object[]{Properties.Listener, MessageListener.class },
new Object[]{Properties.AcceptSubtypes, Boolean.class }
};
+ // ensure types match
for(Object[] property : expectedProperties){
if (properties.get(property[0]) == null || !((Class)property[1]).isAssignableFrom(properties.get(property[0]).getClass()))
throw new IllegalArgumentException("Property " + property[0] + " was expected to be not null and of type " + property[1]
+ " but was: " + properties.get(property[0]));
}
-
-
}
public A getAnnotation(Class annotationType){
@@ -182,7 +185,7 @@ public int getPriority() {
return priority;
}
- public Method getHandler() {
+ public Method getMethod() {
return handler;
}
diff --git a/src/main/java/net/engio/mbassy/subscription/Subscription.java b/src/main/java/net/engio/mbassy/subscription/Subscription.java
index 8677b8ac..4b75ff76 100644
--- a/src/main/java/net/engio/mbassy/subscription/Subscription.java
+++ b/src/main/java/net/engio/mbassy/subscription/Subscription.java
@@ -36,35 +36,28 @@ public class Subscription {
}
/**
- * Check whether this subscription manages a message handler of the given message listener class
- *
- * @param listener
- * @return
+ * Check whether this subscription manages a message handler of the given listener class.
*/
public boolean belongsTo(Class listener){
- return context.getHandlerMetadata().isFromListener(listener);
+ return context.getHandler().isFromListener(listener);
}
/**
- * Check whether this subscriptions manages the given listener instance
- * @param listener
- * @return
+ * Check whether this subscriptions manages the given listener instance.
*/
public boolean contains(Object listener){
return listeners.contains(listener);
}
/**
- * Check whether this subscription manages a message handler
- * @param messageType
- * @return
+ * Check whether this subscription manages a specific message type.
*/
public boolean handlesMessageType(Class messageType) {
- return context.getHandlerMetadata().handlesMessage(messageType);
+ return context.getHandler().handlesMessage(messageType);
}
public Class[] getHandledMessageTypes(){
- return context.getHandlerMetadata().getHandledMessages();
+ return context.getHandler().getHandledMessages();
}
@@ -74,7 +67,7 @@ public void publish(IMessagePublication publication, Object message){
}
public int getPriority() {
- return context.getHandlerMetadata().getPriority();
+ return context.getHandler().getPriority();
}
diff --git a/src/main/java/net/engio/mbassy/subscription/SubscriptionContext.java b/src/main/java/net/engio/mbassy/subscription/SubscriptionContext.java
index 458378b6..59c0f8c9 100644
--- a/src/main/java/net/engio/mbassy/subscription/SubscriptionContext.java
+++ b/src/main/java/net/engio/mbassy/subscription/SubscriptionContext.java
@@ -19,33 +19,30 @@
public class SubscriptionContext implements RuntimeProvider {
// the handler's metadata -> for each handler in a listener, a unique subscription context is created
- private final MessageHandler handlerMetadata;
+ private final MessageHandler handler;
// error handling is first-class functionality
private final Collection errorHandlers;
- private BusRuntime runtime;
+ private final BusRuntime runtime;
- public SubscriptionContext(BusRuntime runtime, MessageHandler handlerMetadata,
- Collection errorHandlers) {
+ public SubscriptionContext(final BusRuntime runtime, final MessageHandler handler,
+ final Collection errorHandlers) {
this.runtime = runtime;
- this.handlerMetadata = handlerMetadata;
+ this.handler = handler;
this.errorHandlers = errorHandlers;
}
/**
* Get the meta data that specifies the characteristics of the message handler
* that is associated with this context
- *
- * @return
*/
- public MessageHandler getHandlerMetadata() {
- return handlerMetadata;
+ public MessageHandler getHandler() {
+ return handler;
}
/**
* Get the error handlers registered with the enclosing bus.
- * @return
*/
public Collection getErrorHandlers(){
return errorHandlers;
diff --git a/src/main/java/net/engio/mbassy/subscription/SubscriptionFactory.java b/src/main/java/net/engio/mbassy/subscription/SubscriptionFactory.java
index 4cc05767..ebb048a9 100644
--- a/src/main/java/net/engio/mbassy/subscription/SubscriptionFactory.java
+++ b/src/main/java/net/engio/mbassy/subscription/SubscriptionFactory.java
@@ -1,6 +1,7 @@
package net.engio.mbassy.subscription;
import net.engio.mbassy.bus.BusRuntime;
+import net.engio.mbassy.bus.common.Properties;
import net.engio.mbassy.bus.error.IPublicationErrorHandler;
import net.engio.mbassy.bus.error.MessageBusException;
import net.engio.mbassy.common.StrongConcurrentSet;
@@ -20,7 +21,7 @@ public class SubscriptionFactory {
public Subscription createSubscription(BusRuntime runtime, MessageHandler handlerMetadata) throws MessageBusException{
try {
- Collection errorHandlers = runtime.get(BusRuntime.Properties.ErrorHandlers);
+ Collection errorHandlers = runtime.get(Properties.Handler.PublicationError);
SubscriptionContext context = new SubscriptionContext(runtime, handlerMetadata, errorHandlers);
IHandlerInvocation invocation = buildInvocationForHandler(context);
IMessageDispatcher dispatcher = buildDispatcher(context, invocation);
@@ -34,10 +35,10 @@ public Subscription createSubscription(BusRuntime runtime, MessageHandler handle
protected IHandlerInvocation buildInvocationForHandler(SubscriptionContext context) throws Exception {
IHandlerInvocation invocation = createBaseHandlerInvocation(context);
- if(context.getHandlerMetadata().isSynchronized()){
+ if(context.getHandler().isSynchronized()){
invocation = new SynchronizedHandlerInvocation(invocation);
}
- if (context.getHandlerMetadata().isAsynchronous()) {
+ if (context.getHandler().isAsynchronous()) {
invocation = new AsynchronousHandlerInvocation(invocation);
}
return invocation;
@@ -45,17 +46,17 @@ protected IHandlerInvocation buildInvocationForHandler(SubscriptionContext conte
protected IMessageDispatcher buildDispatcher(SubscriptionContext context, IHandlerInvocation invocation) {
IMessageDispatcher dispatcher = new MessageDispatcher(context, invocation);
- if (context.getHandlerMetadata().isEnveloped()) {
+ if (context.getHandler().isEnveloped()) {
dispatcher = new EnvelopedMessageDispatcher(dispatcher);
}
- if (context.getHandlerMetadata().isFiltered()) {
+ if (context.getHandler().isFiltered()) {
dispatcher = new FilteredMessageDispatcher(dispatcher);
}
return dispatcher;
}
protected IHandlerInvocation createBaseHandlerInvocation(SubscriptionContext context) throws MessageBusException {
- Class invocation = context.getHandlerMetadata().getHandlerInvocation();
+ Class invocation = context.getHandler().getHandlerInvocation();
if(invocation.isMemberClass() && !Modifier.isStatic(invocation.getModifiers())){
throw new MessageBusException("The handler invocation must be top level class or nested STATIC inner class");
}
diff --git a/src/test/java/net/engio/mbassy/AllTests.java b/src/test/java/net/engio/mbassy/AllTests.java
index 9df884aa..70b5d9df 100644
--- a/src/test/java/net/engio/mbassy/AllTests.java
+++ b/src/test/java/net/engio/mbassy/AllTests.java
@@ -13,7 +13,7 @@
@Suite.SuiteClasses({
StrongConcurrentSetTest.class,
WeakConcurrentSetTest.class,
- MBassadorTest.class,
+ SyncAsyncTest.class,
SyncBusTest.MBassadorTest.class,
SyncBusTest.SyncMessageBusTest.class,
FilterTest.class,
diff --git a/src/test/java/net/engio/mbassy/ConditionalHandlers.java b/src/test/java/net/engio/mbassy/ConditionalHandlers.java
index 0e5f9d1c..945e864d 100644
--- a/src/test/java/net/engio/mbassy/ConditionalHandlers.java
+++ b/src/test/java/net/engio/mbassy/ConditionalHandlers.java
@@ -93,7 +93,7 @@ public void handleEnvelopedMessage(MessageEnvelope envelope) {
************************************************************************/
@Test
public void testSimpleStringCondition() throws Exception {
- MBassador bus = createBus(SyncAsync());
+ MBassador bus = createBus(SyncAsync(false));
bus.subscribe(new ConditionalMessageListener());
TestEvent message = new TestEvent("TEST", 0);
@@ -109,7 +109,7 @@ public void testSimpleStringCondition() throws Exception {
************************************************************************/
@Test
public void testSimpleNumberCondition() throws Exception {
- MBassador bus = new MBassador();
+ MBassador bus = createBus(SyncAsync(false));
bus.subscribe(new ConditionalMessageListener());
TestEvent message = new TestEvent("", 5);
@@ -124,7 +124,7 @@ public void testSimpleNumberCondition() throws Exception {
************************************************************************/
@Test
public void testHandleCombinedEL() throws Exception {
- MBassador bus = createBus(SyncAsync());
+ MBassador bus = createBus(SyncAsync(false));
bus.subscribe(new ConditionalMessageListener());
TestEvent message = new TestEvent("", 3);
@@ -139,7 +139,7 @@ public void testHandleCombinedEL() throws Exception {
************************************************************************/
@Test
public void testNotMatchingAnyCondition() throws Exception {
- MBassador bus = createBus(SyncAsync());
+ MBassador bus = createBus(SyncAsync(false));
bus.subscribe(new ConditionalMessageListener());
TestEvent message = new TestEvent("", 0);
@@ -153,7 +153,7 @@ public void testNotMatchingAnyCondition() throws Exception {
************************************************************************/
@Test
public void testHandleMethodAccessEL() throws Exception {
- MBassador bus = createBus(SyncAsync());
+ MBassador bus = createBus(SyncAsync(false));
bus.subscribe(new ConditionalMessageListener());
TestEvent message = new TestEvent("XYZ", 1);
diff --git a/src/test/java/net/engio/mbassy/CustomHandlerAnnotationTest.java b/src/test/java/net/engio/mbassy/CustomHandlerAnnotationTest.java
index 548f0596..0c6ad16e 100644
--- a/src/test/java/net/engio/mbassy/CustomHandlerAnnotationTest.java
+++ b/src/test/java/net/engio/mbassy/CustomHandlerAnnotationTest.java
@@ -4,6 +4,7 @@
import net.engio.mbassy.common.MessageBusTest;
import net.engio.mbassy.listener.*;
import net.engio.mbassy.subscription.MessageEnvelope;
+import net.engio.mbassy.subscription.SubscriptionContext;
import org.junit.Test;
import java.lang.annotation.*;
@@ -67,13 +68,14 @@ public class CustomHandlerAnnotationTest extends MessageBusTest
public static class NamedMessageFilter implements IMessageFilter
{
@Override
- public boolean accepts( NamedMessage message, MessageHandler metadata ) {
- NamedMessageHandler namedMessageHandler = metadata.getAnnotation(NamedMessageHandler.class);
+ public boolean accepts( NamedMessage message, SubscriptionContext context ) {
+ MessageHandler handler = context.getHandler();
+ NamedMessageHandler namedMessageHandler = handler.getAnnotation(NamedMessageHandler.class);
if ( namedMessageHandler != null ) {
return Arrays.asList( namedMessageHandler.value() ).contains( message.getName() );
}
- EnvelopedNamedMessageHandler envelopedHandler = metadata.getAnnotation(EnvelopedNamedMessageHandler.class);
+ EnvelopedNamedMessageHandler envelopedHandler = handler.getAnnotation(EnvelopedNamedMessageHandler.class);
return envelopedHandler != null && Arrays.asList( envelopedHandler.value() ).contains( message.getName() );
}
diff --git a/src/test/java/net/engio/mbassy/FilterTest.java b/src/test/java/net/engio/mbassy/FilterTest.java
index f87ceef7..b7b31777 100644
--- a/src/test/java/net/engio/mbassy/FilterTest.java
+++ b/src/test/java/net/engio/mbassy/FilterTest.java
@@ -9,6 +9,7 @@
import net.engio.mbassy.listener.*;
import net.engio.mbassy.messages.SubTestMessage;
import net.engio.mbassy.messages.TestMessage;
+import net.engio.mbassy.subscription.SubscriptionContext;
import org.junit.Test;
import java.util.List;
@@ -131,7 +132,7 @@ public void handle(TestMessage message){
public static class RejectFilteredObjects implements IMessageFilter{
@Override
- public boolean accepts(Object message, MessageHandler metadata) {
+ public boolean accepts(Object message, SubscriptionContext context) {
if(message.getClass().equals(FilteredMessage.class) && ((FilteredMessage)message).getMessage().getClass().equals(Object.class)){
return false;
}
@@ -142,7 +143,7 @@ public boolean accepts(Object message, MessageHandler metadata) {
public static final class RejectAll implements IMessageFilter {
@Override
- public boolean accepts(Object event, MessageHandler metadata) {
+ public boolean accepts(Object event, SubscriptionContext context) {
return false;
}
}
diff --git a/src/test/java/net/engio/mbassy/SubscriptionManagerTest.java b/src/test/java/net/engio/mbassy/SubscriptionManagerTest.java
index 0412ea17..5cbe28bb 100644
--- a/src/test/java/net/engio/mbassy/SubscriptionManagerTest.java
+++ b/src/test/java/net/engio/mbassy/SubscriptionManagerTest.java
@@ -1,8 +1,8 @@
package net.engio.mbassy;
import net.engio.mbassy.bus.BusRuntime;
+import net.engio.mbassy.bus.common.Properties;
import net.engio.mbassy.common.*;
-import net.engio.mbassy.listener.Handler;
import net.engio.mbassy.listener.MetadataReader;
import net.engio.mbassy.listeners.*;
import net.engio.mbassy.messages.*;
@@ -214,8 +214,8 @@ public void testPrioritizedMessageHandlers(){
private BusRuntime mockedRuntime(){
return new BusRuntime(null)
- .add(BusRuntime.Properties.ErrorHandlers, Collections.EMPTY_SET)
- .add(BusRuntime.Properties.AsynchronousHandlerExecutor, null);
+ .add(Properties.Handler.PublicationError, Collections.EMPTY_SET)
+ .add(Properties.Handler.AsynchronousHandlerExecutor, null);
}
private ListenerFactory listeners(Class ...listeners){
@@ -248,22 +248,22 @@ private void runTestWith(final ListenerFactory listeners, final SubscriptionVali
public static class PrioritizedListener{
- @Handler(priority = 1)
+ @net.engio.mbassy.listener.Handler(priority = 1)
public void handlePrio1(IMessage message){
message.handled(this.getClass());
}
- @Handler(priority = 2)
+ @net.engio.mbassy.listener.Handler(priority = 2)
public void handlePrio2(IMessage message){
message.handled(this.getClass());
}
- @Handler(priority = 3)
+ @net.engio.mbassy.listener.Handler(priority = 3)
public void handlePrio3(IMessage message){
message.handled(this.getClass());
}
- @Handler(priority = 4)
+ @net.engio.mbassy.listener.Handler(priority = 4)
public void handlePrio4(IMessage message){
message.handled(this.getClass());
}
diff --git a/src/test/java/net/engio/mbassy/MBassadorTest.java b/src/test/java/net/engio/mbassy/SyncAsyncTest.java
similarity index 96%
rename from src/test/java/net/engio/mbassy/MBassadorTest.java
rename to src/test/java/net/engio/mbassy/SyncAsyncTest.java
index a8dc5d5e..a1c3dc5e 100644
--- a/src/test/java/net/engio/mbassy/MBassadorTest.java
+++ b/src/test/java/net/engio/mbassy/SyncAsyncTest.java
@@ -1,7 +1,7 @@
package net.engio.mbassy;
import net.engio.mbassy.bus.MBassador;
-import net.engio.mbassy.bus.config.BusConfiguration;
+import net.engio.mbassy.bus.common.Properties;
import net.engio.mbassy.bus.error.IPublicationErrorHandler;
import net.engio.mbassy.bus.error.PublicationError;
import net.engio.mbassy.common.*;
@@ -19,7 +19,7 @@
* @author bennidi
* Date: 2/8/12
*/
-public class MBassadorTest extends MessageBusTest {
+public class SyncAsyncTest extends MessageBusTest {
@Test
@@ -132,8 +132,8 @@ public void handleError(PublicationError error) {
}
};
- final MBassador bus = new MBassador(SyncAsync());
- bus.addErrorHandler(ExceptionCounter);
+ final MBassador bus = new MBassador(SyncAsync()
+ .setProperty(Properties.Handler.PublicationError, ExceptionCounter));
ListenerFactory listeners = new ListenerFactory()
.create(InstancesPerListener, ExceptionThrowingListener.class);
ConcurrentExecutor.runConcurrent(TestUtil.subscriber(bus, listeners), ConcurrentUnits);
diff --git a/src/test/java/net/engio/mbassy/SyncBusTest.java b/src/test/java/net/engio/mbassy/SyncBusTest.java
index 59508569..1592b93d 100644
--- a/src/test/java/net/engio/mbassy/SyncBusTest.java
+++ b/src/test/java/net/engio/mbassy/SyncBusTest.java
@@ -1,8 +1,12 @@
package net.engio.mbassy;
-import net.engio.mbassy.bus.BusFactory;
import net.engio.mbassy.bus.MBassador;
+import net.engio.mbassy.bus.SyncMessageBus;
import net.engio.mbassy.bus.common.GenericMessagePublicationSupport;
+import net.engio.mbassy.bus.common.Properties;
+import net.engio.mbassy.bus.config.BusConfiguration;
+import net.engio.mbassy.bus.config.Feature;
+import net.engio.mbassy.bus.config.IBusConfiguration;
import net.engio.mbassy.bus.error.IPublicationErrorHandler;
import net.engio.mbassy.bus.error.PublicationError;
import net.engio.mbassy.common.ConcurrentExecutor;
@@ -17,7 +21,6 @@
import net.engio.mbassy.messages.MessageTypes;
import net.engio.mbassy.messages.MultipartMessage;
import net.engio.mbassy.messages.StandardMessage;
-import org.junit.Assert;
import org.junit.Test;
import java.util.concurrent.atomic.AtomicInteger;
@@ -31,12 +34,12 @@
public abstract class SyncBusTest extends MessageBusTest {
- protected abstract GenericMessagePublicationSupport getSyncMessageBus();
+ protected abstract GenericMessagePublicationSupport getSyncMessageBus(boolean failOnException);
@Test
public void testSynchronousMessagePublication() throws Exception {
- final GenericMessagePublicationSupport bus = getSyncMessageBus();
+ final GenericMessagePublicationSupport bus = getSyncMessageBus(true);
ListenerFactory listeners = new ListenerFactory()
.create(InstancesPerListener, IMessageListener.DefaultListener.class)
.create(InstancesPerListener, IMessageListener.DisabledListener.class)
@@ -87,7 +90,7 @@ public void handleError(PublicationError error) {
}
};
- final GenericMessagePublicationSupport bus = getSyncMessageBus();
+ final GenericMessagePublicationSupport bus = getSyncMessageBus(false);
bus.addErrorHandler(ExceptionCounter);
ListenerFactory listeners = new ListenerFactory()
.create(InstancesPerListener, ExceptionThrowingListener.class);
@@ -113,7 +116,7 @@ public void run() {
@Test
public void testCustomHandlerInvocation(){
- final GenericMessagePublicationSupport bus = getSyncMessageBus();
+ final GenericMessagePublicationSupport bus = getSyncMessageBus(true);
ListenerFactory listeners = new ListenerFactory()
.create(InstancesPerListener, CustomInvocationListener.class)
.create(InstancesPerListener, Object.class);
@@ -147,7 +150,7 @@ public void run() {
@Test
public void testHandlerPriorities(){
- final GenericMessagePublicationSupport bus = getSyncMessageBus();
+ final GenericMessagePublicationSupport bus = getSyncMessageBus(true);
ListenerFactory listeners = new ListenerFactory()
.create(InstancesPerListener, PrioritizedListener.class)
.create(InstancesPerListener, Object.class);
@@ -157,9 +160,7 @@ public void testHandlerPriorities(){
Runnable publishAndCheck = new Runnable() {
@Override
public void run() {
-
bus.post(new IncrementingMessage()).now();
-
}
};
@@ -176,8 +177,13 @@ public static class MBassadorTest extends SyncBusTest {
@Override
- protected GenericMessagePublicationSupport getSyncMessageBus() {
- return new MBassador();
+ protected GenericMessagePublicationSupport getSyncMessageBus(boolean failOnException) {
+ IBusConfiguration asyncFIFOConfig = new BusConfiguration()
+ .setProperty(Properties.Handler.PublicationError, new AssertionErrorHandler(failOnException));
+ asyncFIFOConfig.addFeature(Feature.SyncPubSub.Default());
+ asyncFIFOConfig.addFeature(Feature.AsynchronousHandlerInvocation.Default(1, 1));
+ asyncFIFOConfig.addFeature(Feature.AsynchronousMessageDispatch.Default().setNumberOfMessageDispatchers(1));
+ return new MBassador(asyncFIFOConfig);
}
}
@@ -186,13 +192,18 @@ public static class SyncMessageBusTest extends SyncBusTest {
@Override
- protected GenericMessagePublicationSupport getSyncMessageBus() {
- return BusFactory.SynchronousOnly();
+ protected GenericMessagePublicationSupport getSyncMessageBus(boolean failOnException) {
+ IBusConfiguration syncPubSubCfg = new BusConfiguration()
+ .setProperty(Properties.Handler.PublicationError, new AssertionErrorHandler(failOnException));
+ syncPubSubCfg.addFeature(Feature.SyncPubSub.Default());
+ return new SyncMessageBus(syncPubSubCfg);
}
}
+
+
static class IncrementingMessage{
private int count = 1;
@@ -200,7 +211,7 @@ static class IncrementingMessage{
public void markHandled(int newVal){
// only transitions by the next handler are allowed
if(count == newVal || count + 1 == newVal) count = newVal;
- else Assert.fail("Message was handled out of order");
+ else throw new RuntimeException("Message was handled out of order");
}
}
@@ -209,22 +220,22 @@ public static class PrioritizedListener{
@Handler(priority = Integer.MIN_VALUE)
public void handle1(IncrementingMessage message) {
- message.markHandled(1);
+ message.markHandled(4);
}
@Handler(priority = -2)
public void handle2(IncrementingMessage message) {
- message.markHandled(2);
+ message.markHandled(3);
}
@Handler
public void handle3(IncrementingMessage message) {
- message.markHandled(3);
+ message.markHandled(2);
}
@Handler(priority = Integer.MAX_VALUE)
public void handle4(IncrementingMessage message) {
- message.markHandled(4);
+ message.markHandled(1);
}
diff --git a/src/test/java/net/engio/mbassy/SynchronizedHandlerTest.java b/src/test/java/net/engio/mbassy/SynchronizedHandlerTest.java
index 0c3b9dd5..463900a6 100644
--- a/src/test/java/net/engio/mbassy/SynchronizedHandlerTest.java
+++ b/src/test/java/net/engio/mbassy/SynchronizedHandlerTest.java
@@ -29,7 +29,7 @@ public class SynchronizedHandlerTest extends MessageBusTest {
@Test
public void testSynchronizedWithSynchronousInvocation(){
List handlers = new LinkedList();
- IBusConfiguration config = SyncAsync();
+ IBusConfiguration config = SyncAsync(true);
config.getFeature(Feature.AsynchronousMessageDispatch.class)
.setNumberOfMessageDispatchers(6);
IMessageBus bus = createBus(config);
@@ -57,7 +57,7 @@ public void testSynchronizedWithSynchronousInvocation(){
@Test
public void testSynchronizedWithAsSynchronousInvocation(){
List handlers = new LinkedList();
- IBusConfiguration config = SyncAsync();
+ IBusConfiguration config = SyncAsync(true);
config.getFeature(Feature.AsynchronousMessageDispatch.class)
.setNumberOfMessageDispatchers(6);
IMessageBus bus = createBus(config);
diff --git a/src/test/java/net/engio/mbassy/common/ConcurrentExecutor.java b/src/test/java/net/engio/mbassy/common/ConcurrentExecutor.java
index 09c12647..ed5bdac0 100644
--- a/src/test/java/net/engio/mbassy/common/ConcurrentExecutor.java
+++ b/src/test/java/net/engio/mbassy/common/ConcurrentExecutor.java
@@ -11,7 +11,7 @@
*
* Date: 2/14/12
*
- * @Author bennidi
+ * @author bennidi
*/
public class ConcurrentExecutor {
diff --git a/src/test/java/net/engio/mbassy/common/ListenerFactory.java b/src/test/java/net/engio/mbassy/common/ListenerFactory.java
index d188f6c1..0fc9adc3 100644
--- a/src/test/java/net/engio/mbassy/common/ListenerFactory.java
+++ b/src/test/java/net/engio/mbassy/common/ListenerFactory.java
@@ -75,7 +75,6 @@ public synchronized void clear(){
*
* NOTE: Iterator is not perfectly synchronized with mutator methods of the list of generated listeners
* In theory, it is possible that the list is changed while iterators are still running which should be avoided.
- * @return
*/
public Iterator iterator(){
getAll();
diff --git a/src/test/java/net/engio/mbassy/common/MessageBusTest.java b/src/test/java/net/engio/mbassy/common/MessageBusTest.java
index 78e75532..2fe89683 100644
--- a/src/test/java/net/engio/mbassy/common/MessageBusTest.java
+++ b/src/test/java/net/engio/mbassy/common/MessageBusTest.java
@@ -1,8 +1,7 @@
package net.engio.mbassy.common;
-import junit.framework.Assert;
-import net.engio.mbassy.bus.MBassador;
import net.engio.mbassy.bus.IMessagePublication;
+import net.engio.mbassy.bus.MBassador;
import net.engio.mbassy.bus.config.BusConfiguration;
import net.engio.mbassy.bus.config.Feature;
import net.engio.mbassy.bus.config.IBusConfiguration;
@@ -28,14 +27,20 @@ public abstract class MessageBusTest extends AssertSupport {
protected static final int ConcurrentUnits = 10;
protected static final int IterationsPerThread = 100;
- protected static final IPublicationErrorHandler TestFailingHandler = new IPublicationErrorHandler() {
+ public static final class AssertionErrorHandler implements IPublicationErrorHandler{
+
+ private boolean failOnException;
+
+ public AssertionErrorHandler(boolean failOnException) {
+ this.failOnException = failOnException;
+ }
+
@Override
public void handleError(PublicationError error) {
- error.getCause().printStackTrace();
- Assert.fail();
+ if(failOnException)
+ org.junit.Assert.fail(error.getCause().getMessage());
}
- };
-
+ }
private StrongConcurrentSet issuedPublications = new StrongConcurrentSet();
@@ -47,21 +52,24 @@ public void setUp(){
}
public static IBusConfiguration SyncAsync() {
+ return SyncAsync(true);
+ }
+
+ public static IBusConfiguration SyncAsync(boolean failOnError) {
return new BusConfiguration()
.addFeature(Feature.SyncPubSub.Default())
.addFeature(Feature.AsynchronousHandlerInvocation.Default())
- .addFeature(Feature.AsynchronousMessageDispatch.Default());
+ .addFeature(Feature.AsynchronousMessageDispatch.Default())
+ .setProperty(net.engio.mbassy.bus.common.Properties.Handler.PublicationError, new AssertionErrorHandler(failOnError));
}
public MBassador createBus(IBusConfiguration configuration) {
MBassador bus = new MBassador(configuration);
- bus.addErrorHandler(TestFailingHandler);
return bus;
}
public MBassador createBus(IBusConfiguration configuration, ListenerFactory listeners) {
MBassador bus = new MBassador(configuration);
- bus.addErrorHandler(TestFailingHandler);
ConcurrentExecutor.runConcurrent(TestUtil.subscriber(bus, listeners), ConcurrentUnits);
return bus;
}
diff --git a/testNTimes.sh b/testNTimes.sh
index 35c8b125..681ac045 100755
--- a/testNTimes.sh
+++ b/testNTimes.sh
@@ -1,12 +1,12 @@
#!/bin/bash
for (( i = 1; i < $1 ; i++ ))
do
- echo "Attempt $i"
+ echo "Round $i"
mvn test -o -Dtest=$2
exitcode=$?
if [ $exitcode -ne 0 ]
then
- echo "Error at attempt $i"
+ echo "Error at round $i"
exit
fi
done