aboutsummaryrefslogtreecommitdiff
path: root/src/main/java/io/trygvis/spring/SpringQueueService.java
diff options
context:
space:
mode:
Diffstat (limited to 'src/main/java/io/trygvis/spring/SpringQueueService.java')
-rw-r--r--src/main/java/io/trygvis/spring/SpringQueueService.java70
1 files changed, 70 insertions, 0 deletions
diff --git a/src/main/java/io/trygvis/spring/SpringQueueService.java b/src/main/java/io/trygvis/spring/SpringQueueService.java
new file mode 100644
index 0000000..3432e35
--- /dev/null
+++ b/src/main/java/io/trygvis/spring/SpringQueueService.java
@@ -0,0 +1,70 @@
+package io.trygvis.spring;
+
+import io.trygvis.queue.JdbcQueueService;
+import io.trygvis.queue.Queue;
+import io.trygvis.queue.QueueService;
+import org.springframework.dao.DataAccessException;
+import org.springframework.jdbc.core.ConnectionCallback;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.transaction.annotation.Transactional;
+
+import javax.annotation.PostConstruct;
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.util.Date;
+import java.util.List;
+
+import static io.trygvis.queue.JdbcQueueService.createQueueService;
+
+public class SpringQueueService implements QueueService {
+
+ public final JdbcTemplate jdbcTemplate;
+
+ public JdbcQueueService queueService;
+
+ public SpringQueueService(JdbcTemplate jdbcTemplate) {
+ this.jdbcTemplate = jdbcTemplate;
+ }
+
+ @PostConstruct
+ public void postConstruct() {
+ queueService = jdbcTemplate.execute(new ConnectionCallback<JdbcQueueService>() {
+ @Override
+ public JdbcQueueService doInConnection(Connection c) throws SQLException, DataAccessException {
+ return createQueueService(c);
+ }
+ });
+ }
+
+ @Transactional
+ public void consume(final Queue queue, final TaskEffect effect) throws SQLException {
+ jdbcTemplate.execute(new ConnectionCallback<Object>() {
+ @Override
+ public Object doInConnection(Connection c) throws SQLException, DataAccessException {
+ queueService.consume(c, queue, effect);
+ return null;
+ }
+ });
+ }
+
+ @Transactional
+ public Queue getQueue(final String name, final int interval, final boolean autoCreate) throws SQLException {
+ return jdbcTemplate.execute(new ConnectionCallback<Queue>() {
+ @Override
+ public Queue doInConnection(Connection c) throws SQLException, DataAccessException {
+ return queueService.getQueue(c, name, interval, autoCreate);
+ }
+ });
+ }
+
+ @Transactional
+ public void schedule(final Queue queue, final Date scheduled, final List<String> arguments) throws SQLException {
+ jdbcTemplate.execute(new ConnectionCallback<Object>() {
+ @Override
+ public Object doInConnection(Connection c) throws SQLException, DataAccessException {
+ queueService.schedule(c, queue, scheduled, arguments);
+ return null;
+ }
+ });
+ }
+}