Получить данные из kafka и отправить их в RSocket

Я пытаюсь получать сообщения от kafka и отправлять их в RSocket с помощью Spring. Публикация на стороне сервера на Spring Java и на стороне клиента с помощью React

@Configuration
@EnableConfigurationProperties(RsocketConsumerProperties.class)
public class RsocketConsumerConfiguration {

    @Bean
    public Function<Integer, Mono<Integer>> rsocketConsumer(RSocketRequester.Builder builder,
                                                            RsocketConsumerProperties rsocketConsumerProperties) {
        RSocketRequester rSocketRequester = builder.websocket(URI.create("ws://localhost:7000/"));
        return input -> rSocketRequester.route(rsocketConsumerProperties.getRoute()).data(input).retrieveMono(Integer.class);
    }
}

@EnableBinding(Sink.class)
public class Listener {

    @Autowired
    private Function<Integer, Mono<Integer>> rsocketConsumer;


    @StreamListener(Sink.INPUT)
    public void fireAndForget(Integer val) {
        System.out.println(val);
        rsocketConsumer.apply(val).subscribe();
    }
}

@Controller
public class ServerController {

    @MessageMapping("data")
    public Mono<Integer> hello(Integer integer) {
        return Mono.just(integer);
    }

}

Что я делаю неправильно на стороне сервера, потому что мой клиент подключен, но не может получать новые сообщения

  client.connect().subscribe({
    onComplete: socket => {
        socket.fireAndForget({
          data: { message: "hello from javascript!" },
          metadata: null
        });
      },
      onError: error => {
        console.info("got error");
        console.error(error);
      },
      onSubscribe: cancel => {
        /* call cancel() to abort */
        console.info("subscribe!");
        console.info(cancel);
        // cancel.cancel();
      }
    });