Common code to deserialize a state message in the new format

airbyte-protocol/protocol-models/build.gradle

@@ -7,6 +7,7 @@ plugins {
 dependencies {
     implementation 'javax.validation:validation-api:1.1.0.Final'
     implementation 'org.apache.commons:commons-lang3:3.11'
+    implementation 'io.vavr:vavr:0.10.4'
 }
 
 jsonSchema2Pojo {

airbyte-protocol/protocol-models/src/main/java/io/airbyte/protocol/models/StateMessageHelper.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.protocol.models;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JsonNode;
+import io.airbyte.commons.json.Jsons;
+import io.vavr.control.Either;
+import java.util.ArrayList;
+import java.util.List;
+
+public class StateMessageHelper {
+
+  public static class AirbyteStateMessageListTypeReference extends TypeReference<List<AirbyteStateMessage>> {}
+
+  /**
+   * This a takes a json blob state and tries return either a legacy state in the format of a json
+   * object or a state message with the new format which is a list of airbyte state message.
+   *
+   * @param state
+   * @return Either a json blob (on the left) or a structure state message.
+   */
+  public static Either<JsonNode, List<AirbyteStateMessage>> getTypedState(JsonNode state) {
+    if (state == null) {
+      return Either.right(new ArrayList<>());
+    }
+    try {
+      return Either.right(Jsons.object(state, new AirbyteStateMessageListTypeReference()));
+    } catch (final IllegalArgumentException e) {
+      return Either.left(state);
+    }
+  }
+
+}