diff options
Diffstat (limited to 'src/test/java/io/trygvis/test/jdbc/AsyncConsumerExample.java')
-rw-r--r-- | src/test/java/io/trygvis/test/jdbc/AsyncConsumerExample.java | 18 |
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 { + } } |