class 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();

}