@Test void testUnsubscribe() { final String[] result = new String[1]; final boolean[] closed = new boolean[1]; final Processor.Iso<String> proc = create(); final Disposable disp = proc.subscribe(new Subscriber<String>() { @Override public void submit(final String item) { result[0] = item; } @Override public void onClose() { closed[0] = true; } }); proc.submit("Hello"); disp.dispose(); proc.submit("World"); assertEquals("Hello", result[0]); assertFalse(closed[0]); }
@Override public void submit(final V value) { if (this.subscriberCount.get() >= 0) { boolean handled = false; for (Iterator<RegistrationImpl<V>> it = this.registrations.iterator(); it.hasNext(); ) { final RegistrationImpl<V> reg = it.next(); try { handled = true; reg.submit(value); } catch (final Exception t) { it.remove(); this.subscriberCount.getAndDecrement(); reg.subscriber.onError(t); } } if (!handled) { final @Nullable Subscriber<V> fallback = this.fallbackHandler; if (fallback != null) { fallback.submit(value); } } } }
ProcessorImpl extends AbstractProcessor<V, RegistrationImpl<V>> implements Processor.Iso<V> { @Override public void submit(final V value) { if (this.subscriberCount.get() >= 0) { boolean handled = false; for (Iterator<RegistrationImpl<V>> it = this.registrations.iterator(); it.hasNext(); ) { final RegistrationImpl<V> reg = it.next(); try { handled = true; reg.submit(value); } catch (final Exception t) { it.remove(); this.subscriberCount.getAndDecrement(); reg.subscriber.onError(t); } } if (!handled) { final @Nullable Subscriber<V> fallback = this.fallbackHandler; if (fallback != null) { fallback.submit(value); } } } } }
ProcessorImpl extends AbstractProcessor<V, RegistrationImpl<V>> implements Processor.Iso<V> { @Override public void submit(final V value) { if (this.subscriberCount.get() >= 0) { boolean handled = false; for (Iterator<RegistrationImpl<V>> it = this.registrations.iterator(); it.hasNext(); ) { final RegistrationImpl<V> reg = it.next(); try { handled = true; reg.submit(value); } catch (final Exception t) { it.remove(); this.subscriberCount.getAndDecrement(); reg.subscriber.onError(t); } } if (!handled) { final @Nullable Subscriber<V> fallback = this.fallbackHandler; if (fallback != null) { fallback.submit(value); } } } } ProcessorImpl(final Executor exec); }
ProcessorImpl extends AbstractProcessor<V, RegistrationImpl<V>> implements Processor.Iso<V> { @Override public void submit(final V value) { if (this.subscriberCount.get() >= 0) { boolean handled = false; for (Iterator<RegistrationImpl<V>> it = this.registrations.iterator(); it.hasNext(); ) { final RegistrationImpl<V> reg = it.next(); try { handled = true; reg.submit(value); } catch (final Exception t) { it.remove(); this.subscriberCount.getAndDecrement(); reg.subscriber.onError(t); } } if (!handled) { final @Nullable Subscriber<V> fallback = this.fallbackHandler; if (fallback != null) { fallback.submit(value); } } } } ProcessorImpl(final Executor exec); @Override void submit(final V value); }
ProcessorImpl extends AbstractProcessor<V, RegistrationImpl<V>> implements Processor.Iso<V> { @Override public void submit(final V value) { if (this.subscriberCount.get() >= 0) { boolean handled = false; for (Iterator<RegistrationImpl<V>> it = this.registrations.iterator(); it.hasNext(); ) { final RegistrationImpl<V> reg = it.next(); try { handled = true; reg.submit(value); } catch (final Exception t) { it.remove(); this.subscriberCount.getAndDecrement(); reg.subscriber.onError(t); } } if (!handled) { final @Nullable Subscriber<V> fallback = this.fallbackHandler; if (fallback != null) { fallback.submit(value); } } } } ProcessorImpl(final Executor exec); @Override void submit(final V value);  }