2 * Copyright © 2021 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.serialization.StringDeserializer;
22 import org.opendaylight.transportpce.common.converter.JsonStringConverter;
23 import org.opendaylight.transportpce.nbinotifications.serialization.ConfigConstants;
24 import org.opendaylight.transportpce.nbinotifications.serialization.NotificationAlarmServiceDeserializer;
25 import org.opendaylight.transportpce.nbinotifications.utils.NbiNotificationsUtils;
26 import org.opendaylight.yang.gen.v1.nbi.notifications.rev210628.get.notifications.alarm.service.output.NotificationAlarmService;
27 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
28 import org.slf4j.Logger;
29 import org.slf4j.LoggerFactory;
31 public class SubscriberAlarm {
32 private static final Logger LOG = LoggerFactory.getLogger(SubscriberAlarm.class);
34 private final Consumer<String, NotificationAlarmService> consumer;
36 public SubscriberAlarm(String id, String groupId, String subscriberServer,
37 JsonStringConverter<org.opendaylight.yang.gen.v1
38 .nbi.notifications.rev210628.NotificationAlarmService> deserializer) {
39 Properties propsConsumer = NbiNotificationsUtils.loadProperties("subscriber.properties");
40 propsConsumer.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
41 propsConsumer.put(ConsumerConfig.CLIENT_ID_CONFIG, id);
42 propsConsumer.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
43 propsConsumer.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG , NotificationAlarmServiceDeserializer.class);
44 propsConsumer.put(ConfigConstants.CONVERTER , deserializer);
45 if (subscriberServer != null && !subscriberServer.isBlank()) {
46 propsConsumer.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, subscriberServer);
48 LOG.info("Subscribing for group id {}, client config id {} with properties {}", groupId, id, propsConsumer);
49 consumer = new KafkaConsumer<>(propsConsumer);
52 public List<NotificationAlarmService> subscribeAlarm(String topicName) {
53 LOG.info("Subscribe request to topic '{}' ", topicName);
54 consumer.subscribe(Collections.singleton(topicName));
55 final ConsumerRecords<String, NotificationAlarmService> consumerRecords = consumer
56 .poll(Duration.ofMillis(1000));
57 List<NotificationAlarmService> notificationAlarmServiceList = new ArrayList<>();
58 YangInstanceIdentifier.of(NotificationAlarmService.QNAME);
59 for (ConsumerRecord<String, NotificationAlarmService> record : consumerRecords) {
60 if (record.value() != null) {
61 notificationAlarmServiceList.add(record.value());
64 LOG.info("Getting records '{}' ", notificationAlarmServiceList);
65 consumer.unsubscribe();
67 return notificationAlarmServiceList;
70 @VisibleForTesting public SubscriberAlarm(Consumer<String, NotificationAlarmService> consumer) {
71 this.consumer = consumer;