Experimental support for using lambdas for scheduling

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

@@ -1,22 +1,11 @@
 package com.gruelbox.transactionoutbox;
 
-import com.google.gson.Gson;
-import com.google.gson.GsonBuilder;
-import com.google.gson.JsonArray;
-import com.google.gson.JsonDeserializationContext;
-import com.google.gson.JsonDeserializer;
-import com.google.gson.JsonElement;
-import com.google.gson.JsonObject;
-import com.google.gson.JsonParseException;
-import com.google.gson.JsonSerializationContext;
-import com.google.gson.JsonSerializer;
-import com.google.gson.TypeAdapter;
+import com.google.gson.*;
+import com.google.gson.reflect.TypeToken;
 import com.google.gson.stream.JsonReader;
 import com.google.gson.stream.JsonToken;
 import com.google.gson.stream.JsonWriter;
-import java.io.IOException;
-import java.io.Reader;
-import java.io.Writer;
+import java.io.*;
 import java.lang.reflect.Modifier;
 import java.lang.reflect.Type;
 import java.math.BigDecimal;
@@ -25,15 +14,7 @@
 import java.time.*;
 import java.time.format.DateTimeFormatter;
 import java.time.temporal.ChronoUnit;
-import java.util.Calendar;
-import java.util.Date;
-import java.util.GregorianCalendar;
-import java.util.HashMap;
-import java.util.Locale;
-import java.util.Map;
-import java.util.Set;
-import java.util.TimeZone;
-import java.util.UUID;
+import java.util.*;
 import lombok.Builder;
 import lombok.extern.slf4j.Slf4j;
 
@@ -92,6 +73,17 @@ public final class DefaultInvocationSerializer implements InvocationSerializer {
             .registerTypeAdapter(Period.class, new PeriodTypeAdapter())
             .registerTypeAdapter(Year.class, new YearTypeAdapter())
             .registerTypeAdapter(YearMonth.class, new YearMonthAdapter())
+            .registerTypeAdapterFactory(
+                new TypeAdapterFactory() {
+                  @SuppressWarnings("unchecked")
+                  @Override
+                  public <T> TypeAdapter<T> create(Gson gson, TypeToken<T> type) {
+                    if (SerializableLambda.class.isAssignableFrom(type.getRawType())) {
+                      return (TypeAdapter<T>) new SerializableRunnerAdapter();
+                    }
+                    return null;
+                  }
+                })
             .excludeFieldsWithModifiers(Modifier.TRANSIENT, Modifier.STATIC)
             .create();
   }
@@ -298,6 +290,7 @@ public Invocation deserialize(
         JsonElement argType = arg.getAsJsonObject().get("t");
         if (argType != null) {
           JsonElement argValue = arg.getAsJsonObject().get("v");
+          log.info("arg = {}", arg);
           Class<?> argClass = classForName(argType.getAsString());
           try {
             args[i] = context.deserialize(argValue, argClass);
@@ -313,6 +306,9 @@ public Invocation deserialize(
     }
 
     private Class<?> classForName(String name) {
+      if (name.equals(SerializableLambda.class.getName()) || name.contains("$$Lambda")) {
+        return SerializableLambda.class;
+      }
       var clazz = nameToClass.get(name);
       if (clazz == null) {
         throw new IllegalArgumentException("Cannot deserialize class - not found: " + name);
@@ -321,6 +317,9 @@ private Class<?> classForName(String name) {
     }
 
     private String nameForClass(Class<?> clazz) {
+      if (SerializableLambda.class.isAssignableFrom(clazz)) {
+        return clazz.getName();
+      }
       var name = classToName.get(clazz);
       if (name == null) {
         throw new IllegalArgumentException(
@@ -696,4 +695,24 @@ private static int parseInt(String value, int beginIndex, int endIndex)
       return -result;
     }
   }
+
+  private static class SerializableRunnerAdapter extends TypeAdapter<SerializableLambda> {
+    @Override
+    public void write(JsonWriter out, SerializableLambda value) throws IOException {
+      ByteArrayOutputStream bos = new ByteArrayOutputStream();
+      new ObjectOutputStream(bos).writeObject(value);
+      out.value(Base64.getEncoder().encodeToString(bos.toByteArray()));
+    }
+
+    @Override
+    public SerializableLambda read(JsonReader in) throws IOException {
+      byte[] bytes = Base64.getDecoder().decode(in.nextString());
+      ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bytes));
+      try {
+        return (SerializableLambda) ois.readObject();
+      } catch (ClassNotFoundException e) {
+        throw new RuntimeException(e);
+      }
+    }
+  }
 }

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

@@ -65,4 +65,8 @@ static Instantiator using(Function<Class<?>, Object> fn) {
    * @return An instance of the class.
    */
   Object getInstance(String name);
+
+  default <T> T getInstance(Class<T> clazz) {
+    throw new UnsupportedOperationException();
+  }
 }

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

@@ -1,6 +1,9 @@
 package com.gruelbox.transactionoutbox;
 
+import static java.util.stream.Collectors.joining;
+
 import com.google.gson.annotations.SerializedName;
+import com.gruelbox.transactionoutbox.spi.Utils;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.util.Arrays;
@@ -136,4 +139,18 @@ void invoke(Object instance, TransactionOutboxListener listener)
     }
     listener.wrapInvocation(() -> method.invoke(instance, args));
   }
+
+  public String getDescription() {
+    if (getClassName().equals(LambdaRunner.class.getName())) {
+      return ((SerializableLambda) args[0]).getDescription();
+    } else {
+      return String.format(
+          "%s.%s(%s)",
+          className,
+          methodName,
+          args == null
+              ? null
+              : Arrays.stream(args).map(Utils::stringifyArg).collect(joining(", ")));
+    }
+  }
 }

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

@@ -0,0 +1,13 @@
+package com.gruelbox.transactionoutbox;
+
+import lombok.AccessLevel;
+import lombok.AllArgsConstructor;
+
+@AllArgsConstructor(access = AccessLevel.PACKAGE)
+public class LambdaContext {
+  private final Instantiator instantiator;
+
+  public <T> T getInstance(Class<T> clazz) {
+    return instantiator.getInstance(clazz);
+  }
+}

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

@@ -0,0 +1,14 @@
+package com.gruelbox.transactionoutbox;
+
+import lombok.AccessLevel;
+import lombok.AllArgsConstructor;
+
+@AllArgsConstructor(access = AccessLevel.PROTECTED)
+final class LambdaRunner {
+
+  private final LambdaContext lambdaContext;
+
+  void run(SerializableLambda runnable) {
+    runnable.run(lambdaContext);
+  }
+}

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

@@ -0,0 +1,38 @@
+package com.gruelbox.transactionoutbox;
+
+import static java.util.stream.Collectors.joining;
+
+import com.gruelbox.transactionoutbox.spi.Utils;
+import java.io.Serializable;
+import java.util.Arrays;
+
+/** TODO */
+@FunctionalInterface
+public interface SerializableLambda extends Serializable {
+
+  void run(LambdaContext lambdaContext);
+
+  default String getDescription() {
+    var className = getClass().getName();
+    var index = className.indexOf("$$Lambda/");
+    if (index == -1) {
+      return "<invalid lambda>";
+    }
+    var parentClass = className.substring(0, index);
+    var lambdaId = className.substring(index + 9);
+    var args =
+        Arrays.stream(getClass().getDeclaredFields())
+            .map(
+                f -> {
+                  try {
+                    f.setAccessible(true);
+                    return f.get(this);
+                  } catch (IllegalAccessException e) {
+                    throw new RuntimeException(e);
+                  }
+                })
+            .map(Utils::stringifyArg)
+            .collect(joining(","));
+    return String.format("%s.lambda_%s_(%s)", parentClass, lambdaId, args);
+  }
+}

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

@@ -58,6 +58,14 @@ static TransactionOutboxBuilder builder() {
    */
   <T> T schedule(Class<T> clazz);
 
+  /**
+   * TODO
+   *
+   * @param runnable
+   * @return
+   */
+  TransactionOutboxEntry schedule(SerializableLambda runnable);
+
   /**
    * Starts building a schedule request with parameterization. See {@link
    * ParameterizedScheduleBuilder#schedule(Class)} for more information.

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

@@ -1,9 +1,7 @@
 package com.gruelbox.transactionoutbox;
 
-import static java.util.stream.Collectors.joining;
 
 import java.time.Instant;
-import java.util.Arrays;
 import lombok.*;
 import lombok.experimental.SuperBuilder;
 
@@ -122,41 +120,21 @@ public String description() {
     if (!this.initialized) {
       synchronized (this) {
         if (!this.initialized) {
-          String description =
+          this.description =
               String.format(
-                  "%s.%s(%s) [%s]%s%s",
-                  invocation.getClassName(),
-                  invocation.getMethodName(),
-                  invocation.getArgs() == null
-                      ? null
-                      : Arrays.stream(invocation.getArgs())
-                          .map(this::stringify)
-                          .collect(joining(", ")),
+                  "%s [%s]%s%s",
+                  invocation.getDescription(),
                   id,
                   uniqueRequestId == null ? "" : " uid=[" + uniqueRequestId + "]",
                   topic == null ? "" : " seq=[" + topic + "/" + sequence + "]");
-          this.description = description;
           this.initialized = true;
-          return description;
+          return this.description;
         }
       }
     }
     return this.description;
   }
 
-  private String stringify(Object o) {
-    if (o == null) {
-      return "null";
-    }
-    if (o.getClass().isArray()) {
-      return "[" + Arrays.stream((Object[]) o).map(this::stringify).collect(joining(", ")) + "]";
-    }
-    if (o instanceof String) {
-      return "\"" + o + "\"";
-    }
-    return o.toString();
-  }
-
   @Override
   public void validate(Validator validator) {
     validator.notNull("id", id);