aboutsummaryrefslogtreecommitdiff
path: root/src/test
diff options
context:
space:
mode:
Diffstat (limited to 'src/test')
-rw-r--r--src/test/java/io/trygvis/test/DbUtil.java4
-rw-r--r--src/test/java/io/trygvis/test/activemq/AsyncConsumerWithActiveMq.java37
-rw-r--r--src/test/java/io/trygvis/test/jdbc/AsyncConsumerExample.java18
-rwxr-xr-xsrc/test/resources/logback.xml2
4 files changed, 53 insertions, 8 deletions
diff --git a/src/test/java/io/trygvis/test/DbUtil.java b/src/test/java/io/trygvis/test/DbUtil.java
index 025442e..33c2807 100644
--- a/src/test/java/io/trygvis/test/DbUtil.java
+++ b/src/test/java/io/trygvis/test/DbUtil.java
@@ -37,8 +37,8 @@ public class DbUtil {
ds.setIdleConnectionTestPeriodInSeconds(60);
ds.setIdleMaxAgeInSeconds(240);
ds.setMaxConnectionsPerPartition(4);
- ds.setMinConnectionsPerPartition(2);
- ds.setPartitionCount(2);
+ ds.setMinConnectionsPerPartition(1);
+ ds.setPartitionCount(4);
ds.setAcquireIncrement(1);
ds.setStatementsCacheSize(1000);
ds.setReleaseHelperThreads(1);
diff --git a/src/test/java/io/trygvis/test/activemq/AsyncConsumerWithActiveMq.java b/src/test/java/io/trygvis/test/activemq/AsyncConsumerWithActiveMq.java
new file mode 100644
index 0000000..bd86732
--- /dev/null
+++ b/src/test/java/io/trygvis/test/activemq/AsyncConsumerWithActiveMq.java
@@ -0,0 +1,37 @@
+package io.trygvis.test.activemq;
+
+import io.trygvis.activemq.ActiveMqHinter;
+import io.trygvis.async.QueueController;
+import io.trygvis.queue.QueueService;
+import io.trygvis.test.jdbc.AsyncConsumerExample;
+import org.apache.activemq.ActiveMQConnectionFactory;
+
+import javax.jms.JMSException;
+
+public class AsyncConsumerWithActiveMq extends AsyncConsumerExample implements AutoCloseable {
+ private final ActiveMqHinter activeMqHinter;
+
+ public AsyncConsumerWithActiveMq(ActiveMQConnectionFactory cf) throws JMSException {
+ activeMqHinter = new ActiveMqHinter(cf);
+ }
+
+ public static void main(String[] args) throws Exception {
+ ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory("failover:tcp://localhost:61616");
+
+ int poolSize = 4;
+ QueueService.TaskExecutionRequest req = new QueueService.TaskExecutionRequest(1, true).
+ interval(60 * 6000).
+ continueOnFullChunk(false);
+ try (AsyncConsumerWithActiveMq consumer = new AsyncConsumerWithActiveMq(cf)) {
+ consumer.work(poolSize, req);
+ }
+ }
+
+ public void close() throws JMSException {
+ activeMqHinter.close();
+ }
+
+ protected void wrapInputQueue(QueueController input) throws Exception {
+ activeMqHinter.createHinter(input);
+ }
+}
diff --git a/src/test/java/io/trygvis/test/jdbc/AsyncConsumerExample.java b/src/test/java/io/trygvis/test/jdbc/AsyncConsumerExample.java
index 14db21b..16640dd 100644
--- a/src/test/java/io/trygvis/test/jdbc/AsyncConsumerExample.java
+++ b/src/test/java/io/trygvis/test/jdbc/AsyncConsumerExample.java
@@ -3,12 +3,12 @@ package io.trygvis.test.jdbc;
import io.trygvis.async.JdbcAsyncService;
import io.trygvis.async.QueueController;
import io.trygvis.async.QueueStats;
-import io.trygvis.queue.SqlEffect;
-import io.trygvis.queue.SqlEffectExecutor;
import io.trygvis.queue.JdbcQueueService;
import io.trygvis.queue.QueueExecutor;
import io.trygvis.queue.QueueService;
import io.trygvis.queue.QueueSystem;
+import io.trygvis.queue.SqlEffect;
+import io.trygvis.queue.SqlEffectExecutor;
import io.trygvis.queue.Task;
import io.trygvis.queue.TaskEffect;
@@ -43,6 +43,12 @@ public class AsyncConsumerExample {
};
public static void main(String[] args) throws Exception {
+ int poolSize = 4;
+ QueueService.TaskExecutionRequest req = new QueueService.TaskExecutionRequest(100, true);
+ new AsyncConsumerExample().work(poolSize, req);
+ }
+
+ public void work(int poolSize, QueueService.TaskExecutionRequest req) throws Exception {
System.out.println("Starting consumer");
DataSource ds = createDataSource();
@@ -66,9 +72,8 @@ public class AsyncConsumerExample {
final QueueExecutor input = queues[0];
final QueueExecutor output = queues[1];
- QueueService.TaskExecutionRequest req = new QueueService.TaskExecutionRequest(100, true);
-
QueueController controller = asyncService.registerQueue(input, req, adder);
+ wrapInputQueue(controller);
Timer timer = new Timer();
timer.scheduleAtFixedRate(new TimerTask() {
@@ -80,7 +85,7 @@ public class AsyncConsumerExample {
}, 1000, 1000);
long start = currentTimeMillis();
- controller.start(new ScheduledThreadPoolExecutor(4));
+ controller.start(new ScheduledThreadPoolExecutor(poolSize));
Thread.sleep(60 * 1000);
controller.stop();
long end = currentTimeMillis();
@@ -96,4 +101,7 @@ public class AsyncConsumerExample {
double rate = 1000 * ((double) stats.totalMessageCount) / duration;
System.out.println("Consumed " + stats.totalMessageCount + " messages in " + duration + "ms at " + rate + " msg/s");
}
+
+ protected void wrapInputQueue(QueueController input) throws Exception {
+ }
}
diff --git a/src/test/resources/logback.xml b/src/test/resources/logback.xml
index f6e0f47..f964cd3 100755
--- a/src/test/resources/logback.xml
+++ b/src/test/resources/logback.xml
@@ -23,7 +23,7 @@
</encoder>
</appender>
- <root level="INFO">
+ <root level="WARN">
<appender-ref ref="STDOUT"/>
</root>
</configuration>