2 * ========================LICENSE_START=================================
5 * Copyright (C) 2021 Nordix Foundation
7 * Licensed under the Apache License, Version 2.0 (the "License");
8 * you may not use this file except in compliance with the License.
9 * You may obtain a copy of the License at
11 * http://www.apache.org/licenses/LICENSE-2.0
13 * Unless required by applicable law or agreed to in writing, software
14 * distributed under the License is distributed on an "AS IS" BASIS,
15 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16 * See the License for the specific language governing permissions and
17 * limitations under the License.
18 * ========================LICENSE_END===================================
21 package org.oran.dmaapadapter.tasks;
23 import java.util.Collections;
24 import java.util.HashMap;
27 import org.apache.kafka.clients.consumer.ConsumerConfig;
28 import org.apache.kafka.clients.consumer.ConsumerRecord;
29 import org.apache.kafka.common.serialization.StringDeserializer;
30 import org.oran.dmaapadapter.configuration.ApplicationConfig;
31 import org.oran.dmaapadapter.repository.InfoType;
32 import org.slf4j.Logger;
33 import org.slf4j.LoggerFactory;
35 import reactor.core.Disposable;
36 import reactor.core.publisher.Sinks;
37 import reactor.core.publisher.Sinks.Many;
38 import reactor.kafka.receiver.KafkaReceiver;
39 import reactor.kafka.receiver.ReceiverOptions;
42 * The class streams incoming requests from a Kafka topic and sends them further
43 * to a multi cast sink, which several other streams can connect to.
45 @SuppressWarnings("squid:S2629") // Invoke method(s) only conditionally
46 public class KafkaTopicListener {
47 private static final Logger logger = LoggerFactory.getLogger(KafkaTopicListener.class);
48 private final ApplicationConfig applicationConfig;
49 private final InfoType type;
50 private Many<String> output;
51 private Disposable topicReceiverTask;
53 public KafkaTopicListener(ApplicationConfig applicationConfig, InfoType type) {
54 this.applicationConfig = applicationConfig;
59 public Many<String> getOutput() {
65 final int CONSUMER_BACKPRESSURE_BUFFER_SIZE = 1024 * 10;
66 this.output = Sinks.many().multicast().onBackpressureBuffer(CONSUMER_BACKPRESSURE_BUFFER_SIZE);
67 logger.debug("Listening to kafka topic: {} type :{}", this.type.getKafkaInputTopic(), type.getId());
68 topicReceiverTask = KafkaReceiver.create(kafkaInputProperties()) //
70 .doOnNext(this::onReceivedData) //
72 this::onReceivedError, //
73 () -> logger.warn("KafkaTopicReceiver stopped"));
77 if (topicReceiverTask != null) {
78 topicReceiverTask.dispose();
79 topicReceiverTask = null;
83 private void onReceivedData(ConsumerRecord<String, String> input) {
84 logger.debug("Received from kafka topic: {} :{}", this.type.getKafkaInputTopic(), input.value());
85 output.emitNext(input.value(), Sinks.EmitFailureHandler.FAIL_FAST);
88 private void onReceivedError(Throwable t) {
89 logger.error("KafkaTopicReceiver error: {}", t.getMessage());
92 private ReceiverOptions<String, String> kafkaInputProperties() {
93 Map<String, Object> consumerProps = new HashMap<>();
94 if (this.applicationConfig.getKafkaBootStrapServers().isEmpty()) {
95 logger.error("No kafka boostrap server is setup");
97 consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.applicationConfig.getKafkaBootStrapServers());
98 consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "osc-dmaap-adaptor");
99 consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
100 consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
102 return ReceiverOptions.<String, String>create(consumerProps)
103 .subscription(Collections.singleton(this.type.getKafkaInputTopic()));