import reactor.bus.Event; //导入依赖的package包/类
@Override
public void accept(Event<FailedAttemptEvent> event) {
	String ip = event.getData().ip();
	LogFailedEntity entity = repo.findByIp(ip);
	if (entity != null) {
		LOG.info(ip + " :: already in. increasing attempts!");
		entity.increaseAttempts();
		repo.saveAndFlush(entity);
	} else {
		bus.sendAndReceive("geocode", Event.wrap(ip), ev -> {
			LogFailedEntity logEntity = (LogFailedEntity) ev.getData();
			LogFailedEntity existing = repo.findByIp(logEntity.getIp());
			if (existing != null) {
				existing.increaseAttempts();
				repo.saveAndFlush(existing);
			} else {
				repo.saveAndFlush(logEntity);
			}
			LOG.info(ip + " :: geocode and persist!");
		});
	}
}
 

import reactor.bus.Event; //导入依赖的package包/类
@Override
public LogFailedEntity apply(Event<String> ip) {
	if (!this.conditionsMet()) {
		try {
			waitAndReset();
		} catch (InterruptedException e) {
			LOG.error("Something went wrong while waiting");
		}
	}

	RestTemplate temp = new RestTemplate();
	LogFailedEntity entity = temp.getForObject(address + ip.getData(), LogFailedEntity.class);
	entity.setIp(ip.getData());
	requests.incrementAndGet();

	return entity;
}
 

import reactor.bus.Event; //导入依赖的package包/类
@Override
public void accept(Event<GetDiskTypesRequest> getDiskTypesRequestEvent) {
    LOGGER.info("Received event: {}", getDiskTypesRequestEvent);
    GetDiskTypesRequest request = getDiskTypesRequestEvent.getData();
    try {
        Map<Platform, Collection<DiskType>> platformDiskTypes = Maps.newHashMap();
        Map<Platform, DiskType> defaultDiskTypes = Maps.newHashMap();
        Map<Platform, Map<String, VolumeParameterType>> diskMappings = Maps.newHashMap();
        Map<Platform, Map<DiskType, DisplayName>> diskDisplayNames = Maps.newHashMap();

        for (Entry<Platform, Collection<Variant>> connector : cloudPlatformConnectors.getPlatformVariants().getPlatformToVariants().entrySet()) {
            DiskTypes diskTypes = cloudPlatformConnectors.getDefault(connector.getKey()).parameters().diskTypes();
            defaultDiskTypes.put(connector.getKey(), diskTypes.defaultType());
            platformDiskTypes.put(connector.getKey(), diskTypes.types());
            diskMappings.put(connector.getKey(), diskTypes.diskMapping());
            diskDisplayNames.put(connector.getKey(), diskTypes.displayNames());
        }
        GetDiskTypesResult getDiskTypesResult = new GetDiskTypesResult(request,
                new PlatformDisks(platformDiskTypes, defaultDiskTypes, diskMappings, diskDisplayNames));
        request.getResult().onNext(getDiskTypesResult);
        LOGGER.info("Query platform disk types finished.");
    } catch (RuntimeException e) {
        request.getResult().onNext(new GetDiskTypesResult(e.getMessage(), e, request));
    }
}
 

import reactor.bus.Event; //导入依赖的package包/类
@Override
public void accept(Event<CheckImageRequest> event) {
    LOGGER.info("Received event: {}", event);
    CheckImageRequest request = event.getData();
    CloudContext cloudContext = request.getCloudContext();
    try {
        CloudConnector connector = cloudPlatformConnectors.get(request.getCloudContext().getPlatformVariant());
        AuthenticatedContext auth = connector.authentication().authenticate(cloudContext, request.getCloudCredential());
        Image image = request.getImage();
        CloudStack stack = request.getStack();
        ImageStatusResult progress = connector.setup().checkImageStatus(auth, stack, image);
        CheckImageResult imageResult = new CheckImageResult(request, progress.getImageStatus(), progress.getStatusProgressValue());
        request.getResult().onNext(imageResult);
        LOGGER.info("Provision setup finished for {}", cloudContext);
    } catch (RuntimeException e) {
        CheckImageResult failure = new CheckImageResult(e, request, ImageStatus.CREATE_FAILED);
        request.getResult().onNext(failure);
    }
}
 

import reactor.bus.Event; //导入依赖的package包/类
@Override
public void accept(Event<ResourceNotification> event) {
    LOGGER.info("Resource notification event received: {}", event);
    ResourceNotification notification = event.getData();

    RetryUtil.withDefaultRetries()
            .retry(() -> {
                ResourceNotification notificationPersisted;
                switch (notification.getType()) {
                    case CREATE:
                        notificationPersisted = cloudResourcePersisterService.persist(notification);
                        break;
                    case UPDATE:
                        notificationPersisted = cloudResourcePersisterService.update(notification);
                        break;
                    case DELETE:
                        notificationPersisted = cloudResourcePersisterService.delete(notification);
                        break;
                    default:
                        throw new IllegalArgumentException("Unsupported notification type: " + notification.getType());
                }
                notificationPersisted.getPromise().onNext(new ResourcePersisted());
            })
            .checkIfRecoverable(e -> e instanceof TransientDataAccessException)
            .ifNotRecoverable(e -> notification.getPromise().onError(e)).run();
}
 

import reactor.bus.Event; //导入依赖的package包/类
@Override
public void accept(Event<CreateCredentialRequest> credentialRequestEvent) {
    LOGGER.info("Received event: {}", credentialRequestEvent);
    CreateCredentialRequest request = credentialRequestEvent.getData();
    CloudContext cloudContext = request.getCloudContext();
    try {
        CloudConnector connector = cloudPlatformConnectors.get(cloudContext.getPlatformVariant());
        AuthenticatedContext ac = connector.authentication().authenticate(cloudContext, request.getCloudCredential());
        CloudCredentialStatus credentialStatus = connector.credentials().create(ac);
        if (CredentialStatus.FAILED == credentialStatus.getStatus()) {
            if (credentialStatus.getException() != null) {
                throw new CloudConnectorException(credentialStatus.getException());
            }
            throw new CloudConnectorException(credentialStatus.getStatusReason());
        }
        CreateCredentialResult result = new CreateCredentialResult(request);
        request.getResult().onNext(result);
        eventBus.notify(result.selector(), new Event(credentialRequestEvent.getHeaders(), result));
        LOGGER.info("Creating credential successfully finished for {}", cloudContext);
    } catch (RuntimeException e) {
        CreateCredentialResult failure = new CreateCredentialResult(e, request);
        request.getResult().onNext(failure);
        eventBus.notify(failure.selector(), new Event(credentialRequestEvent.getHeaders(), failure));
    }
}
 

import reactor.bus.Event; //导入依赖的package包/类
@Override
public void accept(Event<GetPlatformOrchestratorsRequest> getPlatformOrchestratorsRequest) {
    LOGGER.info("Received event: {}", getPlatformOrchestratorsRequest);
    GetPlatformOrchestratorsRequest request = getPlatformOrchestratorsRequest.getData();
    try {
        Map<Platform, Collection<Orchestrator>> platformCollectionHashMap = Maps.newHashMap();
        Map<Platform, Orchestrator> defaults = Maps.newHashMap();

        for (Entry<Platform, Collection<Variant>> connector : cloudPlatformConnectors.getPlatformVariants().getPlatformToVariants().entrySet()) {
            PlatformOrchestrator platformOrchestrator = cloudPlatformConnectors.getDefault(connector.getKey()).parameters().orchestratorParams();

            platformCollectionHashMap.put(connector.getKey(), platformOrchestrator.types());
            defaults.put(connector.getKey(), platformOrchestrator.defaultType());
        }
        GetPlatformOrchestratorsResult getPlatformOrchestratorsResult = new GetPlatformOrchestratorsResult(request,
                new PlatformOrchestrators(platformCollectionHashMap, defaults));
        request.getResult().onNext(getPlatformOrchestratorsResult);
        LOGGER.info("Query platform orchestrators types finished.");
    } catch (RuntimeException e) {
        request.getResult().onNext(new GetPlatformOrchestratorsResult(e.getMessage(), e, request));
    }
}
 

import reactor.bus.Event; //导入依赖的package包/类
@Override
public void accept(Event<GetPlatformNetworksRequest> getPlatformNetworksRequest) {
    LOGGER.info("Received event: {}", getPlatformNetworksRequest);
    GetPlatformNetworksRequest request = getPlatformNetworksRequest.getData();

    try {
        CloudPlatformVariant cloudPlatformVariant = new CloudPlatformVariant(
                Platform.platform(request.getExtendedCloudCredential().getCloudPlatform()),
                Variant.variant(request.getVariant()));
        CloudNetworks networks = cloudPlatformConnectors.get(cloudPlatformVariant)
                .platformResources()
                .networks(request.getCloudCredential(), Region.region(request.getRegion()), request.getFilters());
        GetPlatformNetworksResult getPlatformNetworksResult = new GetPlatformNetworksResult(request, networks);
        request.getResult().onNext(getPlatformNetworksResult);
        LOGGER.info("Query platform networks types finished.");
    } catch (Exception e) {
        request.getResult().onNext(new GetPlatformNetworksResult(e.getMessage(), e, request));
    }
}
 

import reactor.bus.Event; //导入依赖的package包/类
@Override
public void accept(Event<GetPlatformSshKeysRequest> getPlatformSshKeysRequest) {
    LOGGER.info("Received event: {}", getPlatformSshKeysRequest);
    GetPlatformSshKeysRequest request = getPlatformSshKeysRequest.getData();

    try {
        CloudPlatformVariant cloudPlatformVariant = new CloudPlatformVariant(
                Platform.platform(request.getExtendedCloudCredential().getCloudPlatform()),
                Variant.variant(request.getVariant()));
        CloudSshKeys cloudSshKeys = cloudPlatformConnectors.get(cloudPlatformVariant)
                .platformResources().sshKeys(request.getCloudCredential(), Region.region(request.getRegion()), request.getFilters());
        GetPlatformSshKeysResult getPlatformSshKeysResult = new GetPlatformSshKeysResult(request, cloudSshKeys);
        request.getResult().onNext(getPlatformSshKeysResult);
        LOGGER.info("Query platform networks types finished.");
    } catch (RuntimeException e) {
        request.getResult().onNext(new GetPlatformSshKeysResult(e.getMessage(), e, request));
    }
}
 

import reactor.bus.Event; //导入依赖的package包/类
@Override
public void accept(Event<GetPlatformSecurityGroupsRequest> getPlatformSecurityGroupsRequest) {
    LOGGER.info("Received event: {}", getPlatformSecurityGroupsRequest);
    GetPlatformSecurityGroupsRequest request = getPlatformSecurityGroupsRequest.getData();

    try {
        CloudPlatformVariant cloudPlatformVariant = new CloudPlatformVariant(
                Platform.platform(request.getExtendedCloudCredential().getCloudPlatform()),
                Variant.variant(request.getVariant()));
        CloudSecurityGroups securityGroups = cloudPlatformConnectors.get(cloudPlatformVariant)
                .platformResources()
                .securityGroups(request.getCloudCredential(), Region.region(request.getRegion()), request.getFilters());
        GetPlatformSecurityGroupsResult getPlatformSecurityGroupsResult = new GetPlatformSecurityGroupsResult(request, securityGroups);
        request.getResult().onNext(getPlatformSecurityGroupsResult);
        LOGGER.info("Query platform networks types finished.");
    } catch (Exception e) {
        request.getResult().onNext(new GetPlatformSecurityGroupsResult(e.getMessage(), e, request));
    }
}
 

import reactor.bus.Event; //导入依赖的package包/类
@Override
public void accept(Event<GetPlatformCloudAccessConfigsRequest> getPlatformCloudAccessConfigsRequest) {
    LOGGER.info("Received event: {}", getPlatformCloudAccessConfigsRequest);
    GetPlatformCloudAccessConfigsRequest request = getPlatformCloudAccessConfigsRequest.getData();

    try {
        CloudPlatformVariant cloudPlatformVariant = new CloudPlatformVariant(
                Platform.platform(request.getExtendedCloudCredential().getCloudPlatform()),
                Variant.variant(request.getVariant()));
        CloudAccessConfigs cloudAccessConfigs = cloudPlatformConnectors.get(cloudPlatformVariant)
                .platformResources().accessConfigs(request.getCloudCredential(), Region.region(request.getRegion()), request.getFilters());
        GetPlatformCloudAccessConfigsResult getPlatformCloudAccessConfigsResult = new GetPlatformCloudAccessConfigsResult(request, cloudAccessConfigs);
        request.getResult().onNext(getPlatformCloudAccessConfigsResult);
        LOGGER.info("Query platform access configs finished.");
    } catch (Exception e) {
        request.getResult().onNext(new GetPlatformCloudAccessConfigsResult(e.getMessage(), e, request));
    }
}
 

import reactor.bus.Event; //导入依赖的package包/类
@Override
public void accept(Event<GetPlatformCloudIpPoolsRequest> getPlatformIpPoolsRequest) {
    LOGGER.info("Received event: {}", getPlatformIpPoolsRequest);
    GetPlatformCloudIpPoolsRequest request = getPlatformIpPoolsRequest.getData();

    try {
        CloudPlatformVariant cloudPlatformVariant = new CloudPlatformVariant(
                Platform.platform(request.getExtendedCloudCredential().getCloudPlatform()),
                Variant.variant(request.getVariant()));
        CloudIpPools cloudIpPools = cloudPlatformConnectors.get(cloudPlatformVariant)
                .platformResources().publicIpPool(request.getCloudCredential(), Region.region(request.getRegion()), request.getFilters());
        GetPlatformCloudIpPoolsResult getPlatformIpPoolsResult = new GetPlatformCloudIpPoolsResult(request, cloudIpPools);
        request.getResult().onNext(getPlatformIpPoolsResult);
        LOGGER.info("Query platform ip pool types finished.");
    } catch (Exception e) {
        request.getResult().onNext(new GetPlatformCloudIpPoolsResult(e.getMessage(), e, request));
    }
}
 

import reactor.bus.Event; //导入依赖的package包/类
@Override
public void accept(Event<StartInstancesRequest> event) {
    LOGGER.info("Received event: {}", event);
    StartInstancesRequest request = event.getData();
    CloudContext cloudContext = request.getCloudContext();
    try {
        CloudConnector connector = cloudPlatformConnectors.get(cloudContext.getPlatformVariant());
        AuthenticatedContext authenticatedContext = connector.authentication().authenticate(cloudContext, request.getCloudCredential());
        List<CloudInstance> instances = request.getCloudInstances();
        List<CloudVmInstanceStatus> instanceStatuses = connector.instances().start(authenticatedContext, request.getResources(), instances);
        PollTask<InstancesStatusResult> task = statusCheckFactory.newPollInstanceStateTask(authenticatedContext, instances,
                Sets.newHashSet(InstanceStatus.STARTED, InstanceStatus.FAILED));
        InstancesStatusResult statusResult = new InstancesStatusResult(cloudContext, instanceStatuses);
        if (!task.completed(statusResult)) {
            statusResult = syncPollingScheduler.schedule(task);
        }
        StartInstancesResult result = new StartInstancesResult(request, cloudContext, statusResult);
        request.getResult().onNext(result);
        eventBus.notify(result.selector(), new Event(event.getHeaders(), result));
    } catch (Exception e) {
        StartInstancesResult failure = new StartInstancesResult("Failed to start stack", e, request);
        request.getResult().onNext(failure);
        eventBus.notify(failure.selector(), new Event(event.getHeaders(), failure));
    }
}
 

import reactor.bus.Event; //导入依赖的package包/类
@Override
public void accept(Event<GetInstancesStateRequest> event) {
    LOGGER.info("Received event: {}", event);
    GetInstancesStateRequest request = event.getData();
    CloudContext cloudContext = request.getCloudContext();
    GetInstancesStateResult result;
    try {
        List<CloudVmInstanceStatus> instanceStatuses =
                instanceStateQuery.getCloudVmInstanceStatuses(
                        request.getCloudCredential(), cloudContext, request.getInstances());
        result = new GetInstancesStateResult(request, instanceStatuses);
    } catch (RuntimeException e) {
        result = new GetInstancesStateResult("Instance state synchronizing failed", e, request);
    }
    request.getResult().onNext(result);
    eventBus.notify(result.selector(), new Event(event.getHeaders(), result));
}
 

import reactor.bus.Event; //导入依赖的package包/类
@Override
public void accept(Event<GetVirtualMachineRecommendtaionRequest> gVMRRE) {
    LOGGER.info("Received GetVirtualMachineRecommendtaionRequest event: {}", gVMRRE);
    GetVirtualMachineRecommendtaionRequest request = gVMRRE.getData();
    String cloudPlatform = request.getCloudPlatform();
    try {
        CloudConnector connector = cloudPlatformConnectors.getDefault(Platform.platform(cloudPlatform));
        VmRecommendations recommendations = connector.parameters().recommendedVms();
        GetVirtualMachineRecommendationResponse gVMRResponse = new GetVirtualMachineRecommendationResponse(request, recommendations);
        request.getResult().onNext(gVMRResponse);
        LOGGER.info("Query platform machine recommendations finished.");
    } catch (RuntimeException e) {
        LOGGER.warn("Could not get virtual machine recommendations for platform: " + cloudPlatform, e);
        request.getResult().onNext(new GetVirtualMachineRecommendationResponse(e.getMessage(), e, request));
    }
}
 

import reactor.bus.Event; //导入依赖的package包/类
@Override
public void accept(Event<ResourceDefinitionRequest> getRegionsRequestEvent) {
    LOGGER.info("Received event: {}", getRegionsRequestEvent);
    ResourceDefinitionRequest request = getRegionsRequestEvent.getData();
    try {
        CloudConnector connector = cloudPlatformConnectors.get(request.getPlatform());
        String resource = request.getResource();
        String definition = connector.parameters().resourceDefinition(request.getResource());
        if (definition == null) {
            Exception exception = new Exception("Failed to find resource definition for " + resource);
            request.getResult().onNext(new ResourceDefinitionResult(exception.getMessage(), exception, request));
        } else {
            request.getResult().onNext(new ResourceDefinitionResult(request, definition));
        }
    } catch (RuntimeException e) {
        request.getResult().onNext(new ResourceDefinitionResult(e.getMessage(), e, request));
    }
}
 

import reactor.bus.Event; //导入依赖的package包/类
@Override
public void accept(Event<SetupRequest> event) {
    LOGGER.info("Received event: {}", event);
    SetupRequest request = event.getData();
    CloudContext cloudContext = request.getCloudContext();
    SetupResult result;
    try {
        CloudConnector connector = cloudPlatformConnectors.get(cloudContext.getPlatformVariant());
        AuthenticatedContext auth = connector.authentication().authenticate(cloudContext, request.getCloudCredential());
        CloudStack cloudStack = request.getCloudStack();
        connector.setup().prerequisites(auth, cloudStack, resourceNotifier);
        result = new SetupResult(request);

        LOGGER.info("Provision setup finished for {}", cloudContext);
    } catch (RuntimeException e) {
        result = new SetupResult(e, request);
    }
    request.getResult().onNext(result);
    eventBus.notify(result.selector(), new Event(event.getHeaders(), result));
}
 

import reactor.bus.Event; //导入依赖的package包/类
@Override
public void accept(Event<DownscaleStackCollectResourcesRequest> collectResourcesRequestEvent) {
    LOGGER.info("Received event: {}", collectResourcesRequestEvent);
    DownscaleStackCollectResourcesRequest request = collectResourcesRequestEvent.getData();
    DownscaleStackCollectResourcesResult result;
    try {
        CloudContext cloudContext = request.getCloudContext();
        CloudConnector connector = cloudPlatformConnectors.get(cloudContext.getPlatformVariant());
        AuthenticatedContext ac = connector.authentication().authenticate(cloudContext, request.getCloudCredential());
        Object resourcesToScale = connector.resources().collectResourcesToRemove(ac, request.getCloudStack(),
                request.getCloudResources(), request.getInstances());
        LOGGER.info("Collect resources successfully finished for {}", cloudContext);
        result = new DownscaleStackCollectResourcesResult(request, resourcesToScale);
    } catch (RuntimeException e) {
        LOGGER.error("Failed to handle DownscaleStackCollectResourcesRequest.", e);
        result = new DownscaleStackCollectResourcesResult(e.getMessage(), e, request);
    }
    request.getResult().onNext(result);
    LOGGER.info("DownscaleStackCollectResourcesRequest finished");
    eventBus.notify(result.selector(), new Event(collectResourcesRequestEvent.getHeaders(), result));
}
 

import reactor.bus.Event; //导入依赖的package包/类
@Override
public void accept(Event<GetPlatformTemplateRequest> platformTemplateRequestEvent) {
    LOGGER.info("Received event: {}", platformTemplateRequestEvent);
    GetPlatformTemplateRequest request = platformTemplateRequestEvent.getData();
    String template = null;
    try {
        CloudConnector connector = cloudPlatformConnectors.get(request.getCloudContext().getPlatformVariant());
        if (connector != null) {
            template = connector.resources().getStackTemplate();
        }
    } catch (TemplatingDoesNotSupportedException ignored) {
    }
    GetPlatformTemplateResult getPlatformTemplateResult = new GetPlatformTemplateResult(request, template);
    request.getResult().onNext(getPlatformTemplateResult);
    LOGGER.info("Get template finished.");
}
 

import reactor.bus.Event; //导入依赖的package包/类
@Override
public void accept(Event<PlatformParametersRequest> platformParameterRequestEvent) {
    LOGGER.info("Received event: {}", platformParameterRequestEvent);
    PlatformParametersRequest request = platformParameterRequestEvent.getData();
    Map<Platform, PlatformParameters> platformParameters = new HashMap<>();
    try {
        for (Entry<Platform, Collection<Variant>> connector : cloudPlatformConnectors.getPlatformVariants().getPlatformToVariants().entrySet()) {
            platformParameters.put(connector.getKey(), cloudPlatformConnectors.getDefault(connector.getKey()).parameters());
        }
        PlatformParametersResult platformParameterResult = new PlatformParametersResult(request, platformParameters);
        request.getResult().onNext(platformParameterResult);
        LOGGER.info("Query platform parameters finished.");
    } catch (RuntimeException e) {
        request.getResult().onNext(new PlatformParametersResult(e.getMessage(), e, request));
    }
}