CommandHelper { public static <T> Publisher<T> toPublisher(final Class<T> clazz, final ObjectMapper mapper, final HystrixCommand<InputStream> command) { AtomicInteger count = new AtomicInteger(0); return Flux.create(sink -> { final Observable<InputStream> observable = command.toObservable(); observable .map(i -> { try { return mapper.readerFor(clazz).<T> readValues(i); } catch (IOException e) { throw new TDPException(CommonErrorCodes.UNEXPECTED_EXCEPTION, e); } }) .doOnCompleted(() -> LOGGER.debug("Completed command '{}' (emits '{}') with '{}' records.", command.getClass().getName(), clazz.getName(), count.get())) .toBlocking() .forEach(s -> { while (s.hasNext()) { sink.next(s.next()); count.incrementAndGet(); } sink.complete(); }); }, FluxSink.OverflowStrategy.BUFFER); } private  CommandHelper(); static StreamingResponseBody toStreaming(final HystrixCommand<InputStream> command); static ResponseEntity<Void> async(final GenericCommand<?> command); static ResponseEntity<StreamingResponseBody> toStreaming(final GenericCommand<InputStream> command); static Publisher<T> toPublisher(final Class<T> clazz, final ObjectMapper mapper,
            final HystrixCommand<InputStream> command); static Stream<T> toStream(Class<T> clazz, ObjectMapper mapper, HystrixCommand<InputStream> command);  }
@Test public void testCommandToPublisher() throws Exception { HystrixCommand<InputStream> command = new InputStreamTestCommand(); final Publisher<CommandHelperTestData> publisher = CommandHelper.toPublisher(CommandHelperTestData.class, mapper, command); final List<CommandHelperTestData> receivedData = new ArrayList<>(); Flux.from(publisher).subscribe(receivedData::add); assertEquals(3, receivedData.size()); assertEquals("1", receivedData.get(0).getValue()); assertEquals("2", receivedData.get(1).getValue()); assertEquals("3", receivedData.get(2).getValue()); }
@Test public void testMalformedToPublisher() throws Exception { HystrixCommand<InputStream> command = new MalformedInputStreamTestCommand(); final Publisher<CommandHelperTestData> publisher = CommandHelper.toPublisher(CommandHelperTestData.class, mapper, command); final List<CommandHelperTestData> receivedData = new ArrayList<>(); try { Flux.from(publisher).subscribe(receivedData::add); fail(); } catch (Exception e) { assertTrue(receivedData.isEmpty()); } }