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

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