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.assertj.core.api.Assertions.assertThatIllegalArgumentException;
24 import static org.mockito.ArgumentMatchers.any;
25 import static org.mockito.Mockito.doThrow;
26 import static org.mockito.Mockito.never;
27 import static org.mockito.Mockito.times;
28 import static org.mockito.Mockito.verify;
30 import java.util.Arrays;
32 import java.util.function.BiConsumer;
33 import org.junit.Before;
34 import org.junit.Test;
35 import org.junit.runner.RunWith;
36 import org.mockito.Mock;
37 import org.mockito.junit.MockitoJUnitRunner;
38 import org.onap.policy.common.utils.coder.StandardCoderObject;
39 import org.onap.policy.controlloop.actorserviceprovider.Util;
41 @RunWith(MockitoJUnitRunner.class)
42 public class ForwarderTest {
43 private static final String TEXT = "some text";
45 private static final String KEY1 = "requestId";
46 private static final String KEY2 = "container";
47 private static final String SUBKEY = "subRequestId";
49 private static final String VALUEA_REQID = "hello";
50 private static final String VALUEA_SUBREQID = "world";
52 // request id is shared with value A
53 private static final String VALUEB_REQID = "hello";
54 private static final String VALUEB_SUBREQID = "another world";
57 private static final String VALUEC_REQID = "bye";
58 private static final String VALUEC_SUBREQID = "bye-bye";
61 private BiConsumer<String, StandardCoderObject> listener1;
64 private BiConsumer<String, StandardCoderObject> listener1b;
67 private BiConsumer<String, StandardCoderObject> listener2;
70 private BiConsumer<String, StandardCoderObject> listener3;
72 private Forwarder forwarder;
80 forwarder = new Forwarder(Arrays.asList(new SelectorKey(KEY1), new SelectorKey(KEY2, SUBKEY)));
82 forwarder.register(Arrays.asList(VALUEA_REQID, VALUEA_SUBREQID), listener1);
83 forwarder.register(Arrays.asList(VALUEA_REQID, VALUEA_SUBREQID), listener1b);
84 forwarder.register(Arrays.asList(VALUEB_REQID, VALUEB_SUBREQID), listener2);
85 forwarder.register(Arrays.asList(VALUEC_REQID, VALUEC_SUBREQID), listener3);
89 public void testRegister() {
90 // key size mismatches
91 assertThatIllegalArgumentException().isThrownBy(() -> forwarder.register(Arrays.asList(), listener1))
92 .withMessage("key/value mismatch");
93 assertThatIllegalArgumentException()
94 .isThrownBy(() -> forwarder.register(Arrays.asList(VALUEA_REQID), listener1))
95 .withMessage("key/value mismatch");
99 public void testUnregister() {
101 forwarder.unregister(Arrays.asList(VALUEA_REQID, VALUEA_SUBREQID), listener1b);
103 StandardCoderObject sco = makeMessage(Map.of(KEY1, VALUEA_REQID, KEY2, Map.of(SUBKEY, VALUEA_SUBREQID)));
104 forwarder.onMessage(TEXT, sco);
106 verify(listener1).accept(TEXT, sco);
107 verify(listener1b, never()).accept(any(), any());
110 forwarder.unregister(Arrays.asList(VALUEA_REQID, VALUEA_SUBREQID), listener1);
111 forwarder.onMessage(TEXT, sco);
113 // route a message to listener2
114 sco = makeMessage(Map.of(KEY1, VALUEB_REQID, KEY2, Map.of(SUBKEY, VALUEB_SUBREQID)));
115 forwarder.onMessage(TEXT, sco);
116 verify(listener2).accept(TEXT, sco);
118 // no more messages to listener1 or 1b
119 verify(listener1).accept(any(), any());
120 verify(listener1b, never()).accept(any(), any());
124 public void testOnMessage() {
125 StandardCoderObject sco = makeMessage(Map.of(KEY1, VALUEA_REQID, KEY2, Map.of(SUBKEY, VALUEA_SUBREQID)));
126 forwarder.onMessage(TEXT, sco);
128 verify(listener1).accept(TEXT, sco);
129 verify(listener1b).accept(TEXT, sco);
131 // repeat - counts should increment
132 forwarder.onMessage(TEXT, sco);
134 verify(listener1, times(2)).accept(TEXT, sco);
135 verify(listener1b, times(2)).accept(TEXT, sco);
137 // should not have been invoked
138 verify(listener2, never()).accept(any(), any());
139 verify(listener3, never()).accept(any(), any());
141 // try other listeners now
142 sco = makeMessage(Map.of(KEY1, VALUEB_REQID, KEY2, Map.of(SUBKEY, VALUEB_SUBREQID)));
143 forwarder.onMessage(TEXT, sco);
144 verify(listener2).accept(TEXT, sco);
146 sco = makeMessage(Map.of(KEY1, VALUEC_REQID, KEY2, Map.of(SUBKEY, VALUEC_SUBREQID)));
147 forwarder.onMessage(TEXT, sco);
148 verify(listener3).accept(TEXT, sco);
150 // message has no listeners
151 sco = makeMessage(Map.of(KEY1, "xyzzy", KEY2, Map.of(SUBKEY, VALUEB_SUBREQID)));
152 forwarder.onMessage(TEXT, sco);
154 // message doesn't have both keys
155 sco = makeMessage(Map.of(KEY1, VALUEA_REQID));
156 forwarder.onMessage(TEXT, sco);
158 // counts should not have incremented
159 verify(listener1, times(2)).accept(any(), any());
160 verify(listener1b, times(2)).accept(any(), any());
161 verify(listener2).accept(any(), any());
162 verify(listener3).accept(any(), any());
166 * Tests onMessage() when listener1 throws an exception.
169 public void testOnMessageListenerException1() {
170 doThrow(new IllegalStateException("expected exception")).when(listener1).accept(any(), any());
172 StandardCoderObject sco = makeMessage(Map.of(KEY1, VALUEA_REQID, KEY2, Map.of(SUBKEY, VALUEA_SUBREQID)));
173 forwarder.onMessage(TEXT, sco);
175 verify(listener1b).accept(TEXT, sco);
179 * Tests onMessage() when listener1b throws an exception.
182 public void testOnMessageListenerException1b() {
183 doThrow(new IllegalStateException("expected exception")).when(listener1b).accept(any(), any());
185 StandardCoderObject sco = makeMessage(Map.of(KEY1, VALUEA_REQID, KEY2, Map.of(SUBKEY, VALUEA_SUBREQID)));
186 forwarder.onMessage(TEXT, sco);
188 verify(listener1).accept(TEXT, sco);
192 * Makes a message from a map.
194 private StandardCoderObject makeMessage(Map<String, Object> map) {
195 return Util.translate("", map, StandardCoderObject.class);