2 * ============LICENSE_START=======================================================
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
11 * http://www.apache.org/licenses/LICENSE-2.0
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=========================================================
21 package org.onap.policy.models.sim.dmaap.provider;
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;
33 * Data associated with a consumer group. All consumer instances within a group share the
36 public class ConsumerGroupData {
37 private static final Logger logger = LoggerFactory.getLogger(ConsumerGroupData.class);
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.
44 public static final List<String> UNREADABLE_LIST = Collections.unmodifiableList(Collections.emptyList());
47 * Returned when there are no messages read. Collections.emptyList() is already
48 * unmodifiable, thus no need to wrap it.
50 private static final List<String> EMPTY_LIST = Collections.emptyList();
53 * This is locked while fields other than {@link #messageQueue} are updated.
55 private final Object lockit = new Object();
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.
62 private int nsweeps = 0;
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.
68 private int nreaders = 0;
73 private final BlockingQueue<String> messageQueue = new LinkedBlockingQueue<>();
77 * Constructs the object.
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
82 public ConsumerGroupData(String topicName, String groupName) {
83 logger.info("Topic {}: add consumer group: {}", topicName, groupName);
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
92 * @return {@code true} if this consumer group should be removed, {@code false}
95 public boolean shouldRemove() {
96 synchronized (lockit) {
97 return (nreaders == 0 && ++nsweeps > 1);
102 * Reads messages from the queue, blocking if necessary.
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
112 public List<String> read(int maxRead, long waitMs) throws InterruptedException {
114 synchronized (lockit) {
115 if (nsweeps > 1 && nreaders == 0) {
116 // cannot use this consumer group object anymore
117 return UNREADABLE_LIST;
124 * Note: do EVERYTHING inside of the "try" block, so that the "finally" block can
125 * update the reader count.
127 * Do NOT hold the lockit while we're polling, as poll() may block for a while.
130 // always read at least one message
131 int maxRead2 = Math.max(1, maxRead);
132 long waitMs2 = Math.max(0, waitMs);
134 // perform a blocking read of the queue
135 String obj = getNextMessage(waitMs2);
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.
145 List<String> lst = new ArrayList<>(Math.min(maxRead2, messageQueue.size() + 1));
148 // perform NON-blocking read of subsequent messages
149 for (var x = 1; x < maxRead2; ++x) {
150 if ((obj = messageQueue.poll()) == null) {
160 synchronized (lockit) {
168 * Writes messages to the queue.
170 * @param messages messages to be written to the queue
172 public void write(List<String> messages) {
173 messageQueue.addAll(messages);
176 // the following methods may be overridden by junit tests
179 * Gets the next message from the queue.
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
187 protected String getNextMessage(long waitMs) throws InterruptedException {
188 return messageQueue.poll(waitMs, TimeUnit.MILLISECONDS);