X-Git-Url: https://gerrit.onap.org/r/gitweb?a=blobdiff_plain;ds=sidebyside;f=models-sim%2Fmodels-sim-dmaap%2Fsrc%2Fmain%2Fjava%2Forg%2Fonap%2Fpolicy%2Fmodels%2Fsim%2Fdmaap%2Fprovider%2FConsumerGroupData.java;fp=models-sim%2Fmodels-sim-dmaap%2Fsrc%2Fmain%2Fjava%2Forg%2Fonap%2Fpolicy%2Fmodels%2Fsim%2Fdmaap%2Fprovider%2FConsumerGroupData.java;h=0000000000000000000000000000000000000000;hb=49f07db935d114b72a44e446867b16262dd552aa;hp=3acaf08886fa5752a77bc516719adda5a7cb34da;hpb=71be21fd5b9b52c613bb855f00a79a51e81906dd;p=policy%2Fmodels.git diff --git a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/provider/ConsumerGroupData.java b/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/provider/ConsumerGroupData.java deleted file mode 100644 index 3acaf0888..000000000 --- a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/provider/ConsumerGroupData.java +++ /dev/null @@ -1,190 +0,0 @@ -/*- - * ============LICENSE_START======================================================= - * ONAP Policy Models - * ================================================================================ - * Copyright (C) 2019, 2021 AT&T Intellectual Property. All rights reserved. - * ================================================================================ - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * ============LICENSE_END========================================================= - */ - -package org.onap.policy.models.sim.dmaap.provider; - -import java.util.ArrayList; -import java.util.Collections; -import java.util.List; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.LinkedBlockingQueue; -import java.util.concurrent.TimeUnit; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * Data associated with a consumer group. All consumer instances within a group share the - * same data object. - */ -public class ConsumerGroupData { - private static final Logger logger = LoggerFactory.getLogger(ConsumerGroupData.class); - - /** - * Returned when messages can no longer be read from this consumer group object, - * because it is being removed from the topic. {@link #UNREADABLE_LIST} must not be - * the same list as Collections.emptyList(), thus we wrap it. - */ - public static final List UNREADABLE_LIST = Collections.unmodifiableList(Collections.emptyList()); - - /** - * Returned when there are no messages read. Collections.emptyList() is already - * unmodifiable, thus no need to wrap it. - */ - private static final List EMPTY_LIST = Collections.emptyList(); - - /** - * This is locked while fields other than {@link #messageQueue} are updated. - */ - private final Object lockit = new Object(); - - /** - * Number of sweep cycles that have occurred since a consumer has attempted to read - * from the queue. This consumer group should be removed once this count exceeds - * {@code 1}, provided {@link #nreaders} is zero. - */ - private int nsweeps = 0; - - /** - * Number of consumers that are currently attempting to read from the queue. This - * consumer group should not be removed as long as this is non-zero. - */ - private int nreaders = 0; - - /** - * Message queue. - */ - private final BlockingQueue messageQueue = new LinkedBlockingQueue<>(); - - - /** - * Constructs the object. - * - * @param topicName name of the topic with which this object is associated - * @param groupName name of the consumer group with which this object is associated - */ - public ConsumerGroupData(String topicName, String groupName) { - logger.info("Topic {}: add consumer group: {}", topicName, groupName); - } - - /** - * Determines if this consumer group should be removed. This should be invoked once - * during each sweep cycle. When this returns {@code true}, this consumer group should - * be immediately discarded, as any readers will sit in a spin loop waiting for it to - * be discarded. - * - * @return {@code true} if this consumer group should be removed, {@code false} - * otherwise - */ - public boolean shouldRemove() { - synchronized (lockit) { - return (nreaders == 0 && ++nsweeps > 1); - } - } - - /** - * Reads messages from the queue, blocking if necessary. - * - * @param maxRead maximum number of messages to read - * @param waitMs time to wait, in milliseconds, if the queue is currently empty - * @return a list of messages read from the queue, empty if no messages became - * available before the wait time elapsed, or {@link #UNREADABLE_LIST} if this - * consumer group object is no longer active - * @throws InterruptedException if this thread was interrupted while waiting for the - * first message - */ - public List read(int maxRead, long waitMs) throws InterruptedException { - - synchronized (lockit) { - if (nsweeps > 1 && nreaders == 0) { - // cannot use this consumer group object anymore - return UNREADABLE_LIST; - } - - ++nreaders; - } - - /* - * Note: do EVERYTHING inside of the "try" block, so that the "finally" block can - * update the reader count. - * - * Do NOT hold the lockit while we're polling, as poll() may block for a while. - */ - try { - // always read at least one message - int maxRead2 = Math.max(1, maxRead); - long waitMs2 = Math.max(0, waitMs); - - // perform a blocking read of the queue - String obj = getNextMessage(waitMs2); - if (obj == null) { - return EMPTY_LIST; - } - - /* - * List should hold all messages from the queue PLUS the one we already have. - * Note: it's possible for additional messages to be added to the queue while - * we're reading from it. In that case, the list will grow as needed. - */ - List lst = new ArrayList<>(Math.min(maxRead2, messageQueue.size() + 1)); - lst.add(obj); - - // perform NON-blocking read of subsequent messages - for (var x = 1; x < maxRead2; ++x) { - if ((obj = messageQueue.poll()) == null) { - break; - } - - lst.add(obj); - } - - return lst; - - } finally { - synchronized (lockit) { - --nreaders; - nsweeps = 0; - } - } - } - - /** - * Writes messages to the queue. - * - * @param messages messages to be written to the queue - */ - public void write(List messages) { - messageQueue.addAll(messages); - } - - // the following methods may be overridden by junit tests - - /** - * Gets the next message from the queue. - * - * @param waitMs time to wait, in milliseconds, if the queue is currently empty - * @return the next message, or {@code null} if no messages became available before - * the wait time elapsed - * @throws InterruptedException if this thread was interrupted while waiting for the - * message - */ - protected String getNextMessage(long waitMs) throws InterruptedException { - return messageQueue.poll(waitMs, TimeUnit.MILLISECONDS); - } -}