Introduce unsnapshottableIdentifier() to allow transient WorkflowIdentifiers.

This fixes #79 and unblocks the ability to implement GUWT Workers as Workflows.
`WorkflowIdentifier` can now indicate that it is not serializable by returning
null from `toByteString()`, in which case neither it nor the snapshot of the
workflow it identifies will be serialized. Because these identifiers do not need
to be serialized, they can contain arbitrary values such as `KType`s, which is
how Workers distinguish themselves for example. Workers also do not need to
be snapshotted, so this works out.

Author: zach-klippenstein

workflow-core/api/workflow-core.api

@@ -168,16 +168,16 @@ public final class com/squareup/workflow/WorkflowAction$Updater {
 	public final fun setState (Ljava/lang/Object;)V
 }
 
-public final class com/squareup/workflow/WorkflowIdentifier {
+public abstract class com/squareup/workflow/WorkflowIdentifier {
 	public static final field Companion Lcom/squareup/workflow/WorkflowIdentifier$Companion;
-	public fun equals (Ljava/lang/Object;)Z
-	public fun hashCode ()I
+	protected abstract fun getProxiedIdentifier ()Lcom/squareup/workflow/WorkflowIdentifier;
+	protected abstract fun getTypeName ()Ljava/lang/String;
+	public abstract fun maybeToByteString ()Lokio/ByteString;
 	public fun toString ()Ljava/lang/String;
-	public final fun write (Lokio/BufferedSink;)V
 }
 
 public final class com/squareup/workflow/WorkflowIdentifier$Companion {
-	public final fun read (Lokio/BufferedSource;)Lcom/squareup/workflow/WorkflowIdentifier;
+	public final fun parse (Lokio/ByteString;)Lcom/squareup/workflow/WorkflowIdentifier;
 }
 
 public final class com/squareup/workflow/WorkflowOutput {
@@ -225,6 +225,7 @@ public final class com/squareup/workflow/Workflows {
 	public static synthetic fun stateful$default (Lcom/squareup/workflow/Workflow$Companion;Lkotlin/jvm/functions/Function2;Lkotlin/jvm/functions/Function3;Lkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function3;ILjava/lang/Object;)Lcom/squareup/workflow/StatefulWorkflow;
 	public static final fun stateless (Lcom/squareup/workflow/Workflow$Companion;Lkotlin/jvm/functions/Function2;)Lcom/squareup/workflow/Workflow;
 	public static final fun transform (Lcom/squareup/workflow/Worker;Lkotlin/jvm/functions/Function1;)Lcom/squareup/workflow/Worker;
+	public static final fun unsnapshottableIdentifier (Lkotlin/reflect/KType;)Lcom/squareup/workflow/WorkflowIdentifier;
 	public static final fun workflowAction (Lcom/squareup/workflow/StatefulWorkflow;Ljava/lang/String;Lkotlin/jvm/functions/Function1;)Lcom/squareup/workflow/WorkflowAction;
 	public static final fun workflowAction (Lcom/squareup/workflow/StatefulWorkflow;Lkotlin/jvm/functions/Function0;Lkotlin/jvm/functions/Function1;)Lcom/squareup/workflow/WorkflowAction;
 	public static synthetic fun workflowAction$default (Lcom/squareup/workflow/StatefulWorkflow;Ljava/lang/String;Lkotlin/jvm/functions/Function1;ILjava/lang/Object;)Lcom/squareup/workflow/WorkflowAction;

workflow-core/src/main/java/com/squareup/workflow/WorkflowIdentifier.kt

@@ -18,11 +18,12 @@
 
 package com.squareup.workflow
 
-import okio.BufferedSink
-import okio.BufferedSource
+import okio.Buffer
+import okio.ByteString
 import okio.EOFException
 import kotlin.LazyThreadSafetyMode.PUBLICATION
 import kotlin.reflect.KClass
+import kotlin.reflect.KType
 
 /**
  * Represents a [Workflow]'s "identity" and is used by the runtime to determine whether a workflow
@@ -31,55 +32,104 @@ import kotlin.reflect.KClass
  *
  * A workflow's identity consists primarily of its concrete type (i.e. the class that implements
  * the [Workflow] interface). Two workflows of the same concrete type are considered identical.
- *
  * However, if a workflow class implements [ImpostorWorkflow], the identifier will also include
  * that workflow's [ImpostorWorkflow.realIdentifier].
  *
- * @constructor
- * @param type The [KClass] of the [Workflow] this identifier identifies.
- * @param proxiedIdentifier An optional identifier from [ImpostorWorkflow.realIdentifier] that will
- * be used to further narrow the scope of this identifier.
+ * Instances of this class are [equatable][equals] and [hashable][hashCode].
+ *
+ * ## Identifiers and snapshots
+ *
+ * Since workflows can be [serialized][StatefulWorkflow.snapshotState], workflows' identifiers must
+ * also be serializable in order to match workflows back up with their snapshots when restoring.
+ * However, some [WorkflowIdentifier]s may represent workflows that cannot be snapshotted. When an
+ * identifier is not snapshottable, [maybeToByteString] will return null, and any identifiers that
+ * reference [ImpostorWorkflow]s whose [ImpostorWorkflow.realIdentifier] is not snapshottable will
+ * also not be snapshottable. Such identifiers are created with [unsnapshottableIdentifier], but
+ * should not be used to wrap arbitrary workflows since those workflows may expect to be
+ * snapshotted.
  */
 @ExperimentalWorkflowApi
-class WorkflowIdentifier internal constructor(
-  private val type: KClass<out Workflow<*, *, *>>,
-  private val proxiedIdentifier: WorkflowIdentifier?
-) {
+// TODO not sure i like having protected members, those are part of the ABI – can we make this implementation
+// entirely module-private somehow?
+sealed class WorkflowIdentifier {
+
+  /**
+   * The fully-qualified name of the type of workflow this identifier identifies. Computed lazily
+   * and cached.
+   */
+  protected abstract val typeName: String
+
+  /**
+   * An optional identifier from [ImpostorWorkflow.realIdentifier] that will be used to further
+   * narrow the scope of this identifier.
+   */
+  protected abstract val proxiedIdentifier: WorkflowIdentifier?
 
   /**
-   * The fully-qualified name of [type]. Computed lazily.
+   * If this identifier is snapshottable, returns the serialized form of the identifier.
+   * If it is not snapshottable. returns null.
    */
-  private val typeString: String by lazy(PUBLICATION) { type.java.name }
+  abstract fun maybeToByteString(): ByteString?
 
   /**
    * Returns a description of this identifier including the name of its workflow type and any
-   * [proxiedIdentifier]s. Computes [typeString] if it has not already been computed.
+   * [ImpostorWorkflow.realIdentifier]s.
    */
   override fun toString(): String =
     generateSequence(this) { it.proxiedIdentifier }
-        .joinToString { it.typeString }
+        .joinToString { it.typeName }
         .let { "WorkflowIdentifier($it)" }
 
+  companion object {
+    fun parse(bytes: ByteString): WorkflowIdentifier? = WorkflowClassIdentifier.parse(bytes)
+  }
+}
+
+/**
+ * @param type The [KClass] of the [Workflow] this identifier identifies.
+ */
+@OptIn(ExperimentalWorkflowApi::class)
+private class WorkflowClassIdentifier(
+  private val type: KClass<out Workflow<*, *, *>>,
+  override val proxiedIdentifier: WorkflowIdentifier?
+) : WorkflowIdentifier() {
+
   /**
-   * Serializes this identifier to the sink. It can be read back with [WorkflowIdentifier.read].
+   * The fully-qualified name of the type of workflow this identifier identifies.
+   * Note that Class.name is already cached, so we don't need to cache it ourselves.
    */
-  fun write(sink: BufferedSink) {
-    sink.writeUtf8WithLength(typeString)
-    if (proxiedIdentifier != null) {
-      sink.writeByte(PROXY_IDENTIFIER_TAG.toInt())
-      proxiedIdentifier.write(sink)
-    } else {
-      sink.writeByte(NO_PROXY_IDENTIFIER_TAG.toInt())
+  override val typeName: String get() = type.java.name
+
+  /**
+   * Serializes this identifier to a [ByteString] if its [proxiedIdentifier] is serializable, else
+   * returns null. The non-null [ByteString] can be read back with [WorkflowIdentifier.read].
+   */
+  override fun maybeToByteString(): ByteString? {
+    val proxiedBytes = proxiedIdentifier?.let {
+      // If we have a proxied identifier but it's not serializable, then we can't be serializable
+      // either.
+      it.maybeToByteString() ?: return null
+    }
+
+    return Buffer().let { sink ->
+      sink.writeUtf8WithLength(typeName)
+      if (proxiedBytes != null) {
+        sink.writeByte(PROXY_IDENTIFIER_TAG.toInt())
+        sink.write(proxiedBytes)
+      } else {
+        sink.writeByte(NO_PROXY_IDENTIFIER_TAG.toInt())
+      }
+      sink.readByteString()
     }
   }
 
   /**
-   * Determines equality to another [WorkflowIdentifier] by comparing their [type]s and their
+   * Determines equality to another [WorkflowClassIdentifier] by comparing their [type]s and their
    * [proxiedIdentifier]s.
    */
   override fun equals(other: Any?): Boolean = when {
     this === other -> true
-    other !is WorkflowIdentifier -> false
+    other !is WorkflowClassIdentifier -> false
     else -> type == other.type && proxiedIdentifier == other.proxiedIdentifier
   }
 
@@ -103,31 +153,68 @@ class WorkflowIdentifier internal constructor(
      * @throws ClassNotFoundException if one of the workflow types can't be found in the class
      * loader
      */
-    fun read(source: BufferedSource): WorkflowIdentifier? {
+    fun parse(bytes: ByteString): WorkflowIdentifier? = Buffer().let { source ->
+      source.write(bytes)
+
       try {
         val typeString = source.readUtf8WithLength()
         val proxiedIdentifier = when (source.readByte()) {
           NO_PROXY_IDENTIFIER_TAG -> null
-          PROXY_IDENTIFIER_TAG -> read(source)
+          PROXY_IDENTIFIER_TAG -> parse(source.readByteString())
           else -> throw IllegalArgumentException("Invalid WorkflowIdentifier")
         }
 
         @Suppress("UNCHECKED_CAST")
         val type = Class.forName(typeString) as Class<out Workflow<Nothing, Any, Any>>
-        return WorkflowIdentifier(type.kotlin, proxiedIdentifier)
+        return WorkflowClassIdentifier(type.kotlin, proxiedIdentifier)
       } catch (e: EOFException) {
         throw IllegalArgumentException("Invalid WorkflowIdentifier")
       }
     }
   }
 }
 
+/**
+ * A [WorkflowIdentifier] that
+ */
+@OptIn(ExperimentalWorkflowApi::class)
+private class TransientWorkflowIdentifier(private val type: KType) : WorkflowIdentifier() {
+  override val proxiedIdentifier: WorkflowIdentifier? get() = null
+  override val typeName: String by lazy(PUBLICATION) { type.toString() }
+  override fun maybeToByteString(): ByteString? = null
+
+  /** Determines equality to another [TransientWorkflowIdentifier] by comparing their [type]s. */
+  override fun equals(other: Any?): Boolean = when {
+    this === other -> true
+    other !is TransientWorkflowIdentifier -> false
+    else -> type == other.type
+  }
+
+  /** Derives a hashcode from [type]. */
+  override fun hashCode(): Int = type.hashCode()
+}
+
+/**
+ * Creates a [WorkflowIdentifier] that is not capable of being snapshotted and will cause any
+ * [ImpostorWorkflow] workflow identified by it to also not be snapshotted.
+ *
+ * **This function should not be used for [ImpostorWorkflow]s that wrap arbitrary workflows**, since
+ * those workflows may expect to be on snapshotted. Using such identifiers _anywhere in the
+ * [ImpostorWorkflow.realIdentifier] chain_ will disable snapshotting for that workflow. **This
+ * function should only be used for [ImpostorWorkflow]s that wrap a closed set of known workflow
+ * types.**
+ */
+@ExperimentalWorkflowApi
+@Suppress("unused")
+fun unsnapshottableIdentifier(type: KType): WorkflowIdentifier =
+  TransientWorkflowIdentifier(type)
+
 /**
  * The [WorkflowIdentifier] that identifies this [Workflow].
  */
 @ExperimentalWorkflowApi
 val Workflow<*, *, *>.identifier: WorkflowIdentifier
   get() {
     val proxiedIdentifier = (this as? ImpostorWorkflow)?.realIdentifier
-    return WorkflowIdentifier(type = this::class, proxiedIdentifier = proxiedIdentifier)
+    return WorkflowClassIdentifier(type = this::class, proxiedIdentifier = proxiedIdentifier)
   }