Merge pull request #49 from durron597/master
Overloading factory methods for default bus configuration
This commit is contained in:
commit
bc267cde2b
@ -41,13 +41,57 @@ public class BusConfiguration implements IBusConfiguration {
|
|||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Creates a new instance, using the Default settings of 2 dispatchers, and
|
||||||
|
* asynchronous handlers with an initial count equal to the number of
|
||||||
|
* available processors in the machine, with maximum count equal to
|
||||||
|
* 2 * the number of available processors. Uses {@link Runtime#availableProcessors()} to
|
||||||
|
* determine the number of available processors
|
||||||
|
*
|
||||||
|
* @return a Default BusConfiguration
|
||||||
|
*/
|
||||||
public static BusConfiguration Default() {
|
public static BusConfiguration Default() {
|
||||||
BusConfiguration defaultConfig = new BusConfiguration();
|
return Default(2);
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Creates a new instance, using the specified number of dispatchers, and
|
||||||
|
* asynchronous handlers with an initial count equal to the number of
|
||||||
|
* available processors in the machine, with maximum count equal to
|
||||||
|
* 2 * the number of available processors. Uses {@link Runtime#availableProcessors()} to
|
||||||
|
* determine the number of available processors
|
||||||
|
*
|
||||||
|
* @return a Default BusConfiguration
|
||||||
|
*/
|
||||||
|
public static BusConfiguration Default(int numberOfDispatchers) {
|
||||||
int numberOfCoreThreads = Runtime.getRuntime().availableProcessors();
|
int numberOfCoreThreads = Runtime.getRuntime().availableProcessors();
|
||||||
defaultConfig.setExecutorForAsynchronousHandlers(new ThreadPoolExecutor(numberOfCoreThreads, numberOfCoreThreads*2, 1, TimeUnit.MINUTES, new LinkedBlockingQueue<Runnable>(), AsynchronousHandlerThreadFactory));
|
return Default(numberOfDispatchers, numberOfCoreThreads, numberOfCoreThreads * 2);
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Creates a new instance, using the specified number of dispatchers, and
|
||||||
|
* asynchronous handlers with initial threads and maximum threads specified by the calling
|
||||||
|
* parameters.
|
||||||
|
*
|
||||||
|
* @return a Default BusConfiguration
|
||||||
|
*/
|
||||||
|
public static BusConfiguration Default(int numberOfDispatchers, int initialCoreThreads, int maximumCoreThreads) {
|
||||||
|
ThreadPoolExecutor executor = new ThreadPoolExecutor(initialCoreThreads, maximumCoreThreads, 1, TimeUnit.MINUTES, new LinkedBlockingQueue<Runnable>(), AsynchronousHandlerThreadFactory);
|
||||||
|
return Default(numberOfDispatchers, executor);
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Creates a new instance, using the specified number of dispatchers, and
|
||||||
|
* asynchronous handlers that use the provided ThreadPoolExecutor.
|
||||||
|
*
|
||||||
|
* @return a Default BusConfiguration
|
||||||
|
*/
|
||||||
|
public static BusConfiguration Default(int numberOfDispatchers, ThreadPoolExecutor executor) {
|
||||||
|
BusConfiguration defaultConfig = new BusConfiguration();
|
||||||
|
defaultConfig.setExecutorForAsynchronousHandlers(executor);
|
||||||
defaultConfig.setMetadataReader(new MetadataReader());
|
defaultConfig.setMetadataReader(new MetadataReader());
|
||||||
defaultConfig.setSubscriptionFactory(new SubscriptionFactory());
|
defaultConfig.setSubscriptionFactory(new SubscriptionFactory());
|
||||||
defaultConfig.setNumberOfMessageDispatchers(2);
|
defaultConfig.setNumberOfMessageDispatchers(numberOfDispatchers);
|
||||||
defaultConfig.setMessagePublicationFactory(new MessagePublication.Factory());
|
defaultConfig.setMessagePublicationFactory(new MessagePublication.Factory());
|
||||||
defaultConfig.setPendingMessagesQueue(new LinkedBlockingQueue<MessagePublication>(Integer.MAX_VALUE));
|
defaultConfig.setPendingMessagesQueue(new LinkedBlockingQueue<MessagePublication>(Integer.MAX_VALUE));
|
||||||
defaultConfig.setThreadFactoryForAsynchronousMessageDispatch(DispatcherThreadFactory);
|
defaultConfig.setThreadFactoryForAsynchronousMessageDispatch(DispatcherThreadFactory);
|
||||||
|
@ -4,6 +4,9 @@ import net.engio.mbassy.messages.IMessage;
|
|||||||
|
|
||||||
import java.util.Collection;
|
import java.util.Collection;
|
||||||
|
|
||||||
|
import org.slf4j.Logger;
|
||||||
|
import org.slf4j.LoggerFactory;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Created with IntelliJ IDEA.
|
* Created with IntelliJ IDEA.
|
||||||
* User: benjamin
|
* User: benjamin
|
||||||
@ -12,6 +15,8 @@ import java.util.Collection;
|
|||||||
* To change this template use File | Settings | File Templates.
|
* To change this template use File | Settings | File Templates.
|
||||||
*/
|
*/
|
||||||
public class MessageManager {
|
public class MessageManager {
|
||||||
|
private static final Logger LOG =
|
||||||
|
LoggerFactory.getLogger(MessageManager.class);
|
||||||
|
|
||||||
|
|
||||||
private StrongConcurrentSet<MessageContext> messages = new StrongConcurrentSet();
|
private StrongConcurrentSet<MessageContext> messages = new StrongConcurrentSet();
|
||||||
@ -90,7 +95,7 @@ public class MessageManager {
|
|||||||
}
|
}
|
||||||
|
|
||||||
private void logSuccess(MessageContext mCtx){
|
private void logSuccess(MessageContext mCtx){
|
||||||
System.out.println("Message " + mCtx.getMessage() + " was successfully handled " + mCtx.getExpectedCount() + " times by " + mCtx.printListeners());
|
LOG.info("Message " + mCtx.getMessage() + " was successfully handled " + mCtx.getExpectedCount() + " times by " + mCtx.printListeners());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
@ -100,7 +105,7 @@ public class MessageManager {
|
|||||||
errorMessage.append("Failing messages:\n");
|
errorMessage.append("Failing messages:\n");
|
||||||
for(MessageContext failingMessage : failing)
|
for(MessageContext failingMessage : failing)
|
||||||
errorMessage.append(failingMessage);
|
errorMessage.append(failingMessage);
|
||||||
System.out.println(errorMessage.toString());
|
LOG.info(errorMessage.toString());
|
||||||
}
|
}
|
||||||
|
|
||||||
private class MessageContext{
|
private class MessageContext{
|
||||||
|
Loading…
Reference in New Issue
Block a user