Socket.IO-client Java 的 Reactor Flux 代理答案

【问题标题】：Reactor Flux proxy for Socket.IO-client JavaSocket.IO-client Java 的 Reactor Flux 代理
【发布时间】：2017-11-04 11:41:21
【问题描述】：

我正在实现 Spring WebFlux 端点，它应该从 Socket.IO-client Java 获取数据。

我不明白如何将传入数据收集到 Flux 流中。我可以通过某种方式创建新的 Flux 并订阅该传入数据吗？感谢您的建议。

@GetMapping("/streaming", produces = MediaType.APPLICATION_STREAM_JSON_VALUE)
    public Flux<MyRecourse> getStreaming() {

    URI uri = URI.create("http://localhost/socket.io"); // client
    Socket socket = IO.socket(uri);

    socket.on("event", args -> {    
        JSONObject obj = (JSONObject)args[0]; 
        MyRecourse recource = MyRecourse.create(obj);

        // how to put this recource into Flux stream?
    });

    return fluxStreamOfRecources;

}

【问题讨论】：

标签： project-reactor spring-webflux socket.io-java-client

【解决方案1】：

您可以使用Flux.create() 从事件侦听器生成Flux。

Flux.<MyResource>create(emitter -> {

     URI uri = URI.create("http://localhost/socket.io"); // client
     Socket socket = IO.socket(uri);

     socket.on("event", args -> {    
       JSONObject obj = (JSONObject)args[0]; 
       MyResource resource = MyResource.create(obj);
       emitter.next(resource);
     });

     // subscribe on error events
     socket.on(Socket.EVENT_CONNECT_ERROR, args -> {    
       // get error
       emitter.error(throwable);
     });

     // unsubscribe from events when the client cancels
     emitter.onDispose(() -> {
         // disconnect from socket
         // socket.off(...)
     });
 });

【讨论】：