models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/provider/ConsumerGroupData.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * ONAP Policy Models
   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.models.sim.dmaap.provider;
  22
  23 import java.util.ArrayList;
  24 import java.util.Collections;
  25 import java.util.List;
  26 import java.util.concurrent.BlockingQueue;
  27 import java.util.concurrent.LinkedBlockingQueue;
  28 import java.util.concurrent.TimeUnit;
  29 import org.slf4j.Logger;
  30 import org.slf4j.LoggerFactory;
  31
  32 /**
  33  * Data associated with a consumer group. All consumer instances within a group share the
  34  * same data object.
  35  */
  36 public class ConsumerGroupData {
  37     private static final Logger logger = LoggerFactory.getLogger(ConsumerGroupData.class);
  38
  39     /**
  40      * Returned when messages can no longer be read from this consumer group object,
  41      * because it is being removed from the topic. {@link #UNREADABLE_LIST} must not be
  42      * the same list as Collections.emptyList(), thus we wrap it.
  43      */
  44     public static final List<String> UNREADABLE_LIST = Collections.unmodifiableList(Collections.emptyList());
  45
  46     /**
  47      * Returned when there are no messages read. Collections.emptyList() is already
  48      * unmodifiable, thus no need to wrap it.
  49      */
  50     private static final List<String> EMPTY_LIST = Collections.emptyList();
  51
  52     /**
  53      * This is locked while fields other than {@link #messageQueue} are updated.
  54      */
  55     private final Object lockit = new Object();
  56
  57     /**
  58      * Number of sweep cycles that have occurred since a consumer has attempted to read
  59      * from the queue. This consumer group should be removed once this count exceeds
  60      * {@code 1}, provided {@link #nreaders} is zero.
  61      */
  62     private int nsweeps = 0;
  63
  64     /**
  65      * Number of consumers that are currently attempting to read from the queue. This
  66      * consumer group should not be removed as long as this is non-zero.
  67      */
  68     private int nreaders = 0;
  69
  70     /**
  71      * Message queue.
  72      */
  73     private final BlockingQueue<String> messageQueue = new LinkedBlockingQueue<>();
  74
  75
  76     /**
  77      * Constructs the object.
  78      *
  79      * @param topicName name of the topic with which this object is associated
  80      * @param groupName name of the consumer group with which this object is associated
  81      */
  82     public ConsumerGroupData(String topicName, String groupName) {
  83         logger.info("Topic {}: add consumer group: {}", topicName, groupName);
  84     }
  85
  86     /**
  87      * Determines if this consumer group should be removed. This should be invoked once
  88      * during each sweep cycle. When this returns {@code true}, this consumer group should
  89      * be immediately discarded, as any readers will sit in a spin loop waiting for it to
  90      * be discarded.
  91      *
  92      * @return {@code true} if this consumer group should be removed, {@code false}
  93      *         otherwise
  94      */
  95     public boolean shouldRemove() {
  96         synchronized (lockit) {
  97             return (nreaders == 0 && ++nsweeps > 1);
  98         }
  99     }
 100
 101     /**
 102      * Reads messages from the queue, blocking if necessary.
 103      *
 104      * @param maxRead maximum number of messages to read
 105      * @param waitMs time to wait, in milliseconds, if the queue is currently empty
 106      * @return a list of messages read from the queue, empty if no messages became
 107      *         available before the wait time elapsed, or {@link #UNREADABLE_LIST} if this
 108      *         consumer group object is no longer active
 109      * @throws InterruptedException if this thread was interrupted while waiting for the
 110      *         first message
 111      */
 112     public List<String> read(int maxRead, long waitMs) throws InterruptedException {
 113
 114         synchronized (lockit) {
 115             if (nsweeps > 1 && nreaders == 0) {
 116                 // cannot use this consumer group object anymore
 117                 return UNREADABLE_LIST;
 118             }
 119
 120             ++nreaders;
 121         }
 122
 123         /*
 124          * Note: do EVERYTHING inside of the "try" block, so that the "finally" block can
 125          * update the reader count.
 126          *
 127          * Do NOT hold the lockit while we're polling, as poll() may block for a while.
 128          */
 129         try {
 130             // always read at least one message
 131             int maxRead2 = Math.max(1, maxRead);
 132             long waitMs2 = Math.max(0, waitMs);
 133
 134             // perform a blocking read of the queue
 135             String obj = getNextMessage(waitMs2);
 136             if (obj == null) {
 137                 return EMPTY_LIST;
 138             }
 139
 140             /*
 141              * List should hold all messages from the queue PLUS the one we already have.
 142              * Note: it's possible for additional messages to be added to the queue while
 143              * we're reading from it. In that case, the list will grow as needed.
 144              */
 145             List<String> lst = new ArrayList<>(Math.min(maxRead2, messageQueue.size() + 1));
 146             lst.add(obj);
 147
 148             // perform NON-blocking read of subsequent messages
 149             for (var x = 1; x < maxRead2; ++x) {
 150                 if ((obj = messageQueue.poll()) == null) {
 151                     break;
 152                 }
 153
 154                 lst.add(obj);
 155             }
 156
 157             return lst;
 158
 159         } finally {
 160             synchronized (lockit) {
 161                 --nreaders;
 162                 nsweeps = 0;
 163             }
 164         }
 165     }
 166
 167     /**
 168      * Writes messages to the queue.
 169      *
 170      * @param messages messages to be written to the queue
 171      */
 172     public void write(List<String> messages) {
 173         messageQueue.addAll(messages);
 174     }
 175
 176     // the following methods may be overridden by junit tests
 177
 178     /**
 179      * Gets the next message from the queue.
 180      *
 181      * @param waitMs time to wait, in milliseconds, if the queue is currently empty
 182      * @return the next message, or {@code null} if no messages became available before
 183      *         the wait time elapsed
 184      * @throws InterruptedException if this thread was interrupted while waiting for the
 185      *         message
 186      */
 187     protected String getNextMessage(long waitMs) throws InterruptedException {
 188         return messageQueue.poll(waitMs, TimeUnit.MILLISECONDS);
 189     }
 190 }