Reputation: 2210
I really don't know how to create an object with data from Cassandra without breaking my reactive chain?
I have some private method that is part of the whole reactive chain:
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));
}
and then i want to prepare some object:
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);
}
And getHostname method:
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;
}
My repos look like:
public interface HostRepository extends ReactiveCassandraRepository<Host, MapId> {
Flux<Host> findByAgentId(UUID agentId);
}
Maybe is my approach wrong? I can't of course use
hostRepository
.findByAgentId(security.getAgentId()).subscribe() // or blockFirst()
because I don't want to break my reactive chain...
How can I solve my problem? Please don't hesitate to give any, even very small tips:)
UPDATE
Here I added the missing body of the method 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);
}
Upvotes: 0
Views: 763
Reputation: 11216
You need to chain everything together, your code currently is like a mix of imperative and reactive. Also you should never need to call block.
Something like below should work
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;
}
Upvotes: 1