@Override
    public Mono<Void> writeWith(final ServerWebExchange exchange, final ShenyuPluginChain chain) {
        return chain.execute(exchange).doOnError(throwable -> cleanup(exchange)).then(Mono.defer(() -> {
            Connection connection = exchange.getAttribute(Constants.CLIENT_RESPONSE_CONN_ATTR);
            if (Objects.isNull(connection)) {
                Object error = ShenyuResultWrap.error(exchange, ShenyuResultEnum.SERVICE_RESULT_ERROR);
                return WebFluxResultUtils.result(exchange, error);
            }
            ServerHttpResponse response = exchange.getResponse();
            NettyDataBufferFactory factory = (NettyDataBufferFactory) response.bufferFactory();
            final Flux<NettyDataBuffer> body = connection
                    .inbound()
                    .receive()
                    .retain()
                    .map(factory::wrap);
            MediaType contentType = response.getHeaders().getContentType();

            Mono<Void> responseMono = isStreamingMediaType(contentType)
                    ? response.writeAndFlushWith(body.map(Flux::just))
                    : response.writeWith(body);
            exchange.getAttributes().put(Constants.RESPONSE_MONO, responseMono);
            // watcher httpStatus
            final Consumer<HttpStatusCode> consumer = exchange.getAttribute(Constants.WATCHER_HTTP_STATUS);
            Optional.ofNullable(consumer).ifPresent(c -> c.accept(response.getStatusCode()));
            return responseMono.onErrorResume(error -> releaseIfNotConsumed(body, error));
        })).doOnCancel(() -> cleanup(exchange));
    }
@Test
    public void testWriteWith() {
        ServerWebExchange exchangeNoClient = MockServerWebExchange.from(MockServerHttpRequest.get("/test")
                .build());
        StepVerifier.create(nettyClientMessageWriter.writeWith(exchangeNoClient, chain)).expectSubscription().verifyComplete();

        ServerWebExchange exchange = mock(ServerWebExchange.class);
        when(exchange.getRequest()).thenReturn(mock(ServerHttpRequest.class));
        when(exchange.getResponse()).thenReturn(mock(ServerHttpResponse.class));
        when(exchange.getResponse().getHeaders()).thenReturn(HttpHeaders.EMPTY);
        NettyDataBufferFactory factory = new NettyDataBufferFactory(mock(ByteBufAllocator.class));
        when(exchange.getResponse().bufferFactory()).thenReturn(factory);

        Connection connection = mock(Connection.class);
        when(connection.inbound()).thenReturn(mock(NettyInbound.class));
        when(connection.inbound().receive()).thenReturn(mock(ByteBufFlux.class));
        when(connection.inbound().receive().retain()).thenReturn(mock(ByteBufFlux.class));

        final Flux<NettyDataBuffer> body = new Flux<>() {
            @Override
            public void subscribe(final CoreSubscriber<? super NettyDataBuffer> coreSubscriber) {
                coreSubscriber.onComplete();
            }
        };
        when(connection.inbound().receive().retain().map(factory::wrap)).thenReturn(body);
        when(exchange.getAttribute(Constants.CLIENT_RESPONSE_CONN_ATTR)).thenReturn(connection);

        StepVerifier.create(nettyClientMessageWriter.writeWith(exchange, chain)).expectSubscription().verifyError();
    }