下面列出了怎么用com.google.common.util.concurrent.Callables的API类实例代码及写法,或者点击链接到github查看源代码。
@Override
@NonNull
public synchronized ImmutableList<ApkOutputFile> getOutputSplitFiles() {
if (outputFiles == null) {
ImmutableList.Builder<ApkOutputFile> builder = ImmutableList.builder();
for (String split : splits) {
String apkName = getApkName(split);
ApkOutputFile apkOutput = new ApkOutputFile(
OutputFile.OutputType.SPLIT,
ImmutableList.of(FilterDataImpl.build(OutputFile.ABI, apkName)),
Callables.returning(new File(outputDirectory, apkName)));
builder.add(apkOutput);
}
outputFiles = builder.build();
}
return outputFiles;
}
@Test
public void testStaleFiresFutureListener() throws InterruptedException {
Object obj = new Object();
final PluggableJob<Object> job =
new PluggableJob<>("name", Callables.returning(obj), unused -> true);
assertFalse(job.getFuture().isDone());
final boolean[] listenerRun = new boolean[] {false};
job.getFuture().addListener(() -> {
listenerRun[0] = true;
assertTrue(job.getFuture().isCancelled());
}, MoreExecutors.directExecutor());
assertFalse(listenerRun[0]);
job.schedule();
job.join();
assertTrue(listenerRun[0]);
assertEquals("Should be CANCEL", IStatus.CANCEL, job.getResult().getSeverity());
}
/**
* Check that the {@link Repeater} will stop after a time limit.
*
* The repeater is configured to run every 100ms and never stop until the limit is reached.
* This is given as {@link Repeater#limitTimeTo(org.apache.brooklyn.util.time.Duration)} and the execution time
* is then checked to ensure it is between 100% and 400% of the specified value. Due to scheduling
* delays and other factors in a non RTOS system it is expected that the repeater will take much
* longer to exit occasionally.
*
* @see #runRespectsMaximumIterationLimitAndReturnsFalseIfReached()
*/
@Test(groups="Integration")
public void runRespectsTimeLimitAndReturnsFalseIfReached() {
final long LIMIT = 2000l;
Repeater repeater = new Repeater("runRespectsTimeLimitAndReturnsFalseIfReached")
.every(Duration.millis(100))
.until(Callables.returning(false))
.limitTimeTo(LIMIT, TimeUnit.MILLISECONDS);
Stopwatch stopwatch = Stopwatch.createStarted();
boolean result = repeater.run();
stopwatch.stop();
assertFalse(result);
long difference = stopwatch.elapsed(TimeUnit.MILLISECONDS);
assertTrue(difference >= LIMIT, "Difference was: " + difference);
assertTrue(difference < 4 * LIMIT, "Difference was: " + difference);
}
@Test
@SuppressWarnings("unused")
public void testFunctionPollConfigBuilding() throws Exception {
FunctionPollConfig<Integer, Integer> typeFromCallable = FunctionPollConfig.forSensor(SENSOR_INT)
.period(1)
.callable(Callables.returning(1))
.onSuccess(Functions.constant(-1));
FunctionPollConfig<Integer, Integer> typeFromSupplier = FunctionPollConfig.forSensor(SENSOR_INT)
.period(1)
.supplier(Suppliers.ofInstance(1))
.onSuccess(Functions.constant(-1));
FunctionPollConfig<Integer, Integer> usingConstructor = new FunctionPollConfig<Integer, Integer>(SENSOR_INT)
.period(1)
.supplier(Suppliers.ofInstance(1))
.onSuccess(Functions.constant(-1));
FunctionPollConfig<Integer, Integer> usingConstructorWithFailureOrException = new FunctionPollConfig<Integer, Integer>(SENSOR_INT)
.period(1)
.supplier(Suppliers.ofInstance(1))
.onFailureOrException(Functions.<Integer>constant(null));
}
@Test
public void testRepeater() throws Exception {
Task<?> t;
t = Tasks.requiring(Repeater.create().until(Callables.returning(true)).every(Duration.millis(1))).build();
app.getExecutionContext().submit(t);
t.get(Duration.TEN_SECONDS);
t = Tasks.testing(Repeater.create().until(Callables.returning(true)).every(Duration.millis(1))).build();
app.getExecutionContext().submit(t);
Assert.assertEquals(t.get(Duration.TEN_SECONDS), true);
t = Tasks.requiring(Repeater.create().until(Callables.returning(false)).limitIterationsTo(2).every(Duration.millis(1))).build();
app.getExecutionContext().submit(t);
try {
t.get(Duration.TEN_SECONDS);
Assert.fail("Should have failed");
} catch (Exception e) {
// expected
}
t = Tasks.testing(Repeater.create().until(Callables.returning(false)).limitIterationsTo(2).every(Duration.millis(1))).build();
app.getExecutionContext().submit(t);
Assert.assertEquals(t.get(Duration.TEN_SECONDS), false);
}
@Test
public void testRepeaterDescription() throws Exception{
final String description = "task description";
Repeater repeater = Repeater.create(description)
.repeat(Callables.returning(null))
.every(Duration.ONE_MILLISECOND)
.limitIterationsTo(1)
.until(new Callable<Boolean>() {
@Override
public Boolean call() {
TaskInternal<?> current = (TaskInternal<?>)Tasks.current();
assertEquals(current.getBlockingDetails(), description);
return true;
}
});
Task<Boolean> t = Tasks.testing(repeater).build();
app.getExecutionContext().submit(t);
assertTrue(t.get(Duration.TEN_SECONDS));
}
@Test
public void testGetResultOfQueuedTaskBeforeItExecutes() throws Exception {
final CountDownLatch latch = new CountDownLatch(1);
em.submit(MutableMap.of("tag", "category1"), newLatchAwaiter(latch));
BasicTask<Integer> t = new BasicTask<Integer>(Callables.returning(123));
Future<Integer> future = em.submit(MutableMap.of("tag", "category1"), t);
Thread thread = new Thread(new Runnable() {
@Override
public void run() {
try {
Thread.sleep(10);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
latch.countDown();
}});
thread.start();
assertEquals(future.get(), (Integer)123);
}
@Test
public void testTask() throws Exception {
final TestApplication app = TestApplication.Factory.newManagedInstanceForTests();
mgmt = app.getManagementContext();
Task<String> completedTask = app.getExecutionContext().submit("return myval", Callables.returning("myval"));
completedTask.get();
String loggerName = UnwantedStateLoggingMapper.class.getName();
ch.qos.logback.classic.Level logLevel = ch.qos.logback.classic.Level.WARN;
Predicate<ILoggingEvent> filter = EventPredicates.containsMessage("Task object serialization is not supported or recommended");
String serializedForm;
try (LogWatcher watcher = new LogWatcher(loggerName, logLevel, filter)) {
serializedForm = serializer.toString(completedTask);
watcher.assertHasEvent();
}
assertEquals(serializedForm.trim(), "<"+BasicTask.class.getName()+">myval</"+BasicTask.class.getName()+">");
Object deserialized = serializer.fromString(serializedForm);
assertEquals(deserialized, null, "serializedForm="+serializedForm+"; deserialized="+deserialized);
}
@Override
public void init() {
super.init();
addFeed(FunctionFeed.builder()
.entity(this)
.poll(FunctionPollConfig.forSensor(SENSOR_INT)
.period(POLL_PERIOD)
.callable(Callables.returning(1)))
.build());
addFeed(FunctionFeed.builder()
.entity(this)
.poll(FunctionPollConfig.forSensor(SENSOR_STRING)
.period(POLL_PERIOD)
.callable(Callables.returning("OK")))
.build());
}
private void assertActionWithContentChangingInput(final boolean unconditionalExecution)
throws Exception {
// Assert that a simple, non-skyframe-aware action is executed twice
// if its input's content changes between builds.
assertActionExecutions(
new ExecutionCountingActionFactory() {
@Override
public ExecutionCountingAction create(
Artifact input, Artifact output, AtomicInteger executionCounter) {
return unconditionalExecution
? new ExecutionCountingCacheBypassingAction(input, output, executionCounter)
: new ExecutionCountingAction(input, output, executionCounter);
}
},
ChangeArtifact.CHANGE_MTIME_AND_CONTENT,
Callables.<Void>returning(null),
ExpectActionIs.REEXECUTED);
}
private void assertActionWithMtimeChangingInput(final boolean unconditionalExecution)
throws Exception {
// Assert that a simple, non-skyframe-aware action is executed only once
// if its input's mtime changes but its contents stay the same between builds.
assertActionExecutions(
new ExecutionCountingActionFactory() {
@Override
public ExecutionCountingAction create(
Artifact input, Artifact output, AtomicInteger executionCounter) {
return unconditionalExecution
? new ExecutionCountingCacheBypassingAction(input, output, executionCounter)
: new ExecutionCountingAction(input, output, executionCounter);
}
},
ChangeArtifact.CHANGE_MTIME,
Callables.<Void>returning(null),
unconditionalExecution
? ExpectActionIs.REEXECUTED
: ExpectActionIs.DIRTIED_BUT_VERIFIED_CLEAN);
}
private void assertActionWithNonChangingInput(final boolean unconditionalExecution)
throws Exception {
// Assert that a simple, non-skyframe-aware action is executed only once
// if its input does not change at all between builds.
assertActionExecutions(
new ExecutionCountingActionFactory() {
@Override
public ExecutionCountingAction create(
Artifact input, Artifact output, AtomicInteger executionCounter) {
return unconditionalExecution
? new ExecutionCountingCacheBypassingAction(input, output, executionCounter)
: new ExecutionCountingAction(input, output, executionCounter);
}
},
ChangeArtifact.DONT_CHANGE,
Callables.<Void>returning(null),
ExpectActionIs.NOT_DIRTIED);
}
@Test
public void Instrumentable(){
interestingClasses.clear();
interestingClasses.add(Callables.class.getName());
ClassPathHandler.getInstance().changeTargetCPtoTheSameAsEvoSuite();
List<Class> instrumentedClasses = instrumentation.instrumentClasses(interestingClasses,Callables.class.getName());
}
/**
* Calculates the list of output files, coming from the list of input files, mangling the output
* file name.
*/
@Override
public List<ApkOutputFile> getOutputSplitFiles() {
final ImmutableList.Builder<ApkOutputFile> builder = ImmutableList.builder();
forEachInputFile(new SplitFileHandler() {
@Override
public void execute(String split, File file) {
// find the split identification, if null, the split is not requested any longer.
FilterData filterData = null;
for (String density : densitySplits) {
if (split.startsWith(density)) {
filterData = FilterDataImpl.build(
OutputFile.FilterType.DENSITY.toString(), density);
}
}
if (languageSplits.contains(unMangleSplitName(split))) {
filterData = FilterDataImpl.build(
OutputFile.FilterType.LANGUAGE.toString(), unMangleSplitName(split));
}
if (filterData != null) {
builder.add(new ApkOutputFile(
OutputFile.OutputType.SPLIT,
ImmutableList.of(filterData),
Callables.returning(
new File(outputDirectory, getOutputFileNameForSplit(split)))));
}
}
});
return builder.build();
}
@Test
public void testConstructor_nullStalenessCheck() {
try {
new PluggableJob<>("name", Callables.returning((Void) null), null);
fail("Expected NPE");
} catch (NullPointerException ex) {
}
}
@Test
public void testScheduled() throws InterruptedException, ExecutionException {
Object obj = new Object();
PluggableJob<Object> job = new PluggableJob<>("name", Callables.returning(obj));
assertFalse(job.getFuture().isDone());
assertFalse(job.getFuture().isCancelled());
job.schedule();
job.join();
assertTrue(job.getFuture().isDone());
assertFalse(job.getFuture().isCancelled());
assertSame(obj, job.getFuture().get());
assertEquals("Should be OK", IStatus.OK, job.getResult().getSeverity());
}
@Test
public void testStaleCancelsFuture() throws InterruptedException {
Object obj = new Object();
PluggableJob<Object> job = new PluggableJob<>("name", Callables.returning(obj), unused -> true);
job.schedule();
job.join();
assertEquals("Should be CANCEL", IStatus.CANCEL, job.getResult().getSeverity());
assertTrue(job.getFuture().isCancelled());
}
@Test
public void testCompleteness_normal() throws InterruptedException {
Object obj = new Object();
PluggableJob<Object> job = new PluggableJob<>("name", Callables.returning(obj));
assertFalse(job.isComputationComplete());
job.schedule();
job.join();
assertTrue(job.isComputationComplete());
assertFalse(job.getComputationError().isPresent());
assertTrue(job.getComputationResult().isPresent());
assertEquals(obj, job.getComputationResult().get());
assertEquals(obj, job.getComputation().get());
}
@Test
public void testOnSuccess_normal() throws InterruptedException {
Object obj = new Object();
PluggableJob<Object> job = new PluggableJob<>("name", Callables.returning(obj));
final boolean[] listenerRun = new boolean[] {false};
job.onSuccess(MoreExecutors.directExecutor(), () -> listenerRun[0] = true);
assertFalse(listenerRun[0]);
job.schedule();
job.join();
assertTrue(listenerRun[0]);
assertTrue(job.isComputationComplete());
}
@Test
public void testOnSuccess_abandon() throws InterruptedException {
Object obj = new Object();
PluggableJob<Object> job =
new PluggableJob<>("name", Callables.returning(obj), unused -> true);
final boolean[] listenerRun = new boolean[] {false};
job.onSuccess(MoreExecutors.directExecutor(), () -> listenerRun[0] = true);
assertFalse(listenerRun[0]);
job.schedule(); // should be stale and cancelled
job.join();
assertFalse("onSuccess should not have been called", listenerRun[0]);
}
@Test
public void testIsCurrent_abandon() throws InterruptedException {
Object obj = new Object();
PluggableJob<Object> job = new PluggableJob<>("name", Callables.returning(obj));
assertTrue(job.isCurrent());
job.schedule(); // should be stale and cancelled
job.join();
assertTrue(job.isCurrent());
job.abandon();
assertFalse("Abandoned jobs should not be current", job.isCurrent());
}
@Test
public void testIsCurrent_stale() throws InterruptedException {
Object obj = new Object();
final boolean[] isStale = new boolean[] { false };
PluggableJob<Object> job = new PluggableJob<>("name", Callables.returning(obj),
unused -> isStale[0]);
assertTrue(job.isCurrent());
job.schedule(); // should self-cancel
job.join();
assertTrue(job.isCurrent());
isStale[0] = true;
// should now be stale
assertFalse("Stale jobs should not be current", job.isCurrent());
}
@Test
public void runReturnsTrueIfExitConditionIsTrue() {
assertTrue(new Repeater("runReturnsTrueIfExitConditionIsTrue")
.every(Duration.millis(1))
.until(Callables.returning(true))
.run());
}
@Test
public void runRespectsMaximumIterationLimitAndReturnsFalseIfReached() {
final AtomicInteger iterations = new AtomicInteger();
assertFalse(new Repeater("runRespectsMaximumIterationLimitAndReturnsFalseIfReached")
.repeat(new Runnable() {
@Override public void run() { iterations.incrementAndGet(); }
})
.every(Duration.millis(1))
.until(Callables.returning(false))
.limitIterationsTo(5)
.run());
assertEquals(iterations.get(), 5);
}
@Test(expectedExceptions = { NullPointerException.class })
public void runFailsIfEveryWasNotSet() {
new Repeater("runFailsIfEveryWasNotSet")
.until(Callables.returning(true))
.run();
fail("Expected exception was not thrown");
}
@Test
public void testGetSensorValueOfTypeCompletedTask() throws Exception {
Task<String> task = entity.getExecutionContext().submit("returning myval", Callables.returning("myval"));
task.get();
entity.sensors().set(Sensors.newSensor(Task.class, "myTask"), task);
doGetSensorTest("myTask", String.class, "\"myval\"");
}
private <T> Task<T> gone() {
Task<T> t = Tasks.<T>builder().dynamic(false).displayName(displayName+" (placeholder for "+id+")")
.description("Details of the original task have been forgotten.")
.body(Callables.returning((T)null)).build();
// don't really want anyone executing the "gone" task...
// also if we are GC'ing tasks then cancelled may help with cleanup
// of sub-tasks that have lost their submitted-by-task reference ?
// also don't want warnings when it's finalized, this means we don't need ignoreIfNotRun()
((BasicTask<T>)t).cancelled = true;
return t;
}
@Test
public void testCallsOnSuccessWithResultOfCallable() throws Exception {
feed = FunctionFeed.builder()
.entity(entity)
.poll(new FunctionPollConfig<Integer, Integer>(SENSOR_INT)
.period(1)
.callable(Callables.returning(123))
.onSuccess(new AddOneFunction()))
.build();
EntityAsserts.assertAttributeEqualsEventually(entity, SENSOR_INT, 124);
}
@Test
public void testCallsOnFailureWithResultOfCallable() throws Exception {
feed = FunctionFeed.builder()
.entity(entity)
.poll(new FunctionPollConfig<Integer, Integer>(SENSOR_INT)
.period(1)
.callable(Callables.returning(1))
.checkSuccess(Predicates.alwaysFalse())
.onSuccess(new AddOneFunction())
.onFailure(Functions.constant(-1)))
.build();
EntityAsserts.assertAttributeEqualsEventually(entity, SENSOR_INT, -1);
}
@Test
public void testCallsOnExceptionWhenCheckSuccessIsFalseButNoFailureHandler() throws Exception {
feed = FunctionFeed.builder()
.entity(entity)
.poll(new FunctionPollConfig<Integer, Integer>(SENSOR_INT)
.period(1)
.callable(Callables.returning(1))
.checkSuccess(Predicates.alwaysFalse())
.onSuccess(new AddOneFunction())
.onException(Functions.constant(-1)))
.build();
EntityAsserts.assertAttributeEqualsEventually(entity, SENSOR_INT, -1);
}