import org.oran.dmaapadapter.repository.Jobs;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.springframework.http.MediaType;
import reactor.core.publisher.Flux;
-import reactor.core.publisher.FluxSink;
import reactor.core.publisher.Mono;
/**
* The class fetches incoming requests from DMAAP and sends them further to the
* consumers that has a job for this InformationType.
*/
-
public class DmaapTopicConsumer {
private static final Duration TIME_BETWEEN_DMAAP_RETRIES = Duration.ofSeconds(10);
private static final Logger logger = LoggerFactory.getLogger(DmaapTopicConsumer.class);
private final AsyncRestClient dmaapRestClient;
- private final InfiniteFlux infiniteSubmitter = new InfiniteFlux();
- private final AsyncRestClient consumerRestClient;
protected final ApplicationConfig applicationConfig;
protected final InfoType type;
protected final Jobs jobs;
- /** Submits new elements until stopped */
- private static class InfiniteFlux {
- private FluxSink<Integer> sink;
- private int counter = 0;
-
- public synchronized Flux<Integer> start() {
- stop();
- return Flux.create(this::next).doOnRequest(this::onRequest);
- }
-
- public synchronized void stop() {
- if (this.sink != null) {
- this.sink.complete();
- this.sink = null;
- }
- }
-
- void onRequest(long no) {
- logger.debug("InfiniteFlux.onRequest {}", no);
- for (long i = 0; i < no; ++i) {
- sink.next(counter++);
- }
- }
-
- void next(FluxSink<Integer> sink) {
- logger.debug("InfiniteFlux.next");
- this.sink = sink;
- sink.next(counter++);
- }
- }
-
public DmaapTopicConsumer(ApplicationConfig applicationConfig, InfoType type, Jobs jobs) {
AsyncRestClientFactory restclientFactory = new AsyncRestClientFactory(applicationConfig.getWebClientConfig());
this.dmaapRestClient = restclientFactory.createRestClientNoHttpProxy("");
this.applicationConfig = applicationConfig;
- this.consumerRestClient = type.isUseHttpProxy() ? restclientFactory.createRestClientUseHttpProxy("")
- : restclientFactory.createRestClientNoHttpProxy("");
this.type = type;
this.jobs = jobs;
}
public void start() {
- infiniteSubmitter.start() //
+ Flux.range(0, Integer.MAX_VALUE) //
.flatMap(notUsed -> getFromMessageRouter(getDmaapUrl()), 1) //
.flatMap(this::pushDataToConsumers) //
.subscribe(//
null, //
throwable -> logger.error("DmaapMessageConsumer error: {}", throwable.getMessage()), //
- () -> logger.warn("DmaapMessageConsumer stopped {}", type.getId())); //
+ this::onComplete); //
+ }
+ private void onComplete() {
+ logger.warn("DmaapMessageConsumer completed {}", type.getId());
+ start();
}
private String getDmaapUrl() {
private Mono<String> handleDmaapErrorResponse(Throwable t) {
logger.debug("error from DMAAP {} {}", t.getMessage(), type.getDmaapTopicUrl());
- return Mono.delay(TIME_BETWEEN_DMAAP_RETRIES).flatMap(notUsed -> Mono.empty());
+ return Mono.delay(TIME_BETWEEN_DMAAP_RETRIES) //
+ .flatMap(notUsed -> Mono.empty());
}
private Mono<String> getFromMessageRouter(String topicUrl) {
// Distibute the body to all jobs for this type
return Flux.fromIterable(this.jobs.getJobsForType(this.type)) //
- .doOnNext(job -> logger.debug("Sending to consumer {}", job.getCallbackUrl()))
- .flatMap(job -> consumerRestClient.post(job.getCallbackUrl(), body), CONCURRENCY) //
+ .filter(job -> job.isFilterMatch(body)) //
+ .doOnNext(job -> logger.debug("Sending to consumer {}", job.getCallbackUrl())) //
+ .flatMap(job -> job.getConsumerRestClient().post("", body, MediaType.APPLICATION_JSON), CONCURRENCY) //
.onErrorResume(this::handleConsumerErrorResponse);
}
}