@Override
	@SuppressWarnings("unchecked")
	public Mono<WebSocketCloseStatus> receiveCloseStatus() {
		return onCloseState.asMono().or((Mono) onTerminate());
	}
@Test
	void testWebSocketServerCancelled() throws InterruptedException {
		try (LogTracker lt = new LogTracker(HttpServerOperations.class, WebsocketServerOperations.INBOUND_CANCEL_LOG)) {
			AtomicReference<WebSocketCloseStatus> clientCloseStatus = new AtomicReference<>();
			AtomicReference<WebSocketCloseStatus> serverCloseStatus = new AtomicReference<>();
			CountDownLatch closeLatch = new CountDownLatch(2);
			CountDownLatch cancelled = new CountDownLatch(1);
			AtomicReference<List<String>> serverMsg = new AtomicReference<>(new ArrayList<>());
			Sinks.Empty<Void> empty = Sinks.empty();
			CancelReceiverHandlerTest cancelReceiver = new CancelReceiverHandlerTest(() -> empty.tryEmitEmpty());

			disposableServer = createServer()
					.handle((in, out) -> out.sendWebsocket((i, o) -> {
						i.withConnection(conn -> conn.addHandlerLast(cancelReceiver));

						i.receiveCloseStatus()
								.log("server.closestatus")
								.doOnNext(status -> {
									serverCloseStatus.set(status);
									closeLatch.countDown();
								})
								.subscribe();

						Mono<Void> receive = i.receive()
								.asString()
								.log("server.receive")
								.doOnCancel(cancelled::countDown)
								.doOnNext(s -> serverMsg.get().add(s))
								.then();

						return Flux.zip(receive, empty.asMono())
								.then(Mono.never());
					}))
					.bindNow();

			createClient(disposableServer.port())
					.websocket()
					.uri("/test")
					.handle((in, out) -> {
						in.receiveCloseStatus()
								.log("client.closestatus")
								.doOnNext(status -> {
									clientCloseStatus.set(status);
									closeLatch.countDown();
								})
								.subscribe();

						return out.sendString(Mono.just("PING"))
								.neverComplete();
					})
					.log("client")
					.subscribe();

			assertThat(closeLatch.await(30, TimeUnit.SECONDS)).isTrue();
			// client received closed without any status code
			assertThat(clientCloseStatus.get()).isNotNull().isEqualTo(WebSocketCloseStatus.EMPTY);
			// server locally closed abnormally
			assertThat(serverCloseStatus.get()).isNotNull().isEqualTo(WebSocketCloseStatus.ABNORMAL_CLOSURE);
			assertThat(lt.latch.await(30, TimeUnit.SECONDS)).isTrue();
			assertThat(cancelled.await(30, TimeUnit.SECONDS)).isTrue();

			List<String> serverMessages = serverMsg.get();
			assertThat(serverMessages).isNotNull();
			assertThat(serverMessages.size()).isEqualTo(0);
			assertThat(cancelReceiver.awaitAllReleased(30)).as("cancelReceiver").isTrue();
		}
	}