Skip to content

Commit 046ab3a

Browse files
blabadiblabadi
authored
Merge pull request #146 from overture-stack/rc/3.0.0-fe99363
release 3.0.0 #116 fix junit 5 version to avoid conflicts in class path #118 update analysis centric index fields, add data_type, rename files filed to file, remove data_path and metadata_path from repositories field #119 kafka stream listener configurable status (PUBLISHED, SUPPRESSED, wtv) #120 add support for cram/crai index files #121 change specimen type and sample type to list #127 update mappings fields #129 rename format to file_type #131 remove last_modified from file centric mapping #132 rename file-id to file-object_id #135 fix index analysis response message #136 add maven formatter #139 add ability to remove document by analysis id from analysis_centric #143 add dynamic info indexed field support to donor, Specimen, sample, and file update test document to cover that for file centric document #145 fix NPE for analysis centric index
2 parents 388ac6c + fe99363 commit 046ab3a

File tree

135 files changed

+6466
-5696
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

135 files changed

+6466
-5696
lines changed

‎.mvn/maven.config

+1-1
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,3 @@
1-
-Drevision=2.0.0
1+
-Drevision=3.0.0
22
-Dsha1=
33
-Dchangelist=-SNAPSHOT

‎docs/features.md

+3-3
Original file line numberDiff line numberDiff line change
@@ -13,9 +13,9 @@ As an example, you can see in the `repositories` section of the following docume
1313
"access": "controlled",
1414
"study": "ABCD-CA",
1515
"analysis": {
16-
"id": "EGAZ",
17-
"type": "sequencingRead",
18-
"state": "PUBLISHED",
16+
"analysis_id": "EGAZ",
17+
"analysis_type": "sequencingRead",
18+
"analysis_state": "PUBLISHED",
1919
"study": "ABCD-CA",
2020
"experiment": {
2121
"analysisId": "EGAZ",

‎maestro-app/src/main/java/bio/overture/maestro/app/Maestro.java

+3-3
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@
3131
@SpringBootApplication
3232
@Import({RootConfiguration.class})
3333
public class Maestro {
34-
public static void main(String[] args) {
35-
SpringApplication.run(Maestro.class, args);
36-
}
34+
public static void main(String[] args) {
35+
SpringApplication.run(Maestro.class, args);
36+
}
3737
}

‎maestro-app/src/main/java/bio/overture/maestro/app/infra/adapter/inbound/messaging/IndexAnalysisMessage.java

+5-8
Original file line numberDiff line numberDiff line change
@@ -23,13 +23,10 @@
2323
@ToString
2424
@AllArgsConstructor
2525
class IndexAnalysisMessage {
26-
@NonNull
27-
private String analysisId;
28-
@NonNull
29-
private String studyId;
30-
@NonNull
31-
private String repositoryCode;
26+
@NonNull private String analysisId;
27+
@NonNull private String studyId;
28+
@NonNull private String repositoryCode;
3229

33-
/** if callers set this flag it will do a remove instead of add.*/
34-
private Boolean removeAnalysis = false;
30+
/** if callers set this flag it will do a remove instead of add. */
31+
private Boolean removeAnalysis = false;
3532
}

‎maestro-app/src/main/java/bio/overture/maestro/app/infra/adapter/inbound/messaging/IndexMessage.java

+5-6
Original file line numberDiff line numberDiff line change
@@ -26,10 +26,9 @@
2626
@ToString
2727
@AllArgsConstructor
2828
class IndexMessage {
29-
private String analysisId;
30-
private String studyId;
31-
@NonNull
32-
private String repositoryCode;
33-
/** if callers set this flag it will do a remove instead of add.*/
34-
private Boolean removeAnalysis = false;
29+
private String analysisId;
30+
private String studyId;
31+
@NonNull private String repositoryCode;
32+
/** if callers set this flag it will do a remove instead of add. */
33+
private Boolean removeAnalysis = false;
3534
}

‎maestro-app/src/main/java/bio/overture/maestro/app/infra/adapter/inbound/messaging/IndexMessagesHelper.java

+32-31
Original file line numberDiff line numberDiff line change
@@ -19,45 +19,46 @@
1919

2020
import bio.overture.maestro.domain.api.message.IndexResult;
2121
import io.vavr.Tuple2;
22+
import java.util.function.Supplier;
2223
import lombok.extern.slf4j.Slf4j;
2324
import lombok.val;
2425
import reactor.core.publisher.Flux;
2526
import reactor.core.publisher.Mono;
2627

27-
import java.util.function.Supplier;
28-
2928
@Slf4j
3029
public class IndexMessagesHelper {
31-
public static <T> void handleIndexRepository(Supplier<Mono<Tuple2<T, IndexResult>>> resultSupplier) {
32-
val result = resultSupplier.get().blockOptional();
33-
val tuple = result.orElseThrow(() -> new RuntimeException("failed to obtain result"));
34-
if (!tuple._2().isSuccessful()) {
35-
log.error("failed to process message : {} successfully", tuple._1());
36-
throw new RuntimeException("failed to process the message");
37-
}
30+
public static <T> void handleIndexRepository(
31+
Supplier<Mono<Tuple2<T, IndexResult>>> resultSupplier) {
32+
val result = resultSupplier.get().blockOptional();
33+
val tuple = result.orElseThrow(() -> new RuntimeException("failed to obtain result"));
34+
if (!tuple._2().isSuccessful()) {
35+
log.error("failed to process message : {} successfully", tuple._1());
36+
throw new RuntimeException("failed to process the message");
3837
}
38+
}
3939

40-
public static <T> void handleIndexResult(Supplier<Flux<Tuple2<T, IndexResult>>> resultSupplier) {
41-
/*
42-
* Why Blocking?
43-
*
44-
* - this is a stream consumer, it's supposed to process one message at a time
45-
* the value of reactive processing diminishes since the queue provides a buffering level,
46-
* without blocking it will async process the messages and if one fails we can
47-
* async add it to a DLQ in the subscriber, However, I opted to use blocking because of the next point.
48-
*
49-
* - spring reactive cloud stream is deprecated in favor of spring cloud functions that support
50-
* stream processing: https://cloud.spring.io/spring-cloud-static/spring-cloud-stream/2.2.0.RELEASE/spring-cloud-stream.html#spring_cloud_function
51-
* so I don't want to use a deprecated library, and if needed we can switch to cloud function in future
52-
* https://stackoverflow.com/questions/53438208/spring-cloud-stream-reactive-how-to-do-the-error-handling-in-case-of-reactive
53-
*/
54-
val result = resultSupplier.get().collectList().blockOptional();
55-
val tupleList = result.orElseThrow(() -> new RuntimeException("failed to obtain result"));
56-
tupleList.forEach(tuple -> {
57-
if (!tuple._2().isSuccessful()) {
58-
log.error("failed to process message : {} successfully", tuple._1());
59-
throw new RuntimeException("failed to process the message");
60-
}
40+
public static <T> void handleIndexResult(Supplier<Flux<Tuple2<T, IndexResult>>> resultSupplier) {
41+
/*
42+
* Why Blocking?
43+
*
44+
* - this is a stream consumer, it's supposed to process one message at a time
45+
* the value of reactive processing diminishes since the queue provides a buffering level,
46+
* without blocking it will async process the messages and if one fails we can
47+
* async add it to a DLQ in the subscriber, However, I opted to use blocking because of the next point.
48+
*
49+
* - spring reactive cloud stream is deprecated in favor of spring cloud functions that support
50+
* stream processing: https://cloud.spring.io/spring-cloud-static/spring-cloud-stream/2.2.0.RELEASE/spring-cloud-stream.html#spring_cloud_function
51+
* so I don't want to use a deprecated library, and if needed we can switch to cloud function in future
52+
* https://stackoverflow.com/questions/53438208/spring-cloud-stream-reactive-how-to-do-the-error-handling-in-case-of-reactive
53+
*/
54+
val result = resultSupplier.get().collectList().blockOptional();
55+
val tupleList = result.orElseThrow(() -> new RuntimeException("failed to obtain result"));
56+
tupleList.forEach(
57+
tuple -> {
58+
if (!tuple._2().isSuccessful()) {
59+
log.error("failed to process message : {} successfully", tuple._1());
60+
throw new RuntimeException("failed to process the message");
61+
}
6162
});
62-
}
63+
}
6364
}

‎maestro-app/src/main/java/bio/overture/maestro/app/infra/adapter/inbound/messaging/IndexRepositoryMessage.java

+1-2
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,5 @@
2323
@ToString
2424
@AllArgsConstructor
2525
class IndexRepositoryMessage {
26-
@NonNull
27-
private String repositoryCode;
26+
@NonNull private String repositoryCode;
2827
}

‎maestro-app/src/main/java/bio/overture/maestro/app/infra/adapter/inbound/messaging/IndexStudyMessage.java

+2-4
Original file line numberDiff line numberDiff line change
@@ -23,8 +23,6 @@
2323
@ToString
2424
@AllArgsConstructor
2525
class IndexStudyMessage {
26-
@NonNull
27-
private String studyId;
28-
@NonNull
29-
private String repositoryCode;
26+
@NonNull private String studyId;
27+
@NonNull private String repositoryCode;
3028
}

‎maestro-app/src/main/java/bio/overture/maestro/app/infra/adapter/inbound/messaging/IndexingMessagesStreamListener.java

+105-95
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,9 @@
1717

1818
package bio.overture.maestro.app.infra.adapter.inbound.messaging;
1919

20+
import static bio.overture.maestro.app.infra.adapter.inbound.messaging.IndexMessagesHelper.handleIndexRepository;
21+
import static bio.overture.maestro.app.infra.adapter.inbound.messaging.IndexMessagesHelper.handleIndexResult;
22+
2023
import bio.overture.maestro.domain.api.Indexer;
2124
import bio.overture.maestro.domain.api.exception.FailureData;
2225
import bio.overture.maestro.domain.api.message.*;
@@ -32,111 +35,118 @@
3235
import reactor.core.publisher.Flux;
3336
import reactor.core.publisher.Mono;
3437

35-
import static bio.overture.maestro.app.infra.adapter.inbound.messaging.IndexMessagesHelper.handleIndexRepository;
36-
import static bio.overture.maestro.app.infra.adapter.inbound.messaging.IndexMessagesHelper.handleIndexResult;
37-
3838
@Slf4j
3939
@EnableBinding(Sink.class)
4040
public class IndexingMessagesStreamListener {
4141

42-
private final Indexer indexer;
43-
44-
public IndexingMessagesStreamListener(@NonNull Indexer indexer) {
45-
this.indexer = indexer;
46-
}
47-
48-
@StreamListener(Sink.INPUT)
49-
public void handleAnalysisMessage(@Payload IndexMessage indexMessage) {
50-
if (isAnalysisReq(indexMessage)) {
51-
val indexAnalysisMessage = new IndexAnalysisMessage(indexMessage.getAnalysisId(),
52-
indexMessage.getStudyId(),
53-
indexMessage.getRepositoryCode(),
54-
indexMessage.getRemoveAnalysis());
55-
handleIndexResult(() -> this.indexOrRemoveAnalysis(indexAnalysisMessage));
56-
} else if (isStudyMsg(indexMessage)) {
57-
val indexStudyMessage = new IndexStudyMessage(indexMessage.getStudyId(), indexMessage.getRepositoryCode());
58-
handleIndexResult(() -> this.indexStudy(indexStudyMessage));
59-
} else if (isRepoMsg(indexMessage)) {
60-
val indexRepositoryMessage = new IndexRepositoryMessage(indexMessage.getRepositoryCode());
61-
handleIndexRepository(() -> this.indexRepository(indexRepositoryMessage));
62-
} else {
63-
throw new IllegalArgumentException("invalid message format");
64-
}
65-
}
66-
67-
private boolean isAnalysisReq(IndexMessage indexMessage) {
68-
return !StringUtils.isEmpty(indexMessage.getAnalysisId())
69-
&& !StringUtils.isEmpty(indexMessage.getStudyId())
70-
&& !StringUtils.isEmpty(indexMessage.getRepositoryCode());
71-
}
72-
73-
private boolean isStudyMsg(IndexMessage indexMessage) {
74-
return StringUtils.isEmpty(indexMessage.getAnalysisId())
75-
&& !StringUtils.isEmpty(indexMessage.getStudyId())
76-
&& !StringUtils.isEmpty(indexMessage.getRepositoryCode());
77-
}
78-
79-
private boolean isRepoMsg(IndexMessage indexMessage) {
80-
return StringUtils.isEmpty(indexMessage.getAnalysisId())
81-
&& StringUtils.isEmpty(indexMessage.getStudyId())
82-
&& !StringUtils.isEmpty(indexMessage.getRepositoryCode());
42+
private final Indexer indexer;
43+
44+
public IndexingMessagesStreamListener(@NonNull Indexer indexer) {
45+
this.indexer = indexer;
46+
}
47+
48+
@StreamListener(Sink.INPUT)
49+
public void handleAnalysisMessage(@Payload IndexMessage indexMessage) {
50+
if (isAnalysisReq(indexMessage)) {
51+
val indexAnalysisMessage =
52+
new IndexAnalysisMessage(
53+
indexMessage.getAnalysisId(),
54+
indexMessage.getStudyId(),
55+
indexMessage.getRepositoryCode(),
56+
indexMessage.getRemoveAnalysis());
57+
handleIndexResult(() -> this.indexOrRemoveAnalysis(indexAnalysisMessage));
58+
} else if (isStudyMsg(indexMessage)) {
59+
val indexStudyMessage =
60+
new IndexStudyMessage(indexMessage.getStudyId(), indexMessage.getRepositoryCode());
61+
handleIndexResult(() -> this.indexStudy(indexStudyMessage));
62+
} else if (isRepoMsg(indexMessage)) {
63+
val indexRepositoryMessage = new IndexRepositoryMessage(indexMessage.getRepositoryCode());
64+
handleIndexRepository(() -> this.indexRepository(indexRepositoryMessage));
65+
} else {
66+
throw new IllegalArgumentException("invalid message format");
8367
}
84-
85-
private Flux<Tuple2<IndexAnalysisMessage, IndexResult>> indexOrRemoveAnalysis(IndexAnalysisMessage msg) {
86-
if (msg.getRemoveAnalysis()) {
87-
return Flux.from(removeAnalysis(msg));
88-
} else {
89-
return indexAnalysis(msg);
90-
}
68+
}
69+
70+
private boolean isAnalysisReq(IndexMessage indexMessage) {
71+
return !StringUtils.isEmpty(indexMessage.getAnalysisId())
72+
&& !StringUtils.isEmpty(indexMessage.getStudyId())
73+
&& !StringUtils.isEmpty(indexMessage.getRepositoryCode());
74+
}
75+
76+
private boolean isStudyMsg(IndexMessage indexMessage) {
77+
return StringUtils.isEmpty(indexMessage.getAnalysisId())
78+
&& !StringUtils.isEmpty(indexMessage.getStudyId())
79+
&& !StringUtils.isEmpty(indexMessage.getRepositoryCode());
80+
}
81+
82+
private boolean isRepoMsg(IndexMessage indexMessage) {
83+
return StringUtils.isEmpty(indexMessage.getAnalysisId())
84+
&& StringUtils.isEmpty(indexMessage.getStudyId())
85+
&& !StringUtils.isEmpty(indexMessage.getRepositoryCode());
86+
}
87+
88+
private Flux<Tuple2<IndexAnalysisMessage, IndexResult>> indexOrRemoveAnalysis(
89+
IndexAnalysisMessage msg) {
90+
if (msg.getRemoveAnalysis()) {
91+
return Flux.from(removeAnalysis(msg));
92+
} else {
93+
return indexAnalysis(msg);
9194
}
92-
93-
private Mono<Tuple2<IndexAnalysisMessage, IndexResult>> removeAnalysis(IndexAnalysisMessage msg) {
94-
return indexer.removeAnalysis(RemoveAnalysisCommand.builder()
95-
.analysisIdentifier(AnalysisIdentifier.builder()
96-
.studyId(msg.getStudyId())
97-
.analysisId(msg.getAnalysisId())
98-
.repositoryCode(msg.getRepositoryCode())
99-
.build())
95+
}
96+
97+
private Flux<Tuple2<IndexAnalysisMessage, IndexResult>> removeAnalysis(IndexAnalysisMessage msg) {
98+
return indexer
99+
.removeAnalysis(
100+
RemoveAnalysisCommand.builder()
101+
.analysisIdentifier(
102+
AnalysisIdentifier.builder()
103+
.studyId(msg.getStudyId())
104+
.analysisId(msg.getAnalysisId())
105+
.repositoryCode(msg.getRepositoryCode())
106+
.build())
100107
.build())
101-
.map(out -> new Tuple2<>(msg, out))
102-
.onErrorResume((e) -> catchUnhandledErrors(msg, e));
103-
}
108+
.map(out -> new Tuple2<>(msg, out))
109+
.onErrorResume((e) -> catchUnhandledErrors(msg, e));
110+
}
111+
112+
private Flux<Tuple2<IndexAnalysisMessage, IndexResult>> indexAnalysis(IndexAnalysisMessage msg) {
113+
return indexer
114+
.indexAnalysis(
115+
IndexAnalysisCommand.builder()
116+
.analysisIdentifier(
117+
AnalysisIdentifier.builder()
118+
.studyId(msg.getStudyId())
119+
.analysisId(msg.getAnalysisId())
120+
.repositoryCode(msg.getRepositoryCode())
121+
.build())
122+
.build())
123+
.map(out -> new Tuple2<>(msg, out))
124+
.onErrorResume((e) -> catchUnhandledErrors(msg, e));
125+
}
104126

105-
private Flux<Tuple2<IndexAnalysisMessage, IndexResult>> indexAnalysis(IndexAnalysisMessage msg) {
106-
return indexer.indexAnalysis(IndexAnalysisCommand.builder()
107-
.analysisIdentifier(AnalysisIdentifier.builder()
127+
private Flux<Tuple2<IndexStudyMessage, IndexResult>> indexStudy(IndexStudyMessage msg) {
128+
return indexer
129+
.indexStudy(
130+
IndexStudyCommand.builder()
108131
.studyId(msg.getStudyId())
109-
.analysisId(msg.getAnalysisId())
110132
.repositoryCode(msg.getRepositoryCode())
111-
.build()
112-
).build())
133+
.build())
134+
.map(out -> new Tuple2<>(msg, out));
135+
}
136+
137+
private Mono<Tuple2<IndexRepositoryMessage, IndexResult>> indexRepository(
138+
IndexRepositoryMessage msg) {
139+
return indexer
140+
.indexRepository(
141+
IndexStudyRepositoryCommand.builder().repositoryCode(msg.getRepositoryCode()).build())
113142
.map(out -> new Tuple2<>(msg, out))
114143
.onErrorResume((e) -> catchUnhandledErrors(msg, e));
115-
}
116-
117-
private Flux<Tuple2<IndexStudyMessage, IndexResult>> indexStudy(IndexStudyMessage msg) {
118-
return indexer.indexStudy(IndexStudyCommand.builder()
119-
.studyId(msg.getStudyId())
120-
.repositoryCode(msg.getRepositoryCode())
121-
.build())
122-
.map(out -> new Tuple2<>(msg, out));
123-
}
124-
125-
private Mono<Tuple2<IndexRepositoryMessage, IndexResult>> indexRepository(IndexRepositoryMessage msg) {
126-
return indexer.indexRepository(IndexStudyRepositoryCommand.builder()
127-
.repositoryCode(msg.getRepositoryCode())
128-
.build())
129-
.map(out -> new Tuple2<>(msg, out))
130-
.onErrorResume((e) -> catchUnhandledErrors(msg, e));
131-
}
132-
133-
private <T> Mono<Tuple2<T, IndexResult>> catchUnhandledErrors(T msg, Throwable e) {
134-
log.error("failed processing message: {} ", msg, e);
135-
val indexResult = IndexResult.builder()
136-
.successful(false)
137-
.failureData(FailureData.builder().build())
138-
.build();
139-
return Mono.just(new Tuple2<>(msg, indexResult));
140-
}
141-
144+
}
145+
146+
private <T> Mono<Tuple2<T, IndexResult>> catchUnhandledErrors(T msg, Throwable e) {
147+
log.error("failed processing message: {} ", msg, e);
148+
val indexResult =
149+
IndexResult.builder().successful(false).failureData(FailureData.builder().build()).build();
150+
return Mono.just(new Tuple2<>(msg, indexResult));
151+
}
142152
}

‎maestro-app/src/main/java/bio/overture/maestro/app/infra/adapter/inbound/messaging/MessagingConfig.java

+3-4
Original file line numberDiff line numberDiff line change
@@ -20,9 +20,8 @@
2020
import bio.overture.maestro.app.infra.adapter.inbound.messaging.song.SongAnalysisStreamListener;
2121
import org.springframework.context.annotation.Import;
2222

23-
2423
@Import({
25-
IndexingMessagesStreamListener.class,
26-
SongAnalysisStreamListener.class,
24+
IndexingMessagesStreamListener.class,
25+
SongAnalysisStreamListener.class,
2726
})
28-
public class MessagingConfig { }
27+
public class MessagingConfig {}

0 commit comments

Comments
 (0)