gerrit.onap Code Review - policy/apex-pdp.git/blob

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  *  Copyright (C) 2016-2018 Ericsson. 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
   8  *
   9  *      http://www.apache.org/licenses/LICENSE-2.0
  10  *
  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.
  16  *
  17  * SPDX-License-Identifier: Apache-2.0
  18  * ============LICENSE_END=========================================================
  19  */
  20
  21 package org.onap.policy.apex.apps.uservice.test.adapt.jms;
  22
  23 import static org.junit.Assert.assertTrue;
  24
  25 import java.io.IOException;
  26 import java.nio.file.Paths;
  27 import java.util.ArrayList;
  28 import java.util.Arrays;
  29 import java.util.List;
  30
  31 import javax.jms.JMSException;
  32
  33 import org.apache.activemq.ActiveMQConnectionFactory;
  34 import org.apache.activemq.broker.BrokerPlugin;
  35 import org.apache.activemq.broker.BrokerService;
  36 import org.apache.activemq.security.AuthenticationUser;
  37 import org.apache.activemq.security.SimpleAuthenticationPlugin;
  38 import org.junit.AfterClass;
  39 import org.junit.BeforeClass;
  40 import org.junit.Test;
  41 import org.onap.policy.apex.core.infrastructure.threading.ThreadUtilities;
  42 import org.onap.policy.apex.model.basicmodel.concepts.ApexException;
  43 import org.onap.policy.apex.service.engine.main.ApexMain;
  44 import org.slf4j.ext.XLogger;
  45 import org.slf4j.ext.XLoggerFactory;
  46
  47 public class TestJMS2JMS {
  48     private static final String GROUP_ROLE = "guests";
  49     private static final String PACKAGE_NAME = "org.onap.policy.apex.apps.uservice.test.adapt.jms";
  50     private static final String USERNAME = "guest";
  51     private static final String PASSWORD = "IAmAGuest";
  52     private static final String JMS_TOPIC_APEX_IN = "jms/topic/apexIn";
  53     private static final String JMS_TOPIC_APEX_OUT = "jms/topic/apexOut";
  54     private static final String URL = "tcp://localhost:5445";
  55
  56     private static final String DATA_PARENT_DIR = Paths.get("target", "activemq-data").toString();
  57
  58     private static final XLogger LOGGER = XLoggerFactory.getXLogger(TestJMS2JMS.class);
  59
  60     private static final long MAX_TEST_LENGTH = 10000;
  61     private static final int EVENT_COUNT = 100;
  62     private static final int EVENT_INTERVAL = 20;
  63
  64     private static BrokerService broker;
  65
  66     public static ActiveMQConnectionFactory connectionFactory;
  67
  68
  69     @BeforeClass
  70     public static void setupEmbeddedJMSServer() throws Exception {
  71         final ArrayList<BrokerPlugin> plugins = new ArrayList<BrokerPlugin>();
  72         final BrokerPlugin authenticationPlugin = getAuthenticationBrokerPlugin();
  73         if (authenticationPlugin != null) {
  74             plugins.add(authenticationPlugin);
  75         }
  76
  77         broker = new BrokerService();
  78         broker.setUseJmx(false);
  79         broker.setPersistent(false);
  80         broker.addConnector(URL);
  81         broker.setDeleteAllMessagesOnStartup(true);
  82         broker.setPlugins(plugins.toArray(new BrokerPlugin[0]));
  83         broker.setDataDirectory(DATA_PARENT_DIR);
  84         broker.start();
  85         broker.waitUntilStarted();
  86         connectionFactory = new ActiveMQConnectionFactory(URL);
  87         connectionFactory.setTrustedPackages(Arrays.asList(PACKAGE_NAME));
  88     }
  89
  90     private static BrokerPlugin getAuthenticationBrokerPlugin() {
  91         final List<AuthenticationUser> users = new ArrayList<AuthenticationUser>();
  92         users.add(new AuthenticationUser(USERNAME, PASSWORD, GROUP_ROLE));
  93         final SimpleAuthenticationPlugin authenticationPlugin = new SimpleAuthenticationPlugin(users);
  94         return authenticationPlugin;
  95     }
  96
  97     @AfterClass
  98     public static void shutdownEmbeddedJMSServer() throws IOException {
  99         try {
 100             if (broker != null) {
 101                 broker.stop();
 102             }
 103         } catch (final Exception e) {
 104             LOGGER.warn("Failed to stop JMS server", e);
 105         }
 106
 107     }
 108
 109     @Test
 110     public void testJMSObjectEvents() throws ApexException, JMSException {
 111         final String[] args = {"src/test/resources/prodcons/JMS2JMSObjectEvent.json"};
 112         testJMSEvents(args, true);
 113     }
 114
 115     @Test
 116     public void testJMSJsonEvents() throws ApexException, JMSException {
 117         final String[] args = {"src/test/resources/prodcons/JMS2JMSJsonEvent.json"};
 118         testJMSEvents(args, false);
 119     }
 120
 121     private void testJMSEvents(final String[] args, final Boolean sendObjects) throws ApexException, JMSException {
 122         final JMSEventSubscriber subscriber =
 123                 new JMSEventSubscriber(JMS_TOPIC_APEX_OUT, connectionFactory, USERNAME, PASSWORD);
 124         final JMSEventProducer producer = new JMSEventProducer(JMS_TOPIC_APEX_IN, connectionFactory, USERNAME, PASSWORD,
 125                 EVENT_COUNT, sendObjects, EVENT_INTERVAL);
 126
 127         final ApexMain apexMain = new ApexMain(args);
 128         ThreadUtilities.sleep(3000);
 129
 130         producer.sendEvents();
 131
 132         final long testStartTime = System.currentTimeMillis();
 133
 134         while (System.currentTimeMillis() < testStartTime + MAX_TEST_LENGTH
 135                 && subscriber.getEventsReceivedCount() < EVENT_COUNT) {
 136             ThreadUtilities.sleep(EVENT_INTERVAL);
 137         }
 138
 139         ThreadUtilities.sleep(1000);
 140
 141         System.out.println("sent event count: " + producer.getEventsSentCount());
 142         System.out.println("received event count: " + subscriber.getEventsReceivedCount());
 143         assertTrue(subscriber.getEventsReceivedCount() == producer.getEventsSentCount());
 144
 145         apexMain.shutdown();
 146         subscriber.shutdown();
 147         producer.shutdown();
 148         ThreadUtilities.sleep(1000);
 149     }
 150 }