Java 类org.springframework.scheduling.annotation.AsyncResult 实例源码

@Async
@Override
public Future<List<GetPrizeDTO>> scheduleItems(ScheduleItem item) throws InterruptedException {
    log.info("Start Schedule with : " +item.getRecipientID());
    log.info("query Type " + item.getQueryType());
    Future<List<GetPrizeDTO>> result = new AsyncResult<>(new ArrayList<>());
    if(item.getQueryType() == ConstantUtil.NORMAL_QUERY) {
        result = new AsyncResult<>(resultService.findPrizeByResultType(item.getLotteryType(), item.getParam().toArray(new String[]{})));
    } else if(item.getQueryType() == ConstantUtil.CODE_RANGE_QUERY) {
        result = new AsyncResult<>(resultService.findPrizesByCode(item.getParam().get(0), item.getParam().get(1), item.getParam().get(2), item.getLotteryType()));
    } else if(item.getQueryType() == ConstantUtil.POINT_RANGE_QUERY) {
        result = new AsyncResult<>(resultService.findPrizesByPoints(item.getParam().get(0), item.getParam().get(1), item.getParam().get(2), item.getLotteryType()));
    }
    // remove from db after finding result.
    deleteScheduleItem(item.getRecipientID());
    return result;
}

/**
 * 查询用户在线状态
 * 
 * @param fromUserId 用户ID
 * @param userIdList 查询列表
 * @return
 * @since  1.0
 */
@Async
public ListenableFuture<List<IMBaseDefine.UserStat>> userStatusReq(Long fromUserId, List<Long> userIdList) {

    logger.debug("查询用户在线状态, user_cnt={}", userIdList.size());

    List<IMBaseDefine.UserStat> userStatList = new ArrayList<>();
    for (Long userId: userIdList) {

        UserClientInfoManager.UserClientInfo userClientInfo = userClientInfoManager.getUserInfo(userId);
        IMBaseDefine.UserStat.Builder userStatBuiler = IMBaseDefine.UserStat.newBuilder();
        userStatBuiler.setUserId(userId);
        if (userClientInfo != null) {
            userStatBuiler.setStatus(userClientInfo.getStatus());
        } else {
            userStatBuiler.setStatus(IMBaseDefine.UserStatType.USER_STATUS_OFFLINE);
        }

        userStatList.add(userStatBuiler.build());
    }

    AsyncResult<List<IMBaseDefine.UserStat>> result = new AsyncResult<>(userStatList);
    return result;
}

/**
 * 异步执行,需要返回的Future<>类型
 *
 * @param name
 * @return
 */
@Async
public Future<RxJavaDTO> getRxJavaDTO(String name) {
    try {
        Thread.sleep(100);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    log.info("common service begin to process");
    RxJavaDTO item = new RxJavaDTO();
    item.setName(name);
    String value = MDC.get(MdcConstans.MDC_REMOTE_IP);
    if (!StringUtils.isEmpty(value)) {
        log.info("remoteid id " + value);
    } else {
        log.info("remoteid id is empty");
    }
    value = MDC.get(MdcConstans.MDC_ClientRequest_ID);
    if (!StringUtils.isEmpty(value)) {
        log.info("client id " + value);
    } else {
        log.info("client id is empty");
    }
    log.info("common service end to process");
    return new AsyncResult<>(item);
}

@Override
public Future<List<Event>> execute(Event event, Device device) {

    List<Event> outEvents = new ArrayList<>();

    ServiceResponse<List<EventRoute>> serviceRoutes = eventRouteService.getAll(device.getTenant(), device.getApplication());
    if (!serviceRoutes.isOk()) {
        LOGGER.error("Error listing application events routes", device.toURI(), device.getTenant().getLogLevel());
        return new AsyncResult<>(outEvents);
    }

    List<EventRoute> eventRoutes = serviceRoutes.getResult();
    if (eventRoutes.isEmpty()) {
        return new AsyncResult<>(outEvents);
    }

    eventRoutes.parallelStream().forEach((eventRoute) ->
        processEventRoute(event, device, outEvents, eventRoute)
    );

    return new AsyncResult<>(outEvents);
}

@Async
@Transactional(propagation = Propagation.REQUIRES_NEW)
public Future<Boolean> runTask(Integer jobId) throws AsyncTaskException {
    Job job = this.jobRepo.findOne(jobId);
    try {
        this.restorer.restore(job.getDbDumperServiceInstance(), job.getDatabaseRefTarget(), job.getDumpDate());
    } catch (RestoreException e) {
        logger.error(String.format("Cannot restore dump for '%s' in '%s': %s", job.getDatabaseRefSrc().getDatabaseName(), job.getDatabaseRefTarget().getDatabaseName(), e.getMessage()));
        job.setJobEvent(JobEvent.ERRORED);
        job.setErrorMessage(e.getMessage());
        this.databaseRefManager.deleteServiceKey(job);
        jobRepo.save(job);
        return new AsyncResult<Boolean>(false);
    }
    this.databaseRefManager.deleteServiceKey(job);
    job.setJobEvent(JobEvent.FINISHED);
    jobRepo.save(job);
    return new AsyncResult<Boolean>(true);
}

@Override
@Async
public Future<Void> processBusinessObjectDataNotificationEventAsync(NotificationEventTypeEntity.EventTypesBdata notificationEventType,
    BusinessObjectDataKey businessObjectDataKey, String newBusinessObjectDataStatus, String oldBusinessObjectDataStatus)
{
    /*
     * Need to clear the security context here since the current thread may have been reused, which may might have left over its security context. If we do
     * not clear the security context, any subsequent calls may be restricted by the permissions given to the previous thread's security context.
     */
    SecurityContextHolder.clearContext();

    processBusinessObjectDataNotificationEventSync(notificationEventType, businessObjectDataKey, newBusinessObjectDataStatus, oldBusinessObjectDataStatus);

    // Return an AsyncResult so callers will know the future is "done". They can call "isDone" to know when this method has completed and they
    // can call "get" to see if any exceptions were thrown.
    return new AsyncResult<>(null);
}

@Override
@Async
public Future<Void> processStorageUnitNotificationEventAsync(NotificationEventTypeEntity.EventTypesStorageUnit notificationEventType,
    BusinessObjectDataKey businessObjectDataKey, String storageName, String newStorageUnitStatus, String oldStorageUnitStatus)
{
    /*
     * Need to clear the security context here since the current thread may have been reused, which may might have left over its security context. If we do
     * not clear the security context, any subsequent calls may be restricted by the permissions given to the previous thread's security context.
     */
    SecurityContextHolder.clearContext();

    processStorageUnitNotificationEventSync(notificationEventType, businessObjectDataKey, storageName, newStorageUnitStatus, oldStorageUnitStatus);

    // Return an AsyncResult so callers will know the future is "done". They can call "isDone" to know when this method has completed and they
    // can call "get" to see if any exceptions were thrown.
    return new AsyncResult<>(null);
}

@Override
@Async
public Future<Void> indexValidateAllTags(String indexName)
{
    final String documentType = configurationHelper.getProperty(ConfigurationValue.ELASTICSEARCH_BDEF_DOCUMENT_TYPE, String.class);

    // Get a list of all tags
    final List<TagEntity> tagEntityList = Collections.unmodifiableList(tagDao.getTags());

    // Remove any index documents that are not in the database
    removeAnyIndexDocumentsThatAreNotInTagsList(indexName, documentType, tagEntityList);

    // Validate all Tags
    tagHelper.executeFunctionForTagEntities(indexName, documentType, tagEntityList, indexFunctionsDao::validateDocumentIndex);

    // Return an AsyncResult so callers will know the future is "done". They can call "isDone" to know when this method has completed and they
    // can call "get" to see if any exceptions were thrown.
    return new AsyncResult<>(null);
}

@Override
@Async
public Future<Void> indexValidateAllBusinessObjectDefinitions(String indexName)
{
    final String documentType = configurationHelper.getProperty(ConfigurationValue.ELASTICSEARCH_BDEF_DOCUMENT_TYPE, String.class);

    // Get a list of all business object definitions
    final List<BusinessObjectDefinitionEntity> businessObjectDefinitionEntityList =
        Collections.unmodifiableList(businessObjectDefinitionDao.getAllBusinessObjectDefinitions());

    // Remove any index documents that are not in the database
    removeAnyIndexDocumentsThatAreNotInBusinessObjectsDefinitionsList(indexName, documentType, businessObjectDefinitionEntityList);

    // Validate all Business Object Definitions
    businessObjectDefinitionHelper.executeFunctionForBusinessObjectDefinitionEntities(indexName, documentType, businessObjectDefinitionEntityList,
        indexFunctionsDao::validateDocumentIndex);

    // Return an AsyncResult so callers will know the future is "done". They can call "isDone" to know when this method has completed and they
    // can call "get" to see if any exceptions were thrown.
    return new AsyncResult<>(null);
}

@Override
@Async
public Future<Void> indexAllTags(SearchIndexKey searchIndexKey, String documentType)
{
    // Get a list of all tags
    final List<TagEntity> tagEntities = Collections.unmodifiableList(tagDao.getTags());

    // Index all tags.
    tagHelper.executeFunctionForTagEntities(searchIndexKey.getSearchIndexName(), documentType, tagEntities, indexFunctionsDao::createIndexDocument);

    // Simple count validation, index size should equal entity list size.
    validateSearchIndexSize(searchIndexKey.getSearchIndexName(), documentType, tagEntities.size());

    // Update search index status to READY.
    searchIndexDaoHelper.updateSearchIndexStatus(searchIndexKey, SearchIndexStatusEntity.SearchIndexStatuses.READY.name());

    // Return an AsyncResult so callers will know the future is "done". They can call "isDone" to know when this method has completed and they can call
    // "get" to see if any exceptions were thrown.
    return new AsyncResult<>(null);
}

@Test
public void createVimInstance()
    throws VimException, PluginException, IOException, BadRequestException,
        AlreadyExistingException, ExecutionException, InterruptedException {
  OpenstackVimInstance datacenter = new OpenstackVimInstance();
  datacenter.setId("123");
  datacenter.setName("DC-1");
  datacenter.setType("OpenStack");
  datacenter.setUsername("datacenter_test");
  datacenter.setTenant("tenant");
  datacenter.setKeyPair("keypair");
  datacenter.setPassword("");
  when(mock.add(any(datacenter.getClass()), anyString()))
      .thenReturn(new AsyncResult<>(datacenter));
  log.info("" + restVimInstances.create(datacenter, "pi"));
  BaseVimInstance datacenter2 = restVimInstances.create(datacenter, "pi");
  assertEquals(datacenter, datacenter2);
}

@Override
@Async
public Future<NFVMessage> requestLog(VirtualNetworkFunctionRecord vnfr, String hostname)
    throws NotFoundException, BadFormatException, ExecutionException, InterruptedException {
  VnfmManagerEndpoint endpoint = generator.getVnfm(vnfr.getEndpoint());
  if (endpoint == null)
    throw new NotFoundException(
        "VnfManager of type "
            + vnfr.getType()
            + " (endpoint = "
            + vnfr.getEndpoint()
            + ") is not registered");

  OrVnfmLogMessage orVnfmLogMessage = new OrVnfmLogMessage(vnfr.getName(), hostname);
  VnfmSender vnfmSender;
  try {
    vnfmSender = generator.getVnfmSender(endpoint.getEndpointType());
  } catch (BeansException e) {
    throw new NotFoundException(e);
  }
  Future<NFVMessage> answerFuture = vnfmSender.sendCommand(orVnfmLogMessage, endpoint);
  answerFuture.get();
  NFVMessage message = answerFuture.get();
  return new AsyncResult<>(message);
}

@Override
@Async
public Future<Void> handleVNF(
    NetworkServiceDescriptor networkServiceDescriptor,
    NetworkServiceRecord networkServiceRecord,
    DeployNSRBody body,
    Map<String, Set<String>> vduVimInstances,
    VirtualNetworkFunctionDescriptor vnfd,
    String monitoringIp)
    throws NotFoundException, BadFormatException, ExecutionException, InterruptedException {
  log.debug(
      "Processing VNFD ("
          + vnfd.getName()
          + ") for NSD ("
          + networkServiceDescriptor.getName()
          + ")");

  VnfmSender vnfmSender = generator.getVnfmSender(vnfd);
  NFVMessage message =
      generator.getNextMessage(vnfd, vduVimInstances, networkServiceRecord, body, monitoringIp);
  VnfmManagerEndpoint endpoint = generator.getEndpoint(vnfd);
  log.debug("----------Executing ACTION: " + message.getAction());
  executeAction(vnfmSender.sendCommand(message, endpoint));
  log.info("Sent " + message.getAction() + " to VNF: " + vnfd.getName());
  return new AsyncResult<>(null);
}

@Override
@Async
public Future<Void> send(EventEndpoint endpoint, final ApplicationEventNFVO event) {

  log.debug("Sending message: " + event + " to endpoint: " + endpoint);
  log.info("Sending message: " + event.getAction() + " to endpoint: " + endpoint.getName());
  final String json =
      "{\"action\":\""
          + event.getAction()
          + "\",\"payload\":"
          + new Gson().toJson(event.getPayload())
          + "}";
  log.trace("Event body is: " + json);
  rabbitTemplate.convertAndSend(endpoint.getEndpoint(), json);

  return new AsyncResult<>(null);
}

@Test
public void nsrManagementDeleteTest()
    throws VimException, InterruptedException, ExecutionException, NamingException,
        NotFoundException, WrongStatusException, PluginException, BadFormatException {
  NetworkServiceRecord nsd_exp = createNetworkServiceRecord();
  when(resourceManagement.release(any(VirtualDeploymentUnit.class), any(VNFCInstance.class)))
      .thenReturn(new AsyncResult<Void>(null));
  when(nsrRepository.findFirstByIdAndProjectId(nsd_exp.getId(), projectId)).thenReturn(nsd_exp);
  Configuration system = new Configuration();
  system.setConfigurationParameters(new HashSet<>());
  ConfigurationParameter configurationParameter = new ConfigurationParameter();
  configurationParameter.setConfKey("delete-on-all-status");
  configurationParameter.setValue("true");
  when(configurationManagement.queryByName("system")).thenReturn(system);
  nsrManagement.delete(nsd_exp.getId(), projectId);
}

@Async
public Future<Map<String, Object>> getCredentialsFromSensors(String application, String entity,
        Predicate<? super String> sensorWhitelist,
        Predicate<? super String> sensorBlacklist,
        Predicate<? super String> entityWhitelist,
        Predicate<? super String> entityBlacklist) {

    List<EntitySummary> entities = getRestApi().getEntityApi().getChildren(application, entity);
    if (entities.size() == 0) {
        return new AsyncResult<>(getEntitySensors(application, entity, sensorWhitelist, sensorBlacklist, entityWhitelist, entityBlacklist));
    } else if (entities.size() == 1) {
        String entityId = entities.get(0).getId();
        return new AsyncResult<>(getEntitySensors(application, entityId, sensorWhitelist, sensorBlacklist, entityWhitelist, entityBlacklist));
    }
    return new AsyncResult<>(getApplicationSensors(application, entities, sensorWhitelist, sensorBlacklist, entityWhitelist, entityBlacklist));
}

@Async
public Future<String> getDashboardUrl(String application) {
    // search in breadth first order for first sensor that matches
    List<EntitySummary> entities = getRestApi().getEntityApi().list(application);
    Deque<EntitySummary> q = new ArrayDeque<>(entities);
    while (!q.isEmpty()) {
        EntitySummary e = q.remove();
        List<SensorSummary> sensors = getRestApi().getSensorApi().list(application, e.getId());
        for (SensorSummary sensor : sensors) {
            if (sensor.getName().equals("management.url")) {
                String url = String.valueOf(getRestApi().getSensorApi().get(application, e.getId(), sensor.getName(), false));
                LOG.info("found dashboard url={} for application={}", url, application);
                return new AsyncResult<>(url);
            }
        }
        q.addAll(getRestApi().getEntityApi().getChildren(application, e.getId()));
    }

    LOG.info("no dashboard url found for application={}", application);
    return new AsyncResult<>(null);
}

@Async
public Future<Map<String, Object>> getConfigAsMap(String application, String entity, String key) {
    Object object;
    try {
        object = getRestApi().getEntityConfigApi().get(application, entity, key, false);
    } catch (Exception e) {
        LOG.error("Unable to get config with key={}", key);
        return new AsyncResult<>(null);
    }

    if (object == null || !(object instanceof Map)) {
        LOG.error("Unable to get Map with key={}", key);
        return new AsyncResult<>(null);
    }
    Map<String, Object> map = (Map<String, Object>) object;
    return new AsyncResult<>(map);
}

@Test
public void newServiceInstanceCreatedSuccessfully() 
        throws ServiceInstanceExistsException, ServiceBrokerException {

    when(admin.createApplication(any(String.class))).thenReturn(new AsyncResult<>(entity));
    when(catalogService.getServiceDefinition(any(String.class))).thenReturn(serviceDefinition);
       when(serviceDefinition.getPlans()).thenReturn(ImmutableList.of(new DefaultBlueprintPlan("planId", "test_name", "test_description","Test App", ImmutableMap.of("location", "test_location"))));
       when(serviceDefinition.getId()).thenReturn(SVC_DEFINITION_ID);
       when(admin.getDashboardUrl(any(String.class))).thenReturn(new AsyncResult<>(null));

    CreateServiceInstanceRequest request = new CreateServiceInstanceRequest(serviceDefinition.getId(), "planId", "organizationGuid", "spaceGuid");
    CreateServiceInstanceResponse instance = service.createServiceInstance(request.withServiceInstanceId(SVC_INST_ID));

    assertNotNull(instance);
    // TODO: assert service instance created successfully
    // assertEquals(SVC_INST_ID, instance.getServiceInstanceId());
}

@Test
public void newServiceInstanceBindingCreatedSuccessfully()
        throws ServiceBrokerException, ServiceInstanceBindingExistsException {

   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<>(false));
   when(instanceRepository.findOne(anyString(), anyBoolean())).thenReturn(serviceInstance);
   when(serviceDefinition.getMetadata()).thenReturn(ImmutableMap.of());
   when(brooklynCatalogService.getServiceDefinition(anyString())).thenReturn(serviceDefinition);
   when(serviceInstance.getEntityId()).thenReturn("entityId");
   CreateServiceInstanceBindingRequest request = new CreateServiceInstanceBindingRequest(serviceInstance.getServiceDefinitionId(), "planId", "appGuid", null);
   CreateServiceInstanceBindingResponse binding = bindingService.createServiceInstanceBinding(request.withBindingId(SVC_INST_BIND_ID));

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

@Test
public void testMetadataFromBlueprint() {
    when(admin.getCatalogApplications(Mockito.anyBoolean())).thenReturn(new AsyncResult<>(Arrays.asList(TEST_SUMMARY_WITH_METADATA)));
    when(brooklynConfig.includesAllCatalogVersions()).thenReturn(false);
    List<ServiceDefinition> serviceDefinitions = catalogPlanStrategy.makeServiceDefinitions();
    String expectedKey = "test";
    String expectedValue = "test value";
    Map<String, Object> metadata = serviceDefinitions.get(1).getMetadata();
    assertTrue(metadata.containsKey(expectedKey));
    assertEquals(expectedValue, metadata.get(expectedKey));

    expectedKey = "brooklynCatalogId";
    expectedValue = TEST_SUMMARY_WITH_METADATA.getId();
    assertTrue(metadata.containsKey(expectedKey));
    assertEquals(expectedValue, metadata.get(expectedKey));
}

@Async
private Future<Void> sendEmail(String[] to, String subject, String content, boolean isMultipart,
    boolean isHtml) {
  log.debug("Send e-mail[multipart '{}' and html '{}'] to '{}' with subject '{}' and content={}",
      isMultipart, isHtml, to, subject, content);

  // Prepare message using a Spring helper
  MimeMessage mimeMessage = javaMailSender.createMimeMessage();
  try {
    MimeMessageHelper message =
        new MimeMessageHelper(mimeMessage, isMultipart, CharEncoding.UTF_8);
    message.setTo(to);
    message.setFrom(jhipsterProperties.getMail()
        .getFrom());
    message.setSubject(subject);
    message.setText(content, isHtml);
    javaMailSender.send(mimeMessage);

    log.debug("Sent e-mail to users '{}'", Arrays.toString(to));
  } catch (MessagingException e) {
    log.warn("E-mail could not be sent to users '{}', exception is: {}", 
        Arrays.toString(to), e.getMessage());
  }

  return new AsyncResult<>(null);
}

@Async
@Transactional(propagation = Propagation.REQUIRES_NEW)
public Future<String> processPersonBio(Person person, OrcidOAuthClient clientOrcid){
    List<RelPersonApplication> listRelPersonApp = relPersonApplicationRepository.findAllByPersonIsAndLastIsTrueAndOauthAccessTokenIsNotNull(person);
    OrcidBio orcidBio = null;
    for(RelPersonApplication relPersonApplication: listRelPersonApp){
        String orcid = relPersonApplication.getPerson().getOrcid();
        OrcidAccessToken orcidAccessToken = new OrcidAccessToken();
        orcidAccessToken.setOrcid(orcid);
        orcidAccessToken.setAccess_token(relPersonApplication.getOauthAccessToken());
        try {
            orcidBio = clientOrcid.getOrcidBio(orcidAccessToken);

            manageOrcidBio(person, orcidBio);

            log.info(String.format("Method processPersonBio: save personBio person.id=[%s]", person.getId()));
            break;
        } catch (Exception e) {
            e.printStackTrace();
            log.info(String.format("Method processPersonBio: error personBio person.id=[%s], token=[%s], orcid=[%s]", person.getId(), relPersonApplication.getOauthAccessToken(), orcid));
        }
    }
    person.setNeedUpdate(false);
    personRepository.save(person);
    return new AsyncResult<String>("OK");
}

@Async
public Future<GerritReviewStatsResult> populateReviewStatsAsync(final String changeStatus,
        final List<ChangeInfo> noPeerReviewList, final List<ChangeInfo> onePeerReviewList,
        final List<ChangeInfo> twoPlusPeerReviewList, final List<ChangeInfo> collabrativeDevelopmentList,
        final List<ChangeInfo> changes) throws IOException, URISyntaxException {
    LOGGER.info("Starting Thread To Process Changes");
    GerritReviewStatsResult result = null;
    try {
        populateReviewStats(changeStatus, noPeerReviewList, onePeerReviewList, twoPlusPeerReviewList,
                collabrativeDevelopmentList, changes);
        result = new GerritReviewStatsResult(true, changes);
    } catch(Exception e) {
        LOGGER.info("CAUGHT EXCEPTION");
        result = new GerritReviewStatsResult(false, e, changes);
    }
    LOGGER.info("Thread Finished");
    return new AsyncResult<GerritReviewStatsResult>(result);
}

public Future<LineChartStatisticsResult> getCountOfRegisteredUsersByTimeRange(TimeRangedStatisticsRequest request) {

        final List<UserEntity> userEntityList = userRepository.findByCreatedDateBetween(request.getStartDate(), request.getEndDate());

        final Map<String, Long> map = userEntityList.stream().collect(Collectors.groupingBy(
                p -> formatDate(p.getCreatedDate()),
                Collectors.reducing(
                        0L,
                        t -> 1L,
                        Long::sum
                )
        ));

        final Map<String, Long> resultMap = fillMap(
                getDefaultMap(request),
                map
        );

        final LineChartStatisticsResult result = new LineChartStatisticsResult(SUM_REGISTERED_USER.getDisplayName(), resultMap);

        return new AsyncResult<>(result);
    }

public Future<LineChartStatisticsResult> getCreatedProjectSumByTimeRange(TimeRangedStatisticsRequest request) {

        final List<ProjectEntity> projectEntityList = projectRepository.findByCreatedDateBetween(request.getStartDate(), request.getEndDate());

        final Map<String, Long> map = projectEntityList.stream().collect(Collectors.groupingBy(
                p -> formatDate(p.getCreatedDate()),
                Collectors.reducing(
                        0L,
                        t -> 1L,
                        Long::sum
                )
        ));

        final Map<String, Long> resultMap = fillMap(
                getDefaultMap(request),
                map
        );

        final LineChartStatisticsResult result = new LineChartStatisticsResult(SUM_CREATED_PROJECT.getDisplayName(), resultMap);

        return new AsyncResult<>(result);
    }

/**
 * Asynchroner Empfang von Telegrammrohdaten aus dem InputStream in
 *
 * @param in (Inputstream)
 * @return response
 * in ein AsyncResult<byte[]> verpackte Telegrammrohdaten
 * @throws IOException
 */
@Async
Future<byte[]> parseConnection(InputStream in) throws IOException {
    byte[] response = new byte[TelegramPart.RAW_DATA.maxLength()];
    int pos = 0;
    while (pos < 3) {
        //read one byte and look whether it is 0xFF, which marks beginning of a new Telegram
        in.read(response, pos, 1);
        //throw away invalid data
        if (response[pos] != ByteConversions.toUByte(0xFF)) {
            //reset telegram, start again
            if (pos > 0) {
                pos = 0;
            }
            continue;
        }
        pos++;
    }
    //read length byte
    in.read(response, pos, 1);
    int length = ByteConversions.toUInt(response[pos]);
    //read telegram data
    in.read(response, ++pos, length);
    return new AsyncResult<>(response);
}

@Async
public Future<Iterable<PoliceTicker>> execute(final Iterable<String> detailURLs) {
    final Stopwatch watch = Stopwatch.createStarted();
    logger.info("Start crawling detail pages");
    final List<PoliceTicker> policeTickers = new ArrayList<>();
    for (final Iterator<String> iterator = detailURLs.iterator(); iterator.hasNext();) {
        final PoliceTicker ticker = crawl(iterator.next());
        if (ticker != null) {
            policeTickers.add(ticker);
        }
        if (iterator.hasNext()) {
            try {
                Thread.sleep(WAIT_BEFORE_EACH_ACCESS_TO_PREVENT_BANNING);
            } catch (final InterruptedException e) {
                logger.error(e.toString(), e);
            }
        }
    }
    watch.stop();
    logger.info("Finished crawling {} detail pages in {} ms", policeTickers.size(), watch.elapsed(TimeUnit.MILLISECONDS));
    return new AsyncResult<>(policeTickers);
}

@Async
private Future<Math.SummaryStatisticsDto> getVariableFacet(HarmonizationDataset dataset, String variableName,
  OpalTable table) {
  try {
    String studyId = null;

    if (table instanceof StudyTable) {
      studyId = ((StudyTable)table).getStudyId();
    } else if (table instanceof HarmonizationStudyTable) {
      studyId = ((HarmonizationStudyTable)table).getStudyId();
    }

    return new AsyncResult<>(datasetService
      .getVariableSummary(dataset, variableName, studyId, table.getProject(), table.getTable())
      .getWrappedDto());
  } catch(Exception e) {
    log.warn("Unable to retrieve statistics: " + e.getMessage(), e);
    return new AsyncResult<>(null);
  }
}

@Test
public void testPopulateHarmonizedVariablesMap() {
  List<DatasetVariable> l = new ArrayList<DatasetVariable>() {
    {
      add(new DatasetVariable(dataset, Variable.Builder.newVariable("v1", BooleanType.get(), "test").build(), st));
      add(new DatasetVariable(dataset, Variable.Builder.newVariable("v2", BooleanType.get(), "test").build(), st2));
    }};

  doReturn(dataset).when(datasetService).findById(anyString());
  when(helper.asyncGetDatasetVariables(any(Supplier.class))).thenReturn(new AsyncResult<>(l));
  doReturn(l).when(datasetService).getDatasetVariables(any(HarmonizationDataset.class));
  doReturn(l).when(datasetService).getDatasetVariables(any(HarmonizationDataset.class), any(StudyTable.class));

  Map<String, List<DatasetVariable>> res = datasetService.populateHarmonizedVariablesMap(dataset);

  assertEquals(2, res.keySet().size());
  assertEquals(2, res.get("testds:v1:Dataschema").size());
  assertEquals(2, res.get("testds:v2:Dataschema").size());
}

@Test
public void get_witValueLoaderAndNonExistingValue_createsValueFromValueLoaderAndStoresItInCache() throws Exception {

    //Arrange
    MemcachedClientIF client = mock(MemcachedClientIF.class);
    SimpleSpringMemcached cache = new SimpleSpringMemcached(client, "test");
    cache.setExpiration(42);

    when(client.set("myKey", 42, "createdValue")).thenReturn(new AsyncResult<>(true));

    //Act
    String value = cache.get("myKey", () -> "createdValue");

    //Assert
    assertEquals("createdValue", value);
}

@Async
public Future<Integer> test(int i, int size, Date start) {
    try
    {
        Thread.sleep(33);
    }
    catch(Exception e)
    {
        e.printStackTrace();
    }
    if (0 == i %2) {
        throw new RuntimeException("test");
    }
    System.out.println(i);
    return new AsyncResult<Integer>(i);
}

@Async("myTaskAsyncPool")
public Future<Integer> doAsyncTask(String tableName, List<EventData> dataList, String destination) {
    try {
        MDC.put("destination", destination);
        logger.info("thread: " + Thread.currentThread().getName() + " is doing job :" + tableName);
        for (EventData eventData : dataList) {
            SpringUtil.doEvent(eventData.getPath(), eventData.getDbObject());
        }
    } catch (Exception e) {
        logger.error("thread:" + Thread.currentThread().getName() + " get Exception", e);
        return new AsyncResult(0);
    }
    return new AsyncResult(1);
}

@Async
public Future<Employee> readEmployee(Integer empId) {
    try {
        System.out.println("service:readEmployee(empid) task executor: " + Thread.currentThread().getName());
        System.out.println("processing for 2000 ms");
        System.out.println("readEmployee @Async login: " + SecurityContextHolder.getContext().getAuthentication().getPrincipal());
        Thread.sleep(2000);
    } catch (InterruptedException e) {

        e.printStackTrace();
    }
    return new AsyncResult<>(employeeDaoImpl.getEmployee(empId));
}

@Async
public Future<Employee> readEmployee(Integer empId) {
    try {
        System.out.println("service:readEmployee(empid) task executor: " + Thread.currentThread().getName());
        System.out.println("processing for 2000 ms");
        System.out.println("readEmployee @Async login: " + SecurityContextHolder.getContext().getAuthentication().getPrincipal());
        Thread.sleep(2000);
    } catch (InterruptedException e) {

        e.printStackTrace();
    }
    return new AsyncResult<>(employeeDaoImpl.getEmployee(empId));
}

@Async
public Future<Employee> readEmployee(Integer empId) {
    try {
        System.out.println("service:readEmployee(empid) task executor: " + Thread.currentThread().getName());
        System.out.println("processing for 2000 ms");
        System.out.println("readEmployee @Async login: " + SecurityContextHolder.getContext().getAuthentication().getPrincipal());
        Thread.sleep(2000);
    } catch (InterruptedException e) {

        e.printStackTrace();
    }
    return new AsyncResult<>(employeeDaoImpl.getEmployee(empId));
}

@Async
 Future<String> doTaskOne() throws Exception {
    System.out.println("开始做任务一");
    long start = System.currentTimeMillis();
    Thread.sleep(random.nextInt(10000));
    long end = System.currentTimeMillis();
    System.out.println("完成任务一，耗时：" + (end - start) + "毫秒");
    return new AsyncResult<>("任务一OK");
}

@Async
Future<String> doTaskTwo() throws Exception {
    System.out.println("开始做任务二");
    long start = System.currentTimeMillis();
    Thread.sleep(random.nextInt(10000));
    long end = System.currentTimeMillis();
    System.out.println("完成任务二，耗时：" + (end - start) + "毫秒");
    return new AsyncResult<>("任务二OK");
}

@Async
Future<String> doTaskThree() throws Exception {
    System.out.println("开始做任务三");
    long start = System.currentTimeMillis();
    Thread.sleep(random.nextInt(10000));
    long end = System.currentTimeMillis();
    System.out.println("完成任务三，耗时：" + (end - start) + "毫秒");
    return new AsyncResult<>("任务三OK");
}

@Override
@Async
public Future<List<ProcessModelDTO>> findActiveProcessModels(final Pageable pageable) {
  final List<ProcessModelImpl> results = processModelRepository.findActiveProcesses();

  final List<ProcessModelDTO> processModels = createProcessModelDTO(results);

  return new AsyncResult<List<ProcessModelDTO>>(processModels);
}