gerrit.onap Code Review - policy/common.git/blob

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * policy-endpoints
   4  * ================================================================================
   5  * Copyright (C) 2018-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
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  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=========================================================
  19  */
  20
  21 package org.onap.policy.common.endpoints.event.comm.bus.internal;
  22
  23 import static org.assertj.core.api.Assertions.assertThat;
  24 import static org.assertj.core.api.Assertions.assertThatCode;
  25 import static org.junit.Assert.assertEquals;
  26 import static org.junit.Assert.assertNotNull;
  27 import static org.junit.Assert.assertTrue;
  28 import static org.mockito.Mockito.doThrow;
  29 import static org.mockito.Mockito.mock;
  30 import static org.mockito.Mockito.never;
  31 import static org.mockito.Mockito.times;
  32 import static org.mockito.Mockito.verify;
  33 import static org.mockito.Mockito.when;
  34
  35 import java.io.IOException;
  36 import java.net.MalformedURLException;
  37 import java.util.Arrays;
  38 import java.util.Collections;
  39 import org.junit.After;
  40 import org.junit.Before;
  41 import org.junit.Test;
  42 import org.mockito.invocation.InvocationOnMock;
  43 import org.mockito.stubbing.Answer;
  44 import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
  45 import org.onap.policy.common.endpoints.event.comm.TopicListener;
  46 import org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase;
  47 import org.onap.policy.common.utils.gson.GsonTestUtils;
  48 import org.onap.policy.common.utils.network.NetworkUtil;
  49
  50 public class SingleThreadedBusTopicSourceTest extends TopicTestBase {
  51     private Thread thread;
  52     private BusConsumer cons;
  53     private TopicListener listener;
  54     private SingleThreadedBusTopicSourceImpl source;
  55
  56     /**
  57      * Creates the object to be tested, as well as various mocks.
  58      */
  59     @Before
  60     @Override
  61     public void setUp() {
  62         super.setUp();
  63
  64         thread = mock(Thread.class);
  65         cons = mock(BusConsumer.class);
  66         listener = mock(TopicListener.class);
  67         source = new SingleThreadedBusTopicSourceImpl(makeBuilder().build());
  68     }
  69
  70     @After
  71     public void tearDown() {
  72         source.shutdown();
  73     }
  74
  75     @Test
  76     public void testSerialize() {
  77         assertThatCode(() -> new GsonTestUtils().compareGson(source, SingleThreadedBusTopicSourceTest.class))
  78                         .doesNotThrowAnyException();
  79     }
  80
  81     @Test
  82     public void testRegister() {
  83         source.register(listener);
  84         assertEquals(1, source.initCount);
  85         source.offer(MY_MESSAGE);
  86         verify(listener).onTopicEvent(CommInfrastructure.NOOP, MY_TOPIC, MY_MESSAGE);
  87
  88         // register another - should not re-init
  89         TopicListener listener2 = mock(TopicListener.class);
  90         source.register(listener2);
  91         assertEquals(1, source.initCount);
  92         source.offer(MY_MESSAGE + "z");
  93         verify(listener).onTopicEvent(CommInfrastructure.NOOP, MY_TOPIC, MY_MESSAGE + "z");
  94         verify(listener2).onTopicEvent(CommInfrastructure.NOOP, MY_TOPIC, MY_MESSAGE + "z");
  95
  96         // re-register - should not re-init
  97         source.register(listener);
  98         assertEquals(1, source.initCount);
  99         source.offer(MY_MESSAGE2);
 100         verify(listener).onTopicEvent(CommInfrastructure.NOOP, MY_TOPIC, MY_MESSAGE2);
 101
 102         // lock & register - should not init
 103         source = new SingleThreadedBusTopicSourceImpl(makeBuilder().build());
 104         source.lock();
 105         source.register(listener);
 106         assertEquals(0, source.initCount);
 107
 108         // exception during init
 109         source = new SingleThreadedBusTopicSourceImpl(makeBuilder().build());
 110         source.initEx = true;
 111         source.register(listener);
 112     }
 113
 114     @Test
 115     public void testUnregister() {
 116         TopicListener listener2 = mock(TopicListener.class);
 117         source.register(listener);
 118         source.register(listener2);
 119
 120         // unregister first listener - should NOT invoke close
 121         source.unregister(listener);
 122         verify(cons, never()).close();
 123         assertEquals(Arrays.asList(listener2), source.snapshotTopicListeners());
 124
 125         // unregister same listener - should not invoke close
 126         source.unregister(listener);
 127         verify(cons, never()).close();
 128         assertEquals(Arrays.asList(listener2), source.snapshotTopicListeners());
 129
 130         // unregister second listener - SHOULD invoke close
 131         source.unregister(listener2);
 132         verify(cons).close();
 133         assertTrue(source.snapshotTopicListeners().isEmpty());
 134
 135         // unregister same listener - should not invoke close again
 136         source.unregister(listener2);
 137         verify(cons).close();
 138         assertTrue(source.snapshotTopicListeners().isEmpty());
 139     }
 140
 141     @Test
 142     public void testToString() {
 143         assertTrue(source.toString().startsWith("SingleThreadedBusTopicSource ["));
 144     }
 145
 146     @Test
 147     public void testMakePollerThread() {
 148         SingleThreadedBusTopicSource source2 = new SingleThreadedBusTopicSource(makeBuilder().build()) {
 149             @Override
 150             public CommInfrastructure getTopicCommInfrastructure() {
 151                 return CommInfrastructure.NOOP;
 152             }
 153
 154             @Override
 155             public void init() throws MalformedURLException {
 156                 // do nothing
 157             }
 158         };
 159
 160         assertNotNull(source2.makePollerThread());
 161     }
 162
 163     @Test
 164     public void testSingleThreadedBusTopicSource() {
 165         // Note: if the value contains "-", it's probably a UUID
 166
 167         // verify that different wrappers can be built
 168         source = new SingleThreadedBusTopicSourceImpl(makeBuilder().build());
 169         assertThat(source.getConsumerGroup()).isEqualTo(MY_CONS_GROUP);
 170         assertThat(source.getConsumerInstance()).isEqualTo(MY_CONS_INST);
 171
 172         // group is null => group is UUID, instance is as provided
 173         source = new SingleThreadedBusTopicSourceImpl(makeBuilder().consumerGroup(null).build());
 174         assertThat(source.getConsumerGroup()).contains("-").isNotEqualTo(NetworkUtil.getHostname());
 175         assertThat(source.getConsumerInstance()).isEqualTo(MY_CONS_INST);
 176
 177         // instance is null => group is as provided, instance is UUID
 178         source = new SingleThreadedBusTopicSourceImpl(makeBuilder().consumerInstance(null).build());
 179         assertThat(source.getConsumerGroup()).isEqualTo(MY_CONS_GROUP);
 180         assertThat(source.getConsumerInstance()).contains("-").isNotEqualTo(NetworkUtil.getHostname());
 181
 182         // group & instance are null => group is UUID, instance is hostname
 183         source = new SingleThreadedBusTopicSourceImpl(makeBuilder().consumerGroup(null).consumerInstance(null).build());
 184         assertThat(source.getConsumerGroup()).contains("-").isNotEqualTo(NetworkUtil.getHostname());
 185         assertThat(source.getConsumerInstance()).isEqualTo(NetworkUtil.getHostname());
 186
 187         assertThatCode(() -> new SingleThreadedBusTopicSourceImpl(
 188                         makeBuilder().fetchLimit(-1).fetchTimeout(-1).build())).doesNotThrowAnyException();
 189     }
 190
 191     @Test
 192     public void testStart() {
 193         source.start();
 194         assertTrue(source.isAlive());
 195         assertEquals(1, source.initCount);
 196         verify(thread).start();
 197
 198         // attempt to start again - nothing should be invoked again
 199         source.start();
 200         assertTrue(source.isAlive());
 201         assertEquals(1, source.initCount);
 202         verify(thread).start();
 203
 204         // stop & re-start
 205         source.stop();
 206         source.start();
 207         assertTrue(source.isAlive());
 208         assertEquals(2, source.initCount);
 209         verify(thread, times(2)).start();
 210     }
 211
 212     @Test(expected = IllegalStateException.class)
 213     public void testStart_Locked() {
 214         source.lock();
 215         source.start();
 216     }
 217
 218     @Test(expected = IllegalStateException.class)
 219     public void testStart_InitEx() {
 220         source.initEx = true;
 221         source.start();
 222     }
 223
 224     @Test
 225     public void testStop() {
 226         source.start();
 227         source.stop();
 228         verify(cons).close();
 229
 230         // stop it again - not re-closed
 231         source.stop();
 232         verify(cons).close();
 233
 234         // start & stop again, but with an exception
 235         doThrow(new RuntimeException(EXPECTED)).when(cons).close();
 236         source.start();
 237         source.stop();
 238     }
 239
 240     @Test
 241     public void testRun() throws Exception {
 242         source.register(listener);
 243
 244         /*
 245          * Die in the middle of fetching messages. Also, throw an exception during the
 246          * first fetch attempt.
 247          */
 248         when(cons.fetch()).thenAnswer(new Answer<Iterable<String>>() {
 249             int count = 0;
 250
 251             @Override
 252             public Iterable<String> answer(InvocationOnMock invocation) throws Throwable {
 253                 if (++count > 1) {
 254                     source.alive = false;
 255                     return Arrays.asList(MY_MESSAGE, MY_MESSAGE2);
 256
 257                 } else {
 258                     throw new IOException(EXPECTED);
 259                 }
 260             }
 261         });
 262         source.alive = true;
 263         source.run();
 264         assertEquals(Arrays.asList(MY_MESSAGE), Arrays.asList(source.getRecentEvents()));
 265         verify(listener).onTopicEvent(CommInfrastructure.NOOP, MY_TOPIC, MY_MESSAGE);
 266         verify(listener, never()).onTopicEvent(CommInfrastructure.NOOP, MY_TOPIC, MY_MESSAGE2);
 267
 268         /*
 269          * Die AFTER fetching messages.
 270          */
 271         final String msga = "message-A";
 272         final String msgb = "message-B";
 273         when(cons.fetch()).thenAnswer(new Answer<Iterable<String>>() {
 274             int count = 0;
 275
 276             @Override
 277             public Iterable<String> answer(InvocationOnMock invocation) throws Throwable {
 278                 if (++count > 1) {
 279                     source.alive = false;
 280                     return Collections.emptyList();
 281
 282                 } else {
 283                     return Arrays.asList(msga, msgb);
 284                 }
 285             }
 286         });
 287         source.alive = true;
 288         source.run();
 289         verify(listener).onTopicEvent(CommInfrastructure.NOOP, MY_TOPIC, msga);
 290         verify(listener).onTopicEvent(CommInfrastructure.NOOP, MY_TOPIC, msgb);
 291
 292         assertEquals(Arrays.asList(MY_MESSAGE, msga, msgb), Arrays.asList(source.getRecentEvents()));
 293     }
 294
 295     @Test
 296     public void testOffer() {
 297         source.register(listener);
 298         source.offer(MY_MESSAGE);
 299         verify(listener).onTopicEvent(CommInfrastructure.NOOP, MY_TOPIC, MY_MESSAGE);
 300         assertEquals(Arrays.asList(MY_MESSAGE), Arrays.asList(source.getRecentEvents()));
 301     }
 302
 303     @Test(expected = IllegalStateException.class)
 304     public void testOffer_NotStarted() {
 305         source.offer(MY_MESSAGE);
 306     }
 307
 308     @Test
 309     public void testGetConsumerGroup() {
 310         assertEquals(MY_CONS_GROUP, source.getConsumerGroup());
 311     }
 312
 313     @Test
 314     public void testGetConsumerInstance() {
 315         assertEquals(MY_CONS_INST, source.getConsumerInstance());
 316     }
 317
 318     @Test
 319     public void testShutdown() {
 320         source.register(listener);
 321
 322         source.shutdown();
 323         verify(cons).close();
 324         assertTrue(source.snapshotTopicListeners().isEmpty());
 325     }
 326
 327     @Test
 328     public void testGetFetchTimeout() {
 329         assertEquals(MY_FETCH_TIMEOUT, source.getFetchTimeout());
 330     }
 331
 332     @Test
 333     public void testGetFetchLimit() {
 334         assertEquals(MY_FETCH_LIMIT, source.getFetchLimit());
 335     }
 336
 337     /**
 338      * Implementation of SingleThreadedBusTopicSource that counts the number of times
 339      * init() is invoked.
 340      */
 341     private class SingleThreadedBusTopicSourceImpl extends SingleThreadedBusTopicSource {
 342
 343         private int initCount = 0;
 344         private boolean initEx = false;
 345
 346         public SingleThreadedBusTopicSourceImpl(BusTopicParams busTopicParams) {
 347             super(busTopicParams);
 348         }
 349
 350         @Override
 351         public CommInfrastructure getTopicCommInfrastructure() {
 352             return CommInfrastructure.NOOP;
 353         }
 354
 355         @Override
 356         public void init() throws MalformedURLException {
 357             ++initCount;
 358
 359             if (initEx) {
 360                 throw new MalformedURLException(EXPECTED);
 361             }
 362
 363             consumer = cons;
 364         }
 365
 366         @Override
 367         protected Thread makePollerThread() {
 368             return thread;
 369         }
 370
 371     }
 372 }