diff options
Diffstat (limited to 'src/test')
| -rw-r--r-- | src/test/java/io/trygvis/test/PlainJavaExample.java | 36 |
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(); } } } |