test_class.file: rxjava/src/test/java/io/reactivex/internal/operators/flowable/FlowableReplayTest.java
test_class.identifier: FlowableReplayTest
test_case.body: @SuppressWarnings("unchecked")
      @Test
      public void testIssue2191_UnsubscribeSource() throws Exception {
          // setup mocks
          Consumer<Integer> sourceNext = mock(Consumer.class);
          Action sourceCompleted = mock(Action.class);
          Action sourceUnsubscribed = mock(Action.class);
          Subscriber<Integer> spiedSubscriberBeforeConnect = TestHelper.mockSubscriber();
          Subscriber<Integer> spiedSubscriberAfterConnect = TestHelper.mockSubscriber();
  
          // Flowable under test
          Flowable<Integer> source = Flowable.just(1,2);
  
          ConnectableFlowable<Integer> replay = source
                  .doOnNext(sourceNext)
                  .doOnCancel(sourceUnsubscribed)
                  .doOnComplete(sourceCompleted)
                  .replay();
  
          replay.subscribe(spiedSubscriberBeforeConnect);
          replay.subscribe(spiedSubscriberBeforeConnect);
          replay.connect();
          replay.subscribe(spiedSubscriberAfterConnect);
          replay.subscribe(spiedSubscriberAfterConnect);
  
          verify(spiedSubscriberBeforeConnect, times(2)).onSubscribe((Subscription)any());
          verify(spiedSubscriberAfterConnect, times(2)).onSubscribe((Subscription)any());
  
          // verify interactions
          verify(sourceNext, times(1)).accept(1);
          verify(sourceNext, times(1)).accept(2);
          verify(sourceCompleted, times(1)).run();
          verifyObserverMock(spiedSubscriberBeforeConnect, 2, 4);
          verifyObserverMock(spiedSubscriberAfterConnect, 2, 4);
  
          verify(sourceUnsubscribed, never()).run();
  
          verifyNoMoreInteractions(sourceNext);
          verifyNoMoreInteractions(sourceCompleted);
          verifyNoMoreInteractions(sourceUnsubscribed);
          verifyNoMoreInteractions(spiedSubscriberBeforeConnect);
          verifyNoMoreInteractions(spiedSubscriberAfterConnect);
  
      }
test_case.class_method_signature: FlowableReplayTest.testIssue2191_UnsubscribeSource()
test_case.full_signature: @SuppressWarnings("unchecked") @Test public void testIssue2191_UnsubscribeSource()
test_case.identifier: testIssue2191_UnsubscribeSource
test_case.invocations: mock mock mock mockSubscriber mockSubscriber just replay doOnComplete doOnCancel doOnNext subscribe subscribe connect subscribe subscribe onSubscribe verify times any onSubscribe verify times any accept verify times accept verify times run verify times verifyObserverMock verifyObserverMock run verify never verifyNoMoreInteractions verifyNoMoreInteractions verifyNoMoreInteractions verifyNoMoreInteractions verifyNoMoreInteractions
test_case.modifiers: @SuppressWarnings("unchecked") @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testIssue2191_UnsubscribeSource()
focal_class.declarator: source
focal_class.modifier: final
focal_class.original_string: final Flowable<T> source;
focal_class.type: Flowable<T>
focal_class.var_name: source
focal_class.declarator: current
focal_class.modifier: final
focal_class.original_string: final AtomicReference<ReplaySubscriber<T>> current;
focal_class.type: AtomicReference<ReplaySubscriber<T>>
focal_class.var_name: current
focal_class.declarator: bufferFactory
focal_class.modifier: final
focal_class.original_string: final Callable<? extends ReplayBuffer<T>> bufferFactory;
focal_class.type: Callable<? extends ReplayBuffer<T>>
focal_class.var_name: bufferFactory
focal_class.declarator: onSubscribe
focal_class.modifier: final
focal_class.original_string: final Publisher<T> onSubscribe;
focal_class.type: Publisher<T>
focal_class.var_name: onSubscribe
focal_class.declarator: DEFAULT_UNBOUNDED_FACTORY = new DefaultUnboundedFactory()
focal_class.modifier: @SuppressWarnings("rawtypes")
      static final
focal_class.original_string: @SuppressWarnings("rawtypes")
      static final Callable DEFAULT_UNBOUNDED_FACTORY = new DefaultUnboundedFactory();
focal_class.type: Callable
focal_class.var_name: DEFAULT_UNBOUNDED_FACTORY
focal_class.file: rxjava/src/main/java/io/reactivex/internal/operators/flowable/FlowableReplay.java
focal_class.identifier: FlowableReplay
focal_class.interfaces: implements HasUpstreamPublisher<T>, Disposable
focal_class.class_method_signature: FlowableReplay.multicastSelector(
              final Callable<? extends ConnectableFlowable<U>> connectableFactory,
              final Function<? super Flowable<U>, ? extends Publisher<R>> selector)
focal_class.full_signature: public static Flowable<R> multicastSelector(
              final Callable<? extends ConnectableFlowable<U>> connectableFactory,
              final Function<? super Flowable<U>, ? extends Publisher<R>> selector)
focal_class.identifier: multicastSelector
focal_class.modifiers: public static
focal_class.parameters: (
              final Callable<? extends ConnectableFlowable<U>> connectableFactory,
              final Function<? super Flowable<U>, ? extends Publisher<R>> selector)
focal_class.return: Flowable<R>
focal_class.signature: Flowable<R> multicastSelector(
              final Callable<? extends ConnectableFlowable<U>> connectableFactory,
              final Function<? super Flowable<U>, ? extends Publisher<R>> selector)
focal_class.class_method_signature: FlowableReplay.observeOn(final ConnectableFlowable<T> cf, final Scheduler scheduler)
focal_class.full_signature: public static ConnectableFlowable<T> observeOn(final ConnectableFlowable<T> cf, final Scheduler scheduler)
focal_class.identifier: observeOn
focal_class.modifiers: public static
focal_class.parameters: (final ConnectableFlowable<T> cf, final Scheduler scheduler)
focal_class.return: ConnectableFlowable<T>
focal_class.signature: ConnectableFlowable<T> observeOn(final ConnectableFlowable<T> cf, final Scheduler scheduler)
focal_class.class_method_signature: FlowableReplay.createFrom(Flowable<? extends T> source)
focal_class.full_signature: @SuppressWarnings("unchecked") public static ConnectableFlowable<T> createFrom(Flowable<? extends T> source)
focal_class.identifier: createFrom
focal_class.modifiers: @SuppressWarnings("unchecked") public static
focal_class.parameters: (Flowable<? extends T> source)
focal_class.return: ConnectableFlowable<T>
focal_class.signature: ConnectableFlowable<T> createFrom(Flowable<? extends T> source)
focal_class.class_method_signature: FlowableReplay.create(Flowable<T> source,
              final int bufferSize)
focal_class.full_signature: public static ConnectableFlowable<T> create(Flowable<T> source,
              final int bufferSize)
focal_class.identifier: create
focal_class.modifiers: public static
focal_class.parameters: (Flowable<T> source,
              final int bufferSize)
focal_class.return: ConnectableFlowable<T>
focal_class.signature: ConnectableFlowable<T> create(Flowable<T> source,
              final int bufferSize)
focal_class.class_method_signature: FlowableReplay.create(Flowable<T> source,
              long maxAge, TimeUnit unit, Scheduler scheduler)
focal_class.full_signature: public static ConnectableFlowable<T> create(Flowable<T> source,
              long maxAge, TimeUnit unit, Scheduler scheduler)
focal_class.identifier: create
focal_class.modifiers: public static
focal_class.parameters: (Flowable<T> source,
              long maxAge, TimeUnit unit, Scheduler scheduler)
focal_class.return: ConnectableFlowable<T>
focal_class.signature: ConnectableFlowable<T> create(Flowable<T> source,
              long maxAge, TimeUnit unit, Scheduler scheduler)
focal_class.class_method_signature: FlowableReplay.create(Flowable<T> source,
              final long maxAge, final TimeUnit unit, final Scheduler scheduler, final int bufferSize)
focal_class.full_signature: public static ConnectableFlowable<T> create(Flowable<T> source,
              final long maxAge, final TimeUnit unit, final Scheduler scheduler, final int bufferSize)
focal_class.identifier: create
focal_class.modifiers: public static
focal_class.parameters: (Flowable<T> source,
              final long maxAge, final TimeUnit unit, final Scheduler scheduler, final int bufferSize)
focal_class.return: ConnectableFlowable<T>
focal_class.signature: ConnectableFlowable<T> create(Flowable<T> source,
              final long maxAge, final TimeUnit unit, final Scheduler scheduler, final int bufferSize)
focal_class.class_method_signature: FlowableReplay.create(Flowable<T> source,
              final Callable<? extends ReplayBuffer<T>> bufferFactory)
focal_class.full_signature: static ConnectableFlowable<T> create(Flowable<T> source,
              final Callable<? extends ReplayBuffer<T>> bufferFactory)
focal_class.identifier: create
focal_class.modifiers: static
focal_class.parameters: (Flowable<T> source,
              final Callable<? extends ReplayBuffer<T>> bufferFactory)
focal_class.return: ConnectableFlowable<T>
focal_class.signature: ConnectableFlowable<T> create(Flowable<T> source,
              final Callable<? extends ReplayBuffer<T>> bufferFactory)
focal_class.class_method_signature: FlowableReplay.FlowableReplay(Publisher<T> onSubscribe, Flowable<T> source,
              final AtomicReference<ReplaySubscriber<T>> current,
              final Callable<? extends ReplayBuffer<T>> bufferFactory)
focal_class.full_signature: private  FlowableReplay(Publisher<T> onSubscribe, Flowable<T> source,
              final AtomicReference<ReplaySubscriber<T>> current,
              final Callable<? extends ReplayBuffer<T>> bufferFactory)
focal_class.identifier: FlowableReplay
focal_class.modifiers: private
focal_class.parameters: (Publisher<T> onSubscribe, Flowable<T> source,
              final AtomicReference<ReplaySubscriber<T>> current,
              final Callable<? extends ReplayBuffer<T>> bufferFactory)
focal_class.signature:  FlowableReplay(Publisher<T> onSubscribe, Flowable<T> source,
              final AtomicReference<ReplaySubscriber<T>> current,
              final Callable<? extends ReplayBuffer<T>> bufferFactory)
focal_class.class_method_signature: FlowableReplay.source()
focal_class.full_signature: @Override public Publisher<T> source()
focal_class.identifier: source
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: Publisher<T>
focal_class.signature: Publisher<T> source()
focal_class.class_method_signature: FlowableReplay.subscribeActual(Subscriber<? super T> s)
focal_class.full_signature: @Override protected void subscribeActual(Subscriber<? super T> s)
focal_class.identifier: subscribeActual
focal_class.modifiers: @Override protected
focal_class.parameters: (Subscriber<? super T> s)
focal_class.return: void
focal_class.signature: void subscribeActual(Subscriber<? super T> s)
focal_class.class_method_signature: FlowableReplay.dispose()
focal_class.full_signature: @Override public void dispose()
focal_class.identifier: dispose
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void dispose()
focal_class.class_method_signature: FlowableReplay.isDisposed()
focal_class.full_signature: @Override public boolean isDisposed()
focal_class.identifier: isDisposed
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: boolean
focal_class.signature: boolean isDisposed()
focal_class.class_method_signature: FlowableReplay.connect(Consumer<? super Disposable> connection)
focal_class.full_signature: @Override public void connect(Consumer<? super Disposable> connection)
focal_class.identifier: connect
focal_class.modifiers: @Override public
focal_class.parameters: (Consumer<? super Disposable> connection)
focal_class.return: void
focal_class.signature: void connect(Consumer<? super Disposable> connection)
focal_class.superclass: extends ConnectableFlowable<T>
focal_method.body: @Override
      public void connect(Consumer<? super Disposable> connection) {
          boolean doConnect;
          ReplaySubscriber<T> ps;
          // we loop because concurrent connect/disconnect and termination may change the state
          for (;;) {
              // retrieve the current subscriber-to-source instance
              ps = current.get();
              // if there is none yet or the current was disposed
              if (ps == null || ps.isDisposed()) {
  
                  ReplayBuffer<T> buf;
  
                  try {
                      buf = bufferFactory.call();
                  } catch (Throwable ex) {
                      Exceptions.throwIfFatal(ex);
                      throw ExceptionHelper.wrapOrThrow(ex);
                  }
  
                  // create a new subscriber-to-source
                  ReplaySubscriber<T> u = new ReplaySubscriber<T>(buf);
                  // try setting it as the current subscriber-to-source
                  if (!current.compareAndSet(ps, u)) {
                      // did not work, perhaps a new subscriber arrived
                      // and created a new subscriber-to-source as well, retry
                      continue;
                  }
                  ps = u;
              }
              // if connect() was called concurrently, only one of them should actually
              // connect to the source
              doConnect = !ps.shouldConnect.get() && ps.shouldConnect.compareAndSet(false, true);
              break; // NOPMD
          }
          /*
           * Notify the callback that we have a (new) connection which it can dispose
           * but since ps is unique to a connection, multiple calls to connect() will return the
           * same Subscription and even if there was a connect-disconnect-connect pair, the older
           * references won't disconnect the newer connection.
           * Synchronous source consumers have the opportunity to disconnect via dispose on the
           * Disposable as unsafeSubscribe may never return in its own.
           *
           * Note however, that asynchronously disconnecting a running source might leave
           * child-subscribers without any terminal event; ReplaySubject does not have this
           * issue because the cancellation was always triggered by the child-subscribers
           * themselves.
           */
          try {
              connection.accept(ps);
          } catch (Throwable ex) {
              if (doConnect) {
                  ps.shouldConnect.compareAndSet(true, false);
              }
              Exceptions.throwIfFatal(ex);
              throw ExceptionHelper.wrapOrThrow(ex);
          }
          if (doConnect) {
              source.subscribe(ps);
          }
      }
focal_method.class_method_signature: FlowableReplay.connect(Consumer<? super Disposable> connection)
focal_method.full_signature: @Override public void connect(Consumer<? super Disposable> connection)
focal_method.identifier: connect
focal_method.invocations: get isDisposed call throwIfFatal wrapOrThrow compareAndSet get compareAndSet accept compareAndSet throwIfFatal wrapOrThrow subscribe
focal_method.modifiers: @Override public
focal_method.parameters: (Consumer<? super Disposable> connection)
focal_method.return: void
focal_method.signature: void connect(Consumer<? super Disposable> connection)