Refactored the Dialect to use an interface to generate the DB specific SQL.

transactionoutbox-core/src/main/java/com/gruelbox/transactionoutbox/DefaultMigrationManager.java

@@ -135,25 +135,7 @@ private static int currentVersion(Connection connection, Dialect dialect) throws
   private static void createVersionTableIfNotExists(Connection connection, Dialect dialect)
       throws SQLException {
     try (Statement s = connection.createStatement()) {
-      switch (dialect) {
-        case ORACLE:
-          try {
-            s.execute("CREATE TABLE TXNO_VERSION (version NUMBER)");
-          } catch (SQLException e) {
-            // oracle code for name already used by an existing object
-            if (!e.getMessage().contains("955")) {
-              throw e;
-            }
-          }
-          break;
-        case MY_SQL_5:
-        case H2:
-        case MY_SQL_8:
-        case POSTGRESQL_9:
-        default:
-          s.execute("CREATE TABLE IF NOT EXISTS TXNO_VERSION (version INT)");
-          break;
-      }
+      dialect.createVersionTableIfNotExists(s);
     }
   }
 

transactionoutbox-core/src/main/java/com/gruelbox/transactionoutbox/DefaultPersistor.java

@@ -41,8 +41,8 @@ public class DefaultPersistor implements Persistor, Validatable {
   /**
    * @param writeLockTimeoutSeconds How many seconds to wait before timing out on obtaining a write
    *     lock. There's no point making this long; it's always better to just back off as quickly as
-   *     possible and try another record. Generally these lock timeouts only kick in if {@link
-   *     Dialect#isSupportsSkipLock()} is false.
+   *     possible and try another record. Generally these lock timeouts only kick in if the
+   *     implementation does not support skip locking.
    */
   @SuppressWarnings("JavaDoc")
   @Builder.Default
@@ -62,7 +62,7 @@ public class DefaultPersistor implements Persistor, Validatable {
   private final String tableName = "TXNO_OUTBOX";
 
   /**
-   * @param migrate Set to false to disable automatic database migrations. This may be preferred if
+   * @param migrate Set too false to disable automatic database migrations. This may be preferred if
    *     the default migration behaviour interferes with your existing toolset, and you prefer to
    *     manage the migrations explicitly (e.g. using FlyWay or Liquibase), or your do not give the
    *     application DDL permissions at runtime.
@@ -190,18 +190,7 @@ public void update(Transaction tx, TransactionOutboxEntry entry) throws Exceptio
   @Override
   public boolean lock(Transaction tx, TransactionOutboxEntry entry) throws Exception {
     //noinspection resource
-    try (PreparedStatement stmt =
-        tx.connection()
-            .prepareStatement(
-                dialect.isSupportsSkipLock()
-                    // language=MySQL
-                    ? "SELECT id, invocation FROM "
-                        + tableName
-                        + " WHERE id = ? AND version = ? FOR UPDATE SKIP LOCKED"
-                    // language=MySQL
-                    : "SELECT id, invocation FROM "
-                        + tableName
-                        + " WHERE id = ? AND version = ? FOR UPDATE")) {
+    try (PreparedStatement stmt = tx.connection().prepareStatement(dialect.lock(tableName))) {
       stmt.setString(1, entry.getId());
       stmt.setInt(2, entry.getVersion());
       stmt.setQueryTimeout(writeLockTimeoutSeconds);
@@ -228,44 +217,25 @@ public boolean lock(Transaction tx, TransactionOutboxEntry entry) throws Excepti
   @Override
   public boolean unblock(Transaction tx, String entryId) throws Exception {
     @SuppressWarnings("resource")
-    PreparedStatement stmt =
-        tx.prepareBatchStatement(
-            "UPDATE "
-                + tableName
-                + " SET attempts = 0, blocked = "
-                + dialect.booleanValue(false)
-                + " "
-                + "WHERE blocked = "
-                + dialect.booleanValue(true)
-                + " AND processed = "
-                + dialect.booleanValue(false)
-                + " AND id = ?");
-    stmt.setString(1, entryId);
+    PreparedStatement stmt = tx.prepareBatchStatement(dialect.unblock(tableName));
+    stmt.setInt(1, 0);
+    stmt.setBoolean(2, false);
+    stmt.setBoolean(3, true);
+    stmt.setBoolean(4, false);
+    stmt.setString(5, entryId);
     stmt.setQueryTimeout(writeLockTimeoutSeconds);
     return stmt.executeUpdate() != 0;
   }
 
   @Override
   public List<TransactionOutboxEntry> selectBatch(Transaction tx, int batchSize, Instant now)
       throws Exception {
-    String forUpdate = dialect.isSupportsSkipLock() ? " FOR UPDATE SKIP LOCKED" : "";
     //noinspection resource
     try (PreparedStatement stmt =
-        tx.connection()
-            .prepareStatement(
-                // language=MySQL
-                "SELECT "
-                    + ALL_FIELDS
-                    + " FROM "
-                    + tableName
-                    + " WHERE nextAttemptTime < ? AND blocked = "
-                    + dialect.booleanValue(false)
-                    + " AND processed = "
-                    + dialect.booleanValue(false)
-                    + dialect.getLimitCriteria()
-                    + forUpdate)) {
+        tx.connection().prepareStatement(dialect.selectBatch(tableName, ALL_FIELDS, batchSize))) {
       stmt.setTimestamp(1, Timestamp.from(now));
-      stmt.setInt(2, batchSize);
+      stmt.setBoolean(2, false);
+      stmt.setBoolean(3, false);
       return gatherResults(batchSize, stmt);
     }
   }
@@ -275,10 +245,10 @@ public int deleteProcessedAndExpired(Transaction tx, int batchSize, Instant now)
       throws Exception {
     //noinspection resource
     try (PreparedStatement stmt =
-        tx.connection()
-            .prepareStatement(dialect.getDeleteExpired().replace("{{table}}", tableName))) {
+        tx.connection().prepareStatement(dialect.deleteExpired(tableName, batchSize))) {
       stmt.setTimestamp(1, Timestamp.from(now));
-      stmt.setInt(2, batchSize);
+      stmt.setBoolean(2, true);
+      stmt.setBoolean(3, false);
       return stmt.executeUpdate();
     }
   }

transactionoutbox-core/src/main/java/com/gruelbox/transactionoutbox/Dialect.java

@@ -1,60 +1,29 @@
 package com.gruelbox.transactionoutbox;
 
-import lombok.AllArgsConstructor;
-import lombok.Getter;
-
-/**
- * The SQL dialects supported by {@link DefaultPersistor}. Currently this is only used to determine
- * whether {@code SKIP LOCKED} is available, so using the wrong dialect may work for unsupported
- * database platforms. However, in future this is likely to extend to other SQL features and
- * possibly be expanded to an interface to allow easier extension.
- */
-@AllArgsConstructor
-@Getter
-@Beta
-public enum Dialect {
-  MY_SQL_5(false, Constants.DEFAULT_DELETE_EXPIRED_STMT, Constants.DEFAULT_LIMIT_CRITERIA), //
-  MY_SQL_8(true, Constants.DEFAULT_DELETE_EXPIRED_STMT, Constants.DEFAULT_LIMIT_CRITERIA), //
-  POSTGRESQL_9(
-      true,
-      "DELETE FROM {{table}} WHERE id IN (SELECT id FROM {{table}} WHERE nextAttemptTime < ? AND processed = true AND blocked = false LIMIT ?)",
-      Constants.DEFAULT_LIMIT_CRITERIA), //
-  H2(false, Constants.DEFAULT_DELETE_EXPIRED_STMT, Constants.DEFAULT_LIMIT_CRITERIA), //
-  ORACLE(
-      true,
-      "DELETE FROM {{table}} WHERE nextAttemptTime < ? AND processed = 1 AND blocked = 0 AND ROWNUM <= ?",
-      Constants.ORACLE_LIMIT_CRITERIA);
-
-  /**
-   * @return True if hot row support ({@code SKIP LOCKED}) is available, increasing performance when
-   *     there are multiple instances of the application potentially competing to process the same
-   *     task.
-   */
-  @SuppressWarnings("JavaDoc")
-  private final boolean supportsSkipLock;
-
-  /**
-   * @return Format string for the SQL required to delete expired retained records.
-   */
-  @SuppressWarnings("JavaDoc")
-  private final String deleteExpired;
-
-  private final String limitCriteria;
-
-  private static class Constants {
-    static final String DEFAULT_DELETE_EXPIRED_STMT =
-        "DELETE FROM {{table}} WHERE nextAttemptTime < ? AND processed = true AND blocked = false LIMIT ?";
-
-    static final String DEFAULT_LIMIT_CRITERIA = " LIMIT ?";
-
-    static final String ORACLE_LIMIT_CRITERIA = " AND ROWNUM <= ?";
-  }
-
-  public String booleanValue(boolean criteriaValue) {
-    String valueToReturn;
-    if (this == ORACLE) valueToReturn = criteriaValue ? "1" : "0";
-    else valueToReturn = criteriaValue ? Boolean.TRUE.toString() : Boolean.FALSE.toString();
-
-    return valueToReturn;
-  }
+import java.sql.SQLException;
+import java.sql.Statement;
+
+/** The SQL dialects supported by {@link DefaultPersistor}. */
+public interface Dialect {
+  Dialect MY_SQL_5 = new DialectMySQL5Impl();
+  Dialect MY_SQL_8 = new DialectMySQL8Impl();
+  Dialect POSTGRESQL_9 = new DialectPostgres9Impl();
+  Dialect H2 = new DialectH2Impl();
+  Dialect ORACLE = new DialectOracleImpl();
+
+  String lock(String tableName);
+
+  String unblock(String tableName);
+
+  String selectBatch(String tableName, String allFields, int batchSize);
+
+  String deleteExpired(String tableName, int batchSize);
+
+  boolean isSupportsSkipLock();
+
+  void createVersionTableIfNotExists(Statement s) throws SQLException;
+  // Required so the dialects can be used as keys in a Map.
+  boolean equals(Object o);
+
+  int hashCode();
 }

transactionoutbox-core/src/main/java/com/gruelbox/transactionoutbox/DialectBaseLockingImpl.java

@@ -0,0 +1,52 @@
+package com.gruelbox.transactionoutbox;
+
+import java.sql.SQLException;
+import java.sql.Statement;
+
+/**
+ * A SQL implementation for databases that do not support skip locking.
+ */
+final class DialectBaseLockingImpl implements Dialect {
+  @Override
+  public String lock(String tableName) {
+    return "SELECT id, invocation FROM " + tableName + " WHERE id = ? AND version = ? FOR UPDATE";
+  }
+
+  @Override
+  public String unblock(String tableName) {
+    return "UPDATE "
+        + tableName
+        + " SET attempts = ?, blocked = ? "
+        + "WHERE blocked = ? AND processed = ? AND id = ?";
+  }
+
+  @Override
+  public String selectBatch(String tableName, String allFields, int batchSize) {
+    return "SELECT "
+        + allFields
+        + " FROM "
+        + tableName
+        + " WHERE nextAttemptTime < ? AND blocked = ? AND processed = ? "
+        + "LIMIT "
+        + batchSize;
+  }
+
+  @Override
+  public String deleteExpired(String tableName, int batchSize) {
+    return "DELETE FROM "
+        + tableName
+        + " WHERE nextAttemptTime < ? AND processed = ? AND "
+        + "blocked = ? LIMIT "
+        + batchSize;
+  }
+
+  @Override
+  public boolean isSupportsSkipLock() {
+    return false;
+  }
+
+  @Override
+  public void createVersionTableIfNotExists(Statement s) throws SQLException {
+    s.execute("CREATE TABLE IF NOT EXISTS TXNO_VERSION (version INT)");
+  }
+}

transactionoutbox-core/src/main/java/com/gruelbox/transactionoutbox/DialectBaseSkipLockingImpl.java

@@ -0,0 +1,55 @@
+package com.gruelbox.transactionoutbox;
+
+import java.sql.SQLException;
+import java.sql.Statement;
+
+/**
+ * A SQL implementation for databases that support skip locking.
+ */
+final class DialectBaseSkipLockingImpl implements Dialect {
+  @Override
+  public String lock(String tableName) {
+    return "SELECT id, invocation FROM "
+        + tableName
+        + " WHERE id = ? AND version = ? FOR UPDATE SKIP LOCKED";
+  }
+
+  @Override
+  public String unblock(String tableName) {
+    return "UPDATE "
+        + tableName
+        + " SET attempts = ?, blocked = ? "
+        + "WHERE blocked = ? AND processed = ? AND id = ?";
+  }
+
+  @Override
+  public String selectBatch(String tableName, String allFields, int batchSize) {
+    return "SELECT "
+        + allFields
+        + " FROM "
+        + tableName
+        + " WHERE nextAttemptTime < ? AND blocked = ? AND processed = ? "
+        + "LIMIT "
+        + batchSize
+        + " FOR UPDATE SKIP LOCKED";
+  }
+
+  @Override
+  public String deleteExpired(String tableName, int batchSize) {
+    return "DELETE FROM "
+        + tableName
+        + " WHERE nextAttemptTime < ? AND processed = ? AND "
+        + "blocked = ? LIMIT "
+        + batchSize;
+  }
+
+  @Override
+  public boolean isSupportsSkipLock() {
+    return true;
+  }
+
+  @Override
+  public void createVersionTableIfNotExists(Statement s) throws SQLException {
+    s.execute("CREATE TABLE IF NOT EXISTS TXNO_VERSION (version INT)");
+  }
+}

transactionoutbox-core/src/main/java/com/gruelbox/transactionoutbox/DialectH2Impl.java

@@ -0,0 +1,41 @@
+package com.gruelbox.transactionoutbox;
+
+import java.sql.SQLException;
+import java.sql.Statement;
+import lombok.EqualsAndHashCode;
+
+/** Dialect SQL implementation for H2. */
+@EqualsAndHashCode
+final class DialectH2Impl implements Dialect {
+  private final Dialect lockingBase = new DialectBaseLockingImpl();
+
+  @Override
+  public String lock(String tableName) {
+    return lockingBase.lock(tableName);
+  }
+
+  @Override
+  public String unblock(String tableName) {
+    return lockingBase.unblock(tableName);
+  }
+
+  @Override
+  public String selectBatch(String tableName, String allFields, int batchSize) {
+    return lockingBase.selectBatch(tableName, allFields, batchSize);
+  }
+
+  @Override
+  public String deleteExpired(String tableName, int batchSize) {
+    return lockingBase.deleteExpired(tableName, batchSize);
+  }
+
+  @Override
+  public boolean isSupportsSkipLock() {
+    return lockingBase.isSupportsSkipLock();
+  }
+
+  @Override
+  public void createVersionTableIfNotExists(Statement s) throws SQLException {
+    lockingBase.createVersionTableIfNotExists(s);
+  }
+}