Обработка события закрытия WebSocketSession, если инициализация не удалась. ⇐ JAVA

[Anonymous]

Я использую Spring-graphql 1.3.1 как часть Spring-boot-starter-graphql.
Я хочу создать подписку GraphQL на сервер, который мне не принадлежит (означает, что я не могу ничего сделать на стороне сервера, чтобы решить мою проблему).
Вот как я создаю подписку и обрабатываю события

Код: Выделить всё

        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 clientGraphQlResponseFlux = graphQlClient
.document(getQuery(configs))
.executeSubscription();

AtomicReference 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
});

И вот как я отправляю токен аутентификации — как часть полезных данных Connection_init в GraphQlClientInterceptor

Код: Выделить всё

public class GraphQlClientInterceptor implements WebSocketGraphQlClientInterceptor {

private final SubscriptionManagement configs;

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

В целом работает нормально - могу подключаться к серверу и обрабатывать сообщения. Как только возникают некоторые ошибки, я также могу их обработать.
Но когда я начал тестировать поток с недействительными учетными данными для аутентификации, ошибка не была выдана моему обработчику (означает doOnError()< /код>). Я вижу такое сообщение в журнале:

"GraphQlSession over ReactorNettyWebSocketSession[id=2127633d,
uri=wss://my-url] отключено с CloseStatus

Код: Выделить всё

».

По данным моего расследования — WebSocketGraphQlTransport пытается завершить запрос. Но в то время, когда полезная нагрузка Connection_init была отправлена ​​(и была отклонена из-за недействительного токена аутентификации), соединения еще не было. Аналогичная проблема возникает, когда возникает ошибка тайм-аута соединения. Насколько я понимаю, проблема в том, что когда возникает ошибка на этапе Connection_init, я не получаю никаких ошибок в Flux.
Я ищу способ справиться с любыми проблемами, возникающими в течение жизненного цикла подписки. . Буду признателен за любую помощь. 

Подробнее здесь: [url]https://stackoverflow.com/questions/79265405/handling-the-websocketsession-close-event-if-initialization-was-not-successful[/url]