test_class.declarator: factory = new CancelStageFactory()
test_class.modifier: private final
test_class.original_string: private final CancelStageFactory factory = new CancelStageFactory();
test_class.type: CancelStageFactory
test_class.var_name: factory
test_class.file: implementation/src/test/java/io/smallrye/reactive/streams/stages/CancelStageFactoryTest.java
test_class.identifier: CancelStageFactoryTest
test_class.superclass: extends StageTestBase
test_case.body: @Test
      public void create() throws ExecutionException, InterruptedException {
          TerminalStage<Long, Void> terminal = factory.create(null, new Stage.Cancel() {
          });
          AtomicBoolean cancelled = new AtomicBoolean();
          List<Long> list = new ArrayList<>();
          Flowable<Long> flowable = Flowable.interval(1000, TimeUnit.MILLISECONDS)
                  .observeOn(Schedulers.io())
                  .doOnNext(list::add)
                  .doOnCancel(() -> cancelled.set(true));
          CompletionStage<Void> stage = terminal.apply(flowable);
          stage.toCompletableFuture().get();
  
          await().untilAtomic(cancelled, is(true));
          assertThat(list).isEmpty();
          assertThat(cancelled).isTrue();
      }
test_case.class_method_signature: CancelStageFactoryTest.create()
test_case.full_signature: @Test public void create()
test_case.identifier: create
test_case.invocations: create doOnCancel doOnNext observeOn interval io set apply get toCompletableFuture untilAtomic await is isEmpty assertThat isTrue assertThat
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void create()
focal_class.file: implementation/src/main/java/io/smallrye/reactive/streams/stages/CancelStageFactory.java
focal_class.identifier: CancelStageFactory
focal_class.interfaces: implements TerminalStageFactory<Stage.Cancel>
focal_class.class_method_signature: CancelStageFactory.create(Engine engine, Stage.Cancel stage)
focal_class.full_signature: @Override public TerminalStage<I, O> create(Engine engine, Stage.Cancel stage)
focal_class.identifier: create
focal_class.modifiers: @Override public
focal_class.parameters: (Engine engine, Stage.Cancel stage)
focal_class.return: TerminalStage<I, O>
focal_class.signature: TerminalStage<I, O> create(Engine engine, Stage.Cancel stage)
focal_method.body: @Override
      public <I, O> TerminalStage<I, O> create(Engine engine, Stage.Cancel stage) {
          Objects.requireNonNull(stage);
          return (Flowable<I> flow) -> {
              flow.subscribe(new Subscriber<I>() {
  
                  @Override
                  public void onSubscribe(Subscription s) {
                      s.cancel();
                  }
  
                  @Override
                  public void onNext(I in) {
                      // Do nothing.
                  }
  
                  @Override
                  public void onError(Throwable t) {
                      // Do nothing.
                  }
  
                  @Override
                  public void onComplete() {
                      // Do nothing.
                  }
              });
              return CompletableFuture.completedFuture(null);
          };
      }
focal_method.class_method_signature: CancelStageFactory.create(Engine engine, Stage.Cancel stage)
focal_method.full_signature: @Override public TerminalStage<I, O> create(Engine engine, Stage.Cancel stage)
focal_method.identifier: create
focal_method.invocations: requireNonNull subscribe cancel completedFuture
focal_method.modifiers: @Override public
focal_method.parameters: (Engine engine, Stage.Cancel stage)
focal_method.return: TerminalStage<I, O>
focal_method.signature: TerminalStage<I, O> create(Engine engine, Stage.Cancel stage)