main/src/main/java/org/onap/policy/pap/main/comm/MultiPdpStatusListener.java

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * ONAP PAP
   4  * ================================================================================
   5  * Copyright (C) 2019 AT&T Intellectual Property. All rights reserved.
   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.onap.policy.pap.main.comm;
  22
  23 import java.util.Collection;
  24 import java.util.Set;
  25 import java.util.SortedSet;
  26 import java.util.TreeSet;
  27 import java.util.concurrent.ConcurrentHashMap;
  28 import java.util.concurrent.CountDownLatch;
  29 import java.util.concurrent.TimeUnit;
  30 import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
  31 import org.onap.policy.common.endpoints.listeners.TypedMessageListener;
  32 import org.onap.policy.models.pdp.concepts.PdpStatus;
  33 import org.slf4j.Logger;
  34 import org.slf4j.LoggerFactory;
  35
  36 /**
  37  * Listener for PDP Status messages expected to be received from multiple PDPs. The
  38  * listener "completes" once a message has been seen from all of the PDPs.
  39  */
  40 public abstract class MultiPdpStatusListener implements TypedMessageListener<PdpStatus> {
  41     private static final Logger logger = LoggerFactory.getLogger(MultiPdpStatusListener.class);
  42
  43     /**
  44      * This is decremented once a message has been received from every PDP.
  45      */
  46     private final CountDownLatch allSeen = new CountDownLatch(1);
  47
  48     /**
  49      * IDs for which no message has been received yet.
  50      */
  51     private final Set<String> unseenIds = ConcurrentHashMap.newKeySet();
  52
  53     /**
  54      * Constructs the object.
  55      *
  56      * @param id ID for which to wait
  57      */
  58     public MultiPdpStatusListener(String id) {
  59         unseenIds.add(id);
  60     }
  61
  62     /**
  63      * Constructs the object.
  64      *
  65      * @param ids IDs for which to wait
  66      */
  67     public MultiPdpStatusListener(Collection<String> ids) {
  68         if (ids.isEmpty()) {
  69             allSeen.countDown();
  70
  71         } else {
  72             unseenIds.addAll(ids);
  73         }
  74     }
  75
  76     /**
  77      * Gets the set of IDs for which messages have not yet been received.
  78      *
  79      * @return the IDs that have not been seen yet
  80      */
  81     public SortedSet<String> getUnseenIds() {
  82         return new TreeSet<>(unseenIds);
  83     }
  84
  85     /**
  86      * Waits for messages to be received for all PDPs, or until a timeout is reached.
  87      *
  88      * @param timeout the maximum time to wait
  89      * @param unit the time unit of the timeout argument
  90      * @return {@code true} if messages were received for all PDPs, {@code false} if the
  91      *         timeout was reached first
  92      * @throws InterruptedException if the current thread is interrupted while waiting
  93      */
  94     public boolean await(long timeout, TimeUnit unit) throws InterruptedException {
  95         return allSeen.await(timeout, unit);
  96     }
  97
  98     /**
  99      * After giving the event to the subclass via
 100      * {@link #handleEvent(CommInfrastructure, String, PdpStatus)}, this triggers
 101      * completion of {@link #await(long, TimeUnit)}, if all PDPs have received a message.
 102      */
 103     @Override
 104     public final void onTopicEvent(CommInfrastructure infra, String topic, PdpStatus message) {
 105         String id = null;
 106         try {
 107             id = handleEvent(infra, topic, message);
 108         } catch (RuntimeException e) {
 109             logger.warn("handleEvent failed due to: {}", e.getMessage(), e);
 110         }
 111
 112         if (id == null) {
 113             return;
 114         }
 115
 116         unseenIds.remove(id);
 117
 118         if (unseenIds.isEmpty()) {
 119             allSeen.countDown();
 120         }
 121     }
 122
 123     /**
 124      * Indicates that a message was received for a PDP.
 125      *
 126      * @param infra infrastructure with which the message was received
 127      * @param topic topic on which the message was received
 128      * @param message message that was received
 129      * @return the ID extracted from the message, or {@code null} if it cannot be extracted
 130      */
 131     protected abstract String handleEvent(CommInfrastructure infra, String topic, PdpStatus message);
 132 }