1 files changed, 32 insertions, 11 deletions
diff --git a/src/main/java/io/trygvis/spring/SpringJdbcAsyncService.java b/src/main/java/io/trygvis/spring/SpringJdbcAsyncService.java
index b27e94d..a1c9cda 100644
--- a/src/main/java/io/trygvis/spring/SpringJdbcAsyncService.java
+++ b/src/main/java/io/trygvis/spring/SpringJdbcAsyncService.java
@@ -2,8 +2,12 @@ package io.trygvis.spring;
 
 import io.trygvis.async.AsyncService;
 import io.trygvis.async.JdbcAsyncService;
-import io.trygvis.async.SqlEffectExecutor;
+import io.trygvis.async.QueueController;
+import io.trygvis.async.SqlEffect;
+import io.trygvis.queue.JdbcQueueService;
 import io.trygvis.queue.Queue;
+import io.trygvis.queue.QueueExecutor;
+import io.trygvis.queue.QueueService;
 import io.trygvis.queue.QueueSystem;
 import io.trygvis.queue.Task;
 import io.trygvis.queue.TaskEffect;
@@ -17,6 +21,7 @@ import org.springframework.transaction.support.TransactionSynchronizationAdapter
 
 import java.sql.Connection;
 import java.sql.SQLException;
+import java.util.Date;
 import java.util.List;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
@@ -27,52 +32,68 @@ import static org.springframework.transaction.support.TransactionSynchronization
 public class SpringJdbcAsyncService implements AsyncService {
     private final Logger log = LoggerFactory.getLogger(getClass());
 
-    private final ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(10, Executors.defaultThreadFactory());
+    private final ScheduledThreadPoolExecutor executor;
 
     private final JdbcTemplate jdbcTemplate;
 
     private final JdbcAsyncService jdbcAsyncService;
 
+    private final JdbcQueueService queueService;
+
+    private final QueueSystem queueSystem;
+
     public SpringJdbcAsyncService(QueueSystem queueSystem, JdbcTemplate jdbcTemplate) {
+        this.queueSystem = queueSystem;
         this.jdbcTemplate = jdbcTemplate;
         jdbcAsyncService = new JdbcAsyncService(queueSystem);
+        queueService = queueSystem.createQueueService();
+        executor = new ScheduledThreadPoolExecutor(10, Executors.defaultThreadFactory());
     }
 
     @Transactional(propagation = REQUIRED)
-    public void registerQueue(final Queue queue, final TaskEffect processor) {
-        jdbcAsyncService.registerQueue(queue, processor);
+    public QueueController registerQueue(final Queue queue, final QueueService.TaskExecutionRequest req, final TaskEffect processor) throws SQLException {
+        QueueExecutor queueExecutor = queueSystem.sqlEffectExecutor.transaction(new SqlEffect<QueueExecutor>() {
+            @Override
+            public QueueExecutor doInConnection(Connection c) throws SQLException {
+                return queueService.lookupQueue(c, queue.name, queue.interval, true);
+            }
+        });
+
+        final QueueController queueController = jdbcAsyncService.registerQueue(queueExecutor, req, processor);
 
         registerSynchronization(new TransactionSynchronizationAdapter() {
             public void afterCompletion(int status) {
                 log.info("Transaction completed with status = {}", status);
                 if (status == TransactionSynchronization.STATUS_COMMITTED) {
-                    jdbcAsyncService.startQueue(queue, executor);
+                    queueController.start(executor);
                 }
             }
         });
 
-        log.info("registerQueue: LEAVE");
+        return queueController;
     }
 
-    public Queue getQueue(String name) {
+    public QueueExecutor getQueue(String name) {
         return jdbcAsyncService.getQueue(name);
     }
 
     @Transactional(propagation = REQUIRED)
-    public Task schedule(final Queue queue, final List<String> args) {
+    public Task schedule(final Queue queue, final Date scheduled, final List<String> args) {
         return jdbcTemplate.execute(new ConnectionCallback<Task>() {
             @Override
             public Task doInConnection(Connection c) throws SQLException {
-                return jdbcAsyncService.schedule(c, queue, args);
+                QueueExecutor queueExecutor = queueService.getQueue(queue.name);
+                return queueExecutor.schedule(c, scheduled, args);
             }
         });
     }
 
-    public Task schedule(final long parent, final Queue queue, final List<String> args) {
+    public Task schedule(final Queue queue, final long parent, final Date scheduled, 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);
+                QueueExecutor queueExecutor = queueService.getQueue(queue.name);
+                return queueExecutor.schedule(c, parent, scheduled, args);
             }
         });
     }