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.IntegerDeserializer;
30 import org.apache.kafka.common.serialization.StringDeserializer;
31 import org.oran.dmaapadapter.configuration.ApplicationConfig;
32 import org.oran.dmaapadapter.repository.InfoType;
33 import org.slf4j.Logger;
34 import org.slf4j.LoggerFactory;
36 import reactor.core.Disposable;
37 import reactor.core.publisher.Sinks;
38 import reactor.core.publisher.Sinks.Many;
39 import reactor.kafka.receiver.KafkaReceiver;
40 import reactor.kafka.receiver.ReceiverOptions;
43 * The class streams incoming requests from a Kafka topic and sends them further
44 * to a multi cast sink, which several other streams can connect to.
46 @SuppressWarnings("squid:S2629") // Invoke method(s) only conditionally
47 public class KafkaTopicListener {
48 private static final Logger logger = LoggerFactory.getLogger(KafkaTopicListener.class);
49 private final ApplicationConfig applicationConfig;
50 private final InfoType type;
51 private final Many<String> output;
53 public KafkaTopicListener(ApplicationConfig applicationConfig, InfoType type) {
54 this.applicationConfig = applicationConfig;
56 final int CONSUMER_BACKPRESSURE_BUFFER_SIZE = 1024 * 10;
57 this.output = Sinks.many().multicast().onBackpressureBuffer(CONSUMER_BACKPRESSURE_BUFFER_SIZE);
60 startKafkaTopicReceiver();
63 public Many<String> getOutput() {
67 private Disposable startKafkaTopicReceiver() {
68 return KafkaReceiver.create(kafkaInputProperties()) //
70 .doOnNext(this::onReceivedData) //
72 this::onReceivedError, //
73 () -> logger.warn("KafkaTopicReceiver stopped"));
76 private void onReceivedData(ConsumerRecord<Integer, String> input) {
77 logger.debug("Received from kafka topic: {} :{}", this.type.getKafkaInputTopic(), input.value());
78 output.emitNext(input.value(), Sinks.EmitFailureHandler.FAIL_FAST);
81 private void onReceivedError(Throwable t) {
82 logger.error("KafkaTopicReceiver error: {}", t.getMessage());
85 private ReceiverOptions<Integer, String> kafkaInputProperties() {
86 Map<String, Object> consumerProps = new HashMap<>();
87 if (this.applicationConfig.getKafkaBootStrapServers().isEmpty()) {
88 logger.error("No kafka boostrap server is setup");
90 consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.applicationConfig.getKafkaBootStrapServers());
91 consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "osc-dmaap-adaptor");
92 consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
93 consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
95 return ReceiverOptions.<Integer, String>create(consumerProps)
96 .subscription(Collections.singleton(this.type.getKafkaInputTopic()));