@Test public void testPublishToAllSubscribersWithFailure() { MultiEmitterProcessor<Integer> processor = MultiEmitterProcessor.create(); Multi<Integer> multi = processor.toMulti().broadcast().toAllSubscribers(); AssertSubscriber<Integer> s1 = multi.subscribe().withSubscriber(AssertSubscriber.create(10)); s1.assertHasNotReceivedAnyItem().assertNotTerminated(); processor.emit(1).emit(2).emit(3); s1.assertReceived(1, 2, 3).assertNotTerminated(); AssertSubscriber<Integer> s2 = multi.subscribe().withSubscriber(AssertSubscriber.create(1)); s2.assertHasNotReceivedAnyItem().assertNotTerminated(); processor.emit(4).emit(5); s1.assertReceived(1, 2, 3, 4).assertNotTerminated(); s2.assertReceived(4).assertNotTerminated(); processor.fail(new IOException("boom")); s1.assertHasFailedWith(IOException.class, "boom"); s2.assertHasFailedWith(IOException.class, "boom"); }
public Multi<T> toAllSubscribers() { return Infrastructure.onMultiCreation( MultiBroadcaster.publish(upstream, 0, cancelWhenNoOneIsListening, delayAfterLastDeparture)); }
MultiBroadcast { public Multi<T> toAllSubscribers() { return Infrastructure.onMultiCreation( MultiBroadcaster.publish(upstream, 0, cancelWhenNoOneIsListening, delayAfterLastDeparture)); } }
MultiBroadcast { public Multi<T> toAllSubscribers() { return Infrastructure.onMultiCreation( MultiBroadcaster.publish(upstream, 0, cancelWhenNoOneIsListening, delayAfterLastDeparture)); } MultiBroadcast(Multi<T> upstream); }
MultiBroadcast { public Multi<T> toAllSubscribers() { return Infrastructure.onMultiCreation( MultiBroadcaster.publish(upstream, 0, cancelWhenNoOneIsListening, delayAfterLastDeparture)); } MultiBroadcast(Multi<T> upstream); Multi<T> toAllSubscribers(); Multi<T> toAtLeast(int numberOfSubscribers); MultiBroadcast<T> withCancellationAfterLastSubscriberDeparture(); MultiBroadcast<T> withCancellationAfterLastSubscriberDeparture(Duration delay); }
MultiBroadcast { public Multi<T> toAllSubscribers() { return Infrastructure.onMultiCreation( MultiBroadcaster.publish(upstream, 0, cancelWhenNoOneIsListening, delayAfterLastDeparture)); } MultiBroadcast(Multi<T> upstream); Multi<T> toAllSubscribers(); Multi<T> toAtLeast(int numberOfSubscribers); MultiBroadcast<T> withCancellationAfterLastSubscriberDeparture(); MultiBroadcast<T> withCancellationAfterLastSubscriberDeparture(Duration delay);  }