models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/topic/Forwarder.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * ONAP
   4  * ================================================================================
   5  * Copyright (C) 2020 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.controlloop.actorserviceprovider.topic;
  22
  23 import java.util.ArrayList;
  24 import java.util.List;
  25 import java.util.Map;
  26 import java.util.concurrent.ConcurrentHashMap;
  27 import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
  28 import org.onap.policy.common.endpoints.utils.PropertyUtils.TriConsumer;
  29 import org.onap.policy.common.utils.coder.StandardCoderObject;
  30 import org.onap.policy.controlloop.actorserviceprovider.Util;
  31 import org.slf4j.Logger;
  32 import org.slf4j.LoggerFactory;
  33
  34 /**
  35  * Forwarder that selectively forwards message to listeners based on the content of the
  36  * message. Each forwarder is associated with a single set of selector keys. Listeners are
  37  * then registered with that forwarder for a particular set of values for the given keys.
  38  */
  39 public class Forwarder {
  40     private static final Logger logger = LoggerFactory.getLogger(Forwarder.class);
  41
  42     /**
  43      * Maps a set of field values to one or more listeners.
  44      */
  45     // @formatter:off
  46     private final Map<List<String>, Map<TriConsumer<CommInfrastructure, String, StandardCoderObject>, String>>
  47                 values2listeners = new ConcurrentHashMap<>();
  48     // @formatter:on
  49
  50     /**
  51      * Keys used to extract the field values from the {@link StandardCoderObject}.
  52      */
  53     private final List<SelectorKey> keys;
  54
  55     /**
  56      * Constructs the object.
  57      *
  58      * @param keys keys used to extract the field's value from the
  59      *        {@link StandardCoderObject}
  60      */
  61     public Forwarder(List<SelectorKey> keys) {
  62         this.keys = keys;
  63     }
  64
  65     /**
  66      * Registers a listener for messages containing the given field values.
  67      *
  68      * @param values field values of interest, in one-to-one correspondence with the keys
  69      * @param listener listener to register
  70      */
  71     public void register(List<String> values, TriConsumer<CommInfrastructure, String, StandardCoderObject> listener) {
  72         if (keys.size() != values.size()) {
  73             throw new IllegalArgumentException("key/value mismatch");
  74         }
  75
  76         values2listeners.compute(values, (key, listeners) -> {
  77             Map<TriConsumer<CommInfrastructure, String, StandardCoderObject>, String> map = listeners;
  78             if (map == null) {
  79                 map = new ConcurrentHashMap<>();
  80             }
  81
  82             map.put(listener, "");
  83             return map;
  84         });
  85     }
  86
  87     /**
  88      * Unregisters a listener for messages containing the given field values.
  89      *
  90      * @param values field values of interest, in one-to-one correspondence with the keys
  91      * @param listener listener to unregister
  92      */
  93     public void unregister(List<String> values, TriConsumer<CommInfrastructure, String, StandardCoderObject> listener) {
  94         values2listeners.computeIfPresent(values, (key, listeners) -> {
  95             listeners.remove(listener);
  96             return (listeners.isEmpty() ? null : listeners);
  97         });
  98     }
  99
 100     /**
 101      * Processes a message, forwarding it to the appropriate listeners, if any.
 102      *
 103      * @param infra communication infrastructure on which the response was received
 104      * @param textMessage original text message that was received
 105      * @param scoMessage decoded text message
 106      */
 107     public void onMessage(CommInfrastructure infra, String textMessage, StandardCoderObject scoMessage) {
 108         // extract the key values from the message
 109         List<String> values = new ArrayList<>(keys.size());
 110         for (SelectorKey key : keys) {
 111             String value = key.extractField(scoMessage);
 112             if (value == null) {
 113                 /*
 114                  * No value for this field, so this message is not relevant to this
 115                  * forwarder.
 116                  */
 117                 return;
 118             }
 119
 120             values.add(value);
 121         }
 122
 123         // get the listeners for this set of values
 124         Map<TriConsumer<CommInfrastructure, String, StandardCoderObject>, String> listeners =
 125                         values2listeners.get(values);
 126         if (listeners == null) {
 127             // no listeners for this particular list of values
 128             return;
 129         }
 130
 131
 132         // forward the message to each listener
 133         for (TriConsumer<CommInfrastructure, String, StandardCoderObject> listener : listeners.keySet()) {
 134             try {
 135                 listener.accept(infra, textMessage, scoMessage);
 136             } catch (RuntimeException e) {
 137                 logger.warn("exception thrown by listener {}", Util.ident(listener), e);
 138             }
 139         }
 140     }
 141 }