Test subscription with pure WebSocketClient

[hantsy]

I have dgs-kotlin example, it is based on traditional Spring WebMVC stack, it is already updated to the latest DGS 6/Spring Boot 3.0.

I tried to use a StandardWebSocketClient to connect subscriptions endpoint in integration tests.

// subscribe commentaAdded event
val socketClient = StandardWebSocketClient()
val commentsReplay = ArrayList<String>(2)

val subscriptionQuery = mapOf(
    "payload" to mapOf(
        "query" to "subscription onCommentAdded { commentAdded { id postId content } }",
        "variables" to emptyMap<String, Any>()// have to add this to avoid a NPE exception.
    ),
    "type" to "start",
    "id" to 1
)

// The Dgs webmvc subscription is handled by DgsWebsocketHandler which also implements Apollo subscription websocket protocol.
// see: https://github.com/apollographql/subscriptions-transport-ws/blob/master/PROTOCOL.md
// declare a websocket handler to send message and receive message
val socketHandler: WebSocketHandler = object : TextWebSocketHandler() {
    override fun afterConnectionEstablished(session: org.springframework.web.socket.WebSocketSession) {
        session.sendMessage(
            TextMessage(
                """
                {
                    "type":"init", 
                    "id":1
                }
                """.trimIndent()
            )
        )
        session.sendMessage(TextMessage(objectMapper.writeValueAsBytes(subscriptionQuery)))
    }

    override fun handleTextMessage(
        session: org.springframework.web.socket.WebSocketSession,
        message: TextMessage
    ) {
        log.debug("handling message: {}", message)
        val comment = objectMapper.convertValue(
            JsonPath.read(message.payload, "$.payload.data.commentAdded"),
            Comment::class.java
        )
        log.debug("received comments: {}", comment)
        commentsReplay.add(comment.content)
    }
}

// do shakehands
val uri: URI = URI.create("ws://localhost:$port/subscriptions")
val headers: WebSocketHttpHeaders = WebSocketHttpHeaders(HttpHeaders())

socketClient.execute(socketHandler, headers, uri)
    .thenApply {
        log.debug("session: {}", it)

        it
    }
    .join()

Thread.sleep(1000L)

log.debug("comment replay: {}", commentsReplay)
// limit to the `latest` item in the `Sinks.replay`
assertThat(commentsReplay).isEqualTo(arrayListOf("comment2"))

It uses a custom WebSocketHandler to save a copy of incoming messages, but when I ran the tests, it seems the method handleTextMessage of this webscoket handler never be called.

The complete codes is here.

I remembered this test was working in the earlier commits.

[srinivasankavitha]

Thanks for checking. Could you be more specific in stating the version it was last working with? We did update websocket subscriptions to support the newer graphql-ws protocol (the older one is around for backwards compatibility, but not actively maintained) almost a year ago. The code here has not changed much since then.

[hantsy]

Yes, the old version maybe work with the legacy subscrptions-transport-ws protocol, the original version I just used a start type message.

But now I tried to change the codes to satisfy the new graphql-ws protocol like this.

val initMessage = mapOf(
  "payload" to emptyMap<String, Any>(),
  "type" to "connection_init"
)
val subscribeMessage = mapOf(
  "payload" to subscriptionsQuery,
  "type" to "subscribe",
  "id" to "1"
)

session.sendMessage(TextMessage(initMessage.toJson()))
session.sendMessage(TextMessage(subscribeMessage.toJson()))

After the client is connected, send a connection_init and subscribe type messages in the afterConnectionEstablished method, it did not trigger the handleMessage in the server side(no any log in the console).

I am not sure where is needed to improve?

I need to add some headers to declare where subprotocols I am using to connect?

[srinivasankavitha]

yes, you will need the sub-protocol header to indicate that you are using the new protocol in that case. Else it defaults to the older one. For reference, https://github.com/enisdenjo/graphql-ws/blob/master/PROTOCOL.md#successful-connection-initialisation

…

On Wed, Jan 25, 2023 at 12:35 AM Hantsy Bai ***@***.***> wrote: Yes, the old version maybe work with the legacy subscrptions-transport-ws protocol. But now I tried to change the codes to satisfy the new graphql-ws protocol like this. val initMessage = mapOf( "payload" to emptyMap<String, Any>(), "type" to "connection_init" ) val subscribeMessage = mapOf( "payload" to subscriptionsQuery, "type" to "subscribe", "id" to "1" ) session.sendMessage(TextMessage(initMessage.toJson())) session.sendMessage(TextMessage(subscribeMessage.toJson())) After the client is connected, send a connection_init and subscribe type messages in the afterConnectionEstablished method, it did not trigger the handleMessage in the server side(no any log in the console). I am not sure where is needed to improve? I need to add some headers to declare where subprotocols I am using to connect? — Reply to this email directly, view it on GitHub <#1399 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AJ5JPXLCL2OZSXH2YLBV23DWUDQWNANCNFSM6AAAAAAUCY5P4A> . You are receiving this because you commented.Message ID: ***@***.***>

[hantsy]

Worked after adding this subprotocol in the header. Thanks.