2 files changed, 38 insertions, 24 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);
             }
         });
     }
diff --git a/src/main/java/io/trygvis/spring/SpringQueueService.java b/src/main/java/io/trygvis/spring/SpringQueueService.java
index 271e9bf..2027ab5 100644
--- a/src/main/java/io/trygvis/spring/SpringQueueService.java
+++ b/src/main/java/io/trygvis/spring/SpringQueueService.java
@@ -2,9 +2,9 @@ package io.trygvis.spring;
 
 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.TaskEffect;
 import org.springframework.dao.DataAccessException;
 import org.springframework.jdbc.core.ConnectionCallback;
 import org.springframework.jdbc.core.JdbcTemplate;
@@ -23,21 +23,14 @@ public class SpringQueueService implements QueueService {
 
     public SpringQueueService(QueueSystem queueSystem, JdbcTemplate jdbcTemplate) {
         this.jdbcTemplate = jdbcTemplate;
-        this.queueService = queueSystem.queueService;
-    }
-
-    /**
-     * @see JdbcQueueService#consumeAll(io.trygvis.queue.Queue, io.trygvis.queue.QueueService.TaskExecutionRequest, io.trygvis.queue.TaskEffect)
-     */
-    public void consume(final Queue queue, TaskExecutionRequest req, final TaskEffect effect) throws SQLException {
-        queueService.consumeAll(queue, req, effect);
+        this.queueService = queueSystem.createQueueService();
     }
 
     @Transactional
-    public Queue getQueue(final String name, final int interval, final boolean autoCreate) throws SQLException {
-        return jdbcTemplate.execute(new ConnectionCallback<Queue>() {
+    public QueueExecutor getQueue(final String name, final int interval, final boolean autoCreate) throws SQLException {
+        return jdbcTemplate.execute(new ConnectionCallback<QueueExecutor>() {
             @Override
-            public Queue doInConnection(Connection c) throws SQLException, DataAccessException {
+            public QueueExecutor doInConnection(Connection c) throws SQLException, DataAccessException {
                 return queueService.lookupQueue(c, name, interval, autoCreate);
             }
         });
@@ -48,7 +41,7 @@ public class SpringQueueService implements QueueService {
         jdbcTemplate.execute(new ConnectionCallback<Object>() {
             @Override
             public Object doInConnection(Connection c) throws SQLException, DataAccessException {
-                queueService.schedule(c, queue, scheduled, arguments);
+                queueService.getQueue(queue.name).schedule(c, scheduled, arguments);
                 return null;
             }
         });