aboutsummaryrefslogtreecommitdiff
path: root/src/test/java/io/trygvis/test/PlainJavaExample.java
diff options
context:
space:
mode:
authorTrygve Laugstøl <trygvis@inamo.no>2013-06-12 22:55:18 +0200
committerTrygve Laugstøl <trygvis@inamo.no>2013-06-12 22:55:18 +0200
commit54d7b2ce520e57cc0ffb9582546b80a32fa00682 (patch)
treead68043a4f05780c6b87a494f7498297ff978953 /src/test/java/io/trygvis/test/PlainJavaExample.java
parent4b0bab9e722cf77ca0049c54515e8c93acefa355 (diff)
downloadquartz-based-queue-54d7b2ce520e57cc0ffb9582546b80a32fa00682.tar.gz
quartz-based-queue-54d7b2ce520e57cc0ffb9582546b80a32fa00682.tar.bz2
quartz-based-queue-54d7b2ce520e57cc0ffb9582546b80a32fa00682.tar.xz
quartz-based-queue-54d7b2ce520e57cc0ffb9582546b80a32fa00682.zip
wip
Diffstat (limited to 'src/test/java/io/trygvis/test/PlainJavaExample.java')
-rw-r--r--src/test/java/io/trygvis/test/PlainJavaExample.java126
1 files changed, 0 insertions, 126 deletions
diff --git a/src/test/java/io/trygvis/test/PlainJavaExample.java b/src/test/java/io/trygvis/test/PlainJavaExample.java
deleted file mode 100644
index cad8559..0000000
--- a/src/test/java/io/trygvis/test/PlainJavaExample.java
+++ /dev/null
@@ -1,126 +0,0 @@
-package io.trygvis.test;
-
-import io.trygvis.async.SqlEffect;
-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;
-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;
-import static java.util.Collections.singletonList;
-
-public class PlainJavaExample {
- private static final Random r = new Random();
-
- private static String inputName = "my-input";
- private static String outputName = "my-output";
-
- private static int interval = 10;
-
- public static class Consumer {
- public static void main(String[] args) throws Exception {
- System.out.println("Starting consumer");
-
- DataSource ds = createDataSource();
-
- SqlEffectExecutor sqlEffectExecutor = new SqlEffectExecutor(ds);
-
- 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 {
- 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(1000, false);
-
- queueService.consumeAll(input, req, new TaskEffect() {
- public List<Task> apply(Task task) throws Exception {
- 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(3000) > 0) {
- return singletonList(task.childTask(output.name, new Date(), Long.toString(a + b)));
- }
-
- throw new RuntimeException("Simulated exception while processing task.");
- }
- });
- System.out.println("Done");
- }
- }
-
- public static class Producer {
- public static void main(String[] args) throws Exception {
- System.out.println("Starting producer");
-
- int chunks = 100;
- final int chunk = 10000;
-
- DataSource ds = createDataSource();
-
- SqlEffectExecutor sqlEffectExecutor = new SqlEffectExecutor(ds);
-
- QueueSystem queueSystem = QueueSystem.initialize(sqlEffectExecutor);
- final JdbcQueueService queueService = queueSystem.queueService;
-
- 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();
- sqlEffectExecutor.transaction(new SqlEffect.Void() {
- @Override
- public void doInConnection(Connection c) throws SQLException {
- for (int j = 0; j < chunk; j++) {
- queueService.schedule(c, queue, new Date(), asList("10", "20"));
- }
- }
- });
- long end = currentTimeMillis();
-
- long time = end - start;
- System.out.println("Scheduled " + chunk + " tasks in " + time + "ms, " + (((double) chunk * 1000)) / ((double) time) + " chunks per second");
- }
- }
- }
-}