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

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