bug: problem with multiple subscriptions in one websocket session in reactive (webflux) mode

[bartebor]

I found problems with both graphql-transport-ws and graphql-ws in webflux (reactive) mode. Using graphql-transport-ws it is not possible to subscribe more than once with different ids, while graphql-ws does not check for duplications and "forgets" previous subscriptions.

Expected behavior

It should be possible to subscribe multiple times using different ids. It should be also possible to stop subscriptions with those ids.

Actual behavior

• graphql-transport-ws does not support multiple subscriptions:

$ wscat -s graphql-transport-ws -c ws://localhost:8088/api/subscriptions
Connected (press CTRL+C to quit)
> {"type":"connection_init","payload":{}}
< {"payload":{},"type":"connection_ack"}
> {"id":"1","type":"subscribe","payload":{"variables":{},"extensions":{},"operationName":null,"query":"subscription { configMap(id: \"1J20\") { id } }"} }
< {"id":"1","payload":{"data":{"configMap":{"id":"1J20"}},"errors":[]},"type":"next"}
> {"id":"2","type":"subscribe","payload":{"variables":{},"extensions":{},"operationName":null,"query":"subscription { configMap(id: \"1J20\") { id } }"} }
Disconnected (code: 4409, reason: "Subscriber for 6be19906 already exists")

Note that subscription id returned in error message (6be19906) is not what client supplied (1). It looks like the implementation wrongly uses websocket session id instead of user supplied subscription id: https://github.com/Netflix/dgs-framework/blob/v5.4.3/graphql-dgs-spring-webflux-autoconfigure/src/main/kotlin/com/netflix/graphql/dgs/webflux/handlers/WebsocketGraphQLTransportWSProtocolHandler.kt#L93

• graphql-ws takes whatever user throws at it:

$ wscat -s graphql-ws -c ws://localhost:8088/api/subscriptions
Connected (press CTRL+C to quit)
> {"type":"connection_init","payload":{}}
< {"type":"connection_ack","payload":null,"id":""}
> {"id":"1","type":"start","payload":{"variables":{},"extensions":{},"operationName":null,"query":"subscription { configMap(id: \"1J20\") { id } }"} }
< {"type":"data","payload":{"data":{"configMap":{"id":"1J20"}},"errors":[]},"id":"1"}
> {"id":"1","type":"start","payload":{"variables":{},"extensions":{},"operationName":null,"query":"subscription { configMap(id: \"1J20\") { id } }"} }
< {"type":"data","payload":{"data":{"configMap":{"id":"1J20"}},"errors":[]},"id":"1"}
< {"type":"data","payload":{"data":{"configMap":{"id":"1J20"}},"errors":[]},"id":"1"}
> {"id":"1","type":"stop"}
< {"type":"data","payload":{"data":{"configMap":{"id":"1J20"}},"errors":[]},"id":"1"}

Now we can subscribe twice with the same id and we receive data from all subscriptions. We can stop the last subscription, but previous ones are forgotten and still emit data.
There is no id check around https://github.com/Netflix/dgs-framework/blob/v5.4.3/graphql-dgs-spring-webflux-autoconfigure/src/main/kotlin/com/netflix/graphql/dgs/webflux/handlers/WebsocketGraphQLTransportWSProtocolHandler.kt#L109

Steps to reproduce

You need any webflux subscription server and issue requests as above.

PS. I also found that error returned by the publisher in any subscription closes websocket. I think it should send error message to the client and terminate only one subscription, not whole session.