### Question:
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);  }### Answer:
@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); }
@Test public void testRequestsd() { Multi<Integer> multi = Multi.createFrom().range(0, 1000) .broadcast().toAtLeast(2); AssertSubscriber<Integer> subscriber1 = new AssertSubscriber<>(Long.MAX_VALUE); AssertSubscriber<Integer> subscriber2 = new AssertSubscriber<>(100); multi.subscribe().withSubscriber(subscriber1); multi.subscribe().withSubscriber(subscriber2); await().until(() -> subscriber1.items().size() == 100 && subscriber2.items().size() == 100); subscriber2.request(1000); subscriber1.await().assertCompletedSuccessfully(); assertThat(subscriber1.items()).hasSize(1000); subscriber2.await().assertCompletedSuccessfully(); assertThat(subscriber2.items()).hasSize(1000); }