Make sure the duration can be `null`.
Refactor the method to looks less complex
@@ -11,17 +11,15 @@ public class MultiBroadcaster {
Duration delayAfterLastDeparture) {
if (numberOfSubscribers > 0) {
- if (cancelWhenNoOneIsListening) {
- if (delayAfterLastDeparture != null) {
- return MultiPublishOp.create(upstream)
- .referenceCount(numberOfSubscribers, delayAfterLastDeparture);
+ return createPublishWithSubscribersThreshold(upstream, numberOfSubscribers, cancelWhenNoOneIsListening,
+ delayAfterLastDeparture);
} else {
- return MultiPublishOp.create(upstream).referenceCount(numberOfSubscribers, delayAfterLastDeparture);
+ return createPublishImmediate(upstream, cancelWhenNoOneIsListening, delayAfterLastDeparture);
}
- } else {
- return MultiPublishOp.create(upstream).connectAfter(numberOfSubscribers);
}
- } else {
+
+ private static <T> Multi<T> createPublishImmediate(Multi<T> upstream, boolean cancelWhenNoOneIsListening,
+ Duration delayAfterLastDeparture) {
if (cancelWhenNoOneIsListening) {
if (delayAfterLastDeparture != null) {
return MultiPublishOp.create(upstream).referenceCount(1, delayAfterLastDeparture);
@@ -32,6 +30,19 @@ public class MultiBroadcaster {
return MultiPublishOp.create(upstream).connectAfter(1);
}
}
+
+ private static <T> Multi<T> createPublishWithSubscribersThreshold(Multi<T> upstream, int numberOfSubscribers,
+ boolean cancelWhenNoOneIsListening, Duration delayAfterLastDeparture) {
+ if (cancelWhenNoOneIsListening) {
+ if (delayAfterLastDeparture != null) {
+ return MultiPublishOp.create(upstream).referenceCount(numberOfSubscribers, delayAfterLastDeparture);
+ } else {
+ // the duration can be `null`, it will be validated if not `null`.
+ return MultiPublishOp.create(upstream).referenceCount(numberOfSubscribers, null);
+ }
+ } else {
+ return MultiPublishOp.create(upstream).connectAfter(numberOfSubscribers);
+ }
}
private MultiBroadcaster() {
