Add new Maven module to manage NBI Notifications
[transportpce.git] / nbinotifications / src / main / java / org / opendaylight / transportpce / nbinotifications / consumer / Subscriber.java
1 /*
2  * Copyright © 2020 Orange, Inc. and others.  All rights reserved.
3  *
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
7  */
8 package org.opendaylight.transportpce.nbinotifications.consumer;
9
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.NotificationServiceDeserializer;
25 import org.opendaylight.transportpce.nbinotifications.utils.NbiNotificationsUtils;
26 import org.opendaylight.yang.gen.v1.nbi.notifications.rev201130.get.notifications.service.output.NotificationService;
27 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
28 import org.slf4j.Logger;
29 import org.slf4j.LoggerFactory;
30
31 public class Subscriber {
32     private static final Logger LOG = LoggerFactory.getLogger(Subscriber.class);
33
34     private final Consumer<String, NotificationService> consumer;
35
36     public Subscriber(String id, String groupId, String suscriberServer,
37             JsonStringConverter<org.opendaylight.yang.gen.v1
38                 .nbi.notifications.rev201130.NotificationService> 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 , NotificationServiceDeserializer.class);
44         propsConsumer.put(ConfigConstants.CONVERTER , deserializer);
45         if (suscriberServer != null && !suscriberServer.isBlank()) {
46             propsConsumer.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, suscriberServer);
47         }
48         LOG.info("Suscribing for group id {}, client config id {} with properties {}", groupId, id, propsConsumer);
49         consumer = new KafkaConsumer<>(propsConsumer);
50     }
51
52     public List<NotificationService> subscribeService(String topicName) {
53         LOG.info("Subscribe request to topic '{}' ", topicName);
54         consumer.subscribe(Collections.singleton(topicName));
55         final ConsumerRecords<String, NotificationService> consumerRecords = consumer.poll(Duration.ofMillis(1000));
56         List<NotificationService> notificationServiceList = new ArrayList<>();
57         YangInstanceIdentifier.of(NotificationService.QNAME);
58         for (ConsumerRecord<String, NotificationService> record : consumerRecords) {
59             if (record.value() != null) {
60                 notificationServiceList.add(record.value());
61             }
62         }
63         LOG.info("Getting records '{}' ", notificationServiceList);
64         consumer.unsubscribe();
65         consumer.close();
66         return notificationServiceList;
67     }
68
69     @VisibleForTesting public Subscriber(Consumer<String, NotificationService> consumer) {
70         this.consumer = consumer;
71     }
72 }