nbinotifications/src/main/java/org/opendaylight/transportpce/nbinotifications/impl/NbiNotificationsImpl.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.impl;
   9
  10 import com.google.common.collect.ImmutableClassToInstanceMap;
  11 import com.google.common.util.concurrent.ListenableFuture;
  12 import java.util.ArrayList;
  13 import java.util.HashMap;
  14 import java.util.HashSet;
  15 import java.util.List;
  16 import java.util.Map;
  17 import java.util.NoSuchElementException;
  18 import java.util.Optional;
  19 import java.util.Set;
  20 import java.util.UUID;
  21 import java.util.concurrent.ExecutionException;
  22 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  23 import org.opendaylight.transportpce.common.converter.JsonStringConverter;
  24 import org.opendaylight.transportpce.common.network.NetworkTransactionService;
  25 import org.opendaylight.transportpce.nbinotifications.consumer.Subscriber;
  26 import org.opendaylight.transportpce.nbinotifications.serialization.NotificationAlarmServiceDeserializer;
  27 import org.opendaylight.transportpce.nbinotifications.serialization.NotificationServiceDeserializer;
  28 import org.opendaylight.transportpce.nbinotifications.serialization.TapiNotificationDeserializer;
  29 import org.opendaylight.transportpce.nbinotifications.utils.TopicManager;
  30 import org.opendaylight.yang.gen.v1.nbi.notifications.rev230728.GetNotificationsAlarmService;
  31 import org.opendaylight.yang.gen.v1.nbi.notifications.rev230728.GetNotificationsAlarmServiceInput;
  32 import org.opendaylight.yang.gen.v1.nbi.notifications.rev230728.GetNotificationsAlarmServiceOutput;
  33 import org.opendaylight.yang.gen.v1.nbi.notifications.rev230728.GetNotificationsAlarmServiceOutputBuilder;
  34 import org.opendaylight.yang.gen.v1.nbi.notifications.rev230728.GetNotificationsProcessService;
  35 import org.opendaylight.yang.gen.v1.nbi.notifications.rev230728.GetNotificationsProcessServiceInput;
  36 import org.opendaylight.yang.gen.v1.nbi.notifications.rev230728.GetNotificationsProcessServiceOutput;
  37 import org.opendaylight.yang.gen.v1.nbi.notifications.rev230728.GetNotificationsProcessServiceOutputBuilder;
  38 import org.opendaylight.yang.gen.v1.nbi.notifications.rev230728.NbiNotificationsService;
  39 import org.opendaylight.yang.gen.v1.nbi.notifications.rev230728.NotificationAlarmService;
  40 import org.opendaylight.yang.gen.v1.nbi.notifications.rev230728.NotificationProcessService;
  41 import org.opendaylight.yang.gen.v1.nbi.notifications.rev230728.NotificationTapiService;
  42 import org.opendaylight.yang.gen.v1.nbi.notifications.rev230728.get.notifications.alarm.service.output.NotificationsAlarmService;
  43 import org.opendaylight.yang.gen.v1.nbi.notifications.rev230728.get.notifications.process.service.output.NotificationsProcessService;
  44 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.common.rev221121.Context;
  45 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.common.rev221121.OBJECTTYPE;
  46 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.common.rev221121.OBJECTTYPEPROFILE;
  47 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.common.rev221121.OBJECTTYPESERVICEINTERFACEPOINT;
  48 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.common.rev221121.OBJECTTYPETAPICONTEXT;
  49 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.common.rev221121.Uuid;
  50 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.connectivity.rev221121.CONNECTIVITYOBJECTTYPE;
  51 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.connectivity.rev221121.CONNECTIVITYOBJECTTYPECONNECTION;
  52 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.connectivity.rev221121.CONNECTIVITYOBJECTTYPECONNECTIONENDPOINT;
  53 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.connectivity.rev221121.CONNECTIVITYOBJECTTYPECONNECTIVITYSERVICE;
  54 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.Context1;
  55 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.CreateNotificationSubscriptionService;
  56 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.CreateNotificationSubscriptionServiceInput;
  57 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.CreateNotificationSubscriptionServiceOutput;
  58 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.CreateNotificationSubscriptionServiceOutputBuilder;
  59 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.DeleteNotificationSubscriptionService;
  60 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.DeleteNotificationSubscriptionServiceInput;
  61 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.DeleteNotificationSubscriptionServiceOutput;
  62 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.DeleteNotificationSubscriptionServiceOutputBuilder;
  63 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.GetNotificationList;
  64 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.GetNotificationListInput;
  65 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.GetNotificationListOutput;
  66 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.GetNotificationListOutputBuilder;
  67 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.GetNotificationSubscriptionServiceDetails;
  68 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.GetNotificationSubscriptionServiceDetailsInput;
  69 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.GetNotificationSubscriptionServiceDetailsOutput;
  70 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.GetNotificationSubscriptionServiceDetailsOutputBuilder;
  71 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.GetNotificationSubscriptionServiceList;
  72 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.GetNotificationSubscriptionServiceListInput;
  73 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.GetNotificationSubscriptionServiceListOutput;
  74 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.GetNotificationSubscriptionServiceListOutputBuilder;
  75 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.GetSupportedNotificationTypes;
  76 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.GetSupportedNotificationTypesInput;
  77 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.GetSupportedNotificationTypesOutput;
  78 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.GetSupportedNotificationTypesOutputBuilder;
  79 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.NOTIFICATIONTYPE;
  80 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.NOTIFICATIONTYPEATTRIBUTEVALUECHANGE;
  81 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.NOTIFICATIONTYPEOBJECTCREATION;
  82 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.NOTIFICATIONTYPEOBJECTDELETION;
  83 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.TapiNotificationService;
  84 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.UpdateNotificationSubscriptionService;
  85 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.UpdateNotificationSubscriptionServiceInput;
  86 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.UpdateNotificationSubscriptionServiceOutput;
  87 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.context.NotificationContext;
  88 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.context.NotificationContextBuilder;
  89 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.create.notification.subscription.service.output.SubscriptionService;
  90 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.create.notification.subscription.service.output.SubscriptionServiceBuilder;
  91 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.get.notification.list.output.Notification;
  92 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.get.notification.list.output.NotificationKey;
  93 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.get.notification.subscription.service.list.output.SubscriptionServiceKey;
  94 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.notification.context.NotifSubscription;
  95 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.notification.context.NotifSubscriptionBuilder;
  96 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.notification.context.NotifSubscriptionKey;
  97 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.notification.subscription.service.SubscriptionFilter;
  98 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.notification.subscription.service.SubscriptionFilterBuilder;
  99 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.notification.rev221121.notification.subscription.service.SubscriptionFilterKey;
 100 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.topology.rev221121.TOPOLOGYOBJECTTYPEINTERRULEGROUP;
 101 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.topology.rev221121.TOPOLOGYOBJECTTYPELINK;
 102 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.topology.rev221121.TOPOLOGYOBJECTTYPENODE;
 103 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.topology.rev221121.TOPOLOGYOBJECTTYPENODEEDGEPOINT;
 104 import org.opendaylight.yang.gen.v1.urn.onf.otcc.yang.tapi.topology.rev221121.TOPOLOGYOBJECTTYPENODERULEGROUP;
 105 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
 106 import org.opendaylight.yangtools.yang.binding.Rpc;
 107 import org.opendaylight.yangtools.yang.common.ErrorTag;
 108 import org.opendaylight.yangtools.yang.common.ErrorType;
 109 import org.opendaylight.yangtools.yang.common.RpcResult;
 110 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
 111 import org.slf4j.Logger;
 112 import org.slf4j.LoggerFactory;
 113
 114 public class NbiNotificationsImpl implements NbiNotificationsService, TapiNotificationService {
 115     private static final Logger LOG = LoggerFactory.getLogger(NbiNotificationsImpl.class);
 116     private final JsonStringConverter<NotificationProcessService> converterService;
 117     private final JsonStringConverter<NotificationAlarmService> converterAlarmService;
 118     private final JsonStringConverter<NotificationTapiService> converterTapiService;
 119     private final String server;
 120     private final NetworkTransactionService networkTransactionService;
 121     private final TopicManager topicManager;
 122
 123     public NbiNotificationsImpl(JsonStringConverter<NotificationProcessService> converterService,
 124                                 JsonStringConverter<NotificationAlarmService> converterAlarmService,
 125                                 JsonStringConverter<NotificationTapiService> converterTapiService, String server,
 126                                 NetworkTransactionService networkTransactionService, TopicManager topicManager) {
 127         this.converterService = converterService;
 128         this.converterAlarmService = converterAlarmService;
 129         this.converterTapiService = converterTapiService;
 130         this.server = server;
 131         this.networkTransactionService = networkTransactionService;
 132         this.topicManager = topicManager;
 133     }
 134
 135     @Override
 136     public ListenableFuture<RpcResult<GetNotificationsProcessServiceOutput>> getNotificationsProcessService(
 137             GetNotificationsProcessServiceInput input) {
 138         LOG.info("RPC getNotificationsService received");
 139         if (input == null || input.getIdConsumer() == null || input.getGroupId() == null) {
 140             LOG.warn("Missing mandatory params for input {}", input);
 141             return RpcResultBuilder.success(new GetNotificationsProcessServiceOutputBuilder().build()).buildFuture();
 142         }
 143         Subscriber<NotificationProcessService, NotificationsProcessService> subscriber = new Subscriber<>(
 144                 input.getIdConsumer(), input.getGroupId(), server, converterService,
 145                 NotificationServiceDeserializer.class);
 146         List<NotificationsProcessService> notificationServiceList = subscriber
 147                 .subscribe(input.getConnectionType().getName(), NotificationsProcessService.QNAME);
 148         return RpcResultBuilder.success(new GetNotificationsProcessServiceOutputBuilder()
 149                 .setNotificationsProcessService(notificationServiceList).build()).buildFuture();
 150     }
 151
 152     @Override
 153     public ListenableFuture<RpcResult<GetNotificationsAlarmServiceOutput>> getNotificationsAlarmService(
 154             GetNotificationsAlarmServiceInput input) {
 155         LOG.info("RPC getNotificationsAlarmService received");
 156         if (input == null || input.getIdConsumer() == null || input.getGroupId() == null) {
 157             LOG.warn("Missing mandatory params for input {}", input);
 158             return RpcResultBuilder.success(new GetNotificationsAlarmServiceOutputBuilder().build()).buildFuture();
 159         }
 160         Subscriber<NotificationAlarmService, NotificationsAlarmService> subscriber = new Subscriber<>(
 161                 input.getIdConsumer(), input.getGroupId(), server, converterAlarmService,
 162                 NotificationAlarmServiceDeserializer.class);
 163         List<NotificationsAlarmService> notificationAlarmServiceList = subscriber
 164                 .subscribe("alarm" + input.getConnectionType().getName(), NotificationsAlarmService.QNAME);
 165         return RpcResultBuilder.success(new GetNotificationsAlarmServiceOutputBuilder()
 166                 .setNotificationsAlarmService(notificationAlarmServiceList).build()).buildFuture();
 167     }
 168
 169     @Override
 170     public ListenableFuture<RpcResult<GetSupportedNotificationTypesOutput>>
 171             getSupportedNotificationTypes(GetSupportedNotificationTypesInput input) {
 172         NotificationContext notificationContext = getNotificationContext();
 173         if (notificationContext == null) {
 174             return RpcResultBuilder.<GetSupportedNotificationTypesOutput>failed()
 175                 .withError(ErrorType.APPLICATION, "Couldnt get Notification Context from Datastore")
 176                 .buildFuture();
 177         }
 178         //TAPI 2.4 removes supported notification types from notif-subscription list and notification-context
 179         //No way to store what notification types are supported
 180         //Considers that by default all notification are supported
 181         Set<NOTIFICATIONTYPE> notificationTypeList = new HashSet<>();
 182         notificationTypeList.add(NOTIFICATIONTYPEOBJECTCREATION.VALUE);
 183         notificationTypeList.add(NOTIFICATIONTYPEOBJECTDELETION.VALUE);
 184         notificationTypeList.add(NOTIFICATIONTYPEATTRIBUTEVALUECHANGE.VALUE);
 185 //
 186 //        if (notificationContext.getNotifSubscription() == null) {
 187 //            return RpcResultBuilder.success(new GetSupportedNotificationTypesOutputBuilder()
 188 //                .setSupportedNotificationTypes(new HashSet<>())
 189 //                .setSupportedObjectTypes(new HashSet<>()).build()).buildFuture();
 190 //        }
 191 //        Set<NOTIFICATIONTYPE> notificationTypeList = new HashSet<>();
 192
 193         //TAPI 2.4 removes supported object types from notif-subscription list and notification-context
 194         //No way to store what object types are supported
 195         //Considers that by default all object are supported
 196         Set<OBJECTTYPE> objectTypeList = new HashSet<>();
 197         objectTypeList.add(OBJECTTYPESERVICEINTERFACEPOINT.VALUE);
 198         objectTypeList.add(OBJECTTYPETAPICONTEXT.VALUE);
 199         objectTypeList.add(OBJECTTYPEPROFILE.VALUE);
 200         objectTypeList.add(TOPOLOGYOBJECTTYPENODE.VALUE);
 201         objectTypeList.add(TOPOLOGYOBJECTTYPELINK.VALUE);
 202         objectTypeList.add(TOPOLOGYOBJECTTYPENODEEDGEPOINT.VALUE);
 203         objectTypeList.add(TOPOLOGYOBJECTTYPENODERULEGROUP.VALUE);
 204         objectTypeList.add(TOPOLOGYOBJECTTYPEINTERRULEGROUP.VALUE);
 205         objectTypeList.add(CONNECTIVITYOBJECTTYPE.VALUE);
 206         objectTypeList.add(CONNECTIVITYOBJECTTYPECONNECTIVITYSERVICE.VALUE);
 207         objectTypeList.add(CONNECTIVITYOBJECTTYPECONNECTIONENDPOINT.VALUE);
 208         objectTypeList.add(CONNECTIVITYOBJECTTYPECONNECTION.VALUE);
 209 //        for (NotifSubscription notifSubscription:notificationContext.getNotifSubscription().values()) {
 210 //            if (notifSubscription.getSupportedNotificationTypes() != null) {
 211 //                notificationTypeList.addAll(notifSubscription.getSupportedNotificationTypes());
 212 //            }
 213 //            if (notifSubscription.getSupportedObjectTypes() != null) {
 214 //                objectTypeList.addAll(notifSubscription.getSupportedObjectTypes());
 215 //            }
 216 //        }
 217         return RpcResultBuilder.success(new GetSupportedNotificationTypesOutputBuilder()
 218             .setSupportedNotificationTypes(notificationTypeList)
 219             .setSupportedObjectTypes(objectTypeList).build()).buildFuture();
 220     }
 221
 222     @Override
 223     public ListenableFuture<RpcResult<CreateNotificationSubscriptionServiceOutput>>
 224             createNotificationSubscriptionService(CreateNotificationSubscriptionServiceInput input) {
 225         for (Uuid uuid:input.getSubscriptionFilter().getRequestedObjectIdentifier()) {
 226             LOG.info("Adding T-API topic: {} to Kafka server", uuid.getValue());
 227             this.topicManager.addTapiTopic(uuid.getValue());
 228         }
 229         SubscriptionFilter subscriptionFilter = new SubscriptionFilterBuilder()
 230             .setName(input.getSubscriptionFilter().getName())
 231             .setLocalId(input.getSubscriptionFilter().getLocalId())
 232             .setIncludeContent(input.getSubscriptionFilter().getIncludeContent())
 233             .setRequestedNotificationTypes(input.getSubscriptionFilter().getRequestedNotificationTypes())
 234             .setRequestedLayerProtocols(input.getSubscriptionFilter().getRequestedLayerProtocols())
 235             .setRequestedObjectIdentifier(input.getSubscriptionFilter().getRequestedObjectIdentifier())
 236             .setRequestedObjectTypes(input.getSubscriptionFilter().getRequestedObjectTypes())
 237             .build();
 238         Uuid notifSubscriptionUuid = new Uuid(UUID.randomUUID().toString());
 239         Map<SubscriptionFilterKey, SubscriptionFilter> sfmap = new HashMap<>();
 240         sfmap.put(subscriptionFilter.key(), subscriptionFilter);
 241         SubscriptionService subscriptionService = new SubscriptionServiceBuilder()
 242             .setSubscriptionFilter(sfmap)
 243             .setSubscriptionState(input.getSubscriptionState())
 244             .setUuid(notifSubscriptionUuid)
 245             .build();
 246
 247         NotifSubscriptionKey notifSubscriptionKey = new NotifSubscriptionKey(notifSubscriptionUuid);
 248         NotifSubscription notifSubscription = new NotifSubscriptionBuilder()
 249             .setSubscriptionState(subscriptionService.getSubscriptionState())
 250             .setSubscriptionFilter(subscriptionService.getSubscriptionFilter())
 251             .setUuid(notifSubscriptionUuid)
 252 //            Following 2 items are no more in notification-context with T-API 2.4
 253 //            .setSupportedNotificationTypes(notificationTypes)
 254 //            .setSupportedObjectTypes(objectTypes)
 255             .setName(subscriptionService.getName())
 256             .build();
 257         NotificationContext notificationContext = getNotificationContext();
 258         Map<NotifSubscriptionKey, NotifSubscription> notifSubscriptions = new HashMap<>();
 259         if (notificationContext != null && notificationContext.getNotifSubscription() != null) {
 260             notifSubscriptions.putAll(notificationContext.getNotifSubscription());
 261         }
 262         notifSubscriptions.put(notifSubscriptionKey, notifSubscription);
 263         NotificationContext notificationContext1 = new NotificationContextBuilder()
 264             .setNotification(notificationContext == null ? new HashMap<>() : notificationContext.getNotification())
 265             .setNotifSubscription(notifSubscriptions)
 266             .build();
 267         if (!updateNotificationContext(notificationContext1)) {
 268             LOG.error("Failed to update Notification context");
 269             return RpcResultBuilder.<CreateNotificationSubscriptionServiceOutput>failed()
 270                 .withError(ErrorType.RPC, "Failed to update notification context").buildFuture();
 271         }
 272         CreateNotificationSubscriptionServiceOutput serviceOutput =
 273             new CreateNotificationSubscriptionServiceOutputBuilder()
 274                 .setSubscriptionService(subscriptionService)
 275                 .build();
 276         return RpcResultBuilder.success(serviceOutput).buildFuture();
 277     }
 278
 279     @Override
 280     public ListenableFuture<RpcResult<UpdateNotificationSubscriptionServiceOutput>>
 281             updateNotificationSubscriptionService(UpdateNotificationSubscriptionServiceInput input) {
 282         // TODO --> Not yet implemented
 283         return RpcResultBuilder.<UpdateNotificationSubscriptionServiceOutput>failed()
 284             .withError(ErrorType.RPC, ErrorTag.OPERATION_NOT_SUPPORTED, "RPC not implemented yet")
 285             .buildFuture();
 286     }
 287
 288     @Override
 289     public ListenableFuture<RpcResult<DeleteNotificationSubscriptionServiceOutput>>
 290             deleteNotificationSubscriptionService(DeleteNotificationSubscriptionServiceInput input) {
 291         try {
 292             if (input == null || input.getUuid() == null) {
 293                 LOG.warn("Missing mandatory params for input {}", input);
 294                 return RpcResultBuilder.<DeleteNotificationSubscriptionServiceOutput>failed()
 295                     .withError(ErrorType.RPC, "Missing input parameters").buildFuture();
 296             }
 297             Uuid notifSubsUuid = input.getUuid();
 298             InstanceIdentifier<NotifSubscription> notifSubscriptionIID = InstanceIdentifier.builder(Context.class)
 299                 .augmentation(Context1.class).child(NotificationContext.class).child(NotifSubscription.class,
 300                     new NotifSubscriptionKey(notifSubsUuid)).build();
 301             Optional<NotifSubscription> optionalNotifSub = this.networkTransactionService.read(
 302                 LogicalDatastoreType.OPERATIONAL, notifSubscriptionIID).get();
 303
 304             if (optionalNotifSub.isEmpty()) {
 305                 return RpcResultBuilder.<DeleteNotificationSubscriptionServiceOutput>failed()
 306                     .withError(ErrorType.APPLICATION,
 307                         "Notification subscription doesnt exist").buildFuture();
 308             }
 309             NotifSubscription notifSubscription = optionalNotifSub.orElseThrow();
 310             this.networkTransactionService.delete(LogicalDatastoreType.OPERATIONAL, notifSubscriptionIID);
 311             this.networkTransactionService.commit().get();
 312             for (Map.Entry<SubscriptionFilterKey, SubscriptionFilter> sfEntry : notifSubscription
 313                     .getSubscriptionFilter().entrySet()) {
 314                 for (Uuid objectUuid:sfEntry.getValue().getRequestedObjectIdentifier()) {
 315                     this.topicManager.deleteTapiTopic(objectUuid.getValue());
 316                 }
 317             }
 318 //            for (Uuid objectUuid:notifSubscription.getSubscriptionFilter().getRequestedObjectIdentifier()) {
 319 //                this.topicManager.deleteTapiTopic(objectUuid.getValue());
 320 //            }
 321             return RpcResultBuilder.success(new DeleteNotificationSubscriptionServiceOutputBuilder().build())
 322                 .buildFuture();
 323         } catch (InterruptedException | ExecutionException | NoSuchElementException e) {
 324             LOG.error("Failed to delete Notification subscription service", e);
 325         }
 326         return RpcResultBuilder.<DeleteNotificationSubscriptionServiceOutput>failed()
 327             .withError(ErrorType.APPLICATION,
 328                 "Failed to delete notification subscription service").buildFuture();
 329     }
 330
 331     @Override
 332     public ListenableFuture<RpcResult<GetNotificationSubscriptionServiceDetailsOutput>>
 333             getNotificationSubscriptionServiceDetails(GetNotificationSubscriptionServiceDetailsInput input) {
 334         if (input == null || input.getUuid() == null) {
 335             LOG.warn("Missing mandatory params for input {}", input);
 336             return RpcResultBuilder.<GetNotificationSubscriptionServiceDetailsOutput>failed()
 337                 .withError(ErrorType.RPC, "Missing input parameters").buildFuture();
 338         }
 339         Uuid notifSubsUuid = input.getUuid();
 340         NotificationContext notificationContext = getNotificationContext();
 341         if (notificationContext == null) {
 342             return RpcResultBuilder.<GetNotificationSubscriptionServiceDetailsOutput>failed()
 343                 .withError(ErrorType.APPLICATION, "Notification context is empty")
 344                 .buildFuture();
 345         }
 346         if (notificationContext.getNotifSubscription() == null) {
 347             return RpcResultBuilder.success(new GetNotificationSubscriptionServiceDetailsOutputBuilder()
 348                 .setSubscriptionService(new org.opendaylight.yang.gen.v1
 349                     .urn.onf.otcc.yang.tapi.notification.rev221121.get.notification.subscription.service
 350                         .details.output.SubscriptionServiceBuilder().build()).build()).buildFuture();
 351         }
 352         if (!notificationContext.getNotifSubscription().containsKey(new NotifSubscriptionKey(notifSubsUuid))) {
 353             return RpcResultBuilder.<GetNotificationSubscriptionServiceDetailsOutput>failed()
 354                 .withError(ErrorType.APPLICATION,
 355                     "Notification subscription service doesnt exist").buildFuture();
 356         }
 357         return RpcResultBuilder.success(new GetNotificationSubscriptionServiceDetailsOutputBuilder()
 358             .setSubscriptionService(new org.opendaylight.yang.gen.v1.urn
 359                 .onf.otcc.yang.tapi.notification.rev221121.get.notification.subscription.service.details.output
 360                 .SubscriptionServiceBuilder(notificationContext.getNotifSubscription().get(
 361                     new NotifSubscriptionKey(notifSubsUuid))).build()).build()).buildFuture();
 362     }
 363
 364     @Override
 365     public ListenableFuture<RpcResult<GetNotificationSubscriptionServiceListOutput>>
 366             getNotificationSubscriptionServiceList(GetNotificationSubscriptionServiceListInput input) {
 367         NotificationContext notificationContext = getNotificationContext();
 368         if (notificationContext == null) {
 369             return RpcResultBuilder.<GetNotificationSubscriptionServiceListOutput>failed()
 370                 .withError(ErrorType.APPLICATION, "Notification context is empty")
 371                 .buildFuture();
 372         }
 373         if (notificationContext.getNotifSubscription() == null) {
 374             return RpcResultBuilder.success(new GetNotificationSubscriptionServiceListOutputBuilder()
 375                 .setSubscriptionService(new HashMap<>()).build()).buildFuture();
 376         }
 377         Map<SubscriptionServiceKey, org.opendaylight.yang.gen.v1.urn.onf.otcc.yang
 378             .tapi.notification.rev221121.get.notification.subscription.service.list.output.SubscriptionService>
 379                 notifSubsMap = new HashMap<>();
 380         for (NotifSubscription notifSubscription:notificationContext.getNotifSubscription().values()) {
 381             org.opendaylight.yang.gen.v1.urn.onf.otcc.yang
 382                 .tapi.notification.rev221121.get.notification.subscription.service.list.output.SubscriptionService
 383                     subscriptionService = new org.opendaylight.yang.gen.v1
 384                         .urn.onf.otcc.yang.tapi.notification.rev221121.get.notification.subscription.service
 385                             .list.output.SubscriptionServiceBuilder(notifSubscription).build();
 386             notifSubsMap.put(subscriptionService.key(), subscriptionService);
 387         }
 388         return RpcResultBuilder.success(new GetNotificationSubscriptionServiceListOutputBuilder()
 389             .setSubscriptionService(notifSubsMap).build()).buildFuture();
 390     }
 391
 392     @Override
 393     public ListenableFuture<RpcResult<GetNotificationListOutput>> getNotificationList(GetNotificationListInput input) {
 394         try {
 395             LOG.info("RPC getNotificationList received");
 396             if (input == null || input.getSubscriptionId() == null) {
 397                 LOG.warn("Missing mandatory params for input {}", input);
 398                 return RpcResultBuilder.<GetNotificationListOutput>failed().withError(ErrorType.RPC,
 399                     "Missing input parameters").buildFuture();
 400             }
 401             Uuid notifSubsUuid = input.getSubscriptionId();
 402             InstanceIdentifier<NotifSubscription> notifSubscriptionIID = InstanceIdentifier.builder(Context.class)
 403                 .augmentation(Context1.class).child(NotificationContext.class).child(NotifSubscription.class,
 404                     new NotifSubscriptionKey(notifSubsUuid)).build();
 405             Optional<NotifSubscription> optionalNotifSub = this.networkTransactionService.read(
 406                 LogicalDatastoreType.OPERATIONAL, notifSubscriptionIID).get();
 407
 408             if (optionalNotifSub.isEmpty()) {
 409                 return RpcResultBuilder.<GetNotificationListOutput>failed()
 410                     .withError(ErrorType.APPLICATION,
 411                         "Notification subscription doesnt exist").buildFuture();
 412             }
 413             NotifSubscription notifSubscription = optionalNotifSub.orElseThrow();
 414             List<Notification> notificationTapiList = new ArrayList<>();
 415             for (Map.Entry<SubscriptionFilterKey, SubscriptionFilter> sfEntry : notifSubscription
 416                     .getSubscriptionFilter().entrySet()) {
 417                 for (Uuid objectUuid:sfEntry.getValue().getRequestedObjectIdentifier()) {
 418                     if (!this.topicManager.getTapiTopicMap().containsKey(objectUuid.getValue())) {
 419                         LOG.warn("Topic doesnt exist for {}", objectUuid.getValue());
 420                         continue;
 421                     }
 422                     LOG.info("Going to get notifications for topic {}", objectUuid.getValue());
 423                     Subscriber<NotificationTapiService, Notification> subscriber = new Subscriber<>(
 424                         objectUuid.getValue(), objectUuid.getValue(), server, converterTapiService,
 425                         TapiNotificationDeserializer.class);
 426                     notificationTapiList.addAll(subscriber.subscribe(objectUuid.getValue(), Notification.QNAME));
 427                 }
 428             }
 429 //            for (Uuid objectUuid:notifSubscription.getSubscriptionFilter().getRequestedObjectIdentifier()) {
 430 //                if (!this.topicManager.getTapiTopicMap().containsKey(objectUuid.getValue())) {
 431 //                    LOG.warn("Topic doesnt exist for {}", objectUuid.getValue());
 432 //                    continue;
 433 //                }
 434 //                LOG.info("Going to get notifications for topic {}", objectUuid.getValue());
 435 //                Subscriber<NotificationTapiService, Notification> subscriber = new Subscriber<>(
 436 //                    objectUuid.getValue(), objectUuid.getValue(), server, converterTapiService,
 437 //                    TapiNotificationDeserializer.class);
 438 //                notificationTapiList.addAll(subscriber.subscribe(objectUuid.getValue(), Notification.QNAME));
 439 //            }
 440             LOG.info("TAPI notifications = {}", notificationTapiList);
 441             Map<NotificationKey, Notification> notificationMap = new HashMap<>();
 442             for (Notification notif:notificationTapiList) {
 443                 notificationMap.put(notif.key(), notif);
 444             }
 445             return RpcResultBuilder.success(new GetNotificationListOutputBuilder()
 446                 .setNotification(notificationMap).build()).buildFuture();
 447         } catch (InterruptedException | ExecutionException | NoSuchElementException e) {
 448             LOG.error("Failed to get Notifications from Kafka", e);
 449         }
 450         return RpcResultBuilder.<GetNotificationListOutput>failed()
 451             .withError(ErrorType.APPLICATION,
 452                 "Notifications couldnt be retrieved from Kafka server").buildFuture();
 453     }
 454
 455     public ImmutableClassToInstanceMap<Rpc<?, ?>> registerRPCs() {
 456         return ImmutableClassToInstanceMap.<Rpc<?, ?>>builder()
 457             .put(GetNotificationsProcessService.class, this::getNotificationsProcessService)
 458             .put(GetNotificationsAlarmService.class, this::getNotificationsAlarmService)
 459             .put(GetSupportedNotificationTypes.class, this::getSupportedNotificationTypes)
 460             .put(CreateNotificationSubscriptionService.class, this::createNotificationSubscriptionService)
 461             .put(UpdateNotificationSubscriptionService.class, this::updateNotificationSubscriptionService)
 462             .put(DeleteNotificationSubscriptionService.class, this::deleteNotificationSubscriptionService)
 463             .put(GetNotificationSubscriptionServiceDetails.class, this::getNotificationSubscriptionServiceDetails)
 464             .put(GetNotificationSubscriptionServiceList.class, this::getNotificationSubscriptionServiceList)
 465             .put(GetNotificationList.class, this::getNotificationList)
 466             .build();
 467     }
 468
 469     private NotificationContext getNotificationContext() {
 470         LOG.info("Getting tapi notification context");
 471         try {
 472             InstanceIdentifier<NotificationContext> notificationcontextIID =
 473                 InstanceIdentifier.builder(Context.class).augmentation(Context1.class)
 474                     .child(NotificationContext.class).build();
 475             Optional<NotificationContext> notificationContextOptional
 476                 = this.networkTransactionService.read(LogicalDatastoreType.OPERATIONAL, notificationcontextIID).get();
 477             if (!notificationContextOptional.isPresent()) {
 478                 LOG.error("Could not get TAPI notification context");
 479                 return null;
 480             }
 481             return notificationContextOptional.orElseThrow();
 482         } catch (InterruptedException | ExecutionException e) {
 483             LOG.error("Could not get TAPI notification context");
 484         }
 485         return null;
 486     }
 487
 488     private boolean updateNotificationContext(NotificationContext notificationContext1) {
 489         try {
 490             InstanceIdentifier<NotificationContext> notificationcontextIID =
 491                 InstanceIdentifier.builder(Context.class).augmentation(Context1.class)
 492                     .child(NotificationContext.class).build();
 493             this.networkTransactionService.merge(LogicalDatastoreType.OPERATIONAL, notificationcontextIID,
 494                 notificationContext1);
 495             this.networkTransactionService.commit().get();
 496             return true;
 497         } catch (InterruptedException | ExecutionException e) {
 498             LOG.error("Could not update TAPI notification context");
 499         }
 500         return false;
 501     }
 502
 503 }