2 * ============LICENSE_START=======================================================
3 * Copyright (C) 2020 AT&T Intellectual Property. All rights reserved.
4 * ================================================================================
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
9 * http://www.apache.org/licenses/LICENSE-2.0
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
17 * SPDX-License-Identifier: Apache-2.0
18 * ============LICENSE_END=========================================================
21 package org.onap.policy.models.simulators;
23 import java.io.FileNotFoundException;
24 import java.lang.reflect.InvocationTargetException;
25 import java.util.HashMap;
26 import java.util.List;
28 import java.util.Properties;
29 import java.util.concurrent.atomic.AtomicReference;
30 import lombok.AccessLevel;
32 import org.onap.policy.common.endpoints.event.comm.TopicEndpointManager;
33 import org.onap.policy.common.endpoints.event.comm.TopicSink;
34 import org.onap.policy.common.endpoints.event.comm.TopicSource;
35 import org.onap.policy.common.endpoints.http.server.HttpServletServer;
36 import org.onap.policy.common.endpoints.http.server.HttpServletServerFactoryInstance;
37 import org.onap.policy.common.endpoints.parameters.TopicParameters;
38 import org.onap.policy.common.endpoints.properties.PolicyEndPointProperties;
39 import org.onap.policy.common.gson.GsonMessageBodyHandler;
40 import org.onap.policy.common.parameters.BeanValidationResult;
41 import org.onap.policy.common.utils.coder.Coder;
42 import org.onap.policy.common.utils.coder.CoderException;
43 import org.onap.policy.common.utils.coder.StandardCoder;
44 import org.onap.policy.common.utils.network.NetworkUtil;
45 import org.onap.policy.common.utils.resources.ResourceUtils;
46 import org.onap.policy.common.utils.services.ServiceManagerContainer;
47 import org.onap.policy.models.sim.dmaap.parameters.DmaapSimParameterGroup;
48 import org.onap.policy.models.sim.dmaap.provider.DmaapSimProvider;
49 import org.onap.policy.models.sim.dmaap.rest.CambriaMessageBodyHandler;
50 import org.onap.policy.models.sim.dmaap.rest.TextMessageBodyHandler;
51 import org.onap.policy.simulators.TopicServer;
52 import org.slf4j.Logger;
53 import org.slf4j.LoggerFactory;
56 * This class runs all simulators specified in the parameter file.
58 public class Main extends ServiceManagerContainer {
59 private static final Logger logger = LoggerFactory.getLogger(Main.class);
61 private static final String CANNOT_CONNECT = "cannot connect to port ";
63 @Getter(AccessLevel.PROTECTED)
64 private static Main instance;
68 * Runs the simulators.
70 * @param paramFile parameter file name
72 public Main(String paramFile) {
73 super(Main.class.getPackage().getName());
75 SimulatorParameters params = readParameters(paramFile);
76 BeanValidationResult result = params.validate("simulators");
77 if (!result.isValid()) {
78 logger.error("invalid parameters:\n{}", result.getResult());
79 throw new IllegalArgumentException("invalid simulator parameters");
82 DmaapSimParameterGroup dmaapProv = params.getDmaapProvider();
83 String dmaapName = dmaapProv.getName();
84 String provName = dmaapName.replace("simulator", "provider");
87 AtomicReference<DmaapSimProvider> provRef = new AtomicReference<>();
88 addAction(provName, () -> provRef.set(buildDmaapProvider(dmaapProv)), () -> provRef.get().shutdown());
90 // REST server simulators
92 for (ClassRestServerParameters restsim : params.getRestServers()) {
93 AtomicReference<HttpServletServer> ref = new AtomicReference<>();
94 addAction(restsim.getName(),
95 () -> ref.set(buildRestServer(dmaapName, restsim)),
96 () -> ref.get().shutdown());
99 // NOTE: topics must be started AFTER the (dmaap) rest servers
102 Map<String, TopicSink> sinks = new HashMap<>();
103 for (TopicParameters topicParams : params.getTopicSinks()) {
104 String topic = topicParams.getTopic();
105 addAction("Sink " + topic,
106 () -> sinks.put(topic, startSink(topicParams)),
107 () -> sinks.get(topic).shutdown());
111 Map<String, TopicSource> sources = new HashMap<>();
112 for (TopicParameters topicParams : params.getTopicSources()) {
113 String topic = topicParams.getTopic();
114 addAction("Source " + topic,
115 () -> sources.put(topic, startSource(topicParams)),
116 () -> sources.get(topic).shutdown());
119 // topic server simulators
120 for (TopicServerParameters topicsim : params.getTopicServers()) {
121 AtomicReference<TopicServer<?>> ref = new AtomicReference<>();
122 addAction(topicsim.getName(),
123 () -> ref.set(buildTopicServer(topicsim, sinks, sources)),
124 () -> ref.get().shutdown());
130 * The main method. The arguments are validated, thus adding the NOSONAR.
132 * @param args the arguments, the first of which is the name of the parameter file
134 public static void main(final String[] args) { // NOSONAR
136 if (args.length != 1) {
137 throw new IllegalArgumentException("arg(s): parameter-file-name");
140 instance = new Main(args[0]);
143 } catch (RuntimeException e) {
144 logger.error("failed to start simulators", e);
148 private SimulatorParameters readParameters(String paramFile) {
150 String paramsJson = getResourceAsString(paramFile);
151 if (paramsJson == null) {
152 throw new IllegalArgumentException(new FileNotFoundException(paramFile));
155 String hostName = NetworkUtil.getHostname();
156 logger.info("replacing 'HOST_NAME' with {} in {}", hostName, paramFile);
158 paramsJson = paramsJson.replace("${HOST_NAME}", hostName);
160 return makeCoder().decode(paramsJson, SimulatorParameters.class);
162 } catch (CoderException e) {
163 throw new IllegalArgumentException("cannot decode " + paramFile, e);
167 private DmaapSimProvider buildDmaapProvider(DmaapSimParameterGroup params) {
168 DmaapSimProvider prov = new DmaapSimProvider(params);
169 DmaapSimProvider.setInstance(prov);
175 private TopicSink startSink(TopicParameters params) {
176 TopicSink sink = TopicEndpointManager.getManager().addTopicSinks(List.of(params)).get(0);
181 private TopicSource startSource(TopicParameters params) {
182 TopicSource source = TopicEndpointManager.getManager().addTopicSources(List.of(params)).get(0);
187 private HttpServletServer buildRestServer(String dmaapName, ClassRestServerParameters params) {
189 Properties props = getServerProperties(dmaapName, params);
190 HttpServletServer testServer = makeServer(props);
191 testServer.waitedStart(5000);
193 String svcpfx = PolicyEndPointProperties.PROPERTY_HTTP_SERVER_SERVICES + "." + params.getName();
194 String hostName = props.getProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_HOST_SUFFIX);
196 if (!isTcpPortOpen(hostName, testServer.getPort())) {
197 throw new IllegalStateException(CANNOT_CONNECT + testServer.getPort());
202 } catch (InterruptedException e) {
203 Thread.currentThread().interrupt();
204 throw new IllegalStateException("interrupted while building " + params.getName(), e);
208 private TopicServer<?> buildTopicServer(TopicServerParameters params, Map<String, TopicSink> sinks,
209 Map<String, TopicSource> sources) {
211 // find the desired sink
212 TopicSink sink = sinks.get(params.getSink());
214 throw new IllegalArgumentException("invalid sink topic " + params.getSink());
217 // find the desired source
218 TopicSource source = sources.get(params.getSource());
219 if (source == null) {
220 throw new IllegalArgumentException("invalid source topic " + params.getSource());
223 // create the topic server
224 return (TopicServer<?>) Class.forName(params.getProviderClass())
225 .getDeclaredConstructor(TopicSink.class, TopicSource.class).newInstance(sink, source);
227 } catch (InstantiationException | IllegalAccessException | InvocationTargetException | NoSuchMethodException
228 | SecurityException | ClassNotFoundException e) {
229 throw new IllegalArgumentException("cannot create TopicServer: " + params.getName(), e);
234 * Creates a set of properties, suitable for building a REST server, from the
237 * @param params parameters from which to build the properties
238 * @return a set of properties representing the given parameters
240 private static Properties getServerProperties(String dmaapName, ClassRestServerParameters params) {
241 final Properties props = new Properties();
242 props.setProperty(PolicyEndPointProperties.PROPERTY_HTTP_SERVER_SERVICES, params.getName());
244 final String svcpfx = PolicyEndPointProperties.PROPERTY_HTTP_SERVER_SERVICES + "." + params.getName();
246 props.setProperty(PolicyEndPointProperties.PROPERTY_HTTP_SERVER_SERVICES, params.getName());
247 props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_HOST_SUFFIX, params.getHost());
248 props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_PORT_SUFFIX,
249 Integer.toString(params.getPort()));
250 props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_HTTPS_SUFFIX,
251 Boolean.toString(params.isHttps()));
252 props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_REST_CLASSES_SUFFIX,
253 params.getProviderClass());
254 props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_MANAGED_SUFFIX, "false");
255 props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_SWAGGER_SUFFIX, "false");
256 props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_MANAGED_SUFFIX, "true");
258 if (dmaapName.equals(params.getName())) {
259 props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_SERIALIZATION_PROVIDER,
260 String.join(",", CambriaMessageBodyHandler.class.getName(),
261 GsonMessageBodyHandler.class.getName(),
262 TextMessageBodyHandler.class.getName()));
264 props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_SERIALIZATION_PROVIDER, String.join(",",
265 GsonMessageBodyHandler.class.getName(), TextMessageBodyHandler.class.getName()));
271 // the following methods may be overridden by junit tests
273 protected String getResourceAsString(String resourceName) {
274 return ResourceUtils.getResourceAsString(resourceName);
277 protected Coder makeCoder() {
278 return new StandardCoder();
281 protected HttpServletServer makeServer(Properties props) {
282 return HttpServletServerFactoryInstance.getServerFactory().build(props).get(0);
285 protected boolean isTcpPortOpen(String hostName, int port) throws InterruptedException {
286 return NetworkUtil.isTcpPortOpen(hostName, port, 100, 200L);