@Test void testMap() { final String[] items = new String[2]; final Processor.Iso<String> orig = create(); final Publisher<String> mapped = orig.map(x -> "2" + x); orig.subscribe(item -> items[0] = item); mapped.subscribe(item -> items[1] = item); orig.submit("Fast"); assertEquals("Fast", items[0]); assertEquals("2Fast", items[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);  }