@Test(timeout = 60000) public void testSubscribe() throws Exception { ListenerWithSub l = holderRule.addAListener(); holderRule.assertListenerAdded(l.listener); l.subscription.unsubscribe(); assertThat("Listener not removed on unsubscribe.", holderRule.getHolder().getAllListeners(), is(empty())); }
@Override public Subscription subscribe(final T listener) { final CompositeSubscription cs = new CompositeSubscription(); ListenerHolder.configureRemoval(cs, listener, listeners); final ListenerHolder<T> holder = new ListenerHolder<>(listener, cs); listeners.add(holder); return cs; }
ListenersHolder implements EventSource<T>, EventPublisher { @Override public Subscription subscribe(final T listener) { final CompositeSubscription cs = new CompositeSubscription(); ListenerHolder.configureRemoval(cs, listener, listeners); final ListenerHolder<T> holder = new ListenerHolder<>(listener, cs); listeners.add(holder); return cs; } }
ListenersHolder implements EventSource<T>, EventPublisher { @Override public Subscription subscribe(final T listener) { final CompositeSubscription cs = new CompositeSubscription(); ListenerHolder.configureRemoval(cs, listener, listeners); final ListenerHolder<T> holder = new ListenerHolder<>(listener, cs); listeners.add(holder); return cs; } ListenersHolder();  ListenersHolder(ListenersHolder<T> toCopy); }
ListenersHolder implements EventSource<T>, EventPublisher { @Override public Subscription subscribe(final T listener) { final CompositeSubscription cs = new CompositeSubscription(); ListenerHolder.configureRemoval(cs, listener, listeners); final ListenerHolder<T> holder = new ListenerHolder<>(listener, cs); listeners.add(holder); return cs; } ListenersHolder();  ListenersHolder(ListenersHolder<T> toCopy); @Override Subscription subscribe(final T listener); @Override boolean publishingEnabled(); void dispose(); void invokeListeners(Action1<T> invocationAction); void invokeListeners(Action3<T, Long, TimeUnit> invocationAction, long duration, TimeUnit timeUnit); void invokeListeners(Action4<T, Long, TimeUnit, Throwable> invocationAction, long duration,
                                TimeUnit timeUnit, Throwable throwable); void invokeListeners(Action4<T, Long, TimeUnit, A> invocationAction, long duration,
                                    TimeUnit timeUnit, A arg); void invokeListeners(Action5<T, Long, TimeUnit, Throwable, A> invocationAction, long duration,
                                    TimeUnit timeUnit, Throwable throwable, A arg); void invokeListeners(Action2<T, A> invocationAction, A arg); void invokeListeners(Action3<T, Throwable, A> invocationAction, Throwable throwable, A arg); ListenersHolder<T> copy(); void subscribeAllTo(EventSource<T> lazySource); }
ListenersHolder implements EventSource<T>, EventPublisher { @Override public Subscription subscribe(final T listener) { final CompositeSubscription cs = new CompositeSubscription(); ListenerHolder.configureRemoval(cs, listener, listeners); final ListenerHolder<T> holder = new ListenerHolder<>(listener, cs); listeners.add(holder); return cs; } ListenersHolder();  ListenersHolder(ListenersHolder<T> toCopy); @Override Subscription subscribe(final T listener); @Override boolean publishingEnabled(); void dispose(); void invokeListeners(Action1<T> invocationAction); void invokeListeners(Action3<T, Long, TimeUnit> invocationAction, long duration, TimeUnit timeUnit); void invokeListeners(Action4<T, Long, TimeUnit, Throwable> invocationAction, long duration,
                                TimeUnit timeUnit, Throwable throwable); void invokeListeners(Action4<T, Long, TimeUnit, A> invocationAction, long duration,
                                    TimeUnit timeUnit, A arg); void invokeListeners(Action5<T, Long, TimeUnit, Throwable, A> invocationAction, long duration,
                                    TimeUnit timeUnit, Throwable throwable, A arg); void invokeListeners(Action2<T, A> invocationAction, A arg); void invokeListeners(Action3<T, Throwable, A> invocationAction, Throwable throwable, A arg); ListenersHolder<T> copy(); void subscribeAllTo(EventSource<T> lazySource);  }