+++ /dev/null
-/*-
- * ============LICENSE_START=======================================================
- * ONAP Policy Models
- * ================================================================================
- * Copyright (C) 2019 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 static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertSame;
-import static org.junit.Assert.assertTrue;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.List;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-
-public class ConsumerGroupDataTest {
- private static final int WAIT_MS = 5000;
- private static final int MIN_WAIT_MS = WAIT_MS / 2;
- private static final String MY_TOPIC = "my-topic";
- private static final String MY_CONSUMER = "my-consumer";
- private static final String MSG1 = "hello";
- private static final String MSG2 = "there";
- private static final String MSG3 = "world";
- private static final int MAX_THREADS = 30;
-
- private MyData data;
- private MyReader thread;
- private List<MyReader> threads;
-
- /**
- * Sets up.
- */
- @Before
- public void setUp() {
- data = new MyData();
- thread = null;
- threads = new ArrayList<>(MAX_THREADS);
- }
-
- /**
- * Stops any running thread.
- */
- @After
- public void tearDown() {
- for (MyReader thr : threads) {
- thr.interrupt();
- }
-
- for (MyReader thr : threads) {
- thr.await();
- }
- }
-
- @Test
- public void testShouldRemove() throws InterruptedException {
- assertFalse(data.shouldRemove());
- assertTrue(data.shouldRemove());
-
- data = new MyData();
-
- // start a reader thread and wait for it to poll its queue
- startReader(0, 10);
- assertTrue(data.await());
-
- assertFalse(data.shouldRemove());
- }
-
- @Test
- public void testRead() {
- data.enqueue(MSG1, MSG2, MSG3, MSG1, MSG2, MSG3);
-
- // this reader only wants one
- startReader(1, 1);
- assertTrue(thread.await());
- assertEquals("[hello]", thread.result.toString());
-
- // this reader wants three
- startReader(3, 1);
- assertTrue(thread.await());
- assertEquals("[there, world, hello]", thread.result.toString());
-
- // this reader wants three, but will only get two
- startReader(3, 1);
- assertTrue(thread.await());
- assertEquals("[there, world]", thread.result.toString());
- }
-
- @Test
- public void testRead_Idle() throws InterruptedException {
- // force it to idle
- data.shouldRemove();
- data.shouldRemove();
-
- long tbeg = System.currentTimeMillis();
- assertSame(ConsumerGroupData.UNREADABLE_LIST, data.read(1, WAIT_MS));
-
- // should not have waited
- assertTrue(System.currentTimeMillis() < tbeg + MIN_WAIT_MS);
- }
-
- @Test
- public void testRead_NegativeCount() throws InterruptedException {
- data.enqueue(MSG1, MSG2);
- startReader(-1, 3);
- assertTrue(data.await());
-
- // wait time should be unaffected
- assertEquals(3L, data.waitMs2);
-
- assertTrue(thread.await());
-
- // should only return one message
- assertEquals("[hello]", thread.result.toString());
- }
-
- @Test
- public void testRead_NegativeWait() throws InterruptedException {
- data.enqueue(MSG1, MSG2, MSG3);
- startReader(2, -3);
- assertTrue(data.await());
-
- assertEquals(0L, data.waitMs2);
-
- assertTrue(thread.await());
-
- // should return two messages, as requested
- assertEquals("[hello, there]", thread.result.toString());
- }
-
- @Test
- public void testRead_NoMessages() throws InterruptedException {
- startReader(0, 0);
- assertTrue(data.await());
-
- assertTrue(thread.await());
- assertTrue(thread.result.isEmpty());
- }
-
- @Test
- public void testRead_MultiThreaded() {
- // queue up a bunch of messages
- final int expected = MAX_THREADS * 3;
- for (int x = 0; x < expected; ++x) {
- data.enqueue(MSG1);
- }
-
- for (int x = 0; x < MAX_THREADS; ++x) {
- startReader(4, 1);
- }
-
- int actual = 0;
- for (MyReader thr : threads) {
- thr.await();
- actual += thr.result.size();
- }
-
- assertEquals(expected, actual);
- }
-
-
- /**
- * Starts a reader thread.
- *
- * @param limit number of messages to read at one time
- * @param waitMs wait time, in milliseconds
- */
- private void startReader(int limit, long waitMs) {
- thread = new MyReader(limit, waitMs);
-
- thread.setDaemon(true);
- thread.start();
-
- threads.add(thread);
- }
-
-
- private class MyData extends ConsumerGroupData {
-
- /**
- * Decremented when {@link #getNextMessage(long)} is invoked.
- */
- private final CountDownLatch latch = new CountDownLatch(1);
-
- /**
- * Messages to be added to the queue when {@link #getNextMessage(long)} is
- * invoked.
- */
- private final List<String> messages = new ArrayList<>();
-
- /**
- * Value passed to {@link #getNextMessage(long)}.
- */
- private volatile long waitMs2 = -1;
-
- /**
- * Constructs the object.
- */
- public MyData() {
- super(MY_TOPIC, MY_CONSUMER);
- }
-
- /**
- * Arranges for messages to be injected into the queue the next time
- * {@link #getNextMessage(long)} is invoked.
- *
- * @param messages the messages to be injected
- */
- public void enqueue(String... messages) {
- this.messages.addAll(Arrays.asList(messages));
- }
-
- @Override
- protected String getNextMessage(long waitMs) throws InterruptedException {
- waitMs2 = waitMs;
-
- latch.countDown();
-
- synchronized (messages) {
- write(messages);
- messages.clear();
- }
-
- return super.getNextMessage(waitMs);
- }
-
- /**
- * Waits for {@link #getNextMessage(long)} to be invoked.
- *
- * @return {@code true} if {@link #getNextMessage(long)} was invoked,
- * {@code false} if the timer expired first
- * @throws InterruptedException if the current thread is interrupted while waiting
- */
- public boolean await() throws InterruptedException {
- return latch.await(WAIT_MS, TimeUnit.MILLISECONDS);
- }
- }
-
- /**
- * Thread that will invoke the consumer group's read() method one time.
- */
- private class MyReader extends Thread {
- private final ConsumerGroupData group = data;
- private final int limit;
- private final long waitMs;
-
- /**
- * Result returned by the read() method.
- */
- private List<String> result = Collections.emptyList();
-
- public MyReader(int limit, long waitMs) {
- this.limit = limit;
- this.waitMs = waitMs;
- }
-
- @Override
- public void run() {
- try {
- result = group.read(limit, waitMs);
-
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- }
- }
-
- /**
- * Waits for the thread to complete.
- *
- * @return {@code true} if the thread completed, {@code false} if the thread is
- * still running
- */
- public boolean await() {
- try {
- this.join(WAIT_MS);
-
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- }
-
- return !this.isAlive();
- }
- }
-}
Code Review / policy / models.git / blobdiff