/*
* Copyright 2016 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package io.netty.util.concurrent;
import io.netty.util.internal.
ObjectUtil;
/**
* <p>A promise combiner monitors the outcome of a number of discrete futures, then notifies a final, aggregate promise
* when all of the combined futures are finished. The aggregate promise will succeed if and only if all of the combined
* futures succeed. If any of the combined futures fail, the aggregate promise will fail. The cause failure for the
* aggregate promise will be the failure for one of the failed combined futures; if more than one of the combined
* futures fails, exactly which cause of failure will be assigned to the aggregate promise is undefined.</p>
*
* <p>Callers may populate a promise combiner with any number of futures to be combined via the
* {@link PromiseCombiner#add(Future)} and {@link PromiseCombiner#addAll(Future[])} methods. When all futures to be
* combined have been added, callers must provide an aggregate promise to be notified when all combined promises have
* finished via the {@link PromiseCombiner#finish(Promise)} method.</p>
*/
public final class
PromiseCombiner {
private int
expectedCount;
private int
doneCount;
private boolean
doneAdding;
private
Promise<
Void>
aggregatePromise;
private
Throwable cause;
private final
GenericFutureListener<
Future<?>>
listener = new
GenericFutureListener<
Future<?>>() {
@
Override
public void
operationComplete(
Future<?>
future) throws
Exception {
++
doneCount;
if (!
future.
isSuccess() &&
cause == null) {
cause =
future.
cause();
}
if (
doneCount ==
expectedCount &&
doneAdding) {
tryPromise();
}
}
};
/**
* Adds a new promise to be combined. New promises may be added until an aggregate promise is added via the
* {@link PromiseCombiner#finish(Promise)} method.
*
* @param promise the promise to add to this promise combiner
*
* @deprecated Replaced by {@link PromiseCombiner#add(Future)}.
*/
@
Deprecated
public void
add(
Promise promise) {
add((
Future)
promise);
}
/**
* Adds a new future to be combined. New futures may be added until an aggregate promise is added via the
* {@link PromiseCombiner#finish(Promise)} method.
*
* @param future the future to add to this promise combiner
*/
@
SuppressWarnings({ "unchecked", "rawtypes" })
public void
add(
Future future) {
checkAddAllowed();
++
expectedCount;
future.
addListener(
listener);
}
/**
* Adds new promises to be combined. New promises may be added until an aggregate promise is added via the
* {@link PromiseCombiner#finish(Promise)} method.
*
* @param promises the promises to add to this promise combiner
*
* @deprecated Replaced by {@link PromiseCombiner#addAll(Future[])}
*/
@
Deprecated
public void
addAll(
Promise...
promises) {
addAll((
Future[])
promises);
}
/**
* Adds new futures to be combined. New futures may be added until an aggregate promise is added via the
* {@link PromiseCombiner#finish(Promise)} method.
*
* @param futures the futures to add to this promise combiner
*/
@
SuppressWarnings({ "unchecked", "rawtypes" })
public void
addAll(
Future...
futures) {
for (
Future future :
futures) {
this.
add(
future);
}
}
/**
* <p>Sets the promise to be notified when all combined futures have finished. If all combined futures succeed,
* then the aggregate promise will succeed. If one or more combined futures fails, then the aggregate promise will
* fail with the cause of one of the failed futures. If more than one combined future fails, then exactly which
* failure will be assigned to the aggregate promise is undefined.</p>
*
* <p>After this method is called, no more futures may be added via the {@link PromiseCombiner#add(Future)} or
* {@link PromiseCombiner#addAll(Future[])} methods.</p>
*
* @param aggregatePromise the promise to notify when all combined futures have finished
*/
public void
finish(
Promise<
Void>
aggregatePromise) {
if (
doneAdding) {
throw new
IllegalStateException("Already finished");
}
doneAdding = true;
this.
aggregatePromise =
ObjectUtil.
checkNotNull(
aggregatePromise, "aggregatePromise");
if (
doneCount ==
expectedCount) {
tryPromise();
}
}
private boolean
tryPromise() {
return (
cause == null) ?
aggregatePromise.
trySuccess(null) :
aggregatePromise.
tryFailure(
cause);
}
private void
checkAddAllowed() {
if (
doneAdding) {
throw new
IllegalStateException("Adding promises is not allowed after finished adding");
}
}
}