test_class.declarator: mapper = new ObjectMapper()
test_class.modifier: private final
test_class.original_string: private final ObjectMapper mapper = new ObjectMapper();
test_class.type: ObjectMapper
test_class.var_name: mapper
test_class.file: dataprep-backend-service/src/test/java/org/talend/dataprep/command/CommandHelperTest.java
test_class.identifier: CommandHelperTest
test_case.body: @Test
      public void testMalformedToPublisher() throws Exception {
          // Given
          HystrixCommand<InputStream> command = new MalformedInputStreamTestCommand();
          final Publisher<CommandHelperTestData> publisher =
                  CommandHelper.toPublisher(CommandHelperTestData.class, mapper, command);
          final List<CommandHelperTestData> receivedData = new ArrayList<>();
  
          // When
          try {
              Flux.from(publisher).subscribe(receivedData::add);
              fail();
          } catch (Exception e) {
              assertTrue(receivedData.isEmpty());
          }
      }
test_case.class_method_signature: CommandHelperTest.testMalformedToPublisher()
test_case.full_signature: @Test public void testMalformedToPublisher()
test_case.identifier: testMalformedToPublisher
test_case.invocations: toPublisher subscribe from fail assertTrue isEmpty
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testMalformedToPublisher()
focal_class.declarator: LOGGER = LoggerFactory.getLogger(CommandHelper.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Logger LOGGER = LoggerFactory.getLogger(CommandHelper.class);
focal_class.type: Logger
focal_class.var_name: LOGGER
focal_class.file: dataprep-backend-service/src/main/java/org/talend/dataprep/command/CommandHelper.java
focal_class.identifier: CommandHelper
focal_class.class_method_signature: CommandHelper.CommandHelper()
focal_class.full_signature: private  CommandHelper()
focal_class.identifier: CommandHelper
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.signature:  CommandHelper()
focal_class.class_method_signature: CommandHelper.toStreaming(final HystrixCommand<InputStream> command)
focal_class.full_signature: public static StreamingResponseBody toStreaming(final HystrixCommand<InputStream> command)
focal_class.identifier: toStreaming
focal_class.modifiers: public static
focal_class.parameters: (final HystrixCommand<InputStream> command)
focal_class.return: StreamingResponseBody
focal_class.signature: StreamingResponseBody toStreaming(final HystrixCommand<InputStream> command)
focal_class.class_method_signature: CommandHelper.async(final GenericCommand<?> command)
focal_class.full_signature: public static ResponseEntity<Void> async(final GenericCommand<?> command)
focal_class.identifier: async
focal_class.modifiers: public static
focal_class.parameters: (final GenericCommand<?> command)
focal_class.return: ResponseEntity<Void>
focal_class.signature: ResponseEntity<Void> async(final GenericCommand<?> command)
focal_class.class_method_signature: CommandHelper.toStreaming(final GenericCommand<InputStream> command)
focal_class.full_signature: public static ResponseEntity<StreamingResponseBody> toStreaming(final GenericCommand<InputStream> command)
focal_class.identifier: toStreaming
focal_class.modifiers: public static
focal_class.parameters: (final GenericCommand<InputStream> command)
focal_class.return: ResponseEntity<StreamingResponseBody>
focal_class.signature: ResponseEntity<StreamingResponseBody> toStreaming(final GenericCommand<InputStream> command)
focal_class.class_method_signature: CommandHelper.toPublisher(final Class<T> clazz, final ObjectMapper mapper,
              final HystrixCommand<InputStream> command)
focal_class.full_signature: public static Publisher<T> toPublisher(final Class<T> clazz, final ObjectMapper mapper,
              final HystrixCommand<InputStream> command)
focal_class.identifier: toPublisher
focal_class.modifiers: public static
focal_class.parameters: (final Class<T> clazz, final ObjectMapper mapper,
              final HystrixCommand<InputStream> command)
focal_class.return: Publisher<T>
focal_class.signature: Publisher<T> toPublisher(final Class<T> clazz, final ObjectMapper mapper,
              final HystrixCommand<InputStream> command)
focal_class.class_method_signature: CommandHelper.toStream(Class<T> clazz, ObjectMapper mapper, HystrixCommand<InputStream> command)
focal_class.full_signature: public static Stream<T> toStream(Class<T> clazz, ObjectMapper mapper, HystrixCommand<InputStream> command)
focal_class.identifier: toStream
focal_class.modifiers: public static
focal_class.parameters: (Class<T> clazz, ObjectMapper mapper, HystrixCommand<InputStream> command)
focal_class.return: Stream<T>
focal_class.signature: Stream<T> toStream(Class<T> clazz, ObjectMapper mapper, HystrixCommand<InputStream> command)
focal_method.body: 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);
      }
focal_method.class_method_signature: CommandHelper.toPublisher(final Class<T> clazz, final ObjectMapper mapper,
              final HystrixCommand<InputStream> command)
focal_method.full_signature: public static Publisher<T> toPublisher(final Class<T> clazz, final ObjectMapper mapper,
              final HystrixCommand<InputStream> command)
focal_method.identifier: toPublisher
focal_method.invocations: create toObservable forEach toBlocking doOnCompleted map readValues readerFor debug getName getClass getName get hasNext next next incrementAndGet complete
focal_method.modifiers: public static
focal_method.parameters: (final Class<T> clazz, final ObjectMapper mapper,
              final HystrixCommand<InputStream> command)
focal_method.return: Publisher<T>
focal_method.signature: Publisher<T> toPublisher(final Class<T> clazz, final ObjectMapper mapper,
              final HystrixCommand<InputStream> command)