aboutsummaryrefslogtreecommitdiff
path: root/src/test/java/io
diff options
context:
space:
mode:
Diffstat (limited to 'src/test/java/io')
-rwxr-xr-xsrc/test/java/io/trygvis/test/CreateArticleCallable.java14
-rw-r--r--src/test/java/io/trygvis/test/DbUtil.java5
-rwxr-xr-xsrc/test/java/io/trygvis/test/Main.java9
-rw-r--r--src/test/java/io/trygvis/test/PlainJavaExample.java51
-rw-r--r--src/test/java/io/trygvis/test/PlainJavaExample2.java84
-rwxr-xr-xsrc/test/java/io/trygvis/test/UpdateArticleCallable.java15
-rw-r--r--src/test/java/io/trygvis/test/spring/PlainSpringTest.java11
7 files changed, 158 insertions, 31 deletions
diff --git a/src/test/java/io/trygvis/test/CreateArticleCallable.java b/src/test/java/io/trygvis/test/CreateArticleCallable.java
index a822a51..396fc89 100755
--- a/src/test/java/io/trygvis/test/CreateArticleCallable.java
+++ b/src/test/java/io/trygvis/test/CreateArticleCallable.java
@@ -1,25 +1,31 @@
package io.trygvis.test;
-import io.trygvis.async.AsyncService;
+import io.trygvis.queue.Task;
+import io.trygvis.queue.TaskEffect;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
+import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.Random;
+import static java.util.Collections.emptyList;
import static org.springframework.transaction.annotation.Propagation.MANDATORY;
@Component("createArticle")
@Transactional(propagation = MANDATORY)
-public class CreateArticleCallable implements AsyncService.AsyncCallable {
+public class CreateArticleCallable implements TaskEffect {
private final Logger log = LoggerFactory.getLogger(getClass());
private Random random = new Random();
- public void run(List<String> arguments) throws Exception {
+ @Override
+ public List<Task> apply(Task task) throws Exception {
+ List<String> arguments = task.arguments;
+
log.info("CreateArticeJob.run: BEGIN");
if (random.nextInt() % 3 == 0) {
@@ -34,5 +40,7 @@ public class CreateArticleCallable implements AsyncService.AsyncCallable {
// entityManager.persist(article);
log.info("CreateArticeJob.run: END");
+
+ return emptyList();
}
}
diff --git a/src/test/java/io/trygvis/test/DbUtil.java b/src/test/java/io/trygvis/test/DbUtil.java
index d0e5b47..2362e65 100644
--- a/src/test/java/io/trygvis/test/DbUtil.java
+++ b/src/test/java/io/trygvis/test/DbUtil.java
@@ -30,15 +30,16 @@ public class DbUtil {
ds.setUsername(username);
ds.setPassword(password);
+ ds.setConnectionTestStatement("/* ping*/SELECT 1");
ds.setDefaultAutoCommit(false);
ds.setIdleConnectionTestPeriodInSeconds(60);
ds.setIdleMaxAgeInSeconds(240);
- ds.setMaxConnectionsPerPartition(40);
+ ds.setMaxConnectionsPerPartition(1);
ds.setMinConnectionsPerPartition(0);
ds.setPartitionCount(1);
ds.setAcquireIncrement(1);
ds.setStatementsCacheSize(1000);
- ds.setReleaseHelperThreads(3);
+ ds.setReleaseHelperThreads(1);
ds.setStatisticsEnabled(true);
ds.setLogStatementsEnabled(true);
ds.setLogWriter(new PrintWriter(System.err));
diff --git a/src/test/java/io/trygvis/test/Main.java b/src/test/java/io/trygvis/test/Main.java
index 43ee971..0721ec9 100755
--- a/src/test/java/io/trygvis/test/Main.java
+++ b/src/test/java/io/trygvis/test/Main.java
@@ -4,6 +4,7 @@ import io.trygvis.async.AsyncService;
import io.trygvis.queue.Queue;
import io.trygvis.queue.QueueService;
import io.trygvis.queue.Task;
+import io.trygvis.queue.TaskEffect;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.bridge.SLF4JBridgeHandler;
@@ -62,22 +63,22 @@ public class Main {
@Autowired
@Qualifier("createArticle")
- private AsyncService.AsyncCallable createArticleCallable;
+ private TaskEffect createArticleCallable;
@Autowired
@Qualifier("updateArticle")
- private AsyncService.AsyncCallable updateArticleCallable;
+ private TaskEffect updateArticleCallable;
public void run() throws Exception {
log.info("Main.run");
- final Queue q = null; // queueService.getQueue(c, "create-article", 1);
+ final Queue q = null; // queueService.lookupQueue(c, "create-article", 1);
asyncService.registerQueue(q, createArticleCallable);
// log.info("queue registered: ref = {}", q);
// asyncService.registerQueue("update-queue", 1, updateArticleCallable);
-// q = asyncService.getQueue("create-queue");
+// q = asyncService.lookupQueue("create-queue");
final List<Task> tasks = new ArrayList<>();
diff --git a/src/test/java/io/trygvis/test/PlainJavaExample.java b/src/test/java/io/trygvis/test/PlainJavaExample.java
index 338abad..b09d3e9 100644
--- a/src/test/java/io/trygvis/test/PlainJavaExample.java
+++ b/src/test/java/io/trygvis/test/PlainJavaExample.java
@@ -1,22 +1,26 @@
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.QueueSystem;
import io.trygvis.queue.Task;
+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.Random;
-import static io.trygvis.queue.JdbcQueueService.createQueueService;
-import static io.trygvis.queue.Task.newTask;
import static io.trygvis.test.DbUtil.createDataSource;
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";
@@ -28,26 +32,40 @@ public class PlainJavaExample {
System.out.println("Starting consumer");
DataSource ds = createDataSource();
- Connection c = ds.getConnection();
- JdbcQueueService queueService = createQueueService(c);
+ SqlEffectExecutor sqlEffectExecutor = new SqlEffectExecutor(ds);
+
+ 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[]{
+ queueService.lookupQueue(c, inputName, interval, true),
+ queueService.lookupQueue(c, outputName, interval, true)};
+ }
+ });
- final Queue input = queueService.getQueue(c, inputName, interval, true);
- final Queue output = queueService.getQueue(c, outputName, interval, true);
+ final Queue input = queues[0];
+ final Queue output = queues[1];
- queueService.consume(c, input, new QueueService.TaskEffect() {
- public List<Task> consume(Task task) throws Exception {
- System.out.println("PlainJavaExample$Consumer.consume");
+ queueService.consumeAll(input, 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));
- return singletonList(newTask(output.name, new Date(), Long.toString(a + b)));
+ if(r.nextInt(3) == 0) {
+ return singletonList(task.childTask(output.name, new Date(), Long.toString(a + b)));
+ }
+
+ throw new RuntimeException("Simulated exception while processing task.");
}
});
-
- c.commit();
+ System.out.println("Done");
}
}
@@ -58,9 +76,12 @@ public class PlainJavaExample {
DataSource ds = createDataSource();
Connection c = ds.getConnection();
- JdbcQueueService queueService = createQueueService(c);
+ SqlEffectExecutor sqlEffectExecutor = new SqlEffectExecutor(ds);
+
+ QueueSystem queueSystem = QueueSystem.initialize(sqlEffectExecutor);
+ JdbcQueueService queueService = queueSystem.queueService;
- Queue queue = queueService.getQueue(c, inputName, interval, true);
+ Queue queue = queueService.lookupQueue(c, inputName, interval, true);
queueService.schedule(c, queue, new Date(), asList("10", "20"));
diff --git a/src/test/java/io/trygvis/test/PlainJavaExample2.java b/src/test/java/io/trygvis/test/PlainJavaExample2.java
new file mode 100644
index 0000000..faeebb2
--- /dev/null
+++ b/src/test/java/io/trygvis/test/PlainJavaExample2.java
@@ -0,0 +1,84 @@
+package io.trygvis.test;
+
+import io.trygvis.async.JdbcAsyncService;
+import io.trygvis.async.SqlEffectExecutor;
+import io.trygvis.queue.JdbcQueueService;
+import io.trygvis.queue.Queue;
+import io.trygvis.queue.QueueSystem;
+import io.trygvis.queue.Task;
+import io.trygvis.queue.TaskEffect;
+
+import javax.sql.DataSource;
+import java.sql.Connection;
+import java.util.Date;
+import java.util.List;
+
+import static io.trygvis.queue.Task.newTask;
+import static io.trygvis.test.DbUtil.createDataSource;
+import static java.util.Arrays.asList;
+import static java.util.Collections.singletonList;
+
+public class PlainJavaExample2 {
+
+ private static String inputName = "my-input";
+ private static String outputName = "my-output";
+
+ private static int interval = 10;
+
+ private static final TaskEffect adder = new TaskEffect() {
+ public List<Task> apply(Task task) throws Exception {
+ System.out.println("PlainJavaExample$Consumer.consumeAll");
+ 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));
+
+ return singletonList(newTask(outputName, new Date(), Long.toString(a + b)));
+ }
+ };
+
+ public static class Consumer {
+ public static void main(String[] args) throws Exception {
+ System.out.println("Starting consumer");
+
+ DataSource ds = createDataSource();
+ Connection c = ds.getConnection();
+
+ SqlEffectExecutor sqlEffectExecutor = new SqlEffectExecutor(ds);
+
+ QueueSystem queueSystem = QueueSystem.initialize(sqlEffectExecutor);
+ JdbcQueueService queueService = queueSystem.queueService;
+
+ final Queue input = queueService.lookupQueue(c, inputName, interval, true);
+ final Queue output = queueService.lookupQueue(c, outputName, interval, true);
+
+ JdbcAsyncService asyncService = new JdbcAsyncService(queueSystem);
+
+ asyncService.registerQueue(input, adder);
+
+// queueService.consumeAll(c, input, adder);
+
+ c.commit();
+ }
+ }
+
+ public static class Producer {
+ public static void main(String[] args) throws Exception {
+ System.out.println("Starting producer");
+
+ DataSource ds = createDataSource();
+ Connection c = ds.getConnection();
+
+ SqlEffectExecutor sqlEffectExecutor = new SqlEffectExecutor(ds);
+
+ QueueSystem queueSystem = QueueSystem.initialize(sqlEffectExecutor);
+ JdbcQueueService queueService = queueSystem.queueService;
+
+ Queue queue = queueService.lookupQueue(c, inputName, interval, true);
+
+ queueService.schedule(c, queue, new Date(), asList("10", "20"));
+
+ c.commit();
+ }
+ }
+}
diff --git a/src/test/java/io/trygvis/test/UpdateArticleCallable.java b/src/test/java/io/trygvis/test/UpdateArticleCallable.java
index f50c10a..6aff20f 100755
--- a/src/test/java/io/trygvis/test/UpdateArticleCallable.java
+++ b/src/test/java/io/trygvis/test/UpdateArticleCallable.java
@@ -1,6 +1,7 @@
package io.trygvis.test;
-import io.trygvis.async.AsyncService;
+import io.trygvis.queue.Task;
+import io.trygvis.queue.TaskEffect;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
@@ -9,14 +10,18 @@ import java.util.Date;
import java.util.List;
import java.util.Random;
+import static java.util.Collections.emptyList;
+
@Component("updateArticle")
-public class UpdateArticleCallable
- implements AsyncService.AsyncCallable {
+public class UpdateArticleCallable implements TaskEffect {
private final Logger log = LoggerFactory.getLogger(getClass());
private final Random r = new Random();
- public void run(List<String> arguments) throws Exception {
+ @Override
+ public List<Task> apply(Task task) throws Exception {
+ List<String> arguments = task.arguments;
+
log.info("UpdateArticeJob.run: BEGIN");
Date now = new Date();
@@ -36,5 +41,7 @@ public class UpdateArticleCallable
*/
log.info("UpdateArticeJob.run: END");
+
+ return emptyList();
}
}
diff --git a/src/test/java/io/trygvis/test/spring/PlainSpringTest.java b/src/test/java/io/trygvis/test/spring/PlainSpringTest.java
index 07e67fb..d06d8d6 100644
--- a/src/test/java/io/trygvis/test/spring/PlainSpringTest.java
+++ b/src/test/java/io/trygvis/test/spring/PlainSpringTest.java
@@ -3,6 +3,8 @@ package io.trygvis.test.spring;
import io.trygvis.async.AsyncService;
import io.trygvis.queue.Queue;
import io.trygvis.queue.QueueService;
+import io.trygvis.queue.Task;
+import io.trygvis.queue.TaskEffect;
import io.trygvis.spring.DefaultConfig;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -17,6 +19,7 @@ import java.util.concurrent.atomic.AtomicReference;
import static java.lang.System.getProperty;
import static java.lang.System.setProperty;
import static java.util.Arrays.asList;
+import static java.util.Collections.emptyList;
import static org.fest.assertions.Assertions.assertThat;
import static org.junit.Assert.assertNotNull;
@@ -41,13 +44,15 @@ public class PlainSpringTest {
public void testBasic() throws SQLException, InterruptedException {
Queue test = queueService.getQueue("test", 10, true);
final AtomicReference<List<String>> ref = new AtomicReference<>();
- asyncService.registerQueue(test, new AsyncService.AsyncCallable() {
- public void run(List<String> arguments) throws Exception {
+ asyncService.registerQueue(test, new TaskEffect() {
+ @Override
+ public List<Task> apply(Task task) throws Exception {
System.out.println("PlainSpringTest.run");
- ref.set(arguments);
+ ref.set(task.arguments);
synchronized (ref) {
ref.notify();
}
+ return emptyList();
}
});
generated by cgit v1.2.3 (git 2.39.1) at 2024-05-18 00:12:13 +0000