Default concurrency based on destination type [IMMUTANT-360]

projectodd · Apr 7, 2015 · eafab77 · eafab77
1 parent 2e59311
commit eafab77
Show file tree

Hide file tree

Showing 5 changed files with 14 additions and 2 deletions.
diff --git a/modules/messaging/src/main/java/org/projectodd/wunderboss/messaging/Destination.java b/modules/messaging/src/main/java/org/projectodd/wunderboss/messaging/Destination.java
@@ -27,7 +27,7 @@ public interface Destination {
 
     class ListenOption extends Option {
         public static final ListenOption CONTEXT     = opt("context", ListenOption.class);
-        public static final ListenOption CONCURRENCY = opt("concurrency", Runtime.getRuntime().availableProcessors(), ListenOption.class);
+        public static final ListenOption CONCURRENCY = opt("concurrency", ListenOption.class);
         public static final ListenOption SELECTOR    = opt("selector", ListenOption.class);
         public static final ListenOption MODE        = opt("mode", Context.Mode.TRANSACTED, ListenOption.class);
     }

diff --git a/...es/messaging/src/main/java/org/projectodd/wunderboss/messaging/hornetq/HQDestination.java b/...es/messaging/src/main/java/org/projectodd/wunderboss/messaging/hornetq/HQDestination.java
@@ -57,6 +57,8 @@ public Destination jmsDestination() {
 
     public abstract String jmsName();
 
+    public abstract int defaultConcurrency();
+
     @Override
     public Listener listen(MessageHandler handler, Codecs codecs, Map<ListenOption, Object> options) throws Exception {
         Options<ListenOption> opts = new Options<>(options);

diff --git a/modules/messaging/src/main/java/org/projectodd/wunderboss/messaging/hornetq/HQQueue.java b/modules/messaging/src/main/java/org/projectodd/wunderboss/messaging/hornetq/HQQueue.java
@@ -118,4 +118,8 @@ public String fullName() {
         return fullName(name());
     }
 
+    @Override
+    public int defaultConcurrency() {
+        return Runtime.getRuntime().availableProcessors();
+    }
 }
diff --git a/modules/messaging/src/main/java/org/projectodd/wunderboss/messaging/hornetq/HQTopic.java b/modules/messaging/src/main/java/org/projectodd/wunderboss/messaging/hornetq/HQTopic.java
@@ -107,6 +107,11 @@ public Destination jmsDestination() {
         return broker().lookupTopic(name());
     }
 
+    @Override
+    public int defaultConcurrency() {
+        return 1;
+    }
+
     protected HQSpecificContext context(final String id, final Object context) throws Exception {
         if (context != null) {
             return ((HQSpecificContext)context).asNonCloseable();

diff --git a/...saging/src/main/java/org/projectodd/wunderboss/messaging/hornetq/MessageHandlerGroup.java b/...saging/src/main/java/org/projectodd/wunderboss/messaging/hornetq/MessageHandlerGroup.java
@@ -45,7 +45,8 @@ public MessageHandlerGroup(HQSpecificContext context,
 
     public synchronized MessageHandlerGroup start() throws Exception {
         if (!this.started) {
-            int concurrency = this.options.getInt(ListenOption.CONCURRENCY);
+            Integer option = this.options.getInt(ListenOption.CONCURRENCY);
+            int concurrency = option != null ? option : this.destination.defaultConcurrency();
             log.info("Starting listener for '" + this.destination.name() + "' concurrency=" + concurrency);
             while(concurrency-- > 0) {
                 HQSpecificContext subContext =