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

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * ONAP PAP
   4  * ================================================================================
   5  * Copyright (C) 2019-2021 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.concurrent.BlockingQueue;
  24 import java.util.concurrent.LinkedBlockingQueue;
  25 import org.onap.policy.common.endpoints.event.comm.client.TopicSinkClient;
  26 import org.onap.policy.common.endpoints.event.comm.client.TopicSinkClientException;
  27 import org.onap.policy.common.utils.coder.StandardCoder;
  28 import org.onap.policy.pap.main.PolicyPapException;
  29 import org.slf4j.Logger;
  30 import org.slf4j.LoggerFactory;
  31
  32 /**
  33  * Publishes messages to a topic. Maintains a queue of references to data that is to be
  34  * published. Once the publisher removes a reference from a queue, it sets it to
  35  * {@link null} to indicate that it is being processed. Until it has been set to
  36  * {@link null}, clients are free to atomically update the reference to new values, thus
  37  * maintaining their place in the queue.
  38  *
  39  * <p>This class has not been tested for multiple threads invoking {@link #run()}
  40  * simultaneously.
  41  *
  42  * @param <T> type of message published by this publisher
  43  */
  44 public class Publisher<T> implements Runnable {
  45     private static final Logger logger = LoggerFactory.getLogger(Publisher.class);
  46
  47     /**
  48      * Used to send to the topic.
  49      */
  50     private final TopicSinkClient client;
  51
  52     /**
  53      * Request queue. The references may contain {@code null}.
  54      */
  55     private final BlockingQueue<QueueToken<T>> queue = new LinkedBlockingQueue<>();
  56
  57     /**
  58      * Set to {@code true} to cause the publisher to stop running.
  59      */
  60     private volatile boolean stopNow = false;
  61
  62     /**
  63      * Constructs the object.
  64      *
  65      * @param topic name of the topic to which to publish
  66      * @throws PolicyPapException if the topic sink does not exist
  67      */
  68     public Publisher(String topic) throws PolicyPapException {
  69         try {
  70             this.client = new TopicSinkClient(topic);
  71         } catch (TopicSinkClientException e) {
  72             throw new PolicyPapException(e);
  73         }
  74     }
  75
  76     /**
  77      * Stops the publisher, if it's running.
  78      */
  79     public void stop() {
  80         stopNow = true;
  81
  82         // add an empty reference so the thread doesn't block on the queue
  83         queue.add(new QueueToken<>(null));
  84     }
  85
  86     /**
  87      * Adds an item to the queue. The referenced objects are assumed to be POJOs and will
  88      * be converted to JSON via the {@link StandardCoder} prior to publishing.
  89      *
  90      * @param ref reference to the message to be published
  91      */
  92     public void enqueue(QueueToken<T> ref) {
  93         queue.add(ref);
  94     }
  95
  96     /**
  97      * Continuously publishes items in the queue until {@link #stop()} is invoked.
  98      */
  99     @Override
 100     public void run() {
 101         for (;;) {
 102             QueueToken<T> token = getNext();
 103
 104             if (stopNow) {
 105                 // unblock any other publisher threads
 106                 queue.add(new QueueToken<>(null));
 107                 break;
 108             }
 109
 110             var data = token.replaceItem(null);
 111             if (data != null) {
 112                 client.send(data);
 113             }
 114         }
 115     }
 116
 117     /**
 118      * Gets the next item from the queue. If the thread is interrupted, then it sets
 119      * {@link #stopNow}.
 120      *
 121      * @return the next item, or a reference containing {@code null} if this is
 122      *         interrupted
 123      */
 124     private QueueToken<T> getNext() {
 125         try {
 126             return queue.take();
 127
 128         } catch (InterruptedException e) {
 129             logger.warn("Publisher stopping due to interrupt");
 130             stopNow = true;
 131             Thread.currentThread().interrupt();
 132             return new QueueToken<>(null);
 133         }
 134     }
 135 }