o Adding an ActiveMQ connection to messages can be sent as a hint to the consumer.HEADmaster

1 files changed, 13 insertions, 5 deletions

diff --git a/src/test/java/io/trygvis/test/jdbc/AsyncConsumerExample.java b/src/test/java/io/trygvis/test/jdbc/AsyncConsumerExample.java
index 14db21b..16640dd 100644
--- a/src/test/java/io/trygvis/test/jdbc/AsyncConsumerExample.java
+++ b/src/test/java/io/trygvis/test/jdbc/AsyncConsumerExample.java
@@ -3,12 +3,12 @@ package io.trygvis.test.jdbc;
 import io.trygvis.async.JdbcAsyncService;
 import io.trygvis.async.QueueController;
 import io.trygvis.async.QueueStats;
-import io.trygvis.queue.SqlEffect;
-import io.trygvis.queue.SqlEffectExecutor;
 import io.trygvis.queue.JdbcQueueService;
 import io.trygvis.queue.QueueExecutor;
 import io.trygvis.queue.QueueService;
 import io.trygvis.queue.QueueSystem;
+import io.trygvis.queue.SqlEffect;
+import io.trygvis.queue.SqlEffectExecutor;
 import io.trygvis.queue.Task;
 import io.trygvis.queue.TaskEffect;
 
@@ -43,6 +43,12 @@ public class AsyncConsumerExample {
     };
 
     public static void main(String[] args) throws Exception {
+        int poolSize = 4;
+        QueueService.TaskExecutionRequest req = new QueueService.TaskExecutionRequest(100, true);
+        new AsyncConsumerExample().work(poolSize, req);
+    }
+
+    public void work(int poolSize, QueueService.TaskExecutionRequest req) throws Exception {
         System.out.println("Starting consumer");
 
         DataSource ds = createDataSource();
@@ -66,9 +72,8 @@ public class AsyncConsumerExample {
         final QueueExecutor input = queues[0];
         final QueueExecutor output = queues[1];
 
-        QueueService.TaskExecutionRequest req = new QueueService.TaskExecutionRequest(100, true);
-
         QueueController controller = asyncService.registerQueue(input, req, adder);
+        wrapInputQueue(controller);
 
         Timer timer = new Timer();
         timer.scheduleAtFixedRate(new TimerTask() {
@@ -80,7 +85,7 @@ public class AsyncConsumerExample {
         }, 1000, 1000);
 
         long start = currentTimeMillis();
-        controller.start(new ScheduledThreadPoolExecutor(4));
+        controller.start(new ScheduledThreadPoolExecutor(poolSize));
         Thread.sleep(60 * 1000);
         controller.stop();
         long end = currentTimeMillis();
@@ -96,4 +101,7 @@ public class AsyncConsumerExample {
         double rate = 1000 * ((double) stats.totalMessageCount) / duration;
         System.out.println("Consumed " + stats.totalMessageCount + " messages in " + duration + "ms at " + rate + " msg/s");
     }
+
+    protected void wrapInputQueue(QueueController input) throws Exception {
+    }
 }