@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);
}