From 1c0fc92c719f3856653d0efcc5fe4a1fa30b7bac Mon Sep 17 00:00:00 2001 From: Trygve Laugstøl Date: Sun, 9 Jun 2013 23:22:39 +0200 Subject: wip --- .../java/io/trygvis/async/SqlEffectExecutor.java | 58 +++++++--------------- .../java/io/trygvis/queue/JdbcQueueService.java | 22 +++++--- src/main/java/io/trygvis/queue/TaskDao.java | 2 +- 3 files changed, 33 insertions(+), 49 deletions(-) (limited to 'src/main/java/io/trygvis') diff --git a/src/main/java/io/trygvis/async/SqlEffectExecutor.java b/src/main/java/io/trygvis/async/SqlEffectExecutor.java index e03cf5e..51ad31d 100644 --- a/src/main/java/io/trygvis/async/SqlEffectExecutor.java +++ b/src/main/java/io/trygvis/async/SqlEffectExecutor.java @@ -2,7 +2,9 @@ package io.trygvis.async; import javax.sql.DataSource; import java.sql.Connection; +import java.sql.ResultSet; import java.sql.SQLException; +import java.sql.Statement; public class SqlEffectExecutor { @@ -12,41 +14,13 @@ public class SqlEffectExecutor { this.dataSource = dataSource; } - public A transaction(SqlEffect effect) { -/* + public A transaction(SqlEffect effect) throws SQLException { int pid; try (Connection c = dataSource.getConnection()) { - - try (Statement statement = c.createStatement()) { - ResultSet rs = statement.executeQuery("SELECT pg_backend_pid()"); - rs.next(); - pid = rs.getInt(1); - } - + pid = getPid(c); System.out.println("pid = " + pid); - try { - effect.doInConnection(c); - c.commit(); - } catch (SQLException e) { - c.rollback(); - e.printStackTrace(); - } finally { - System.out.println("Closing pid=" + pid); - try { - c.close(); - } catch (Exception e) { - e.printStackTrace(); - } - } - } catch (SQLException e) { - e.printStackTrace(); - throw new SqlExecutionException(e); - } -*/ - - try (Connection c = dataSource.getConnection()) { boolean ok = false; try { A a = effect.doInConnection(c); @@ -54,16 +28,19 @@ public class SqlEffectExecutor { ok = true; return a; } finally { + System.out.println("Closing, pid = " + pid); if (!ok) { - c.rollback(); + try { + c.rollback(); + } catch (SQLException e) { + // ignore + } } } - } catch (SQLException e) { - throw new SqlExecutionException(e); } } - public void transaction(final SqlEffect.Void effect) { + public void transaction(final SqlEffect.Void effect) throws SQLException { transaction(new SqlEffect() { @Override public Object doInConnection(Connection c) throws SQLException { @@ -73,12 +50,13 @@ public class SqlEffectExecutor { }); } - public static class SqlExecutionException extends RuntimeException { - public final SQLException exception; - - public SqlExecutionException(SQLException ex) { - super(ex); - this.exception = ex; + private int getPid(Connection c) throws SQLException { + int pid; + try (Statement statement = c.createStatement()) { + ResultSet rs = statement.executeQuery("SELECT pg_backend_pid()"); + rs.next(); + pid = rs.getInt(1); } + return pid; } } diff --git a/src/main/java/io/trygvis/queue/JdbcQueueService.java b/src/main/java/io/trygvis/queue/JdbcQueueService.java index d284287..c99bf2e 100644 --- a/src/main/java/io/trygvis/queue/JdbcQueueService.java +++ b/src/main/java/io/trygvis/queue/JdbcQueueService.java @@ -26,7 +26,7 @@ public class JdbcQueueService { this.sqlEffectExecutor = queueSystem.sqlEffectExecutor; } - public void consumeAll(final Queue queue, final TaskEffect effect) { + public void consumeAll(final Queue queue, final TaskEffect effect) throws SQLException { final List tasks = sqlEffectExecutor.transaction(new SqlEffect>() { @Override public List doInConnection(Connection c) throws SQLException { @@ -47,7 +47,7 @@ public class JdbcQueueService { }); } - public void executeTask(final TaskEffect taskEffect, final List tasks) { + public void executeTask(final TaskEffect taskEffect, final List tasks) throws SQLException { sqlEffectExecutor.transaction(new SqlEffect.Void() { @Override public void doInConnection(Connection connection) throws SQLException { @@ -102,12 +102,6 @@ public class JdbcQueueService { final Date now = new Date(); log.error("Unable to execute task, id=" + task.id(), e); - try { - taskDao.rollback(); - } catch (SQLException e2) { - log.error("Error rolling back transaction after failed apply.", e2); - } - final Task t = task; sqlEffectExecutor.transaction(new SqlEffect.Void() { @Override @@ -117,6 +111,12 @@ public class JdbcQueueService { taskDao.update(task); } }); + + if(e instanceof SQLException) { + throw ((SQLException) e); + } + + throw new RuntimeException("Error while executing task, id=" + task.id(), e); } } @@ -156,4 +156,10 @@ public class JdbcQueueService { return new Task(id, parent, queue, NEW, scheduled, null, 0, null, arguments); } + + public static class TaskExecutionFailed extends Throwable { + public TaskExecutionFailed(Exception e) { + super(e); + } + } } diff --git a/src/main/java/io/trygvis/queue/TaskDao.java b/src/main/java/io/trygvis/queue/TaskDao.java index 025823b..8b58585 100644 --- a/src/main/java/io/trygvis/queue/TaskDao.java +++ b/src/main/java/io/trygvis/queue/TaskDao.java @@ -111,7 +111,7 @@ public class TaskDao { rs.getLong(i++), rs.getLong(i++), rs.getString(i++), - valueOf(rs.getString(i++)), + TaskState.valueOf(rs.getString(i++).trim()), rs.getTimestamp(i++), rs.getTimestamp(i++), rs.getInt(i++), -- cgit v1.2.3