3 files changed, 168 insertions, 4 deletions
diff --git a/src/main/java/io/trygvis/spring/DefaultConfig.java b/src/main/java/io/trygvis/spring/DefaultConfig.java
index af8f644..68761f2 100644
--- a/src/main/java/io/trygvis/spring/DefaultConfig.java
+++ b/src/main/java/io/trygvis/spring/DefaultConfig.java
@@ -1,17 +1,21 @@
 package io.trygvis.spring;
 
 import io.trygvis.async.AsyncService;
-import io.trygvis.async.spring.SpringJdbcAsyncService;
+import io.trygvis.queue.QueueService;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.jdbc.core.JdbcTemplate;
-import org.springframework.transaction.support.TransactionTemplate;
 
 @Configuration
 public class DefaultConfig {
 
     @Bean
-    public AsyncService asyncService(TransactionTemplate transactionTemplate, JdbcTemplate jdbcTemplate) {
-        return new SpringJdbcAsyncService(transactionTemplate, jdbcTemplate);
+    public AsyncService asyncService(JdbcTemplate jdbcTemplate) {
+        return new SpringJdbcAsyncService(jdbcTemplate);
+    }
+
+    @Bean
+    public QueueService queueService(JdbcTemplate jdbcTemplate) {
+        return new SpringQueueService(jdbcTemplate);
     }
 }
diff --git a/src/main/java/io/trygvis/spring/SpringJdbcAsyncService.java b/src/main/java/io/trygvis/spring/SpringJdbcAsyncService.java
new file mode 100644
index 0000000..6702642
--- /dev/null
+++ b/src/main/java/io/trygvis/spring/SpringJdbcAsyncService.java
@@ -0,0 +1,90 @@
+package io.trygvis.spring;
+
+import io.trygvis.async.AsyncService;
+import io.trygvis.async.JdbcAsyncService;
+import io.trygvis.async.SqlEffectExecutor;
+import io.trygvis.queue.Queue;
+import io.trygvis.queue.Task;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.jdbc.core.ConnectionCallback;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.transaction.annotation.Transactional;
+import org.springframework.transaction.support.TransactionSynchronization;
+import org.springframework.transaction.support.TransactionSynchronizationAdapter;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+
+import static org.springframework.transaction.annotation.Propagation.REQUIRED;
+import static org.springframework.transaction.support.TransactionSynchronizationManager.registerSynchronization;
+
+public class SpringJdbcAsyncService implements AsyncService {
+    private final Logger log = LoggerFactory.getLogger(getClass());
+
+    private final ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(10, Executors.defaultThreadFactory());
+
+    private final JdbcTemplate jdbcTemplate;
+
+    private final SqlEffectExecutor sqlEffectExecutor;
+
+    private final JdbcAsyncService jdbcAsyncService;
+
+    public SpringJdbcAsyncService(JdbcTemplate jdbcTemplate) {
+        this.jdbcTemplate = jdbcTemplate;
+        jdbcAsyncService = new JdbcAsyncService();
+        sqlEffectExecutor = new SqlEffectExecutor(this.jdbcTemplate.getDataSource());
+    }
+
+    @Transactional(propagation = REQUIRED)
+    public void registerQueue(final Queue queue, final AsyncService.AsyncCallable callable) {
+        jdbcAsyncService.registerQueue(sqlEffectExecutor, queue, callable);
+
+        registerSynchronization(new TransactionSynchronizationAdapter() {
+            public void afterCompletion(int status) {
+                log.info("Transaction completed with status = {}", status);
+                if (status == TransactionSynchronization.STATUS_COMMITTED) {
+                    jdbcAsyncService.startQueue(executor, queue.name);
+                }
+            }
+        });
+
+        log.info("registerQueue: LEAVE");
+    }
+
+    public Queue getQueue(String name) {
+        return jdbcAsyncService.getQueue(name);
+    }
+
+    @Transactional(propagation = REQUIRED)
+    public Task schedule(final Queue queue, final List<String> args) {
+        return jdbcTemplate.execute(new ConnectionCallback<Task>() {
+            @Override
+            public Task doInConnection(Connection c) throws SQLException {
+                return jdbcAsyncService.schedule(c, queue, args);
+            }
+        });
+    }
+
+    public Task schedule(final long parent, final Queue queue, final List<String> args) {
+        return jdbcTemplate.execute(new ConnectionCallback<Task>() {
+            @Override
+            public Task doInConnection(Connection c) throws SQLException {
+                return jdbcAsyncService.schedule(c, parent, queue, args);
+            }
+        });
+    }
+
+    @Transactional(readOnly = true)
+    public Task update(final Task ref) {
+        return jdbcTemplate.execute(new ConnectionCallback<Task>() {
+            @Override
+            public Task doInConnection(Connection c) throws SQLException {
+                return jdbcAsyncService.update(c, ref);
+            }
+        });
+    }
+}
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;
+            }
+        });
+    }
+}