enrichment-coordinator-service/src/main/java/org/oransc/enrichment/repository/InfoTypeSubscriptions.java

   1 /*-
   2  * ========================LICENSE_START=================================
   3  * O-RAN-SC
   4  * %%
   5  * Copyright (C) 2019 Nordix Foundation
   6  * %%
   7  * Licensed under the Apache License, Version 2.0 (the "License");
   8  * you may not use this file except in compliance with the License.
   9  * You may obtain a copy of the License at
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  * ========================LICENSE_END===================================
  19  */
  20
  21 package org.oransc.enrichment.repository;
  22
  23 import java.lang.invoke.MethodHandles;
  24 import java.util.Collection;
  25 import java.util.HashMap;
  26 import java.util.Map;
  27 import java.util.Vector;
  28 import java.util.function.Function;
  29
  30 import lombok.Builder;
  31 import lombok.Getter;
  32
  33 import org.oransc.enrichment.exceptions.ServiceException;
  34 import org.slf4j.Logger;
  35 import org.slf4j.LoggerFactory;
  36 import org.springframework.stereotype.Component;
  37
  38 import reactor.core.publisher.Flux;
  39 import reactor.core.publisher.Mono;
  40
  41 /**
  42  * Subscriptions of callbacks for type registrations
  43  */
  44 @SuppressWarnings("squid:S2629") // Invoke method(s) only conditionally
  45 @Component
  46 public class InfoTypeSubscriptions {
  47     private final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
  48     private final Map<String, SubscriptionInfo> allSubscriptions = new HashMap<>();
  49     private final MultiMap<SubscriptionInfo> subscriptionsByOwner = new MultiMap<>();
  50
  51     public interface Callbacks {
  52         Mono<String> notifyTypeRegistered(InfoType type, SubscriptionInfo subscriptionInfo);
  53
  54         Mono<String> notifyTypeRemoved(InfoType type, SubscriptionInfo subscriptionInfo);
  55     }
  56
  57     @Builder
  58     @Getter
  59     public static class SubscriptionInfo {
  60         private String id;
  61
  62         private String callbackUrl;
  63
  64         private String owner;
  65
  66         private Callbacks callback;
  67     }
  68
  69     public synchronized void put(SubscriptionInfo subscription) {
  70         allSubscriptions.put(subscription.getId(), subscription);
  71         subscriptionsByOwner.put(subscription.owner, subscription.id, subscription);
  72         logger.debug("Added type status subscription {}", subscription.id);
  73     }
  74
  75     public synchronized Collection<SubscriptionInfo> getAllSubscriptions() {
  76         return new Vector<>(allSubscriptions.values());
  77     }
  78
  79     /**
  80      * Get a subscription and throw if not fond.
  81      *
  82      * @param id the ID of the subscription to get.
  83      * @return SubscriptionInfo
  84      * @throws ServiceException if not found
  85      */
  86     public synchronized SubscriptionInfo getSubscription(String id) throws ServiceException {
  87         SubscriptionInfo p = allSubscriptions.get(id);
  88         if (p == null) {
  89             throw new ServiceException("Could not find Information subscription: " + id);
  90         }
  91         return p;
  92     }
  93
  94     /**
  95      * Get a subscription or return null if not found. Equivalent to get in all java
  96      * collections.
  97      *
  98      * @param id the ID of the subscription to get.
  99      * @return SubscriptionInfo
 100      */
 101     public synchronized SubscriptionInfo get(String id) {
 102         return allSubscriptions.get(id);
 103     }
 104
 105     public synchronized int size() {
 106         return allSubscriptions.size();
 107     }
 108
 109     public synchronized void clear() {
 110         allSubscriptions.clear();
 111         subscriptionsByOwner.clear();
 112     }
 113
 114     public void remove(SubscriptionInfo subscription) {
 115         allSubscriptions.remove(subscription.getId());
 116         subscriptionsByOwner.remove(subscription.owner, subscription.id);
 117         logger.debug("Removed type status subscription {}", subscription.id);
 118     }
 119
 120     /**
 121      * returns all subscriptions for an owner. The colllection can contain 0..n
 122      * subscriptions.
 123      *
 124      * @param owner
 125      * @return
 126      */
 127     public synchronized Collection<SubscriptionInfo> getSubscriptionsForOwner(String owner) {
 128         return subscriptionsByOwner.get(owner);
 129     }
 130
 131     public synchronized void notifyTypeRegistered(InfoType type) {
 132         notifyAllSubscribers(subscription -> subscription.callback.notifyTypeRegistered(type, subscription));
 133     }
 134
 135     public synchronized void notifyTypeRemoved(InfoType type) {
 136         notifyAllSubscribers(subscription -> subscription.callback.notifyTypeRemoved(type, subscription));
 137     }
 138
 139     private synchronized void notifyAllSubscribers(Function<? super SubscriptionInfo, Mono<String>> notifyFunc) {
 140         final int CONCURRENCY = 5;
 141         Flux.fromIterable(allSubscriptions.values()) //
 142             .flatMap(notifyFunc::apply, CONCURRENCY) //
 143             .onErrorResume(throwable -> {
 144                 logger.warn("Post failed for consumer callback {}", throwable.getMessage());
 145                 return Flux.empty();
 146             }) //
 147             .subscribe();
 148     }
 149
 150 }