aboutsummaryrefslogtreecommitdiff
path: root/src/test
diff options
context:
space:
mode:
Diffstat (limited to 'src/test')
-rw-r--r--src/test/java/io/trygvis/test/PlainJavaExample.java36
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();
}
}
}