2 * ============LICENSE_START=======================================================
4 * ================================================================================
5 * Copyright (C) 2020-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
11 * http://www.apache.org/licenses/LICENSE-2.0
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=========================================================
21 package org.onap.policy.controlloop.actorserviceprovider.topic;
23 import static org.junit.Assert.assertNotNull;
24 import static org.junit.Assert.assertNotSame;
25 import static org.junit.Assert.assertSame;
26 import static org.mockito.ArgumentMatchers.any;
27 import static org.mockito.ArgumentMatchers.eq;
28 import static org.mockito.Mockito.never;
29 import static org.mockito.Mockito.verify;
31 import java.util.Arrays;
33 import java.util.function.BiConsumer;
34 import org.junit.Before;
35 import org.junit.Test;
36 import org.junit.runner.RunWith;
37 import org.mockito.Mock;
38 import org.mockito.junit.MockitoJUnitRunner;
39 import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
40 import org.onap.policy.common.utils.coder.CoderException;
41 import org.onap.policy.common.utils.coder.StandardCoder;
42 import org.onap.policy.common.utils.coder.StandardCoderObject;
44 @RunWith(MockitoJUnitRunner.class)
45 public class TopicListenerImplTest {
46 private static final StandardCoder coder = new StandardCoder();
47 private static final CommInfrastructure INFRA = CommInfrastructure.NOOP;
48 private static final String MY_TOPIC = "my-topic";
49 private static final String KEY1 = "requestId";
50 private static final String KEY2 = "container";
51 private static final String SUBKEY = "subRequestId";
53 private static final String VALUEA_REQID = "hello";
54 private static final String VALUEA_SUBREQID = "world";
56 private static final String VALUEB_REQID = "bye";
58 private Forwarder forwarder1;
59 private Forwarder forwarder2;
60 private TopicListenerImpl topic;
63 private BiConsumer<String, StandardCoderObject> listener1;
66 private BiConsumer<String, StandardCoderObject> listener1b;
69 private BiConsumer<String, StandardCoderObject> listener2;
77 topic = new TopicListenerImpl();
79 forwarder1 = topic.addForwarder(new SelectorKey(KEY1));
80 forwarder2 = topic.addForwarder(new SelectorKey(KEY1), new SelectorKey(KEY2, SUBKEY));
82 assertNotNull(forwarder1);
83 assertNotNull(forwarder2);
84 assertNotSame(forwarder1, forwarder2);
86 forwarder1.register(Arrays.asList(VALUEA_REQID), listener1);
87 forwarder1.register(Arrays.asList(VALUEB_REQID), listener1b);
88 forwarder2.register(Arrays.asList(VALUEA_REQID, VALUEA_SUBREQID), listener2);
92 public void testShutdown() {
93 // shut it down, which should clear all forwarders
96 // should get a new forwarder now
97 Forwarder forwarder = topic.addForwarder(new SelectorKey(KEY1));
98 assertNotSame(forwarder1, forwarder);
99 assertNotSame(forwarder2, forwarder);
101 // new forwarder should be unchanged
102 assertSame(forwarder, topic.addForwarder(new SelectorKey(KEY1)));
106 public void testAddForwarder() {
107 assertSame(forwarder1, topic.addForwarder(new SelectorKey(KEY1)));
108 assertSame(forwarder2, topic.addForwarder(new SelectorKey(KEY1), new SelectorKey(KEY2, SUBKEY)));
112 public void testOnTopicEvent() {
114 * send a message that should go to listener1 on forwarder1 and listener2 on
117 String msg = makeMessage(Map.of(KEY1, VALUEA_REQID, KEY2, Map.of(SUBKEY, VALUEA_SUBREQID)));
118 topic.onTopicEvent(INFRA, MY_TOPIC, msg);
120 verify(listener1).accept(eq(msg), any());
121 verify(listener2).accept(eq(msg), any());
124 verify(listener1b, never()).accept(any(), any());
127 * now send a message that should only go to listener1b on forwarder1
129 msg = makeMessage(Map.of(KEY1, VALUEB_REQID, KEY2, Map.of(SUBKEY, VALUEA_SUBREQID)));
130 topic.onTopicEvent(INFRA, MY_TOPIC, msg);
132 // should route to listener1 on forwarder1 and listener2 on forwarder2
133 verify(listener1b).accept(eq(msg), any());
135 // try one where the coder throws an exception
136 topic.onTopicEvent(INFRA, MY_TOPIC, "{invalid-json");
138 // no extra invocations
139 verify(listener1).accept(any(), any());
140 verify(listener1b).accept(any(), any());
141 verify(listener2).accept(any(), any());
145 * Makes a message from a map.
147 private String makeMessage(Map<String, Object> map) {
149 return coder.encode(map);
150 } catch (CoderException e) {
151 throw new IllegalArgumentException(e);