Handling the WebSocketSession close event if initialization was not successful

[AliakseiKakora]

I use spring-graphql 1.3.1 as part of spring-boot-starter-graphql. I want to create a GraphQL subscription to the server which does not belong to me (means I can not do anything on server side to solve my issue). This is how I create subscription and handle events

    WebSocketClient client = new ReactorNettyWebSocketClient(HttpClient.create(),
            () -> WebsocketClientSpec.builder().maxFramePayloadLength(256 * 1024));

    String url = configs.getWsUrl();
    WebSocketGraphQlClient graphQlClient = WebSocketGraphQlClient
            .builder(url, client)
            .keepAlive(Duration.ofSeconds(10))
            .interceptor(new GraphQlClientInterceptor(configs))
            .build();

    Flux<ClientGraphQlResponse> clientGraphQlResponseFlux = graphQlClient
            .document(getQuery(configs))
            .executeSubscription();

    AtomicReference<String> offset = new AtomicReference<>(configs.getOffset());
    Disposable disposable = clientGraphQlResponseFlux
            .map(data -> data.toEntity(BusinessEvent.class))
            .doOnError(error -> {
                log.error("Error in subscription");
            })
            .subscribe(message -> {
                //business logic
            });

And this is how I send Auth token - as part of connection_init payload in the GraphQlClientInterceptor

public class GraphQlClientInterceptor implements WebSocketGraphQlClientInterceptor {

private final SubscriptionManagement configs;

/**
 * Create payload property data. Connection Init is in following format
 * <pre>
 * {
 *   "id": "{GUID}"
 *   "type": "connection_init",
 *   "payload": {
 *     "Authorization": "Bearer {OAUTH TOKEN}
 *     "x-app-key": "{APPLICATION KEY}"
 *   }
 * }<pre/>
 */
@Override
public Mono<Object> connectionInitPayload() {
    Map<String, String> payload = Map.of(
            "x-app-key", configs.getAppKey(),
            "Authorization", configs.getAuthorization()
    );
    return Mono.just(payload);
}

}

In general it works fine - I'm able to connect to the server and process messages. Once some errors occurs I'm able to handle them as well.

But when I started testing flow with invalid auth credentials - the error was not throw to my handler (means doOnError()). I see just log message like this -

"GraphQlSession over ReactorNettyWebSocketSession[id=2127633d, uri=wss://my-url] disconnected with CloseStatus[code=4401, reason=Unauthorised - Invalid credentials]".

Based on my investigation - WebSocketGraphQlTransport tries to terminate request. But at the time when connection_init payload send (and was rejected because of invalid auth token) there no connection yet. Similar issue happens when connection timeout error occurs. In my understanding the issue is - when error occurs on the connection_init phase I do not receive any Errors in the Flux. I'm looking a way to be able to handle any issues that occurs during subscription lifecycle. I would appreciate any help here.