1 files changed, 25 insertions, 11 deletions

diff --git a/src/test/java/io/trygvis/test/PlainJavaExample.java b/src/test/java/io/trygvis/test/PlainJavaExample.java
index 788d8a0..cad8559 100644
--- a/src/test/java/io/trygvis/test/PlainJavaExample.java
+++ b/src/test/java/io/trygvis/test/PlainJavaExample.java
@@ -5,8 +5,10 @@ import io.trygvis.async.SqlEffectExecutor;
 import io.trygvis.queue.JdbcQueueService;
 import io.trygvis.queue.Queue;
 import io.trygvis.queue.QueueService;
+import io.trygvis.queue.QueueStats;
 import io.trygvis.queue.QueueSystem;
 import io.trygvis.queue.Task;
+import io.trygvis.queue.TaskDao;
 import io.trygvis.queue.TaskEffect;
 
 import javax.sql.DataSource;
@@ -14,8 +16,10 @@ import java.sql.Connection;
 import java.sql.SQLException;
 import java.util.Date;
 import java.util.List;
+import java.util.Map;
 import java.util.Random;
 
+import static io.trygvis.queue.Task.TaskState;
 import static io.trygvis.test.DbUtil.createDataSource;
 import static java.lang.System.currentTimeMillis;
 import static java.util.Arrays.asList;
@@ -37,32 +41,41 @@ public class PlainJavaExample {
 
             SqlEffectExecutor sqlEffectExecutor = new SqlEffectExecutor(ds);
 
-            QueueSystem queueSystem = QueueSystem.initialize(sqlEffectExecutor);
+            final QueueSystem queueSystem = QueueSystem.initialize(sqlEffectExecutor);
             final JdbcQueueService queueService = queueSystem.queueService;
 
             Queue[] queues = sqlEffectExecutor.transaction(new SqlEffect<Queue[]>() {
                 @Override
                 public Queue[] doInConnection(Connection c) throws SQLException {
-                    return new Queue[]{
+                    Queue[] queues = {
                             queueService.lookupQueue(c, inputName, interval, true),
                             queueService.lookupQueue(c, outputName, interval, true)};
+
+                    TaskDao taskDao = queueSystem.createTaskDao(c);
+
+                    QueueStats stats = taskDao.findQueueStatsByName(inputName);
+                    System.out.println("Queue stats for " + stats.name + ". Total number of tasks: " + stats.totalTaskCount);
+                    for (Map.Entry<TaskState, Long> entry : stats.states.entrySet()) {
+                        System.out.println(entry.getKey() + " = " + entry.getValue());
+                    }
+
+                    return queues;
                 }
             });
 
             final Queue input = queues[0];
             final Queue output = queues[1];
 
-            QueueService.TaskExecutionRequest req = new QueueService.TaskExecutionRequest(false);
+            QueueService.TaskExecutionRequest req = new QueueService.TaskExecutionRequest(1000, false);
 
             queueService.consumeAll(input, req, new TaskEffect() {
                 public List<Task> apply(Task task) throws Exception {
-                    System.out.println("PlainJavaExample$Consumer.consumeAll: arguments = " + task.arguments);
                     Long a = Long.valueOf(task.arguments.get(0));
                     Long b = Long.valueOf(task.arguments.get(1));
 
                     System.out.println("a + b = " + a + " + " + b + " = " + (a + b));
 
-                    if (r.nextInt(3) == 0) {
+                    if (r.nextInt(3000) > 0) {
                         return singletonList(task.childTask(output.name, new Date(), Long.toString(a + b)));
                     }
 
@@ -77,18 +90,21 @@ public class PlainJavaExample {
         public static void main(String[] args) throws Exception {
             System.out.println("Starting producer");
 
-            int chunks = 10;
-            final int chunk = 2000;
+            int chunks = 100;
+            final int chunk = 10000;
 
             DataSource ds = createDataSource();
-            Connection c = ds.getConnection();
 
             SqlEffectExecutor sqlEffectExecutor = new SqlEffectExecutor(ds);
 
             QueueSystem queueSystem = QueueSystem.initialize(sqlEffectExecutor);
             final JdbcQueueService queueService = queueSystem.queueService;
 
-            final Queue queue = queueService.lookupQueue(c, inputName, interval, true);
+            final Queue queue;
+            try (Connection c = ds.getConnection()) {
+                queue = queueService.lookupQueue(c, inputName, interval, true);
+                c.commit();
+            }
 
             for (int i = 0; i < chunks; i++) {
                 long start = currentTimeMillis();
@@ -105,8 +121,6 @@ public class PlainJavaExample {
                 long time = end - start;
                 System.out.println("Scheduled " + chunk + " tasks in " + time + "ms, " + (((double) chunk * 1000)) / ((double) time) + " chunks per second");
             }
-
-            c.commit();
         }
     }
 }