从kafka获取数据并将其发送到RSocket

Question

我正在尝试从kafka获取消息，并使用Spring将其发送到RSocket。在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.log("got error");
console.error(error);
},
onSubscribe: cancel => {
/* call cancel() to abort */
console.log("subscribe!");
console.log(cancel);
// cancel.cancel();
}
});

Answer 1

您可以在我们有的requester.route("input").data("Welcome to Rsocket").send();中执行此操作

/**
* Perform a {@link RSocket#fireAndForget fireAndForget} sending the
* provided data and metadata.
* @return a completion that indicates if the payload was sent
* successfully or not. Note, however that is a one-way send and there
* is no indication of whether or how the event was handled on the
* remote end.
*/
Mono<Void> send();

你看到Mono了吗？这意味着必须订阅才能启动反应流处理。有关更多信息，请参阅项目Reactor：https://projectreactor.io/

另一方面，在您的情况下，还不清楚什么是服务器，什么是客户端。。。你这样做：

/**
* Build an {@link RSocketRequester} with an
* {@link io.rsocket.core.RSocketClient} that connects over WebSocket to
* the given URL. The requester can be used to make requests
* concurrently. Requests are made over a shared connection that is also
* re-established as needed when further requests are made.
* @param uri the URL to connect to
* @return the created {@code RSocketRequester}
* @since 5.3
*/
RSocketRequester websocket(URI uri);

我想说，在您显示的代码中，这意味着客户端。服务器位于为ws://协议打开7000端口的另一侧。因此，请确保您理解并正确配置所有部件。例如，我不明白你为什么在Listener类中需要@RestController。。。