1 files changed, 17 insertions, 67 deletions

diff --git a/src/main/java/io/trygvis/async/QueueThread.java b/src/main/java/io/trygvis/async/QueueThread.java
index 00c46b4..33753a3 100644
--- a/src/main/java/io/trygvis/async/QueueThread.java
+++ b/src/main/java/io/trygvis/async/QueueThread.java
@@ -1,24 +1,27 @@
 package io.trygvis.async;
 
+import io.trygvis.queue.JdbcQueueService;
 import io.trygvis.queue.Queue;
+import io.trygvis.queue.QueueSystem;
 import io.trygvis.queue.Task;
-import io.trygvis.queue.TaskDao;
+import io.trygvis.queue.TaskEffect;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.sql.Connection;
 import java.sql.SQLException;
-import java.util.Date;
 import java.util.List;
 
-import static io.trygvis.async.SqlEffectExecutor.SqlExecutionException;
-
 class QueueThread implements Runnable {
     private final Logger log = LoggerFactory.getLogger(getClass());
 
+    private final QueueSystem queueSystem;
+
+    private final JdbcQueueService queueService;
+
     private final SqlEffectExecutor sqlEffectExecutor;
 
-    private final AsyncService.AsyncCallable callable;
+    private final TaskEffect taskEffect;
 
     public final Queue queue;
 
@@ -28,9 +31,11 @@ class QueueThread implements Runnable {
 
     private boolean busy;
 
-    QueueThread(SqlEffectExecutor sqlEffectExecutor, AsyncService.AsyncCallable callable, Queue queue) {
-        this.sqlEffectExecutor = sqlEffectExecutor;
-        this.callable = callable;
+    QueueThread(QueueSystem queueSystem, TaskEffect taskEffect, Queue queue) {
+        this.queueSystem = queueSystem;
+        this.sqlEffectExecutor = queueSystem.sqlEffectExecutor;
+        this.queueService = queueSystem.createQueueService();
+        this.taskEffect = taskEffect;
         this.queue = queue;
     }
 
@@ -48,28 +53,17 @@ class QueueThread implements Runnable {
     public void run() {
         while (shouldRun) {
             try {
-//                List<Task> tasks = transactionTemplate.execute(new TransactionCallback<List<Task>>() {
-//                    public List<Task> doInTransaction(TransactionStatus status) {
-//                        return taskDao.findByNameAndCompletedIsNull(queue.name);
-//                    }
-//                });
-                List<Task> tasks = sqlEffectExecutor.execute(new SqlEffect<List<Task>>() {
+                List<Task> tasks = sqlEffectExecutor.transaction(new SqlEffect<List<Task>>() {
                     @Override
-                    public List<Task> doInConnection(Connection connection) throws SQLException {
-                        return new TaskDao(connection).findByNameAndCompletedIsNull(queue.name);
+                    public List<Task> doInConnection(Connection c) throws SQLException {
+                        return queueSystem.createTaskDao(c).findByNameAndCompletedIsNull(queue.name);
                     }
                 });
 
                 log.info("Found {} tasks on queue {}", tasks.size(), queue.name);
 
                 if (tasks.size() > 0) {
-                    for (final Task task : tasks) {
-                        try {
-                            executeTask(task);
-                        } catch (SqlExecutionException | TaskFailureException e) {
-                            log.warn("Task execution failed", e);
-                        }
-                    }
+                    queueService.executeTask(taskEffect, tasks);
                 }
             } catch (Throwable e) {
                 log.warn("Error while executing tasks.", e);
@@ -94,48 +88,4 @@ class QueueThread implements Runnable {
             }
         }
     }
-
-    private void executeTask(final Task task) {
-        final Date run = new Date();
-        log.info("Setting last run on task. date = {}, task = {}", run, task);
-        sqlEffectExecutor.execute(new SqlEffect.Void() {
-            @Override
-            public void doInConnection(Connection connection) throws SQLException {
-                new TaskDao(connection).update(task.registerRun());
-            }
-        });
-//        transactionTemplate.execute(new TransactionCallbackWithoutResult() {
-//            protected void doInTransactionWithoutResult(TransactionStatus status) {
-//                taskDao.update(task.registerRun());
-//            }
-//        });
-
-        sqlEffectExecutor.execute(new SqlEffect.Void() {
-            @Override
-            public void doInConnection(Connection c) throws SQLException {
-                try {
-                    callable.run(task.arguments);
-                    Date completed = new Date();
-                    Task t = task.registerComplete(completed);
-                    log.info("Completed task: {}", t);
-                    new TaskDao(c).update(t);
-                } catch (Exception e) {
-                    throw new TaskFailureException(e);
-                }
-            }
-        });
-//        transactionTemplate.execute(new TransactionCallbackWithoutResult() {
-//            protected void doInTransactionWithoutResult(TransactionStatus status) {
-//                try {
-//                    callable.run(task.arguments);
-//                    Date completed = new Date();
-//                    Task t = task.registerComplete(completed);
-//                    log.info("Completed task: {}", t);
-//                    taskDao.update(t);
-//                } catch (Exception e) {
-//                    throw new TaskFailureException(e);
-//                }
-//            }
-//        });
-    }
 }