2 * ========================LICENSE_START=================================
5 * Copyright (C) 2023 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.pmlog;
23 import java.io.ByteArrayInputStream;
24 import java.io.IOException;
25 import java.util.Collections;
26 import java.util.HashMap;
28 import java.util.zip.GZIPInputStream;
30 import org.apache.kafka.clients.consumer.ConsumerConfig;
31 import org.apache.kafka.common.serialization.ByteArrayDeserializer;
32 import org.oran.pmlog.configuration.ApplicationConfig;
33 import org.slf4j.Logger;
34 import org.slf4j.LoggerFactory;
36 import reactor.core.publisher.Flux;
37 import reactor.core.publisher.Mono;
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 {
48 private static final Logger logger = LoggerFactory.getLogger(KafkaTopicListener.class);
49 private final ApplicationConfig applicationConfig;
50 private Flux<DataFromKafkaTopic> dataFromTopic;
51 private static com.google.gson.Gson gson = new com.google.gson.GsonBuilder().disableHtmlEscaping().create();
53 public KafkaTopicListener(ApplicationConfig applConfig) {
54 this.applicationConfig = applConfig;
57 public Flux<DataFromKafkaTopic> getFlux() {
58 if (this.dataFromTopic == null) {
59 this.dataFromTopic = startReceiveFromTopic(this.applicationConfig.getKafkaClientId());
61 return this.dataFromTopic;
64 private Flux<DataFromKafkaTopic> startReceiveFromTopic(String clientId) {
65 logger.debug("Listening to kafka topic: {}", this.applicationConfig.getKafkaInputTopic());
67 return KafkaReceiver.create(kafkaInputProperties(clientId)) //
69 .concatMap(consumerRecord -> consumerRecord) //
70 .doOnNext(input -> logger.trace("Received from kafka topic: {}",
71 this.applicationConfig.getKafkaInputTopic())) //
72 .doOnError(t -> logger.error("Received error: {}", t.getMessage())) //
73 .onErrorResume(t -> Mono.empty()) //
74 .doFinally(sig -> logger.error("KafkaTopicListener stopped, reason: {}", sig)) //
75 .filter(t -> t.value().length > 0 || t.key().length > 0) //
76 .map(input -> new DataFromKafkaTopic(input.headers(), input.key(), input.value())) //
81 private ReceiverOptions<byte[], byte[]> kafkaInputProperties(String clientId) {
82 Map<String, Object> consumerProps = new HashMap<>();
83 if (this.applicationConfig.getKafkaBootStrapServers().isEmpty()) {
84 logger.error("No kafka boostrap server is setup");
87 consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
88 consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.applicationConfig.getKafkaBootStrapServers());
89 consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, applicationConfig.getKafkaGroupId());
90 consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
91 consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
92 consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
94 consumerProps.put(ConsumerConfig.CLIENT_ID_CONFIG, clientId + "_" + applicationConfig.getKafkaGroupId());
96 return ReceiverOptions.<byte[], byte[]>create(consumerProps)
97 .subscription(Collections.singleton(this.applicationConfig.getKafkaInputTopic()));
100 public static byte[] unzip(byte[] bytes) throws IOException {
101 try (final GZIPInputStream gzipInput = new GZIPInputStream(new ByteArrayInputStream(bytes))) {
102 return gzipInput.readAllBytes();
106 private static byte[] unzip(byte[] bytes, String fileName) {
108 return fileName.endsWith(".gz") ? unzip(bytes) : bytes;
109 } catch (IOException e) {
110 logger.error("Error while decompression, file: {}, reason: {}", fileName, e.getMessage());