2 * Copyright © 2020 Orange, Inc. and others. All rights reserved.
4 * This program and the accompanying materials are made available under the
5 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
6 * and is available at http://www.eclipse.org/legal/epl-v10.html
8 package org.opendaylight.transportpce.nbinotifications.consumer;
10 import com.google.common.annotations.VisibleForTesting;
11 import java.time.Duration;
12 import java.util.ArrayList;
13 import java.util.Collections;
14 import java.util.List;
15 import java.util.Properties;
16 import org.apache.kafka.clients.consumer.Consumer;
17 import org.apache.kafka.clients.consumer.ConsumerConfig;
18 import org.apache.kafka.clients.consumer.ConsumerRecord;
19 import org.apache.kafka.clients.consumer.ConsumerRecords;
20 import org.apache.kafka.clients.consumer.KafkaConsumer;
21 import org.apache.kafka.common.TopicPartition;
22 import org.apache.kafka.common.serialization.StringDeserializer;
23 import org.eclipse.jdt.annotation.NonNull;
24 import org.opendaylight.transportpce.common.converter.JsonStringConverter;
25 import org.opendaylight.transportpce.nbinotifications.serialization.ConfigConstants;
26 import org.opendaylight.transportpce.nbinotifications.utils.NbiNotificationsUtils;
27 import org.opendaylight.yangtools.yang.binding.DataObject;
28 import org.opendaylight.yangtools.yang.common.QName;
29 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
30 import org.slf4j.Logger;
31 import org.slf4j.LoggerFactory;
33 public class Subscriber<T extends DataObject, D> {
34 private static final Logger LOG = LoggerFactory.getLogger(Subscriber.class);
36 private final Consumer<String, D> consumer;
38 public Subscriber(String id, String groupId, String subscriberServer, JsonStringConverter<T> deserializer,
39 Class<?> deserializerConf) {
40 Properties propsConsumer = NbiNotificationsUtils.loadProperties("subscriber.properties");
41 propsConsumer.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
42 propsConsumer.put(ConsumerConfig.CLIENT_ID_CONFIG, id);
43 propsConsumer.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
44 propsConsumer.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
45 propsConsumer.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG , deserializerConf);
46 propsConsumer.put(ConfigConstants.CONVERTER , deserializer);
47 if (subscriberServer != null && !subscriberServer.isBlank()) {
48 propsConsumer.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, subscriberServer);
50 LOG.info("Subscribing for group id {}, client config id {} with properties {}", groupId, id, propsConsumer);
51 consumer = new KafkaConsumer<>(propsConsumer);
54 public List<D> subscribe(String topicName, @NonNull QName name) {
55 LOG.info("Subscribe request to topic '{}' ", topicName);
56 consumer.subscribe(Collections.singleton(topicName));
57 final ConsumerRecords<String, D> consumerRecords = consumer.poll(Duration.ofMillis(1000));
58 List<D> notificationServiceList = new ArrayList<>();
59 YangInstanceIdentifier.of(name);
60 for (ConsumerRecord<String, D> record : consumerRecords.records(new TopicPartition(topicName, 0))) {
61 if (record.value() != null) {
62 notificationServiceList.add(record.value());
65 LOG.info("Getting records '{}' ", notificationServiceList);
66 consumer.unsubscribe();
68 return notificationServiceList;
71 @VisibleForTesting public Subscriber(Consumer<String, D> consumer) {
72 this.consumer = consumer;