main/src/test/java/org/onap/policy/pap/main/comm/PublisherTest.java

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * ONAP PAP
   4  * ================================================================================
   5  * Copyright (C) 2019 AT&T Intellectual Property. All rights reserved.
   6  * Modifications Copyright (C) 2021 Bell Canada. All rights reserved.
   7  * ================================================================================
   8  * Licensed under the Apache License, Version 2.0 (the "License");
   9  * you may not use this file except in compliance with the License.
  10  * You may obtain a copy of the License at
  11  *
  12  *      http://www.apache.org/licenses/LICENSE-2.0
  13  *
  14  * Unless required by applicable law or agreed to in writing, software
  15  * distributed under the License is distributed on an "AS IS" BASIS,
  16  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  17  * See the License for the specific language governing permissions and
  18  * limitations under the License.
  19  * ============LICENSE_END=========================================================
  20  */
  21
  22 package org.onap.policy.pap.main.comm;
  23
  24 import static org.assertj.core.api.Assertions.assertThatThrownBy;
  25 import static org.junit.Assert.assertEquals;
  26 import static org.junit.Assert.assertNotNull;
  27 import static org.junit.Assert.assertNull;
  28 import static org.junit.Assert.assertTrue;
  29
  30 import java.util.concurrent.ConcurrentLinkedQueue;
  31 import java.util.concurrent.Semaphore;
  32 import java.util.concurrent.TimeUnit;
  33 import org.junit.After;
  34 import org.junit.AfterClass;
  35 import org.junit.Before;
  36 import org.junit.BeforeClass;
  37 import org.junit.Test;
  38 import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
  39 import org.onap.policy.common.endpoints.event.comm.TopicEndpointManager;
  40 import org.onap.policy.common.endpoints.event.comm.TopicListener;
  41 import org.onap.policy.common.utils.coder.Coder;
  42 import org.onap.policy.common.utils.coder.CoderException;
  43 import org.onap.policy.common.utils.coder.StandardCoder;
  44 import org.onap.policy.models.pdp.concepts.PdpMessage;
  45 import org.onap.policy.models.pdp.concepts.PdpStateChange;
  46 import org.onap.policy.pap.main.PapConstants;
  47 import org.onap.policy.pap.main.PolicyPapException;
  48 import org.onap.policy.pap.main.parameters.CommonTestData;
  49 import org.onap.policy.pap.main.parameters.PapParameterGroup;
  50
  51 public class PublisherTest extends Threaded {
  52
  53     // these messages will have different request IDs
  54     private static final PdpStateChange MSG1 = new PdpStateChange();
  55     private static final PdpStateChange MSG2 = new PdpStateChange();
  56
  57     // MSG1 & MSG2, respectively, encoded as JSON
  58     private static final String JSON1;
  59     private static final String JSON2;
  60
  61     static {
  62         try {
  63             Coder coder = new StandardCoder();
  64             JSON1 = coder.encode(MSG1);
  65             JSON2 = coder.encode(MSG2);
  66
  67         } catch (CoderException e) {
  68             throw new ExceptionInInitializerError(e);
  69         }
  70     }
  71
  72     /**
  73      * Max time to wait, in milliseconds, for a thread to terminate or for a message to be
  74      * published.
  75      */
  76     private static final long MAX_WAIT_MS = 5000;
  77
  78     private Publisher<PdpMessage> pub;
  79     private MyListener listener;
  80
  81     /**
  82      * Configures the topic and attaches a listener.
  83      *
  84      * @throws Exception if an error occurs
  85      */
  86     @BeforeClass
  87     public static void setUpBeforeClass() throws Exception {
  88         final PapParameterGroup parameterGroup = new CommonTestData().getPapParameterGroup(6969);
  89         TopicEndpointManager.getManager().shutdown();
  90
  91         TopicEndpointManager.getManager().addTopics(parameterGroup.getTopicParameterGroup());
  92         TopicEndpointManager.getManager().start();
  93     }
  94
  95     @AfterClass
  96     public static void tearDownAfterClass() throws Exception {
  97         TopicEndpointManager.getManager().shutdown();
  98     }
  99
 100     /**
 101      * Set up.
 102      *
 103      * @throws Exception if an error occurs
 104      */
 105     @Before
 106     public void setUp() throws Exception {
 107         super.setUp();
 108
 109         pub = new Publisher<>(PapConstants.TOPIC_POLICY_PDP_PAP);
 110
 111         listener = new MyListener();
 112         TopicEndpointManager.getManager().getNoopTopicSink(PapConstants.TOPIC_POLICY_PDP_PAP).register(listener);
 113     }
 114
 115     /**
 116      * Tear down.
 117      *
 118      * @throws Exception if an error occurs
 119      */
 120     @After
 121     public void tearDown() throws Exception {
 122         TopicEndpointManager.getManager().getNoopTopicSink(PapConstants.TOPIC_POLICY_PDP_PAP).unregister(listener);
 123
 124         super.tearDown();
 125     }
 126
 127     @Override
 128     protected void stopThread() {
 129         if (pub != null) {
 130             pub.stop();
 131         }
 132     }
 133
 134     @Test
 135     public void testPublisher_testStop() throws Exception {
 136         startThread(pub);
 137         pub.stop();
 138
 139         assertTrue(waitStop());
 140
 141         // ensure we can call "stop" a second time
 142         pub.stop();
 143     }
 144
 145     @Test
 146     public void testPublisher_Ex() throws Exception {
 147         assertThatThrownBy(() -> new Publisher<>("unknwon-topic")).isInstanceOf(PolicyPapException.class);
 148     }
 149
 150     @Test
 151     public void testEnqueue() throws Exception {
 152         // enqueue before running
 153         pub.enqueue(new QueueToken<>(MSG1));
 154
 155         // enqueue another after running
 156         startThread(pub);
 157         pub.enqueue(new QueueToken<>(MSG2));
 158
 159         String json = listener.await(MAX_WAIT_MS);
 160         assertEquals(JSON1, json);
 161
 162         json = listener.await(MAX_WAIT_MS);
 163         assertEquals(JSON2, json);
 164     }
 165
 166     @Test
 167     public void testRun_StopBeforeProcess() throws Exception {
 168         // enqueue before running
 169         QueueToken<PdpMessage> token = new QueueToken<>(MSG1);
 170         pub.enqueue(token);
 171
 172         // stop before running
 173         pub.stop();
 174
 175         // start the thread and then wait for it to stop
 176         startThread(pub);
 177         assertTrue(waitStop());
 178
 179         // message should not have been processed
 180         assertTrue(listener.isEmpty());
 181         assertNotNull(token.get());
 182     }
 183
 184     @Test
 185     public void testRun() throws Exception {
 186         startThread(pub);
 187
 188         // should skip token with null message
 189         QueueToken<PdpMessage> token1 = new QueueToken<>(null);
 190         pub.enqueue(token1);
 191
 192         QueueToken<PdpMessage> token2 = new QueueToken<>(MSG2);
 193         pub.enqueue(token2);
 194
 195         // only the second message should have been processed
 196         String json = listener.await(MAX_WAIT_MS);
 197         assertEquals(JSON2, json);
 198         assertNull(token2.get());
 199
 200         pub.stop();
 201         assertTrue(waitStop());
 202
 203         // no more messages
 204         assertTrue(listener.isEmpty());
 205     }
 206
 207     @Test
 208     public void testGetNext() throws Exception {
 209         startThread(pub);
 210
 211         // wait for a message to be processed
 212         pub.enqueue(new QueueToken<>(MSG1));
 213         assertNotNull(listener.await(MAX_WAIT_MS));
 214
 215         // now interrupt
 216         interruptThread();
 217
 218         assertTrue(waitStop());
 219     }
 220
 221     /**
 222      * Listener for messages published to the topic.
 223      */
 224     private static class MyListener implements TopicListener {
 225
 226         /**
 227          * Released every time a message is added to the queue.
 228          */
 229         private final Semaphore sem = new Semaphore(0);
 230
 231         private final ConcurrentLinkedQueue<String> messages = new ConcurrentLinkedQueue<>();
 232
 233         public boolean isEmpty() {
 234             return messages.isEmpty();
 235         }
 236
 237         /**
 238          * Waits for a message to be published to the topic.
 239          *
 240          * @param waitMs time to wait, in milli-seconds
 241          * @return the next message in the queue, or {@code null} if there are no messages
 242          *         or if the timeout was reached
 243          * @throws InterruptedException if this thread was interrupted while waiting
 244          */
 245         public String await(long waitMs) throws InterruptedException {
 246             if (sem.tryAcquire(waitMs, TimeUnit.MILLISECONDS)) {
 247                 return messages.poll();
 248             }
 249
 250             return null;
 251         }
 252
 253         @Override
 254         public void onTopicEvent(CommInfrastructure commType, String topic, String event) {
 255             messages.add(event);
 256             sem.release();
 257         }
 258     }
 259 }