TcpClient только возвращает первый результат в Spring WebFlux Controller - PullRequest
Новая
       42

TcpClient только возвращает первый результат в Spring WebFlux Controller

0 голосов
/ 13 апреля 2020

У меня есть служба HTTP, предоставляющая конечную точку GET, которая подключается к простому эхо-серверу через TCP. Служба HTTP работает в Netty. 

@RestController
public class OurTcpClient {

    private Connection connection;

    @GetMapping("echo1")
    public Mono<String> echo(@RequestParam("value") final String value) {
        this.connection.outbound()
            .sendString(Mono.just(String.format("%04d", value.length()) + value)) // prepend length
            .then()
            .subscribe();
        return this.connection.inbound()
            .receive()
            .asString()
            .next();
    }

    @PostConstruct
    public void init() {
        this.connection = TcpClient.create()
            .host("localhost")
            .port(10002)
            .wiretap(true)
            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000)
            .option(ChannelOption.SO_KEEPALIVE, true)
            .connectNow();
    }
}

Я ожидаю, что я могу запросить службу, например, по адресу http://localhost: 8081 / echo1? Value = hi , как столько раз, сколько мне нравится, и получаю «привет» обратно в каждом ответе. Это работает для первого запроса. Второй запрос висит бесконечно. Если затем я отменяю второй запрос и пытаюсь выполнить другой, я получаю следующую ошибку: 

{
    "timestamp": "2020-04-13T18:56:40.221+0000",
    "path": "/echo1",
    "status": 500,
    "error": "Internal Server Error",
    "message": "Only one connection receive subscriber allowed."
}

Любая помощь будет принята с благодарностью.

1 Ответ

1 голос
/ 14 апреля 2020

В примере вы используете next()

return this.connection.inbound()
            .receive()
            .asString()
            .next();

Согласно Flux#next javado c Излучайте только первый элемент, испускаемый этим потоком, в новый Mono. , тогда подписка будет отменена.

В контексте Reactor Netty при использовании next, timeout, take et c. операторы, которые отменяют подписку, это означает, что соединение будет закрыто.

...