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

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * Copyright (C) 2018-2021 AT&T Intellectual Property. All rights reserved.
   4  * Modifications Copyright (C) 2024 Nordix Foundation
   5  * ================================================================================
   6  * Licensed under the Apache License, Version 2.0 (the "License");
   7  * you may not use this file except in compliance with the License.
   8  * You may obtain a copy of the License at
   9  *
  10  *      http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing, software
  13  * distributed under the License is distributed on an "AS IS" BASIS,
  14  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  * See the License for the specific language governing permissions and
  16  * limitations under the License.
  17  * ============LICENSE_END=========================================================
  18  */
  19
  20 package org.onap.policy.common.message.bus.event.base;
  21
  22 import static org.assertj.core.api.Assertions.assertThat;
  23 import static org.assertj.core.api.Assertions.assertThatCode;
  24 import static org.assertj.core.api.Assertions.assertThatThrownBy;
  25 import static org.junit.jupiter.api.Assertions.assertEquals;
  26 import static org.junit.jupiter.api.Assertions.assertNotNull;
  27 import static org.junit.jupiter.api.Assertions.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.jupiter.api.AfterEach;
  40 import org.junit.jupiter.api.BeforeEach;
  41 import org.junit.jupiter.api.Test;
  42 import org.mockito.invocation.InvocationOnMock;
  43 import org.mockito.stubbing.Answer;
  44 import org.onap.policy.common.message.bus.event.Topic.CommInfrastructure;
  45 import org.onap.policy.common.message.bus.event.TopicListener;
  46 import org.onap.policy.common.parameters.topic.BusTopicParams;
  47 import org.onap.policy.common.utils.gson.GsonTestUtils;
  48 import org.onap.policy.common.utils.network.NetworkUtil;
  49
  50 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     @BeforeEach
  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     @AfterEach
  71     public void tearDown() {
  72         source.shutdown();
  73     }
  74
  75     @Test
  76     void testSerialize() {
  77         assertThatCode(() -> new GsonTestUtils().compareGson(source, SingleThreadedBusTopicSourceTest.class))
  78                         .doesNotThrowAnyException();
  79     }
  80
  81     @Test
  82     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     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     void testToString() {
 143         assertTrue(source.toString().startsWith("SingleThreadedBusTopicSource ["));
 144     }
 145
 146     @Test
 147     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     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     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
 213     void testStart_Locked() {
 214         source.lock();
 215         assertThatThrownBy(() -> source.start()).isInstanceOf(IllegalStateException.class);
 216     }
 217
 218     @Test
 219     void testStart_InitEx() {
 220         assertThatThrownBy(() -> {
 221             source.initEx = true;
 222
 223             source.start();
 224         }).isInstanceOf(IllegalStateException.class);
 225     }
 226
 227     @Test
 228     void testStop() {
 229         source.start();
 230         source.stop();
 231         verify(cons).close();
 232
 233         // stop it again - not re-closed
 234         source.stop();
 235         verify(cons).close();
 236
 237         // start & stop again, but with an exception
 238         doThrow(new RuntimeException(EXPECTED)).when(cons).close();
 239         source.start();
 240         source.stop();
 241     }
 242
 243     @Test
 244     void testRun() throws Exception {
 245         source.register(listener);
 246
 247         /*
 248          * Die in the middle of fetching messages. Also, throw an exception during the
 249          * first fetch attempt.
 250          */
 251         when(cons.fetch()).thenAnswer(new Answer<Iterable<String>>() {
 252             int count = 0;
 253
 254             @Override
 255             public Iterable<String> answer(InvocationOnMock invocation) throws Throwable {
 256                 if (++count > 1) {
 257                     source.alive = false;
 258                     return Arrays.asList(MY_MESSAGE, MY_MESSAGE2);
 259
 260                 } else {
 261                     throw new IOException(EXPECTED);
 262                 }
 263             }
 264         });
 265         source.alive = true;
 266         source.run();
 267         assertEquals(Arrays.asList(MY_MESSAGE), Arrays.asList(source.getRecentEvents()));
 268         verify(listener).onTopicEvent(CommInfrastructure.NOOP, MY_TOPIC, MY_MESSAGE);
 269         verify(listener, never()).onTopicEvent(CommInfrastructure.NOOP, MY_TOPIC, MY_MESSAGE2);
 270
 271         /*
 272          * Die AFTER fetching messages.
 273          */
 274         final String msga = "message-A";
 275         final String msgb = "message-B";
 276         when(cons.fetch()).thenAnswer(new Answer<Iterable<String>>() {
 277             int count = 0;
 278
 279             @Override
 280             public Iterable<String> answer(InvocationOnMock invocation) throws Throwable {
 281                 if (++count > 1) {
 282                     source.alive = false;
 283                     return Collections.emptyList();
 284
 285                 } else {
 286                     return Arrays.asList(msga, msgb);
 287                 }
 288             }
 289         });
 290         source.alive = true;
 291         source.run();
 292         verify(listener).onTopicEvent(CommInfrastructure.NOOP, MY_TOPIC, msga);
 293         verify(listener).onTopicEvent(CommInfrastructure.NOOP, MY_TOPIC, msgb);
 294
 295         assertEquals(Arrays.asList(MY_MESSAGE, msga, msgb), Arrays.asList(source.getRecentEvents()));
 296     }
 297
 298     @Test
 299     void testOffer() {
 300         source.register(listener);
 301         source.offer(MY_MESSAGE);
 302         verify(listener).onTopicEvent(CommInfrastructure.NOOP, MY_TOPIC, MY_MESSAGE);
 303         assertEquals(Arrays.asList(MY_MESSAGE), Arrays.asList(source.getRecentEvents()));
 304     }
 305
 306     @Test
 307     void testOffer_NotStarted() {
 308         assertThatThrownBy(() -> source.offer(MY_MESSAGE)).isInstanceOf(IllegalStateException.class);
 309     }
 310
 311     @Test
 312     void testGetConsumerGroup() {
 313         assertEquals(MY_CONS_GROUP, source.getConsumerGroup());
 314     }
 315
 316     @Test
 317     void testGetConsumerInstance() {
 318         assertEquals(MY_CONS_INST, source.getConsumerInstance());
 319     }
 320
 321     @Test
 322     void testShutdown() {
 323         source.register(listener);
 324
 325         source.shutdown();
 326         verify(cons).close();
 327         assertTrue(source.snapshotTopicListeners().isEmpty());
 328     }
 329
 330     @Test
 331     void testGetFetchTimeout() {
 332         assertEquals(MY_FETCH_TIMEOUT, source.getFetchTimeout());
 333     }
 334
 335     @Test
 336     void testGetFetchLimit() {
 337         assertEquals(MY_FETCH_LIMIT, source.getFetchLimit());
 338     }
 339
 340     /**
 341      * Implementation of SingleThreadedBusTopicSource that counts the number of times
 342      * init() is invoked.
 343      */
 344     private class SingleThreadedBusTopicSourceImpl extends SingleThreadedBusTopicSource {
 345
 346         private int initCount = 0;
 347         private boolean initEx = false;
 348
 349         public SingleThreadedBusTopicSourceImpl(BusTopicParams busTopicParams) {
 350             super(busTopicParams);
 351         }
 352
 353         @Override
 354         public CommInfrastructure getTopicCommInfrastructure() {
 355             return CommInfrastructure.NOOP;
 356         }
 357
 358         @Override
 359         public void init() throws MalformedURLException {
 360             ++initCount;
 361
 362             if (initEx) {
 363                 throw new MalformedURLException(EXPECTED);
 364             }
 365
 366             consumer = cons;
 367         }
 368
 369         @Override
 370         protected Thread makePollerThread() {
 371             return thread;
 372         }
 373
 374     }
 375 }