1 files changed, 0 insertions, 190 deletions
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<String> 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<String> 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<String> 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<String> 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<String> 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<String> 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);
-    }
-}