Spring WebFlux - 如何从数据库中获取值并设置此值以创建对象

Question

我真的不知道如何在不破坏我的反应链的情况下使用来自 Cassandra 的数据创建一个对象？

我有一些私有方法，它是整个反应链的一部分：

private Mono<SecurityData> createSecurityData(Security securityOfType) {
return jobsProgressRepository
.findByAgentId(securityOfType.getAgentId()) //Flux<JobsProgress>
.collectList() //Mono<List<JobsProgress>>
.flatMap(this::getJobsProgressSummary) //Mono<JobsProgressSummary>
.flatMap(job -> mapToSecurityData(job, securityOfType));
}

然后我想准备一些对象：

private Mono<SecurityData> mapToSecurityData(JobsProgressSummary job, Security security ) {
SecurityData securityData = new SecurityData();
securityData.setAgentId(security.getAgentId());
securityData.setGroupId(security.getGroupId());
securityData.setHostname(getHostname(security)); --> here is the problem!!!
return Mono.just(securityData);
}

和获取主机名方法：

private String getHostname(Security security) {
String hostname = "";
switch(security.getProductType()){
case VM: hostname = vmRepository
.findByAgentId(security.getAgentId()).blockFirst().getHostname();
case HYPER: hostname = hyperRepository
.findByAgentId(security.getAgentId()).blockFirst().getHostname();
default: ""
}
return hostname;
}

我的存储库如下所示：

public interface HostRepository extends ReactiveCassandraRepository<Host, MapId> {
Flux<Host> findByAgentId(UUID agentId);
}

也许我的做法是错误的？我当然不能使用

hostRepository
.findByAgentId(security.getAgentId()).subscribe() // or blockFirst()

因为我不想打破我的反应链...

如何解决我的问题？请不要犹豫，提供任何，即使是非常小的提示：(

更新

在这里，我添加了方法getJobsProgressSummary的缺失正文：

private Mono<JobsProgressSummary> getJobsProgressSummary(List<JobsProgress> jobs) {
JobsProgressSummary jobsProgressSummary = new JobsProgressSummary();
jobs.forEach(
job -> {
if (job.getStatus().toUpperCase(Locale.ROOT).equals(StatusEnum.RUNNING.name())) {
jobsProgressSummary.setRunningJobs(jobsProgressSummary.getRunningJobs() + 1);
} else if (job.getStatus().toUpperCase(Locale.ROOT).equals(StatusEnum.FAILED.name())) {
jobsProgressSummary.setAmountOfErrors(jobsProgressSummary.getAmountOfErrors() + 1);
} else if (isScheduledJob(job.getStartTime())) {
jobsProgressSummary.setScheduledJobs(jobsProgressSummary.getScheduledJobs() + 1);
}
});
Instant lastActivity =
jobs.stream()
.map(JobsProgress::getStartTime)
.map(startTime -> Instant.ofEpochMilli(Long.parseLong(startTime)))
.max(Instant::compareTo)
.orElseGet(null);
jobsProgressSummary.setLastActivity(lastActivity);
return Mono.just(jobsProgressSummary);
}

Answer 1

你需要将所有内容链接在一起，你的代码目前就像命令式和响应式的混合体。此外，您永远不需要调用块。

像下面这样的东西应该可以工作

private Mono<SecurityData> mapToSecurityData(JobsProgressSummary job, Security security ) {
//Try to get hostname first, then process result
return getHostname(security)
//Map it. Probz should use builder or all args constructor to reduce code here
.map(hostname -> {
SecurityData securityData = new SecurityData();
securityData.setAgentId(security.getAgentId());
securityData.setGroupId(security.getGroupId());
securityData.setHostname(hostname);
return securityData;
});
}
private Mono<String> getHostname(Security security) {
Mono<String> hostname = Mono.empty();
switch(security.getProductType()){
//Also assuming hostname is a field in Security
//just change Security to class name if not
case VM: hostname = vmRepository.findByAgentId(security.getAgentId())
.next()
.map(Security::getHostname);
case HYPER: hostname = hyperRepository.findByAgentId(security.getAgentId())
.next()
.map(Security::getHostname);
}
return hostname;
}