Add hints for subscription OperationMessage's payload serialization.

The idea of this commit is to aid Jackson infer the content of the
`OperationMessage` payload by adding the `@JsonSubType` hint. With this
we can tell Jackson that the type could be either...

* empty, via the EmptyPayload class.
* data, via the DataPayload class.
* query, via the QueryPayload class

graphql-dgs-subscription-types/src/main/kotlin/com/netflix/graphql/types/subscription/OperationMessage.kt

@@ -16,7 +16,10 @@
 
 package com.netflix.graphql.types.subscription
 
+import com.fasterxml.jackson.annotation.JsonCreator
 import com.fasterxml.jackson.annotation.JsonProperty
+import com.fasterxml.jackson.annotation.JsonSubTypes
+import com.fasterxml.jackson.annotation.JsonTypeInfo
 
 // OperationMessage types
 const val GQL_CONNECTION_INIT = "connection_init"
@@ -33,28 +36,46 @@ const val GQL_CONNECTION_KEEP_ALIVE = "ka"
 data class OperationMessage(
     @JsonProperty("type")
     val type: String,
+
     @JsonProperty("payload")
+    @JsonTypeInfo(use = JsonTypeInfo.Id.DEDUCTION, defaultImpl = EmptyPayload::class)
+    @JsonSubTypes(
+        JsonSubTypes.Type(value = EmptyPayload::class),
+        JsonSubTypes.Type(value = DataPayload::class),
+        JsonSubTypes.Type(value = QueryPayload::class)
+    )
     val payload: Any? = null,
     @JsonProperty("id", required = false)
     val id: String? = ""
 )
 
+sealed interface MessagePayload
+
+object EmptyPayload : HashMap<String, Any?>(), MessagePayload {
+    @JvmStatic
+    @JsonCreator
+    @SuppressWarnings("unused")
+    fun emptyPayload(): EmptyPayload {
+        return EmptyPayload
+    }
+}
+
 data class DataPayload(
     @JsonProperty("data")
     val data: Any?,
     @JsonProperty("errors")
     val errors: List<Any>? = emptyList()
-)
+) : MessagePayload
 
 data class QueryPayload(
     @JsonProperty("variables")
-    val variables: Map<String, Any>?,
+    val variables: Map<String, Any>? = emptyMap(),
     @JsonProperty("extensions")
-    val extensions: Map<String, Any> = emptyMap(),
+    val extensions: Map<String, Any>? = emptyMap(),
     @JsonProperty("operationName")
-    val operationName: String?,
+    val operationName: String? = null,
     @JsonProperty("query")
     val query: String
-)
+) : MessagePayload
 
 data class Error(@JsonProperty val message: String = "")

graphql-dgs-subscription-types/src/test/kotlin/OperationMessageTest.kt

@@ -19,8 +19,12 @@ import com.fasterxml.jackson.databind.JsonMappingException
 import com.fasterxml.jackson.module.kotlin.MissingKotlinParameterException
 import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
 import com.netflix.graphql.types.subscription.*
+import org.assertj.core.api.Assertions.assertThat
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.assertThrows
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.Arguments
+import org.junit.jupiter.params.provider.MethodSource
 
 /*
  * Copyright 2021 Netflix, Inc.
@@ -39,15 +43,19 @@ import org.junit.jupiter.api.assertThrows
  */
 
 class OperationMessageTest {
-    companion object {
-        val MAPPER = jacksonObjectMapper()
-    }
 
     @Test
     fun rejectsMessageWithoutType() {
         assertFailsToDeserialize<MissingKotlinParameterException>("""{"id": "2"}""")
     }
 
+    @ParameterizedTest
+    @MethodSource("validMessages")
+    fun deserializes(message: String, expected: OperationMessage) {
+        val deserialized = deserialize(message)
+        assertThat(expected).isEqualTo(deserialized)
+    }
+
     @Test
     fun rejectsQueryMessageWithoutQuery() {
         assertFailsToDeserialize<JsonMappingException>(
@@ -69,4 +77,143 @@ class OperationMessageTest {
 
     private fun deserialize(message: String) =
         MAPPER.readValue(message, object : TypeReference<OperationMessage>() {})
+
+    companion object {
+        val MAPPER = jacksonObjectMapper()
+
+        @JvmStatic
+        fun validMessages() = listOf(
+            Arguments.of(
+                """{"type": "connection_init"}""",
+                OperationMessage(GQL_CONNECTION_INIT, null, "")
+            ),
+            Arguments.of(
+                """
+                {"type": "connection_init",
+                 "payload": {}}
+                """.trimIndent(),
+                OperationMessage(GQL_CONNECTION_INIT, EmptyPayload)
+            ),
+            Arguments.of(
+                """
+                {"type": "stop",
+                 "id": "3"}
+                """.trimIndent(),
+                OperationMessage(GQL_STOP, null, "3")
+            ),
+            Arguments.of(
+                """
+                {"type": "stop",
+                 "id": 3}
+                """.trimIndent(),
+                OperationMessage(GQL_STOP, null, "3")
+            ),
+            Arguments.of(
+                """
+                {"type": "start",
+                 "payload": {
+                    "query": "my-query"
+                 },
+                 "id": "3"}
+                """.trimIndent(),
+                OperationMessage(GQL_START, QueryPayload(query = "my-query"), "3")
+            ),
+            Arguments.of(
+                """
+                {"type": "start",
+                 "payload": {
+                    "operationName": "query",
+                    "query": "my-query"
+                 },
+                 "id": "3"}
+                """.trimIndent(),
+                OperationMessage(GQL_START, QueryPayload(operationName = "query", query = "my-query"), "3")
+            ),
+            Arguments.of(
+                """
+                {"type": "start",
+                 "payload": {
+                    "operationName": "query",
+                    "extensions": {"a": "b"},
+                    "query": "my-query"
+                 },
+                 "id": "3"}
+                """.trimIndent(),
+                OperationMessage(
+                    GQL_START,
+                    QueryPayload(
+                        extensions = mapOf(Pair("a", "b")),
+                        operationName = "query",
+                        query = "my-query"
+                    ),
+                    "3"
+                )
+            ),
+            Arguments.of(
+                """
+                {"type": "start",
+                 "payload": {
+                    "operationName": "query",
+                    "extensions": {"a": "b"},
+                    "variables": {"c": "d"},
+                    "query": "my-query"
+                 },
+                 "id": "3"}
+                """.trimIndent(),
+                OperationMessage(
+                    GQL_START,
+                    QueryPayload(
+                        variables = mapOf(Pair("c", "d")),
+                        extensions = mapOf(Pair("a", "b")),
+                        operationName = "query",
+                        query = "my-query"
+                    ),
+                    "3"
+                )
+            ),
+            Arguments.of(
+                """
+                {"type": "data",
+                 "payload": {
+                    "data": {
+                        "a": 1,
+                        "b": "hello",
+                        "c": false
+                    }
+                 },
+                 "id": "3"}
+                """.trimIndent(),
+                OperationMessage(
+                    GQL_DATA,
+                    DataPayload(data = mapOf(Pair("a", 1), Pair("b", "hello"), Pair("c", false))),
+                    "3"
+                )
+            ),
+            Arguments.of(
+                """
+                {"type": "data",
+                 "payload": {
+                    "errors": ["an-error"]
+                 },
+                 "id": "3"}
+                """.trimIndent(),
+                OperationMessage(GQL_DATA, DataPayload(data = null, listOf("an-error")), "3")
+            ),
+            Arguments.of(
+                """
+                {"type": "data",
+                 "payload": {
+                    "data": {
+                        "a": 1,
+                        "b": "hello",
+                        "c": false
+                    },
+                    "errors": ["an-error"]
+                 },
+                 "id": "3"}
+                """.trimIndent(),
+                OperationMessage(GQL_DATA, DataPayload(mapOf(Pair("a", 1), Pair("b", "hello"), Pair("c", false)), listOf("an-error")), "3")
+            ),
+        )
+    }
 }