o Creating a QueueService on top of the DAOs.

9 files changed, 263 insertions, 78 deletions

diff --git a/src/main/java/io/trygvis/async/AsyncService.java b/src/main/java/io/trygvis/async/AsyncService.java
index 57c1af8..17d53e9 100755
--- a/src/main/java/io/trygvis/async/AsyncService.java
+++ b/src/main/java/io/trygvis/async/AsyncService.java
@@ -3,7 +3,6 @@ package io.trygvis.async;
 import io.trygvis.queue.Queue;
 import io.trygvis.queue.Task;
 
-import java.sql.SQLException;
 import java.util.List;
 
 /**
@@ -11,19 +10,13 @@ import java.util.List;
  */
 public interface AsyncService {
 
-    /**
-     * @param name
-     * @param interval how often the queue should be polled for missed tasks in seconds.
-     * @param callable
-     * @return
-     */
-    Queue registerQueue(final String name, final int interval, AsyncCallable callable);
+    void registerQueue(Queue queue, final AsyncService.AsyncCallable callable);
 
     Queue getQueue(String name);
 
-    Task schedule(Queue queue, String... args);
+    Task schedule(Queue queue, List<String> args);
 
-    Task schedule(long parent, Queue queue, String... args);
+    Task schedule(long parent, Queue queue, List<String> args);
 
     /**
      * Polls for a new state of the execution.
diff --git a/src/main/java/io/trygvis/async/JdbcAsyncService.java b/src/main/java/io/trygvis/async/JdbcAsyncService.java
index c34330e..310c59b 100644
--- a/src/main/java/io/trygvis/async/JdbcAsyncService.java
+++ b/src/main/java/io/trygvis/async/JdbcAsyncService.java
@@ -1,7 +1,6 @@
 package io.trygvis.async;
 
 import io.trygvis.queue.Queue;
-import io.trygvis.queue.QueueDao;
 import io.trygvis.queue.Task;
 import io.trygvis.queue.TaskDao;
 import org.slf4j.Logger;
@@ -11,12 +10,12 @@ import java.sql.Connection;
 import java.sql.SQLException;
 import java.util.Date;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 
 import static java.lang.System.currentTimeMillis;
 import static java.lang.Thread.sleep;
-import static java.util.Arrays.asList;
 import static java.util.concurrent.TimeUnit.SECONDS;
 
 public class JdbcAsyncService {
@@ -24,25 +23,12 @@ public class JdbcAsyncService {
 
     private final Map<String, QueueThread> queues = new HashMap<>();
 
-    public Queue registerQueue(Connection c, SqlEffectExecutor sqlEffectExecutor, final String name, final int interval, AsyncService.AsyncCallable callable) throws SQLException {
-        QueueDao queueDao = new QueueDao(c);
+    public void registerQueue(SqlEffectExecutor sqlEffectExecutor, Queue queue, AsyncService.AsyncCallable callable) {
+        final QueueThread queueThread = new QueueThread(sqlEffectExecutor, callable, queue);
 
-        log.info("registerQueue: ENTER");
-
-        Queue q = queueDao.findByName(name);
-
-        log.info("q = {}", q);
-
-        if (q == null) {
-            q = new Queue(name, interval);
-            queueDao.insert(q);
-        }
-
-        final QueueThread queueThread = new QueueThread(sqlEffectExecutor, callable, q);
-        queues.put(name, queueThread);
+        queues.put(queue.name, queueThread);
 
         log.info("registerQueue: LEAVE");
-        return q;
     }
 
     public void startQueue(ScheduledThreadPoolExecutor executor, String name) {
@@ -74,26 +60,21 @@ public class JdbcAsyncService {
         return queueThread.queue;
     }
 
-    public Task schedule(Connection c, final Queue queue, String... args) throws SQLException {
+    public Task schedule(Connection c, final Queue queue, List<String> args) throws SQLException {
         return scheduleInner(c, null, queue, args);
     }
 
-    public Task schedule(Connection c, long parent, Queue queue, String... args) throws SQLException {
+    public Task schedule(Connection c, long parent, Queue queue, List<String> args) throws SQLException {
         return scheduleInner(c, parent, queue, args);
     }
 
-    private Task scheduleInner(Connection c, Long parent, final Queue queue, String... args) throws SQLException {
+    private Task scheduleInner(Connection c, Long parent, final Queue queue, List<String> args) throws SQLException {
         TaskDao taskDao = new TaskDao(c);
 
         Date scheduled = new Date();
 
-        StringBuilder arguments = new StringBuilder();
-        for (String arg : args) {
-            arguments.append(arg).append(' ');
-        }
-
-        long id = taskDao.insert(parent, queue.name, scheduled, arguments.toString());
-        Task task = new Task(id, parent, queue.name, scheduled, null, 0, null, asList(args));
+        long id = taskDao.insert(parent, queue.name, scheduled, args);
+        Task task = new Task(id, parent, queue.name, scheduled, null, 0, null, args);
         log.info("Created task = {}", task);
 
         return task;
@@ -123,6 +104,6 @@ public class JdbcAsyncService {
     public Task update(Connection c, Task ref) throws SQLException {
         TaskDao taskDao = new TaskDao(c);
 
-        return taskDao.findById(ref.id);
+        return taskDao.findById(ref.id());
     }
 }
diff --git a/src/main/java/io/trygvis/queue/JdbcQueueService.java b/src/main/java/io/trygvis/queue/JdbcQueueService.java
new file mode 100644
index 0000000..793333d
--- /dev/null
+++ b/src/main/java/io/trygvis/queue/JdbcQueueService.java
@@ -0,0 +1,91 @@
+package io.trygvis.queue;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.SQLException;
+import java.util.Date;
+import java.util.List;
+
+public class JdbcQueueService {
+
+    private Logger log = LoggerFactory.getLogger(getClass());
+
+    private JdbcQueueService(Connection c) throws SQLException {
+        if (c.getAutoCommit()) {
+            throw new SQLException("The connection cannot be in auto-commit mode.");
+        }
+
+        DatabaseMetaData metaData = c.getMetaData();
+        String productName = metaData.getDatabaseProductName();
+        String productVersion = metaData.getDatabaseProductVersion();
+
+        log.info("productName = " + productName);
+        log.info("productVersion = " + productVersion);
+    }
+
+    public void consume(Connection c, Queue queue, QueueService.TaskEffect effect) throws SQLException {
+        TaskDao taskDao = createTaskDao(c);
+
+        List<Task> tasks = taskDao.findByNameAndCompletedIsNull(queue.name);
+        log.trace("Got {} tasks.", tasks.size());
+
+        for (Task task : tasks) {
+            log.trace("Executing task {}", task.id());
+            try {
+                List<Task> newTasks = effect.consume(task);
+                log.trace("Executed task {}, newTasks: ", task.id(), newTasks.size());
+
+                Date now = new Date();
+
+                task = task.registerComplete(now);
+
+                for (Task newTask : newTasks) {
+                    taskDao.insert(task.id(), newTask.queue, now, newTask.arguments);
+                }
+
+                taskDao.update(task);
+            } catch (Throwable e) {
+                log.error("Unable to execute task, id=" + task.id(), e);
+            }
+            c.commit();
+        }
+    }
+
+    public Queue getQueue(Connection c, String name, int interval, boolean autoCreate) throws SQLException {
+        QueueDao queueDao = createQueueDao(c);
+
+        Queue q = queueDao.findByName(name);
+
+        if (q == null) {
+            if (!autoCreate) {
+                throw new SQLException("No such queue: '" + name + "'.");
+            }
+
+            q = new Queue(name, interval);
+            queueDao.insert(q);
+        }
+
+        return q;
+    }
+
+    public void schedule(Connection c, Queue queue, Date scheduled, List<String> arguments) throws SQLException {
+        TaskDao taskDao = createTaskDao(c);
+
+        taskDao.insert(queue.name, scheduled, arguments);
+    }
+
+    public static JdbcQueueService createQueueService(Connection c) throws SQLException {
+        return new JdbcQueueService(c);
+    }
+
+    public QueueDao createQueueDao(Connection c) {
+        return new QueueDao(c);
+    }
+
+    public TaskDao createTaskDao(Connection c) {
+        return new TaskDao(c);
+    }
+}
diff --git a/src/main/java/io/trygvis/queue/QueueService.java b/src/main/java/io/trygvis/queue/QueueService.java
new file mode 100644
index 0000000..2111013
--- /dev/null
+++ b/src/main/java/io/trygvis/queue/QueueService.java
@@ -0,0 +1,17 @@
+package io.trygvis.queue;
+
+import java.sql.SQLException;
+import java.util.Date;
+import java.util.List;
+
+public interface QueueService {
+    void consume(Queue queue, TaskEffect effect) throws SQLException;
+
+    Queue getQueue(String name, int interval, boolean autoCreate) throws SQLException;
+
+    void schedule(Queue queue, Date scheduled, List<String> arguments) throws SQLException;
+
+    public static interface TaskEffect {
+        List<Task> consume(Task task) throws Exception;
+    }
+}
diff --git a/src/main/java/io/trygvis/queue/Task.java b/src/main/java/io/trygvis/queue/Task.java
index 2b1103b..1af40d7 100755
--- a/src/main/java/io/trygvis/queue/Task.java
+++ b/src/main/java/io/trygvis/queue/Task.java
@@ -3,9 +3,11 @@ package io.trygvis.queue;
 import java.util.Date;
 import java.util.List;
 
+import static java.util.Arrays.asList;
+
 public class Task {
 
-    public final long id;
+    private final long id;
 
     public final Long parent;
 
@@ -54,7 +56,42 @@ public class Task {
                 '}';
     }
 
+    public long id() {
+        if (id == 0) {
+            throw new RuntimeException("This task has not been persisted yet.");
+        }
+
+        return id;
+    }
+
     public boolean isDone() {
         return completed != null;
     }
+
+    public static Task newTask(String name, Date scheduled, String... arguments) {
+        return new Task(0, 0l, name, scheduled, null, 0, null, asList(arguments));
+    }
+
+    public static Task newTask(String name, Date scheduled, List<String> arguments) {
+        return new Task(0, 0l, name, scheduled, null, 0, null, arguments);
+    }
+
+    public static List<String> stringToArguments(String arguments) {
+        return asList(arguments.split(","));
+    }
+
+    public static String argumentsToString(List<String> arguments) {
+        StringBuilder builder = new StringBuilder();
+        for (int i = 0, argumentsLength = arguments.size(); i < argumentsLength; i++) {
+            String argument = arguments.get(i);
+            if (argument.contains(",")) {
+                throw new RuntimeException("The argument string can't contain a comma.");
+            }
+            if (i > 0) {
+                builder.append(',');
+            }
+            builder.append(argument);
+        }
+        return builder.toString();
+    }
 }
diff --git a/src/main/java/io/trygvis/queue/TaskDao.java b/src/main/java/io/trygvis/queue/TaskDao.java
index 5459933..3aa2ac2 100644
--- a/src/main/java/io/trygvis/queue/TaskDao.java
+++ b/src/main/java/io/trygvis/queue/TaskDao.java
@@ -11,7 +11,8 @@ import java.util.Collections;
 import java.util.Date;
 import java.util.List;
 
-import static java.util.Arrays.asList;
+import static io.trygvis.queue.Task.argumentsToString;
+import static io.trygvis.queue.Task.stringToArguments;
 
 public class TaskDao {
 
@@ -23,11 +24,11 @@ public class TaskDao {
         this.connection = connection;
     }
 
-    public long insert(String queue, Date scheduled, String arguments) throws SQLException {
+    public long insert(String queue, Date scheduled, List<String> arguments) throws SQLException {
         return insert(null, queue, scheduled, arguments);
     }
 
-    public long insert(Long parent, String queue, Date scheduled, String arguments) throws SQLException {
+    public long insert(Long parent, String queue, Date scheduled, List<String> arguments) throws SQLException {
         String sql = "INSERT INTO task(id, parent, run_count, queue, scheduled, arguments) " +
                 "VALUES(nextval('task_seq'), ?, 0, ?, ?, ?)";
         try (PreparedStatement stmt = connection.prepareStatement(sql)) {
@@ -39,7 +40,7 @@ public class TaskDao {
             }
             stmt.setString(i++, queue);
             stmt.setTimestamp(i++, new Timestamp(scheduled.getTime()));
-            stmt.setString(i, arguments);
+            stmt.setString(i, argumentsToString(arguments));
             stmt.executeUpdate();
         }
         try (PreparedStatement stmt = connection.prepareStatement("SELECT currval('task_seq')")) {
@@ -51,6 +52,7 @@ public class TaskDao {
 
     public Task findById(long id) throws SQLException {
         try (PreparedStatement stmt = connection.prepareStatement("SELECT " + fields + " FROM task WHERE id=?")) {
+            stmt.setLong(1, id);
             ResultSet rs = stmt.executeQuery();
             return rs.next() ? mapRow(rs) : null;
         }
@@ -76,7 +78,7 @@ public class TaskDao {
             setTimestamp(stmt, i++, task.lastRun);
             stmt.setInt(i++, task.runCount);
             setTimestamp(stmt, i++, task.completed);
-            stmt.setLong(i, task.id);
+            stmt.setLong(i, task.id());
             stmt.executeUpdate();
         }
     }
@@ -99,6 +101,6 @@ public class TaskDao {
                 rs.getTimestamp(5),
                 rs.getInt(6),
                 rs.getTimestamp(7),
-                arguments != null ? asList(arguments.split(" ")) : Collections.<String>emptyList());
+                arguments != null ? stringToArguments(arguments) : Collections.<String>emptyList());
     }
 }
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/async/spring/SpringJdbcAsyncService.java b/src/main/java/io/trygvis/spring/SpringJdbcAsyncService.java
index 327dffa..6702642 100644
--- a/src/main/java/io/trygvis/async/spring/SpringJdbcAsyncService.java
+++ b/src/main/java/io/trygvis/spring/SpringJdbcAsyncService.java
@@ -1,4 +1,4 @@
-package io.trygvis.async.spring;
+package io.trygvis.spring;
 
 import io.trygvis.async.AsyncService;
 import io.trygvis.async.JdbcAsyncService;
@@ -12,10 +12,10 @@ 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 org.springframework.transaction.support.TransactionTemplate;
 
 import java.sql.Connection;
 import java.sql.SQLException;
+import java.util.List;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 
@@ -27,42 +27,32 @@ public class SpringJdbcAsyncService implements AsyncService {
 
     private final ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(10, Executors.defaultThreadFactory());
 
-    private final TransactionTemplate transactionTemplate;
-
     private final JdbcTemplate jdbcTemplate;
 
-    private SqlEffectExecutor sqlEffectExecutor;
+    private final SqlEffectExecutor sqlEffectExecutor;
 
-    final JdbcAsyncService jdbcAsyncService;
+    private final JdbcAsyncService jdbcAsyncService;
 
-    public SpringJdbcAsyncService(TransactionTemplate transactionTemplate, JdbcTemplate jdbcTemplate) {
-        this.transactionTemplate = transactionTemplate;
+    public SpringJdbcAsyncService(JdbcTemplate jdbcTemplate) {
         this.jdbcTemplate = jdbcTemplate;
         jdbcAsyncService = new JdbcAsyncService();
         sqlEffectExecutor = new SqlEffectExecutor(this.jdbcTemplate.getDataSource());
     }
 
     @Transactional(propagation = REQUIRED)
-    public Queue registerQueue(final String name, final int interval, final AsyncService.AsyncCallable callable) {
-        return jdbcTemplate.execute(new ConnectionCallback<Queue>() {
-            @Override
-            public Queue doInConnection(Connection c) throws SQLException {
-
-                Queue q = jdbcAsyncService.registerQueue(c, sqlEffectExecutor, name, interval, callable);
-
-                registerSynchronization(new TransactionSynchronizationAdapter() {
-                    public void afterCompletion(int status) {
-                        log.info("Transaction completed with status = {}", status);
-                        if (status == TransactionSynchronization.STATUS_COMMITTED) {
-                            jdbcAsyncService.startQueue(executor, name);
-                        }
-                    }
-                });
-
-                log.info("registerQueue: LEAVE");
-                return q;
+    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) {
@@ -70,7 +60,7 @@ public class SpringJdbcAsyncService implements AsyncService {
     }
 
     @Transactional(propagation = REQUIRED)
-    public Task schedule(final Queue queue, final String... args) {
+    public Task schedule(final Queue queue, final List<String> args) {
         return jdbcTemplate.execute(new ConnectionCallback<Task>() {
             @Override
             public Task doInConnection(Connection c) throws SQLException {
@@ -79,7 +69,7 @@ public class SpringJdbcAsyncService implements AsyncService {
         });
     }
 
-    public Task schedule(final long parent, final Queue queue, final String... 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 {
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;
+            }
+        });
+    }
+}