@Test void testFallbackHandler() { final Processor.Iso<String> handler = create(); final String[] values = new String[2]; handler.setFallbackHandler(val -> values[0] = val); handler.submit("Hello"); assertEquals("Hello", values[0]); final Disposable disp = handler.subscribe(val -> values[1] = val); handler.submit("World"); assertEquals("Hello", values[0]); assertEquals("World", values[1]); disp.dispose(); handler.submit("Goodbye"); assertEquals("Goodbye", values[0]); assertEquals("World", values[1]); }
@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);  }