wip

5 files changed, 62 insertions, 6 deletions

diff --git a/src/main/java/io/trygvis/queue/JdbcQueueService.java b/src/main/java/io/trygvis/queue/JdbcQueueService.java
index a366838..ef0b5bb 100644
--- a/src/main/java/io/trygvis/queue/JdbcQueueService.java
+++ b/src/main/java/io/trygvis/queue/JdbcQueueService.java
@@ -1,7 +1,5 @@
 package io.trygvis.queue;
 
-import io.trygvis.async.SqlEffectExecutor;
-
 import java.sql.Connection;
 import java.sql.SQLException;
 import java.util.HashMap;
diff --git a/src/main/java/io/trygvis/queue/QueueExecutor.java b/src/main/java/io/trygvis/queue/QueueExecutor.java
index 3739532..468059d 100644
--- a/src/main/java/io/trygvis/queue/QueueExecutor.java
+++ b/src/main/java/io/trygvis/queue/QueueExecutor.java
@@ -1,8 +1,6 @@
 package io.trygvis.queue;
 
 import io.trygvis.async.QueueStats;
-import io.trygvis.async.SqlEffect;
-import io.trygvis.async.SqlEffectExecutor;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
diff --git a/src/main/java/io/trygvis/queue/QueueSystem.java b/src/main/java/io/trygvis/queue/QueueSystem.java
index 6710bf4..5954526 100644
--- a/src/main/java/io/trygvis/queue/QueueSystem.java
+++ b/src/main/java/io/trygvis/queue/QueueSystem.java
@@ -1,8 +1,6 @@
 package io.trygvis.queue;
 
 import io.trygvis.async.JdbcAsyncService;
-import io.trygvis.async.SqlEffect;
-import io.trygvis.async.SqlEffectExecutor;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
diff --git a/src/main/java/io/trygvis/queue/SqlEffect.java b/src/main/java/io/trygvis/queue/SqlEffect.java
new file mode 100644
index 0000000..7864bcd
--- /dev/null
+++ b/src/main/java/io/trygvis/queue/SqlEffect.java
@@ -0,0 +1,12 @@
+package io.trygvis.queue;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+
+public interface SqlEffect<A> {
+    A doInConnection(Connection c) throws SQLException;
+
+    interface Void {
+        void doInConnection(Connection c) throws SQLException;
+    }
+}
diff --git a/src/main/java/io/trygvis/queue/SqlEffectExecutor.java b/src/main/java/io/trygvis/queue/SqlEffectExecutor.java
new file mode 100644
index 0000000..92802da
--- /dev/null
+++ b/src/main/java/io/trygvis/queue/SqlEffectExecutor.java
@@ -0,0 +1,50 @@
+package io.trygvis.queue;
+
+import javax.sql.DataSource;
+import java.sql.Connection;
+import java.sql.SQLException;
+
+public class SqlEffectExecutor {
+
+    private final DataSource dataSource;
+
+    public SqlEffectExecutor(DataSource dataSource) {
+        this.dataSource = dataSource;
+    }
+
+    public <A> A transaction(SqlEffect<A> effect) throws SQLException {
+//        int pid;
+
+        try (Connection c = dataSource.getConnection()) {
+//            pid = getPid(c);
+//            System.out.println("pid = " + pid);
+
+            boolean ok = false;
+            try {
+                A a = effect.doInConnection(c);
+                c.commit();
+                ok = true;
+                return a;
+            } finally {
+//                System.out.println("Closing, pid = " + pid);
+                if (!ok) {
+                    try {
+                        c.rollback();
+                    } catch (SQLException e) {
+                        // ignore
+                    }
+                }
+            }
+        }
+    }
+
+    public void transaction(final SqlEffect.Void effect) throws SQLException {
+        transaction(new SqlEffect<Object>() {
+            @Override
+            public Object doInConnection(Connection c) throws SQLException {
+                effect.doInConnection(c);
+                return null;
+            }
+        });
+    }
+}