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.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;
31 public class Subscriber {
32 private static final Logger LOG = LoggerFactory.getLogger(Subscriber.class);
34 private final Consumer<String, NotificationService> consumer;
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);
48 LOG.info("Suscribing for group id {}, client config id {} with properties {}", groupId, id, propsConsumer);
49 consumer = new KafkaConsumer<>(propsConsumer);
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());
63 LOG.info("Getting records '{}' ", notificationServiceList);
64 consumer.unsubscribe();
66 return notificationServiceList;
69 @VisibleForTesting public Subscriber(Consumer<String, NotificationService> consumer) {
70 this.consumer = consumer;