Socket.IO-client Java 용 Reactor Flux 프록시

Question

Socket.IO-client Java에서 데이터를 가져와야하는 Spring WebFlux 엔드 포인트를 구현 중입니다.

들어오는 데이터를 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; 

} 

Answer 1

당신은 이벤트 리스너에서 Flux을 생성하는 Flux.create()를 사용할 수 있습니다.

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(...) 
    }); 
});