public void start() {
    synchronized (startStopMutex) {
        if (state==ListenerState.RUNNING || (scheduledTask!=null && !scheduledTask.isDone())) {
            LOG.warn("Request to start "+this+" when already running - "+scheduledTask+"; ignoring");
            return;
        }
        state = ListenerState.RUNNING;

        Callable<Task<?>> taskFactory = new Callable<Task<?>>() {
            @Override public Task<Void> call() {
                return Tasks.<Void>builder().dynamic(false).displayName("periodic-persister").body(new Callable<Void>() {
                    @Override
                    public Void call() {
                        persistNowSafely();
                        return null;
                    }}).build();
            }
        };
        scheduledTask = (ScheduledTask) executionContext.submit(
            ScheduledTask.builder(taskFactory).displayName("scheduled:[periodic-persister]").tagTransient().period(period).delay(period).build() );
    }
}
 

protected boolean deleteTaskNonRecursive(Task<?> task) {
    Set<?> tags = TaskTags.getTagsFast(checkNotNull(task, "task"));
    for (Object tag : tags) {
        synchronized (tasksByTag) {
            Set<Task<?>> tasks = tasksWithTagLiveOrNull(tag);
            if (tasks != null) {
                tasks.remove(task);
                if (tasks.isEmpty()) {
                    tasksByTag.remove(tag);
                }
            }
        }
    }
    Task<?> removed = tasksById.remove(task.getId());
    incompleteTaskIds.remove(task.getId());
    if (removed!=null && removed.isSubmitted() && !removed.isDone(true)) {
        Entity context = BrooklynTaskTags.getContextEntity(removed);
        if (context!=null && !Entities.isManaged(context)) {
            log.debug("Forgetting about active task on unmanagement of "+context+": "+removed);
        } else {
            log.warn("Deleting submitted task before completion: "+removed+"; this task will continue to run in the background outwith "+this+", but perhaps it should have been cancelled?");
        }
    }
    return removed != null;
}
 

public static <T> Task<T> invokeEffectorAsync(Entity entity, Effector<T> eff, Map<String,?> parameters) {
    String name = eff.getName();

    if (log.isDebugEnabled()) log.debug("Invoking-async effector {} on {}", new Object[] { name, entity });
    if (log.isTraceEnabled()) log.trace("Invoking-async effector {} on {} with args {}", new Object[] { name, entity, Sanitizer.sanitize(parameters) });
    EntityManagementSupport mgmtSupport = ((EntityInternal)entity).getManagementSupport();
    if (!mgmtSupport.isDeployed()) {
        mgmtSupport.attemptLegacyAutodeployment(name);
    }
    ManagementContextInternal mgmtContext = (ManagementContextInternal) ((EntityInternal)entity).getManagementContext();

    // FIXME seems brittle to have the listeners in the Utils method; better to move into the context.invokeEff
    // (or whatever the last mile before invoking the effector is - though currently there is not such a canonical place!)
    mgmtSupport.getEntityChangeListener().onEffectorStarting(eff, parameters);
    try {
        return mgmtContext.invokeEffector(entity, eff, parameters);
    } finally {
        // FIXME this is really Effector submitted
        mgmtSupport.getEntityChangeListener().onEffectorCompleted(eff);
    }
}
 

private Function<SourceAndDestination, Task<?>> newCopyResourceFunction() {
    return new Function<SourceAndDestination, Task<?>>() {
        @Override
        public Task<?> apply(final SourceAndDestination input) {
            return Tasks.builder()
                    .displayName("Copying file: source=" + input.source + ", destination=" + input.destination)
                    .body(new Callable<Object>() {
                        @Override
                        public Integer call() {
                            return copyResource(input.source, input.destination, true);
                        }
                    })
                    .build();
        }
    };
}
 

@Test
public void testAttributeWhenReadyWithAbort() throws Exception {
    final Task<String> t = submit(DependentConfiguration.builder()
            .attributeWhenReady(entity, TestEntity.NAME)
            .abortIf(entity2, TestEntity.SEQUENCE, Predicates.equalTo(1))
            .build());

    assertNotDoneContinually(t);

    entity2.sensors().set(TestEntity.SEQUENCE, 321);
    assertNotDoneContinually(t);

    entity2.sensors().set(TestEntity.SEQUENCE, 1);
    try {
        assertDoneEventually(t);
        fail();
    } catch (Exception e) {
        if (!e.toString().contains("Aborted waiting for ready")) throw e;
    }
}
 

private void assertEffectorBlockingDetailsEventually(final Entity entity, final String effectorName, final String blockingDetailsSnippet) {
    Asserts.succeedsEventually(new Runnable() {
        @Override public void run() {
            final Set<Task<?>> tasksWithAllTags = mgmt.getExecutionManager().getTasksWithAllTags(ImmutableList.of(BrooklynTaskTags.EFFECTOR_TAG, BrooklynTaskTags.tagForContextEntity(entity)));
            Task<?> entityTask = null;
            for (Task<?> item : tasksWithAllTags) {
                final String itemName = getEffectorName(item);
                entityTask = itemName.equals(effectorName) ? item : entityTask;
            }
            if (entityTask == null) {
                Asserts.fail("Could not find task for effector " + effectorName);
            }
            String blockingDetails = getBlockingDetails(entityTask);
            assertTrue(blockingDetails.contains(blockingDetailsSnippet));
        }});
}
 

@Test
public void testErrorsResolvingPropagatesOrSwallowedAllCorrectly() throws Exception {
    app.config().set(TestEntity.CONF_OBJECT, ValueResolverTest.newThrowTask(Duration.ZERO));
    Task<Object> t = Tasks.builder().body(Functionals.callable(EntityFunctions.config(TestEntity.CONF_OBJECT), app)).build();
    ValueResolver<Object> v = Tasks.resolving(t).as(Object.class).context(app);
    
    ValueResolverTest.assertThrowsOnGetMaybe(v);
    ValueResolverTest.assertThrowsOnGet(v);
    
    v.swallowExceptions();
    ValueResolverTest.assertMaybeIsAbsent(v);
    ValueResolverTest.assertThrowsOnGet(v);
    
    v.defaultValue("foo");
    ValueResolverTest.assertMaybeIsAbsent(v);
    assertEquals(v.clone().get(), "foo");
    assertResolvesValue(v, Object.class, "foo");
}
 

static Maybe<?> execDslImmediately(final BrooklynDslDeferredSupplier<?> dsl, final TypeToken<?> type, final Entity context, boolean execInTask) throws Exception {
    // Exec'ing immediately will call DSL in current thread. It needs to find the context entity,
    // and does this using BrooklynTaskTags.getTargetOrContextEntity(Tasks.current()).
    // If we are not in a task executed by the context entity, then this lookup will fail. 
    Callable<Maybe<?>> job = new Callable<Maybe<?>>() {
        @Override
        public Maybe<?> call() throws Exception {
            return Tasks.resolving(dsl).as(type)
                    .context(context)
                    .description("Computing "+dsl)
                    .immediately(true)
                    .getMaybe();
        }
    };
    if (execInTask) {
        Task<Maybe<?>> task = ((EntityInternal)context).getExecutionContext().submit("Resolving DSL for test: "+dsl, job);
        task.get(Asserts.DEFAULT_LONG_TIMEOUT);
        assertTrue(task.isDone());
        return task.get();
        
    } else {
        return job.call();
    }
}
 

@Test
public void testAttributeWhenReadyWithAbortWaitingNow() throws Exception {
    final Task<String> t = submit(new Callable<String>() {
        @Override
        public String call() {
            return DependentConfiguration.builder()
                    .attributeWhenReady(entity, TestEntity.NAME)
                    .abortIf(entity2, TestEntity.SEQUENCE, Predicates.equalTo(1))
                    .runNow();
        }});

    assertNotDoneContinually(t);

    entity2.sensors().set(TestEntity.SEQUENCE, 321);
    assertNotDoneContinually(t);

    entity2.sensors().set(TestEntity.SEQUENCE, 1);
    try {
        assertDoneEventually(t);
        fail();
    } catch (Exception e) {
        if (!e.toString().contains("Aborted waiting for ready")) throw e;
    }
}
 

@Override
public Task<Entity> newTask() {
    return TaskBuilder.<Entity>builder()
            .displayName(toString())
            .tag(BrooklynTaskTags.TRANSIENT_TASK_TAG)
            .body(new Callable<Entity>() {
                @Override
                public Entity call() {
                    EntityInternal entity = entity();
                    Collection<Entity> entitiesToSearch = entity.getManagementContext().getEntityManager().getEntities();
                    Optional<Entity> result = Iterables.tryFind(entitiesToSearch, EntityPredicates.configEqualTo(TAG, tag));
                    
                    if (result.isPresent()) {
                        return result.get();
                    } else {
                        throw new NoSuchElementException("No entity matching id " + tag+" in "+entitiesToSearch);
                    }
                }})
            .build();
}
 

/** 
 * Method which returns a Future containing an escaped URL string (see {@link Urls#encode(String)}).
 * The arguments can be normal objects, tasks or {@link DeferredSupplier}s.
 * tasks will be waited on (submitted if necessary) and their results substituted.
 */
@SuppressWarnings("unchecked")
public static Task<String> urlEncode(final Object arg) {
    List<TaskAdaptable<Object>> taskArgs = Lists.newArrayList();
    if (arg instanceof TaskAdaptable) taskArgs.add((TaskAdaptable<Object>)arg);
    else if (arg instanceof TaskFactory) taskArgs.add( ((TaskFactory<TaskAdaptable<Object>>)arg).newTask() );
    
    return transformMultiple(
            MutableMap.<String,String>of("displayName", "url-escaping '"+arg), 
            new Function<List<Object>, String>() {
                @Override
                @Nullable
                public String apply(@Nullable List<Object> input) {
                    Object resolvedArg;
                    if (arg instanceof TaskAdaptable || arg instanceof TaskFactory) resolvedArg = Iterables.getOnlyElement(input);
                    else if (arg instanceof DeferredSupplier) resolvedArg = ((DeferredSupplier<?>) arg).get();
                    else resolvedArg = arg;
                    
                    if (resolvedArg == null) return null;
                    String resolvedString = resolvedArg.toString();
                    return Urls.encode(resolvedString);
                }
            },
            taskArgs);
}
 

@Test
public void testDisplayNameEqualTo() throws Exception {
    Task<Object> task = execManager.submit(TaskBuilder.builder()
            .body(Callables.<Object>returning("val"))
            .displayName("myname")
            .build());
    assertTrue(TaskPredicates.displayNameEqualTo("myname").apply(task));
    assertFalse(TaskPredicates.displayNameEqualTo("wrong").apply(task));
}
 

@Override
public void update() {
    try {
        Task<?> task = updateAsync();
        if (task != null) task.getUnchecked();
        ServiceStateLogic.ServiceProblemsLogic.clearProblemsIndicator(this, "update");
    } catch (Exception e) {
        ServiceStateLogic.ServiceProblemsLogic.updateProblemsIndicator(this, "update", "update failed with: "+Exceptions.collapseText(e));
        throw Exceptions.propagate(e);
    }
}
 

@Test
public void testStopSequentiallyContinuesOnFailureInSubTask() throws Exception {
    try {
        entity = app.createAndManageChild(EntitySpec.create(FailingEntity.class)
                .configure(FailingEntity.FAIL_ON_STOP, true)
                .configure(FailingEntity.FAIL_IN_SUB_TASK, true)
                .configure(FailingEntity.LISTENER, listener));
        entity2 = app.createAndManageChild(EntitySpec.create(FailingEntity.class)
                .configure(FailingEntity.LISTENER, listener));
        app.start(ImmutableList.of(loc));
        listener.events.clear();
        
        try {
            Task<?> task = Tasks.builder().displayName("stopSequentially")
                    .body(new Runnable() {
                        @Override public void run() {
                            StartableMethods.stopSequentially(ImmutableList.of(entity, entity2));
                        }})
                    .build();
            Entities.submit(app, task).getUnchecked();
            fail();
        } catch (Exception e) {
            // success; expected exception to be propagated
            if (!(e.toString().contains("Error stopping"))) throw e;
        }
        
        assertEquals(listener.events.get(0)[0], entity);
        assertEquals(listener.events.get(1)[0], entity2);
    } finally {
        // get rid of entity that will fail on stop, so that tearDown won't encounter exception
        Entities.unmanage(entity);
    }
}
 

@Test
public void testRetrievingTasksWithTagsReturnsExpectedTask() throws Exception {
    Task<?> t = new BasicTask<Void>(newNoop());
    em.submit(MutableMap.of("tag", "A"), t);
    t.get();

    assertEquals(em.getTasksWithTag("A"), ImmutableList.of(t));
    assertEquals(em.getTasksWithAnyTag(ImmutableList.of("A")), ImmutableList.of(t));
    assertEquals(em.getTasksWithAnyTag(ImmutableList.of("A", "B")), ImmutableList.of(t));
    assertEquals(em.getTasksWithAllTags(ImmutableList.of("A")), ImmutableList.of(t));
}
 

@Test
public void testSingleExecutionContextEntityWithTaskAndExternalFlags() {
    // Should cause an exception to be thrown in future releases. For now will log a warning.
    // Until then make sure the task is tagged only with the context of the executor.
    final TestEntity entity = app.createAndManageChild(EntitySpec.create(TestEntity.class));
    Task<Void> task = Tasks.<Void>builder()
        .body(new AssertContextRunnable(ImmutableList.of(app))).build();
    ImmutableMap<String,?> flags = ImmutableMap.of(
            "tags", ImmutableList.of(BrooklynTaskTags.tagForContextEntity(entity)));
    app.getExecutionContext().submit(flags, task).getUnchecked();
}
 

/**
 * Invoke effector with parameters on the entity that the policy is attached to.
 */
protected <T> Task<T> invoke(Effector<T> effector, Map<String, ?> parameters) {
    if (isBusySensorEnabled()) {
        getTaskCounter().incrementAndGet();
        publishIsBusy();
    }
    Task<T> task = entity.invoke(effector, parameters);
    if (isBusySensorEnabled()) {
        task.addListener(new EffectorListener(), MoreExecutors.sameThreadExecutor());
    }
    return task;
}
 

@Test
public void testRetrievingTasksWithAllTagsWhenFirstNotMatched() throws Exception {
    Task<?> t = new BasicTask<Void>(newNoop());
    em.submit(MutableMap.of("tags", ImmutableList.of("A")), t);
    t.get();

    assertEquals(em.getTasksWithAllTags(ImmutableList.of("not_there","A")), ImmutableSet.of());
}
 

protected void expireTransientTasks() {
    Set<Task<?>> transientTasks = executionManager.getTasksWithTag(BrooklynTaskTags.TRANSIENT_TASK_TAG);
    for (Task<?> t: transientTasks) {
        if (!t.isDone(true)) continue;
        executionManager.deleteTask(t);
    }
}
 

public void testUnsubmittedTaskWhenNoExecutionContextFails() {
    Task<String> t = newSleepTask(Duration.ZERO, "foo");
    
    // Below, we call ValueResolver.getMaybe() with no execution context. Therefore it will not execute the task.
    Maybe<String> result = Tasks.resolving(t).as(String.class).timeout(Duration.ZERO).getMaybe();
    
    Assert.assertTrue(result.isAbsent(), "result="+result);
    Exception exception = Maybe.getException(result);
    Asserts.assertStringContains(exception.toString(), "no execution context available");
    
    Asserts.assertThat(t, (tt) -> !tt.isBegun());
}
 

/** Invokes in parallel if multiple, but otherwise invokes the item directly. */
public static Task<?> invokeEffector(Entity callingEntity, Iterable<? extends Entity> entitiesToCall,
        final Effector<?> effector, final Map<String,?> parameters) {
    if (Iterables.size(entitiesToCall)==1)
        return invokeEffector(callingEntity, entitiesToCall.iterator().next(), effector, parameters);
    else
        return invokeEffectorList(callingEntity, entitiesToCall, effector, parameters);
}
 

@Test
public void testAttributeWhenReadyWithPredicate() throws Exception {
    final Task<String> t = submit(DependentConfiguration.attributeWhenReady(entity, TestEntity.NAME, Predicates.equalTo("myval2")));
    
    entity.sensors().set(TestEntity.NAME, "myval");
    assertNotDoneContinually(t);
    
    entity.sensors().set(TestEntity.NAME, "myval2");
    assertEquals(assertDoneEventually(t), "myval2");
}
 

@Override
public Task<Object> newTask() {
    boolean immediate = false;
    
    Callable<Object> job = new Callable<Object>() {
        @Override
        public Object call() throws Exception {
            Entity targetEntity = component.get();
            
            int indexI = resolveIndex(immediate);
            
            // this is always run in a new dedicated task (possibly a fake task if immediate), so no need to clear
            String tag = "DSL:entity('"+targetEntity.getId()+"').location('"+indexI+"')";
            checkAndTagForRecursiveReference(targetEntity, tag);

            // TODO Try repeatedly if no location(s)?
            Collection<Location> locations = getLocations(targetEntity);
            if (locations.size() < (indexI + 1)) {
                throw new IndexOutOfBoundsException("Target entity ("+component+") has "+locations.size()+" location(s), but requested index "+index);
            }
            Location result = Iterables.get(locations, indexI);
            if (result == null) {
                throw new NullPointerException("Target entity ("+component+") has null location at index "+index);
            }
            return result;
        }
    };
    
    return Tasks.builder()
            .displayName("retrieving locations["+index+"] for "+component)
            .tag(BrooklynTaskTags.TRANSIENT_TASK_TAG)
            .dynamic(false)
            .body(job).build();
}
 

public static LinkWithMetadata asLink(Task<?> t, UriBuilder ub) {
    if (t==null) return null;
    MutableMap<String,Object> data = new MutableMap<String,Object>();
    data.put("id", t.getId());
    if (t.getDisplayName()!=null) data.put("taskName", t.getDisplayName());
    Entity entity = BrooklynTaskTags.getContextEntity(t);
    if (entity!=null) {
        data.put("entityId", entity.getId());
        if (entity.getDisplayName()!=null) data.put("entityDisplayName", entity.getDisplayName());
    }
    URI taskUri = serviceUriBuilder(ub, ActivityApi.class, "get").build(t.getId());
    return new LinkWithMetadata(taskUri.toString(), data);
}
 

public void testExecutionContextGetImmediatelyBasicTaskSucceeds() throws Exception {
    final Task<String> t = newSleepTask(Duration.ZERO, "foo");
    
    // Extracts job from task, tries to run it; because it doesn't block we'll get the result.
    // It will also render the task unusable (cancelled); calling `t.get()` will throw CancellationException.
    Maybe<String> result = app.getExecutionContext().getImmediately(t);
    Assert.assertTrue(result.isPresent(), "result="+result);
    Assert.assertEquals(result.get(), "foo", "result="+result);
}
 

@Test
public void testTaskGatherOutput() {
    Task<Integer> task = executeSampleScript(new Function<ScriptHelper, Void>() {
        @Override
        public Void apply(ScriptHelper scriptHelper) {
            return null;
        }
    });

    String stdOut = getStreamOrFail(task, BrooklynTaskTags.STREAM_STDOUT);
    String stdErr = getStreamOrFail(task, BrooklynTaskTags.STREAM_STDERR);
    Assert.assertEquals(stdOut, output);
    Assert.assertEquals(stdErr, errorStd);
}
 

@Test
public void testSequentialTaskFailsWhenIntermediateTaskThrowsException() throws Exception {
    BasicTask<String> t1 = taskReturning("a");
    BasicTask<String> t2 = new BasicTask<String>(new Callable<String>() {
            @Override public String call() throws Exception {
                throw new IllegalArgumentException("forced exception");
            }
        });
    BasicTask<String> t3 = taskReturning("c");
    SequentialTask<String> task = new SequentialTask<String>(t1, t2, t3);
    Task<List<String>> tSequence = ec.submit(task);

    try {
        tSequence.get();
        fail("t2 should have thrown an exception");
    } catch (Exception e) {}

    assertTrue(task.isDone());
    assertTrue(task.isError());
    assertTrue(t1.isDone());
    assertFalse(t1.isError());
    assertTrue(t2.isDone());
    assertTrue(t2.isError());
    // t3 not run because of t2 exception
    assertFalse(t3.isDone());
    assertFalse(t3.isBegun());
}
 

@Override
public Task<Object> newTask() {
    return Tasks.<Object>builder()
        .displayName("resolving external configuration: '" + key + "' from provider '" + providerName + "'")
        .dynamic(false)
        .body(new Callable<Object>() {
            @Override
            public Object call() throws Exception {
                return getImmediately().get();
            }
        })
        .build();
}
 

@Override
public Task<Object> newTask() {
    return Tasks.builder()
            .displayName("Deferred function call " + object + "." + toStringF(fnName, args))
            .tag(BrooklynTaskTags.TRANSIENT_TASK_TAG)
            .dynamic(false)
            .body(new Callable<Object>() {
                @Override
                public Object call() throws Exception {
                    return invokeOnDeferred(object, false).get();
                }

            }).build();
}
 

public static void setEffectorParameters(Task<?> task, ConfigBag parameters) {
    EffectorCallTag result = getEffectorCallTag(task, true);
    if (result == null) {
        throw new IllegalStateException("No EffectorCallTag found, is the task an effector? Task: " + task);
    }
    result.setParameters(parameters);
}