@Test public void testPublishAtLeast() { AssertSubscriber<Integer> s1 = AssertSubscriber.create(10); AssertSubscriber<Integer> s2 = AssertSubscriber.create(10); Multi<Integer> multi = Multi.createFrom().range(1, 5).broadcast().toAtLeast(2); multi.subscribe(s1); s1.assertNotTerminated() .assertHasNotReceivedAnyItem(); multi.subscribe(s2); s1.assertCompletedSuccessfully() .assertReceived(1, 2, 3, 4); s2.assertCompletedSuccessfully() .assertReceived(1, 2, 3, 4); }
public Multi<T> toAtLeast(int numberOfSubscribers) { positive(numberOfSubscribers, "numberOfSubscribers"); return Infrastructure.onMultiCreation( MultiBroadcaster.publish(upstream, numberOfSubscribers, cancelWhenNoOneIsListening, delayAfterLastDeparture)); }
MultiBroadcast { public Multi<T> toAtLeast(int numberOfSubscribers) { positive(numberOfSubscribers, "numberOfSubscribers"); return Infrastructure.onMultiCreation( MultiBroadcaster.publish(upstream, numberOfSubscribers, cancelWhenNoOneIsListening, delayAfterLastDeparture)); } }
MultiBroadcast { public Multi<T> toAtLeast(int numberOfSubscribers) { positive(numberOfSubscribers, "numberOfSubscribers"); return Infrastructure.onMultiCreation( MultiBroadcaster.publish(upstream, numberOfSubscribers, cancelWhenNoOneIsListening, delayAfterLastDeparture)); } MultiBroadcast(Multi<T> upstream); }
MultiBroadcast { public Multi<T> toAtLeast(int numberOfSubscribers) { positive(numberOfSubscribers, "numberOfSubscribers"); return Infrastructure.onMultiCreation( MultiBroadcaster.publish(upstream, numberOfSubscribers, 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> toAtLeast(int numberOfSubscribers) { positive(numberOfSubscribers, "numberOfSubscribers"); return Infrastructure.onMultiCreation( MultiBroadcaster.publish(upstream, numberOfSubscribers, cancelWhenNoOneIsListening, delayAfterLastDeparture)); } MultiBroadcast(Multi<T> upstream); Multi<T> toAllSubscribers(); Multi<T> toAtLeast(int numberOfSubscribers); MultiBroadcast<T> withCancellationAfterLastSubscriberDeparture(); MultiBroadcast<T> withCancellationAfterLastSubscriberDeparture(Duration delay);  }