import org.apache.brooklyn.util.time.Duration; //导入依赖的package包/类
@Override
public void run() {
    boolean done = Repeater.create(String.format("Request %s status check", builder.request.toString()))
            .every(Duration.ONE_SECOND)
            .until(new Callable<Boolean>() {
                @Override
                public Boolean call() throws Exception {
                    final String json = HttpTool.httpGet(builder.httpClient, URI.create(builder.request.getHref()), builder.headers).getContentAsString();
                    final String status = JsonPath.read(json, "$.Requests.request_status");
                    if (!VALID_STATES.contains(status)) {
                        throw new RuntimeException(
                                "Request fails with state " + status +
                                        ". Check here for details " + builder.request.getHref());
                    }
                    return StringUtils.equals(status, "COMPLETED");
                }
            })
            .limitTimeTo(builder.timeout)
            .rethrowExceptionImmediately()
            .run();

    if (!done) {
        throw new RuntimeException(builder.errorMessage);
    }
}
 

import org.apache.brooklyn.util.time.Duration; //导入依赖的package包/类
public RequestCheckRunnable build() {
    Preconditions.checkNotNull(this.request);

    if (this.headers == null) {
        this.headers = ImmutableMap.of();
    }
    if (this.timeout == null) {
        this.timeout = Duration.FIVE_MINUTES;
    }
    if (StringUtils.isEmpty(this.errorMessage)) {
        this.errorMessage = String.format("The request did not finish with the status \"COMPLETED\" or within %s", this.timeout.toString());
    }
    if (this.httpClient == null) {
        this.httpClient = HttpTool.httpClientBuilder().build();
    }

    return new RequestCheckRunnable(this);
}
 

import org.apache.brooklyn.util.time.Duration; //导入依赖的package包/类
@Override
public void addHostsToHostGroup(final String blueprintName, final String hostgroupName, final List<String> hosts, final String cluster) {
    Iterable<Map> hostGroupMapping = Iterables.transform(hosts, fqdnsToMaps(blueprintName, hostgroupName));
    LOG.info("hosts " + hostGroupMapping.iterator().hasNext());

    HostEndpoint hostEndpoint = restAdapter.create(HostEndpoint.class);
    Request request = hostEndpoint.addHosts(
            cluster,
            Lists.newArrayList(hostGroupMapping));

    RequestCheckRunnable.check(request)
            .headers(ImmutableMap.of(HttpHeaders.AUTHORIZATION, HttpTool.toBasicAuthorizationValue(usernamePasswordCredentials)))
            .timeout(Duration.ONE_HOUR)
            .errorMessage(String.format("Error during adding %s to %s", hosts, hostgroupName))
            .build()
            .run();
}
 

import org.apache.brooklyn.util.time.Duration; //导入依赖的package包/类
void assertNoFires(final Entity app) {
    EntityAsserts.assertAttributeEqualsEventually(app, Attributes.SERVICE_UP, Boolean.valueOf(true));
    EntityAsserts.assertAttributeEqualsEventually(app, Attributes.SERVICE_STATE_ACTUAL, Lifecycle.RUNNING);
    Asserts.succeedsEventually(ImmutableMap.of("timeout", Duration.FIVE_MINUTES), new Runnable() {
        public void run() {
            Iterator i$ = Entities.descendants(app).iterator();

            while (i$.hasNext()) {
                Entity entity = (Entity) i$.next();
                Assert.assertNotEquals(entity.getAttribute(Attributes.SERVICE_STATE_ACTUAL), Lifecycle.ON_FIRE);
                Assert.assertNotEquals(entity.getAttribute(Attributes.SERVICE_UP), Boolean.valueOf(false));
                if (entity instanceof SoftwareProcess) {
                    EntityAsserts.assertAttributeEquals(entity, Attributes.SERVICE_STATE_ACTUAL, Lifecycle.RUNNING);
                    EntityAsserts.assertAttributeEquals(entity, Attributes.SERVICE_UP, Boolean.TRUE);
                }
            }

        }
    });
}
 

import org.apache.brooklyn.util.time.Duration; //导入依赖的package包/类
@Test
public void testSuccessfulRequestFinishes() {
    Throwable throwable = null;

    try {
        new RequestCheckRunnable.Builder(mockRequest("http://www.example.com"))
                .client(mockHttpClient("{\"Requests\":{\"request_status\": \"COMPLETED\"}}"))
                .timeout(Duration.FIVE_SECONDS)
                .build()
                .run();
    } catch (Throwable t) {
        throwable = t;
    }

    assertNull(throwable);
}
 

import org.apache.brooklyn.util.time.Duration; //导入依赖的package包/类
public void blockUntilTaskCompletes(String id, Duration timeout, Object[] results) throws PollingException {
    try {
        Repeater.create()
                .every(Duration.ONE_SECOND)
                .until(() -> {
                    TaskSummary summary = getRestApi().getActivityApi().get(id);
                    if (summary.isError() || summary.isCancelled() || (summary.getSubmitTimeUtc() == null)) {
                        throw new PollingException(new IllegalStateException("Effector call failed: " + summary));
                    }
                    if (summary.getEndTimeUtc() != null) {
                        results[0] = summary.getResult();
                        return true;
                    }
                    return false;
                })
                .rethrowExceptionImmediately()
                .limitTimeTo(timeout)
                .runRequiringTrue();
    } catch (Exception e) {
        LOG.error("Failed to get task summary: " + e);
        throw new PollingException(e);
    }
}
 

import org.apache.brooklyn.util.time.Duration; //导入依赖的package包/类
/**
 * Waits for the given endpoint to respond true to {@link ServerApi#isUp}
 * or for the forked process to have exited.
 */
private void waitForServerStartOrExit(final ForkedServer forkedServer) throws MojoFailureException {
    final URL url = forkedServer.getServer();
    final ServerApi api = getApi(url).getServerApi();
    getLog().info("Waiting for server at " + url + " to be ready within " + getTimeout());
    boolean isUp = Repeater.create("Waiting for server at " + url + " to be ready within " + getTimeout())
            .every(Duration.ONE_SECOND)
            .limitTimeTo(getTimeout())
            .until(new Callable<Boolean>() {
                @Override
                public Boolean call() throws Exception {
                    // The order of these is important - if forkedServer.hasExited() then api.isUp()
                    // will throw an exception! The exception is thrown by RESTEasy.
                    return forkedServer.hasExited() || api.isUp();
                }
            })
            .run();
    if (forkedServer.hasExited()) {
        throw new MojoFailureException("Forked server exited unexpectedly (exit code " + forkedServer.getExitCode() + ")");
    } else if (!isUp) {
        throw new MojoFailureException("Server at " + url + " does not appear to be running after " + getTimeout());
    }
}
 

import org.apache.brooklyn.util.time.Duration; //导入依赖的package包/类
private void connectServiceLatencySensor() {
    serverLatencyFeed = FunctionFeed.builder()
            .entity(this)
            .period(Duration.seconds(2))
            .poll(new FunctionPollConfig<Double, Double>(SERVER_LATENCY)
                    .onException(Functions.constant(0.0))
                    .callable(new Callable<Double>() {
                        public Double call() {

                            Double total = getAttribute(SERVER_PROCESSING_TIME);
                            Double num = getAttribute(SERVER_REQUESTS);
                            return total / num;
                        }
                    }))
            .build();
}
 

import org.apache.brooklyn.util.time.Duration; //导入依赖的package包/类
private void connectServiceRequestPerSecondSensor() {
    requestPerSecondLatencyFeed = FunctionFeed.builder()
            .entity(this)
            .period(Duration.seconds(1))
            .poll(new FunctionPollConfig<Double, Double>(REQUEST_PER_SECOND)
                    .onException(Functions.constant(0.0))
                    .callable(new Callable<Double>() {
                        public Double call() {
                            Double currentRequestPerSecond = getAttribute(SERVER_REQUESTS) - lastTotalRequest;
                            if (currentRequestPerSecond >= 0) {
                                lastTotalRequest = getAttribute(SERVER_REQUESTS);
                                return currentRequestPerSecond;
                            } else {
                                return getAttribute(SERVER_REQUESTS);
                            }
                        }
                    }))
            .build();
}
 

import org.apache.brooklyn.util.time.Duration; //导入依赖的package包/类
protected void assertHadoopClusterEventuallyDeployed(Application app) {
    AmbariServer ambariServer = Entities.descendants(app, AmbariServer.class).iterator().next();
    EntityAsserts.assertAttributeEventually(
            ImmutableMap.of("timeout", Duration.minutes(60)),
            ambariServer,
            AmbariServer.CLUSTER_STATE,
            Predicates.not(Predicates.or(Predicates.equalTo("ABORTED"), Predicates.equalTo("FAILED"), Predicates.equalTo("TIMEDOUT")))
    );
    EntityAsserts.assertAttributeEventually(
            ImmutableMap.of("timeout", Duration.minutes(60)),
            ambariServer,
            AmbariServer.CLUSTER_STATE,
            Predicates.equalTo("COMPLETED"));
}
 

import org.apache.brooklyn.util.time.Duration; //导入依赖的package包/类
@Test(expectedExceptions = RuntimeException.class, expectedExceptionsMessageRegExp = ERROR_MESSAGE)
public void testTimeoutRequestThrowsRuntimeEx() throws IOException {
    new RequestCheckRunnable.Builder(mockRequest("http://www.example.com"))
            .client(mockHttpClient("{\"Requests\":{\"request_status\": \"IN_PROGRESS\"}}"))
            .timeout(Duration.FIVE_SECONDS)
            .errorMessage(ERROR_MESSAGE)
            .build()
            .run();
}
 

import org.apache.brooklyn.util.time.Duration; //导入依赖的package包/类
@Test(expectedExceptions = PathNotFoundException.class)
public void testInvalidReponseThrowsParseEx() throws IOException {
    new RequestCheckRunnable.Builder(mockRequest("http://www.example.com"))
            .client(mockHttpClient("Invalid body: No Json"))
            .timeout(Duration.FIVE_SECONDS)
            .build()
            .run();
}
 

import org.apache.brooklyn.util.time.Duration; //导入依赖的package包/类
@Test
public void newServiceInstanceBindingCreatedSuccessfullyWithBindEffector()
        throws ServiceBrokerException, ServiceInstanceBindingExistsException, PollingException {
   when(admin.getRestApi()).thenReturn(brooklynApi);
   when(admin.getCredentialsFromSensors(
           anyString(),
           anyString(),
           any(Predicate.class),
           any(Predicate.class),
           any(Predicate.class),
           any(Predicate.class)
   )).thenReturn(new AsyncResult<>(Collections.<String, Object>emptyMap()));
   when(admin.hasEffector(anyString(), anyString(), anyString())).thenReturn(new AsyncResult<>(true));
   when(admin.invokeEffector(anyString(), anyString(), anyString(), anyString(), anyMap())).thenReturn(new AsyncResult<>(TASK_RESPONSE_COMPLETE));
   when(brooklynApi.getActivityApi()).thenReturn(activityApi);
   when(activityApi.get(anyString()))
           .thenReturn(TASK_SUMMARY_INCOMPLETE)
           .thenReturn(TASK_SUMMARY_INCOMPLETE)
           .thenReturn(TASK_SUMMARY_INCOMPLETE)
           .thenReturn(TASK_SUMMARY_INCOMPLETE)
           .thenReturn(TASK_SUMMARY_COMPLETE);
   doCallRealMethod().when(admin).blockUntilTaskCompletes(anyString());
   doCallRealMethod().when(admin).blockUntilTaskCompletes(anyString(), any(Duration.class), any(Object[].class));
   when(instanceRepository.findOne(anyString(), anyBoolean())).thenReturn(serviceInstance);
   when(serviceDefinition.getMetadata()).thenReturn(ImmutableMap.of());
   when(brooklynCatalogService.getServiceDefinition(anyString())).thenReturn(serviceDefinition);
   CreateServiceInstanceBindingRequest request = new CreateServiceInstanceBindingRequest(serviceInstance.getServiceDefinitionId(), "planId", "appGuid", null);
   CreateServiceInstanceBindingResponse binding = bindingService.createServiceInstanceBinding(request.withBindingId(SVC_INST_BIND_ID));

   // TODO assert binding was completed successfully
   //assertEquals(SVC_INST_BIND_ID, binding.getServiceBindingId());
}
 

import org.apache.brooklyn.util.time.Duration; //导入依赖的package包/类
@Test(expected = RuntimeException.class)
public void testServiceInstanceBindingFailureWithBindEffector()
        throws ServiceBrokerException, ServiceInstanceBindingExistsException, PollingException {
   when(admin.getRestApi()).thenReturn(brooklynApi);
   when(admin.getCredentialsFromSensors(
           anyString(),
           anyString(),
           any(Predicate.class),
           any(Predicate.class),
           any(Predicate.class),
           any(Predicate.class)
   )).thenReturn(new AsyncResult<>(Collections.<String, Object>emptyMap()));
   when(admin.hasEffector(anyString(), anyString(), anyString())).thenReturn(new AsyncResult<>(true));
   when(admin.invokeEffector(anyString(), anyString(), anyString(), anyString(), anyMap())).thenReturn(new AsyncResult<>(null));
   when(brooklynApi.getActivityApi()).thenReturn(activityApi);
   when(activityApi.get(anyString()))
           .thenReturn(TASK_SUMMARY_INCOMPLETE)
           .thenReturn(TASK_SUMMARY_INCOMPLETE)
           .thenReturn(TASK_SUMMARY_INCOMPLETE)
           .thenReturn(TASK_SUMMARY_INCOMPLETE)
           .thenReturn(TASK_SUMMARY_COMPLETE);
   doCallRealMethod().when(admin).blockUntilTaskCompletes(anyString());
   doCallRealMethod().when(admin).blockUntilTaskCompletes(anyString(), any(Duration.class), any(Object[].class));
   when(instanceRepository.findOne(anyString(), anyBoolean())).thenReturn(serviceInstance);
   when(serviceDefinition.getMetadata()).thenReturn(ImmutableMap.of());
   when(brooklynCatalogService.getServiceDefinition(anyString())).thenReturn(serviceDefinition);
   CreateServiceInstanceBindingRequest request = new CreateServiceInstanceBindingRequest(serviceInstance.getServiceDefinitionId(), "planId", "appGuid", null);
   bindingService.createServiceInstanceBinding(request.withBindingId(SVC_INST_BIND_ID));
}
 

import org.apache.brooklyn.util.time.Duration; //导入依赖的package包/类
public static ApplicationContext newApplicationContext(ManagementContext mgmt, ResourceLoader resourceLoader) throws Exception {
    log.info("Loading Alien4Cloud platform...");
    // TODO if ES cannot find a config file, it will hang waiting for peers; should warn if does not complete in 1m
    try {
        Stopwatch s = Stopwatch.createStarted();

        AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext();
        
        if (null != resourceLoader) {
            ctx.setResourceLoader(resourceLoader);
        }

        // messy, but seems we must manually load the properties before loading the beans; otherwise we get e.g.
        // Caused by: java.lang.IllegalArgumentException: Could not resolve placeholder 'directories.alien' in string value "${directories.alien}/plugins"
        final YamlPropertiesFactoryBean yamlPropertiesFactoryBean = AlienBrooklynYamlPropertiesFactoryBeanFactory.get(mgmt, ctx);
        if (yamlPropertiesFactoryBean == null) {
            throw new IllegalStateException("Could not load configuration for A4C. Expected either a value for ConfigKey " +
                    AlienBrooklynYamlPropertiesFactoryBeanFactory.ALIEN_CONFIG_FILE.getName() + " or for a resource named " +
                    AlienYamlPropertiesFactoryBeanFactory.ALIEN_CONFIGURATION_YAML + " to be available.");
        }


        ctx.getEnvironment().getPropertySources().addFirst(new PropertiesPropertySource("user",
                yamlPropertiesFactoryBean.getObject()));
        ctx.getBeanFactory().registerSingleton("brooklynManagementContext", mgmt);

        ctx.register(Alien4CloudSpringContext.class, Alien4CloudSpringConfig.class);
        ctx.refresh();
        ctx.registerShutdownHook();
        log.info("Finished loading Alien4Cloud platform (" + Duration.of(s) + ")");
        return ctx;

    } catch (Throwable t) {
        log.warn("Errors loading Alien4Cloud platform (rethrowing): " + t, t);
        throw Exceptions.propagate(t);
    }
}
 

import org.apache.brooklyn.util.time.Duration; //导入依赖的package包/类
private void addPropertyDefinitions(CatalogEntitySummary brooklynEntity, IndexedNodeType toscaType) {
    Set<EntityConfigSummary> entityConfigSummaries = brooklynEntity.getConfig(); // properties in TOSCA
    Map<String, PropertyDefinition> properties = Maps.newHashMap();
    toscaType.setProperties(properties);
    if (entityConfigSummaries == null) {
        return;
    }
    for (EntityConfigSummary entityConfigSummary : entityConfigSummaries) {
        String propertyType = TYPE_MAPPING.get(entityConfigSummary.getType());
        if (propertyType == null) {
            log.warn("Skipping entityConfigSummary {} as type {} is not recognized", entityConfigSummary, entityConfigSummary.getType());
        } else {
            PropertyDefinition propertyDefinition = new PropertyDefinition();
            propertyDefinition.setDescription(entityConfigSummary.getDescription());
            propertyDefinition.setType(propertyType);
            if (entityConfigSummary.getDefaultValue() != null) {
                if (propertyType.equals(ToscaType.TIME)) {
                    propertyDefinition.setDefault(Duration.of(entityConfigSummary.getDefaultValue()).toSeconds() + " s");
                } else {
                    propertyDefinition.setDefault(entityConfigSummary.getDefaultValue().toString());
                }
            }
            if (ToscaType.MAP.equals(propertyType)) {
                PropertyDefinition mapDefinition = new PropertyDefinition();
                // TODO: More complex map types. Unfortunately the type is not available from EntityConfigSummary
                mapDefinition.setType(ToscaType.STRING);
                propertyDefinition.setEntrySchema(mapDefinition);
            }
            propertyDefinition.setRequired(false);
            toscaType.getProperties().put(entityConfigSummary.getName(), propertyDefinition);
        }
    }
}
 

import org.apache.brooklyn.util.time.Duration; //导入依赖的package包/类
/**
 * For connecting the {@link #INSTANCES_NUM} sensor.
 * <p>
 * Should be called inside {@link #connectSensors()}.
 *
 * @see #disconnectInstancesNumberSensor()
 */
protected void connectInstancesNumberSensor() {
    usedNumberOfInstances = FunctionFeed.builder()
            .entity(this)
            .period(Duration.seconds(2))
            .poll(new FunctionPollConfig<Integer, Integer>(INSTANCES_NUM)
                    .onException(Functions.constant(0))
                    .callable(new Callable<Integer>() {
                        public Integer call() {
                            return getDriver().getInstancesNumber();
                        }
                    }))
            .build();
}
 

import org.apache.brooklyn.util.time.Duration; //导入依赖的package包/类
/**
 * For connecting the {@link #DISK} sensor.
 * <p>
 * Should be called inside {@link #connectSensors()}.
 *
 * @see #disconnectDiskQuotaSensor()
 */
protected void connectDiskQuotaSensor() {
    usedDisk = FunctionFeed.builder()
            .entity(this)
            .period(Duration.seconds(2))
            .poll(new FunctionPollConfig<Integer, Integer>(DISK)
                    .onException(Functions.constant(0))
                    .callable(new Callable<Integer>() {
                        public Integer call() {
                            return getDriver().getDisk();
                        }
                    }))
            .build();
}
 

import org.apache.brooklyn.util.time.Duration; //导入依赖的package包/类
/**
 * For connecting the {@link #MEMORY} sensor.
 * <p>
 * Should be called inside {@link #connectSensors()}.
 *
 * @see #disconnectMemorySensor()
 */
protected void connectMemorySensor() {
    usedMemory = FunctionFeed.builder()
            .entity(this)
            .period(Duration.seconds(2))
            .poll(new FunctionPollConfig<Integer, Integer>(MEMORY)
                    .onException(Functions.constant(0))
                    .callable(new Callable<Integer>() {
                        public Integer call() {
                            return getDriver().getMemory();
                        }
                    }))
            .build();
}
 

import org.apache.brooklyn.util.time.Duration; //导入依赖的package包/类
private void connectResourceLatencySensor() {
    resourceLatencyFeed = FunctionFeed.builder()
            .entity(this)
            .period(Duration.seconds(2))
            .poll(new FunctionPollConfig<Double, Double>(RESOURCE_LATENCY)
                    .onException(Functions.constant(0.0))
                    .callable(new Callable<Double>() {
                        public Double call() {
                            Double total = getAttribute(DURATION_SUM);
                            Double num = getAttribute(RESOURCE_HITS);
                            return total / num;
                        }
                    }))
            .build();
}