Crate Policy

CratePolicy.java

package com.acme.crate;

import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.brooklyn.api.entity.EntityLocal;
import org.apache.brooklyn.api.sensor.SensorEvent;
import org.apache.brooklyn.api.sensor.SensorEventListener;
import org.apache.brooklyn.core.entity.trait.Startable;
import org.apache.brooklyn.core.mgmt.internal.EffectorUtils;
import org.apache.brooklyn.core.policy.AbstractPolicy;
import org.apache.brooklyn.feed.function.FunctionFeed;
import org.apache.brooklyn.feed.function.FunctionPollConfig;
import org.apache.brooklyn.util.time.Duration;
import org.python.google.common.collect.ImmutableMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class CratePolicy extends AbstractPolicy {

    private static final Logger LOG = LoggerFactory.getLogger(CratePolicy.class);

    private static final AtomicInteger UPDATE_COUNT = new AtomicInteger(0);

    private FunctionFeed feed;

    @Override
    public void setEntity(EntityLocal entity) {
        super.setEntity(entity);

        subscribe();
    }

    private void subscribe() {
        // Perform a one-off task
        LOG.info("Hello from the CratePolicy");

        // Take action when a sensor changes
        subscriptions().subscribe(entity, CrateNode.SERVER_NAME, new SensorEventListener<String>() {
            @Override
            public void onEvent(SensorEvent<String> event) {
                if (UPDATE_COUNT.getAndIncrement() == 6) {
                    EffectorUtils.invokeEffectorAsync(entity, Startable.STOP, ImmutableMap.<String, Object>of());
                    UPDATE_COUNT.set(0);
                }
            }
        });

        // Do something periodically
        feed = FunctionFeed.builder()
                .entity(entity)
                .period(Duration.TEN_SECONDS)
                .poll(FunctionPollConfig.forSensor(CrateNode.SERVER_NAME)
                    .callable(new Callable<Object>() {
                        @Override
                        public Object call() throws Exception {
                            LOG.info("CratePolicy polling");
                            return null;
                        }
                    }))
                .build();
    }

    @Override
    public void destroy() {
        super.destroy();
        if (feed != null) {
            feed.stop();
        }
    }
}