Support for checking entry state

Author: Graham Crockford

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

@@ -217,37 +217,39 @@ private Class<?> toClass(ClassLoader classLoader, String name) {
     public JsonElement serialize(Invocation src, Type typeOfSrc, JsonSerializationContext context) {
       if (version == 1) {
         log.warn("Serializing as deprecated version {}", version);
-        return serializeV1(src, typeOfSrc, context);
+        return serializeV1(src, context);
       }
       JsonObject obj = new JsonObject();
       obj.addProperty("c", src.getClassName());
       obj.addProperty("m", src.getMethodName());
       JsonArray params = new JsonArray();
-      JsonArray args = new JsonArray();
-      int i = 0;
-      for (Class<?> parameterType : src.getParameterTypes()) {
-        params.add(nameForClass(parameterType));
-        Object arg = src.getArgs()[i];
-        if (arg == null) {
-          JsonObject jsonObject = new JsonObject();
-          jsonObject.add("t", null);
-          jsonObject.add("v", null);
-          args.add(jsonObject);
-        } else {
-          JsonObject jsonObject = new JsonObject();
-          jsonObject.addProperty("t", nameForClass(arg.getClass()));
-          jsonObject.add("v", context.serialize(arg));
-          args.add(jsonObject);
+      if (src.getParameterTypes().length > 0) {
+        JsonArray args = new JsonArray();
+        int i = 0;
+        for (Class<?> parameterType : src.getParameterTypes()) {
+          params.add(nameForClass(parameterType));
+          Object arg = src.getArgs()[i];
+          if (arg == null) {
+            JsonObject jsonObject = new JsonObject();
+            jsonObject.add("t", null);
+            jsonObject.add("v", null);
+            args.add(jsonObject);
+          } else {
+            JsonObject jsonObject = new JsonObject();
+            jsonObject.addProperty("t", nameForClass(arg.getClass()));
+            jsonObject.add("v", context.serialize(arg));
+            args.add(jsonObject);
+          }
+          i++;
         }
-        i++;
+        obj.add("a", args);
       }
       obj.add("p", params);
-      obj.add("a", args);
       obj.add("x", context.serialize(src.getMdc()));
       return obj;
     }
 
-    JsonElement serializeV1(Invocation src, Type typeOfSrc, JsonSerializationContext context) {
+    JsonElement serializeV1(Invocation src, JsonSerializationContext context) {
       JsonObject obj = new JsonObject();
       obj.addProperty("c", src.getClassName());
       obj.addProperty("m", src.getMethodName());
@@ -274,39 +276,46 @@ public Invocation deserialize(
       String className = jsonObject.get("c").getAsString();
       String methodName = jsonObject.get("m").getAsString();
 
-      JsonArray jsonParams = jsonObject.get("p").getAsJsonArray();
-      Class<?>[] params = new Class<?>[jsonParams.size()];
-      for (int i = 0; i < jsonParams.size(); i++) {
-        JsonElement param = jsonParams.get(i);
-        if (param.isJsonObject()) {
-          // For backwards compatibility
-          params[i] = classForName(param.getAsJsonObject().get("t").getAsString());
-        } else {
-          params[i] = classForName(param.getAsString());
+      Class<?>[] params = null;
+      if (jsonObject.has("p")) {
+        JsonArray jsonParams = jsonObject.get("p").getAsJsonArray();
+        params = new Class<?>[jsonParams.size()];
+        for (int i = 0; i < jsonParams.size(); i++) {
+          JsonElement param = jsonParams.get(i);
+          if (param.isJsonObject()) {
+            // For backwards compatibility
+            params[i] = classForName(param.getAsJsonObject().get("t").getAsString());
+          } else {
+            params[i] = classForName(param.getAsString());
+          }
         }
       }
 
-      JsonElement argsElement = jsonObject.get("a");
-      if (argsElement == null) {
-        // For backwards compatibility
-        argsElement = jsonObject.get("p");
-      }
-      JsonArray jsonArgs = argsElement.getAsJsonArray();
-      Object[] args = new Object[jsonArgs.size()];
-      for (int i = 0; i < jsonArgs.size(); i++) {
-        JsonElement arg = jsonArgs.get(i);
-        JsonElement argType = arg.getAsJsonObject().get("t");
-        if (argType != null) {
-          JsonElement argValue = arg.getAsJsonObject().get("v");
-          Class<?> argClass = classForName(argType.getAsString());
-          try {
-            args[i] = context.deserialize(argValue, argClass);
-          } catch (Exception e) {
-            throw new RuntimeException(
-                "Failed to deserialize arg [" + argValue + "] of type [" + argType + "]", e);
+      Object[] args = null;
+      if (jsonObject.has("a")) {
+        JsonElement argsElement = jsonObject.get("a");
+        if (argsElement == null) {
+          // For backwards compatibility
+          argsElement = jsonObject.get("p");
+        }
+        JsonArray jsonArgs = argsElement.getAsJsonArray();
+        args = new Object[jsonArgs.size()];
+        for (int i = 0; i < jsonArgs.size(); i++) {
+          JsonElement arg = jsonArgs.get(i);
+          JsonElement argType = arg.getAsJsonObject().get("t");
+          if (argType != null) {
+            JsonElement argValue = arg.getAsJsonObject().get("v");
+            Class<?> argClass = classForName(argType.getAsString());
+            try {
+              args[i] = context.deserialize(argValue, argClass);
+            } catch (Exception e) {
+              throw new RuntimeException(
+                  "Failed to deserialize arg [" + argValue + "] of type [" + argType + "]", e);
+            }
           }
         }
       }
+
       Map<String, String> mdc = context.deserialize(jsonObject.get("x"), Map.class);
 
       return new Invocation(className, methodName, params, args, mdc);

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

@@ -427,4 +427,17 @@ public boolean checkConnection(Transaction tx) throws SQLException {
       return rs.next() && (rs.getInt(1) == 1);
     }
   }
+
+  @Override
+  public Optional<TransactionOutboxEntry> load(Transaction tx, String entryId) throws Exception {
+    //noinspection resource
+    try (PreparedStatement stmt =
+        tx.connection()
+            .prepareStatement("SELECT " + ALL_FIELDS + " FROM " + tableName + " WHERE id = ?")) {
+      stmt.setString(1, entryId);
+      List<TransactionOutboxEntry> results = new ArrayList<>(1);
+      gatherResults(stmt, results);
+      return results.stream().findFirst();
+    }
+  }
 }

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

@@ -150,4 +150,13 @@ List<TransactionOutboxEntry> selectBatch(Transaction tx, int batchSize, Instant
    * @param tx The current {@link Transaction}.
    */
   void clear(Transaction tx) throws Exception;
+
+  /**
+   * Loads the specified entry.
+   *
+   * @param tx The current {@link Transaction}.
+   * @param entryId The entry id.
+   * @return The entry, or empty if it does not exist.
+   */
+  Optional<TransactionOutboxEntry> load(Transaction tx, String entryId) throws Exception;
 }

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

@@ -64,6 +64,11 @@ public int deleteProcessedAndExpired(Transaction tx, int batchSize, Instant now)
   @Override
   public void clear(Transaction tx) {}
 
+  @Override
+  public Optional<TransactionOutboxEntry> load(Transaction tx, String entryId) {
+    return Optional.empty();
+  }
+
   @Override
   public boolean checkConnection(Transaction tx) {
     return true;

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

@@ -2,7 +2,9 @@
 
 import java.time.Clock;
 import java.time.Duration;
+import java.util.Optional;
 import java.util.concurrent.Executor;
+import java.util.function.Consumer;
 import java.util.function.Supplier;
 import lombok.ToString;
 import org.slf4j.MDC;
@@ -135,6 +137,51 @@ default boolean flush() {
   @SuppressWarnings("WeakerAccess")
   void processNow(TransactionOutboxEntry entry);
 
+  /**
+   * Loads the specified queued test. This can be used to check the status.
+   *
+   * <p>Requires the transaction manager in use to be a {@link
+   * ThreadLocalContextTransactionManager}.
+   *
+   * <p>Note that nothing may be returned for four distinct reasons:
+   *
+   * <ul>
+   *   <li>Such a task has never been created and is therefore unknown
+   *   <li>The task was created, and successfully completed, but had no {@code uniqueRequestId} so
+   *       was deleted.
+   *   <li>The task was created, and completed, and had a {@code uniqueRequestId}, but the {@code
+   *       retentionThreshold} has passed so the record has been deleted
+   *   <li>The task was created, and may have succeeded or failed but was deleted from the database
+   *       by a user for some reason.
+   * </ul>
+   *
+   * @param entryId The entry id.
+   * @return Empty if no such entry exists, otherwise the entry.
+   */
+  Optional<TransactionOutboxEntry> fetchEntry(String entryId);
+
+  /**
+   * Loads the specified queued request. This can be used to check the status of a request.
+   *
+   * <p>Note that nothing may be returned for four distinct reasons:
+   *
+   * <ul>
+   *   <li>Such a task has never been created and is therefore unknown
+   *   <li>The task was created, and successfully completed, but had no {@code uniqueRequestId} so
+   *       was deleted.
+   *   <li>The task was created, and completed, and had a {@code uniqueRequestId}, but the {@code
+   *       retentionThreshold} has passed so the record has been deleted
+   *   <li>The task was created, and may have succeeded or failed but was deleted from the database
+   *       by a user for some reason.
+   * </ul>
+   *
+   * @param entryId The entry id.
+   * @param transactionContext The transaction context (if using a {@link
+   *     ParameterContextTransactionManager}).
+   * @return Empty if no such entry exists, otherwise the entry.
+   */
+  Optional<TransactionOutboxEntry> fetchEntry(String entryId, Object transactionContext);
+
   /** Builder for {@link TransactionOutbox}. */
   @ToString
   abstract class TransactionOutboxBuilder {
@@ -368,6 +415,22 @@ interface ParameterizedScheduleBuilder {
      */
     ParameterizedScheduleBuilder ordered(String topic);
 
+    /**
+     * If set, {@code callback} will be called in the current thread immediately after writing the
+     * entry to the database, so that the details of the record can be inspected or the {@code id}
+     * retained to be able to check the status with {@link TransactionOutbox#fetchEntry(String)}.
+     *
+     * <p>Note that a successful callback does not necessarily mean that the entry has, or will
+     * ever, be processed. The current transaction could be rolled back rather than committed. If
+     * this occurs, {@link TransactionOutbox#fetchEntry(String)} will simply return empty.
+     *
+     * @param callback Receives the {@link TransactionOutboxEntry} once it has been inserted into
+     *     the database (though this insert may be rolled back and will not be processed if this
+     *     happens).
+     * @return Builder.
+     */
+    ParameterizedScheduleBuilder entryCallback(Consumer<TransactionOutboxEntry> callback);
+
     /**
      * Equivalent to {@link TransactionOutbox#schedule(Class)}, but applying additional parameters
      * to the request as configured using {@link TransactionOutbox#with()}.