<|start_filename|>modules/integration/tests-kubernetes-integration/src/main/java/org/sp/integration/tests/core/beans/Port.java<|end_filename|>
/*
*  Copyright (c) 2017, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
*
*  WSO2 Inc. licenses this file to you under the Apache License,
*  Version 2.0 (the "License"); you may not use this file except
*  in compliance with the License.
*  You may obtain a copy of the License at
*
*    http://www.apache.org/licenses/LICENSE-2.0
*
*  Unless required by applicable law or agreed to in writing,
*  software distributed under the License is distributed on an
*  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
*  KIND, either express or implied.  See the License for the
*  specific language governing permissions and limitations
*  under the License.
*/

package org.sp.integration.tests.core.beans;

/**
 * Ports definitions
 */
public class Port {
    private String name;
    private int port;
    private int targetPort;
    private int nodePort;
    private String protocol;

    public Port() {
    }

    public Port(String name, int port, int targetPort, int nodePort, String protocol) {
        this.name = name;
        this.port = port;
        this.targetPort = targetPort;
        this.nodePort = nodePort;
        this.protocol = protocol;
    }

    public Port(String name, int port, int targetPort, String protocol) {
        this.name = name;
        this.port = port;
        this.targetPort = targetPort;
        this.protocol = protocol;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public int getPort() {
        return port;
    }

    public void setPort(int port) {
        this.port = port;
    }

    public int getTargetPort() {
        return targetPort;
    }

    public void setTargetPort(int targetPort) {
        this.targetPort = targetPort;
    }

    public int getNodePort() {
        return nodePort;
    }

    public void setNodePort(int nodePort) {
        this.nodePort = nodePort;
    }

    public String getProtocol() {
        return protocol;
    }

    public void setProtocol(String protocol) {
        this.protocol = protocol;
    }
}


<|start_filename|>modules/integration/tests-kubernetes-integration/src/test/java/org/sp/tests/ha/HighAvailabilityIT.java<|end_filename|>
/*
 * Copyright (c) 2017, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.sp.tests.ha;

import org.apache.log4j.Logger;
import org.sp.tests.base.SPBaseTest;
import org.sp.tests.util.HTTPResponse;
import org.testng.Assert;
import org.testng.annotations.BeforeTest;
import org.testng.annotations.Test;

import java.io.IOException;
import java.net.URI;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import static org.sp.tests.ha.util.SiddhiAppUtil.deployPassThroughSiddhiApp;
import static org.sp.tests.ha.util.SiddhiAppUtil.sendEvent;
import static org.sp.tests.ha.util.SiddhiAppUtil.sendEvents;
import static org.sp.tests.util.Constants.DEFAULT_PASSWORD;
import static org.sp.tests.util.Constants.DEFAULT_USER_NAME;
import static org.sp.tests.util.Constants.HEADER_CONTTP_JSON;
import static org.sp.tests.util.Constants.HEADER_CONTTP_TEXT;
import static org.sp.tests.util.Constants.HTTP_GET;
import static org.sp.tests.util.Constants.HTTP_RESP_200;
import static org.sp.tests.util.Constants.HTTP_RESP_201;
import static org.sp.tests.util.Constants.SINGLE_EVENT_SIMULATION_STARTED_SUCCESSFULLY;
import static org.sp.tests.util.TestUtil.sendHRequest;
import static org.sp.tests.util.TestUtil.waitThread;

public class HighAvailabilityIT extends SPBaseTest {
    private static final Logger log = Logger.getLogger(HighAvailabilityIT.class);
    private static final String SIDDHI_APP_NAME = "TestMinimumHA";
    private static final String TEST_NAME = "cclassName:com.sp.test.HighAvailabilityIT";

    private URI nodeOneURI;
    private URI nodeTwoURI;
    private URI msf4jBaseURI;

    @BeforeTest
    public void testInit() {
        log.info("Starting test " + this.getClass().getCanonicalName());
        nodeOneURI = URI.create(haNodeOneURL);
        nodeTwoURI = URI.create(haNodeTwoURL);
        msf4jBaseURI = URI.create(haNodeTwoMsf4jURL);
    }

    @Test
    public void haFailOverTest() throws IOException {

        HTTPResponse httpResponseNodeOne = deployPassThroughSiddhiApp(nodeOneURI, SIDDHI_APP_NAME);
        Assert.assertEquals(httpResponseNodeOne.getResponseCode(), HTTP_RESP_201, httpResponseNodeOne.getMessage());

        HTTPResponse httpResponseNodeTwo = deployPassThroughSiddhiApp(nodeTwoURI, SIDDHI_APP_NAME);
        Assert.assertEquals(httpResponseNodeTwo.getResponseCode(), HTTP_RESP_201, httpResponseNodeTwo.getMessage());

        waitThread(1000); //Wait for Siddhi applications to deploy

        // Checking if both Siddhi Applications are receiving events
        HTTPResponse sendEventResponseNodeOne = sendEvent(nodeOneURI, TEST_NAME, SIDDHI_APP_NAME, "FooStream",
                "First", "1");

        Assert.assertEquals(sendEventResponseNodeOne.getResponseCode(), HTTP_RESP_200,
                sendEventResponseNodeOne.getMessage());
        Assert.assertEquals(sendEventResponseNodeOne.getContentType(), HEADER_CONTTP_JSON);
        Assert.assertEquals(sendEventResponseNodeOne.getMessage(), SINGLE_EVENT_SIMULATION_STARTED_SUCCESSFULLY);
        waitThread(2000); //Wait for event to publish

        HTTPResponse sendEventResponseNodeTwo = sendEvent(nodeTwoURI, TEST_NAME, SIDDHI_APP_NAME, "FooStream",
                "First", "1");

        Assert.assertEquals(sendEventResponseNodeTwo.getResponseCode(), HTTP_RESP_200,
                sendEventResponseNodeTwo.getMessage());
        Assert.assertEquals(sendEventResponseNodeTwo.getContentType(), HEADER_CONTTP_JSON);
        Assert.assertEquals(sendEventResponseNodeTwo.getMessage(),
                SINGLE_EVENT_SIMULATION_STARTED_SUCCESSFULLY);
        waitThread(2000); //Wait for event to publish

        EventSender eventSenderOne = new EventSender(nodeOneURI, TEST_NAME, SIDDHI_APP_NAME);
        EventSender eventSenderTwo = new EventSender(nodeTwoURI, TEST_NAME, SIDDHI_APP_NAME);
        eventSenderOne.start();
        eventSenderTwo.start();

        waitThread(40000); //Wait before shutting down node 1

        super.runBashScript("ha-scripts", "shutdown-node-1-server.sh");
        while (eventSenderTwo.isAlive()) {
            waitThread(5000);
        }

        log.info("Waiting for events to finish publishing");
        waitThread(5000);
        HTTPResponse msf4jResponse = sendHRequest("", msf4jBaseURI,
                "/testresults/com.sp.test.HighAvailabilityI/count",
                HEADER_CONTTP_TEXT, HTTP_GET, DEFAULT_USER_NAME, DEFAULT_PASSWORD);

        Pattern p = Pattern.compile("-?\\d+");
        Matcher m = p.matcher(msf4jResponse.getMessage());
        if (m.find()) {
            Integer eventCount = Integer.valueOf(m.group());
            Assert.assertTrue(eventCount >= 10000);
        } else {
            Assert.fail("Error in getting event count");
        }

    }
}

class EventSender extends Thread {

    private final URI nodeURI;
    private final String testName;
    private final String siddhiAppName;

    EventSender(URI nodeURI, String testName, String siddhiAppName) {
        this.nodeURI = nodeURI;
        this.testName = testName;
        this.siddhiAppName = siddhiAppName;
    }

    @Override
    public void run() {
        try {
            sendEvents(10000, nodeURI, testName, siddhiAppName, "FooStream", true);
        } catch (IOException e) {
            this.stop();
        }
    }
}



<|start_filename|>modules/integration/tests-kubernetes-integration/src/test/java/org/sp/tests/base/SPInit.java<|end_filename|>
/*
 *  Copyright (c) 2017, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 *  WSO2 Inc. licenses this file to you under the Apache License,
 *  Version 2.0 (the "License"); you may not use this file except
 *  in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing,
 *  software distributed under the License is distributed on an
 *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 *  KIND, either express or implied.  See the License for the
 *  specific language governing permissions and limitations
 *  under the License.
 */
package org.sp.tests.base;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.sp.integration.tests.core.beans.InstanceUrls;
import org.sp.integration.tests.core.beans.Port;
import org.sp.integration.tests.core.commons.DeploymentConfigurationReader;
import org.sp.integration.tests.core.commons.DeploymentDataReader;
import org.sp.integration.tests.core.utills.ScriptExecutorUtil;

import java.io.IOException;
import java.util.HashMap;
import java.util.List;

/**
 * SPInit class
 */
public class SPInit {

    private static final Log log = LogFactory.getLog(SPInit.class);
    public static String spURL;
    public static String msf4jURL;
    public static String haNodeOneMsf4jURL;
    public static String haNodeTwoMsf4jURL;
    public static String mysqlURL;
    public static String haNodeOneURL;
    public static String haNodeTwoURL;

    protected HashMap<String, String> instanceMap;

    /**
     * This method will initialize test environment
     * based on the configuration given at testng.xml
     */
    protected void init(String pattern) throws Exception {
        setURLs(pattern);
    }

    //set the url's as specified in the deployment.json
    protected void setURLs(String patternName) {

        HashMap<String, String> instanceMap = null;
        try {
            DeploymentConfigurationReader depconf = DeploymentConfigurationReader.readConfiguration();
            instanceMap = depconf.getDeploymentInstanceMap(patternName);
        } catch (IOException e) {
            log.error("Exception occured while getting the deployment instance map : " + e.getMessage(), e);
        }
        DeploymentDataReader dataJsonReader = new DeploymentDataReader();
        List<InstanceUrls> urlList = dataJsonReader.getInstanceUrlsList();
        for (InstanceUrls url : urlList) {
            if (instanceMap != null) {
                if (url.getLabel().equals(instanceMap.get(SPConstants.POD_TAG_NAME))) {
                    spURL = getHTTPSUrl(SPConstants.SP_PORT_NAME, url.getHostIP(), url.getPorts(), "");
                    msf4jURL = getHTTPSUrl(SPConstants.MSF4J_PORT_NAME, url.getHostIP(), url.getPorts(), "");
                } else if (url.getLabel().equals(instanceMap.get(SPConstants.HA_NODE_1_POD_NAME))) {
                    haNodeOneURL = getHTTPSUrl(SPConstants.SP_PORT_NAME, url.getHostIP(), url.getPorts(), "");
                    haNodeOneMsf4jURL = getHTTPSUrl(SPConstants.MSF4J_PORT_NAME, url.getHostIP(), url.getPorts(), "");
                } else if (url.getLabel().equals(instanceMap.get(SPConstants.HA_NODE_2_POD_NAME))) {
                    haNodeTwoURL = getHTTPSUrl(SPConstants.SP_PORT_NAME, url.getHostIP(), url.getPorts(), "");
                    haNodeTwoMsf4jURL = getHTTPSUrl(SPConstants.MSF4J_PORT_NAME, url.getHostIP(), url.getPorts(), "");
                } else if (url.getLabel().equals(instanceMap.get(SPConstants.MYSQL_POD_NAME))) {
                    mysqlURL = getHTTPSUrl(SPConstants.MYSQL_PORT_NAME, url.getHostIP(), url.getPorts(), "");
                }
            }
        }
    }

    protected String getHTTPSUrl(String protocol, String hostIP, List<Port> ports, String context) {

        String url = "http://" + hostIP + ":";
        for (Port port : ports) {
            if (port.getProtocol().equals(protocol)) {
                url = url + port.getPort() + context;
                break;
            }
        }
        return url;
    }

    protected String getJDBCUrl(String protocol, String hostIP, List<Port> ports, String databasename) {

        // 192.168.48.44:30306/DAS_DB
        String url = hostIP + ":";
        for (Port port : ports) {
            if (port.getProtocol().equals(protocol)) {
                url = url + port.getPort() + databasename;
                break;
            }
        }
        return url;
    }

    private boolean isURLRemapEnabled() {
        log.info("URL Remap Enabled is set to : " + System.getenv(SPConstants.ENABLE_URL_REMAP));
        return Boolean.parseBoolean((System.getenv(SPConstants.ENABLE_URL_REMAP)));
    }

    private String getRemappedURL(String localIP) {

        String remappedURL = System.getenv("IP_" + localIP.replace(".", "_"));

        if (remappedURL.equals("") | remappedURL == null) {
            log.info("No remap value found for the Local IP : " + localIP);
        }
        return remappedURL;
    }

    //deploy environment
    protected void setTestSuite(String pattern) throws IOException {
        ScriptExecutorUtil.deployScenario(pattern);
    }

    //Undeploy environment
    protected void unSetTestSuite(String pattern) throws Exception {
        ScriptExecutorUtil.unDeployScenario(pattern);
    }

    //Run a script
    public void runBashScript(String patternName, String scriptName) throws IOException {
        ScriptExecutorUtil.execute(patternName, scriptName);
    }
}


<|start_filename|>modules/performance/sample-clients/tcp-client/src/main/java/org/wso2/sp/tcp/client/TCPClient.java<|end_filename|>
/*
 * Copyright (c) 2017, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * WSO2 Inc. licenses this file to you under the Apache License,
 * Version 2.0 (the "License"); you may not use this file except
 * in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package org.wso2.sp.tcp.client;

import org.apache.log4j.Logger;
import org.wso2.extension.siddhi.io.tcp.transport.TCPNettyClient;
import org.wso2.extension.siddhi.map.binary.sinkmapper.BinaryEventConverter;
import org.wso2.siddhi.core.event.Event;
import org.wso2.siddhi.core.exception.ConnectionUnavailableException;
import org.wso2.siddhi.query.api.definition.Attribute;

import java.io.IOException;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.Random;

/**
 * Test client for TCP source
 */
public class TCPClient {
    private static final long EVENT_COUNT = 100000000L;
    private static final String STREAM_NAME = "inputStream"; //TCP_Benchmark
    private static final Attribute.Type[] TYPES = new Attribute.Type[]{Attribute.Type.LONG, Attribute.Type.FLOAT};
    private static final Logger LOG = Logger.getLogger(TCPClient.class);
    private static final int BATCH_SIZE = 1000;
    private static long startTime = 0;

    /**
     * Main method to start the test client
     */
    public static void main(String[] args) {
        try {
            TCPNettyClient tcpNettyClient = new TCPNettyClient();
            tcpNettyClient.connect("localhost", Integer.parseInt("9892"));
            LOG.info("TCP client connected");
            Random rand = new Random();
            long i = 0;
            for (; i < EVENT_COUNT; i += BATCH_SIZE) {
                startTime = System.currentTimeMillis();
                ArrayList<Event> arrayList = new ArrayList<Event>(BATCH_SIZE);
                for (int j = 0; j < BATCH_SIZE; j++) {
                    Event event = new Event();
                    event.setData(new Object[]{System.currentTimeMillis(), rand.nextFloat()});
                    long time = System.currentTimeMillis();
                    if (time - startTime <= 1) {
                        arrayList.add(event);
                    }
                }
                tcpNettyClient.send(STREAM_NAME, BinaryEventConverter.convertToBinaryMessage(
                        arrayList.toArray(new Event[0]), TYPES).array());
                long currentTime = System.currentTimeMillis();
                if (currentTime - startTime <= 1) {
                    try {

                        Thread.sleep(1 - (currentTime - startTime));
                    } catch (InterruptedException ex) {
                        LOG.error(ex);
                    }
                }
            }
            LOG.info("TCP client finished sending events");
            try {
                LOG.info("TCP client finished sending events");
                Thread.sleep(1000);
            } catch (InterruptedException e) {
            }
            tcpNettyClient.disconnect();
            tcpNettyClient.shutdown();
        } catch (ConnectionUnavailableException ex) {
            LOG.error(ex);
        } catch (IOException e) {
            LOG.error(e);
        }
    }

    private static String getCurrentTimestamp() {
        DateFormat dateFormat = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");
        Date date = new Date();
        return dateFormat.format(date);
    }
}


<|start_filename|>modules/integration/tests-kubernetes-integration/src/test/java/org/sp/tests/util/Constants.java<|end_filename|>
/**
 *  Copyright (c) 2017, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 *  WSO2 Inc. licenses this file to you under the Apache License,
 *  Version 2.0 (the "License"); you may not use this file except
 *  in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing,
 *  software distributed under the License is distributed on an
 *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 *  KIND, either express or implied.  See the License for the
 *  specific language governing permissions and limitations
 *  under the License.
 */
package org.sp.tests.util;
/**
 * Constants class
 * */
public class Constants {
    private Constants() {

    }

    public static final String SIDDHI_APP_API = "http://localhost:9090";
    public static final String MSF4J_TEST_API = "http://localhost:8080";
    public static final Integer SP_PORT = 9090;
    public static final String HTTP_POST = "POST";
    public static final String HTTP_GET = "GET";
    public static final String HTTP_DELETE = "DELETE";
    public static final String HEADER_ACCEPT = "Accept";
    public static final String HEADER_CONTENT_TYPE = "Content-Type";
    public static final String HEADER_CONTTP_JSON = "application/json";
    public static final String HEADER_CONTTP_XML = "application/xml";
    public static final String HEADER_CONTTP_TEXT = "text/plain";
    public static final String DEFAULT_USER_NAME = "admin";
    public static final String DEFAULT_PASSWORD = "<PASSWORD>";
    public static final int HTTP_RESP_200 = 200;
    public static final int HTTP_RESP_201 = 201;
    public static final int HTTP_RESP_204 = 204;
    public static final int HTTP_RESP_401 = 401;
    public static final int HTTP_RESP_400 = 400;
    public static final String MYSQL_USERNAME = "spintegrationtest";
    public static final String MYSQL_PASSWORD = "<PASSWORD>";
    public static final String SINGLE_EVENT_SIMULATION_STARTED_SUCCESSFULLY =
            "{\"status\":\"OK\",\"message\":\"Single Event simulation started successfully\"}";

}


<|start_filename|>modules/integration/tests-kubernetes-integration/src/test/java/org/sp/tests/routing/EchoService.java<|end_filename|>
/*
*  Copyright (c) 2017, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
*
*  WSO2 Inc. licenses this file to you under the Apache License,
*  Version 2.0 (the "License"); you may not use this file except
*  in compliance with the License.
*  You may obtain a copy of the License at
*
*    http://www.apache.org/licenses/LICENSE-2.0
*
*  Unless required by applicable law or agreed to in writing,
*  software distributed under the License is distributed on an
*  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
*  KIND, either express or implied.  See the License for the
*  specific language governing permissions and limitations
*  under the License.
*/
package org.sp.tests.routing;

import org.sp.tests.base.SPBaseTest;
import org.testng.annotations.Test;
/**
* EchoService class
 */
public class EchoService extends SPBaseTest {

    @Test public void getPostToEcho() throws Exception {

       String serviceURL = spURL + "/echo";
//
//        StringRequestEntity requestEntity = new StringRequestEntity("{ \"Hello\":\"DAS\" };", "application/json",
//                "UTF-8");
//
//        HttpClient client = new HttpClient();
//        PostMethod post = new PostMethod(serviceURL);
//
//        post.setRequestEntity(requestEntity);
//
//        int statuscode = client.executeMethod(post);
//        assertEquals(statuscode, 201);
    }
}


<|start_filename|>modules/integration/tests-kubernetes-integration/src/test/java/org/sp/tests/util/TestResults.java<|end_filename|>
/*
*  Copyright (c) 2017, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
*
*  WSO2 Inc. licenses this file to you under the Apache License,
*  Version 2.0 (the "License"); you may not use this file except
*  in compliance with the License.
*  You may obtain a copy of the License at
*
*    http://www.apache.org/licenses/LICENSE-2.0
*
*  Unless required by applicable law or agreed to in writing,
*  software distributed under the License is distributed on an
*  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
*  KIND, either express or implied.  See the License for the
*  specific language governing permissions and limitations
*  under the License.
*/
package org.sp.tests.util;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
/**
 *
 * TestUtil class
 */
public abstract class TestResults {

    private static final Log log = LogFactory.getLog(TestResults.class);
    //method for msf4j service listener implementation
    public abstract void waitForResults(int retryCount, long interval);

    //This will verify the actual result with expected.
    public abstract boolean resultsFound(String eventMessage);

    //msf4j service listener - wait for and verify test results
    public synchronized void verifyResult(long interval, int maxRetry, String eventMessage) {
        boolean arrived = resultsFound(eventMessage);
        int i;
        while (!arrived) {
            for (i = 1; i <= maxRetry; i++) {
                arrived = resultsFound(eventMessage);
                try {
                    if (arrived) {
                        break;
                    }
                    wait(interval);
                } catch (InterruptedException e) {
                    log.error(e.getMessage());
                }
            }
            if (i == maxRetry + 1) {
                break;
            }
        }
    }

}


<|start_filename|>modules/integration/tests-kubernetes-integration/src/test/java/org/sp/tests/ha/util/SiddhiAppUtil.java<|end_filename|>
/*
 * Copyright (c) 2017, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.sp.tests.ha.util;

import org.apache.log4j.Logger;
import org.sp.tests.util.HTTPResponse;
import org.sp.tests.util.TestUtil;

import java.io.IOException;
import java.net.URI;
import java.util.Date;

import static org.sp.tests.util.Constants.DEFAULT_PASSWORD;
import static org.sp.tests.util.Constants.DEFAULT_USER_NAME;
import static org.sp.tests.util.Constants.HEADER_CONTTP_TEXT;
import static org.sp.tests.util.Constants.HTTP_POST;
import static org.sp.tests.util.TestUtil.waitThread;

public class SiddhiAppUtil {

    private static final Logger log = Logger.getLogger(SiddhiAppUtil.class);

    public static HTTPResponse deployPassThroughSiddhiApp(URI nodeURI, String siddhiAppName) throws IOException {

        String publisherURL = "http://sp-ha-node-2:8080/testresults";
        String appbody = "@App:name('" + siddhiAppName + "')\n" +
                "@source(type='inMemory', topic='symbol', @map(type='xml'))\n" +
                "define stream FooStream (symbol string, price float, class string);\n" +
                "@sink(type='http', publisher.url='" + publisherURL + "', method='{{method}}', " +
                "headers='{{headers}}',\n" +
                "@map(type='json'))\n" +
                "define stream BarStream (message string, value float, method string, headers string);\n" +

                "from FooStream#log()\n" +
                "select symbol as message, price as value, 'POST' as method, class as headers\n" +
                "insert into BarStream;";

        log.info("Deploying Siddhi Application on " + nodeURI);
        HTTPResponse httpResponse = TestUtil.sendHRequest(appbody, nodeURI, "/siddhi-apps", HEADER_CONTTP_TEXT,
                HTTP_POST, DEFAULT_USER_NAME, DEFAULT_PASSWORD);

        waitThread(2000);
        return httpResponse;
    }

    public static HTTPResponse deployAggregateSiddhiApp(URI nodeURI, String siddhiAppName) throws IOException {

        String publisherURL = "http://sp-ha-node-2:8080/testresults";
        String appbody = "@App:name('" + siddhiAppName + "')\n" +
                "@source(type='inMemory', topic='symbol', @map(type='xml'))\n" +
                "define stream FooStream (symbol string, price float, class string);\n" +
                "@sink(type='http', publisher.url='" + publisherURL + "', method='{{method}}', " +
                "headers='{{headers}}',\n" +
                "@map(type='json'))\n" +
                "define stream BarStream (message string, value float, method string, headers string);\n" +

                "from FooStream#window.length(5)\n" +
                "select symbol as message, max(price) as value, 'POST' as method, class as headers\n" +
                "insert into BarStream;\n" +

                "from BarStream#log(\"Events: \")\n" +
                "insert into OutputStream;";

        log.info("Deploying Siddhi Application on " + nodeURI);
        HTTPResponse httpResponse = TestUtil.sendHRequest(appbody, nodeURI, "/siddhi-apps", HEADER_CONTTP_TEXT,
                HTTP_POST, DEFAULT_USER_NAME, DEFAULT_PASSWORD);

        waitThread(2000);
        return httpResponse;
    }

    public static HTTPResponse sendEvent(URI nodeURI, String testName, String siddhiAppName, String streamName,
                                         String message, String timestamp) throws IOException {

        String queryBody = "{\n" +
                "  \"streamName\": \"" + streamName + "\",\n" +
                "  \"siddhiAppName\": \"" + siddhiAppName + "\",\n" +
                "  \"timestamp\": \"" + timestamp + "\",\n" +
                "  \"data\": [\n" +
                message + ",\n" +
                "   0.0f,\n" +
                "   \"" + testName + "\"\n" +
                "  ]\n" +
                "}";

        return TestUtil.sendHRequest(queryBody, nodeURI, "/simulation/single",
                HEADER_CONTTP_TEXT, HTTP_POST, DEFAULT_USER_NAME, DEFAULT_PASSWORD);
    }

    public static HTTPResponse sendEvent(URI nodeURI, String testName, String siddhiAppName, String streamName,
                                         String message, float price) throws IOException {

        String queryBody = "{\n" +
                "  \"streamName\": \"" + streamName + "\",\n" +
                "  \"siddhiAppName\": \"" + siddhiAppName + "\",\n" +
                "  \"timestamp\": null,\n" +
                "  \"data\": [\n" +
                message + ",\n" +
                price + ",\n" +
                "   \"" + testName + "\"\n" +
                "  ]\n" +
                "}";

        return TestUtil.sendHRequest(queryBody, nodeURI, "/simulation/single",
                HEADER_CONTTP_TEXT, HTTP_POST, DEFAULT_USER_NAME, DEFAULT_PASSWORD);
    }

    public static void sendEvents(int count, URI uri, String testName, String siddhiAppName, String streamName,
                                  boolean withTimestamp)
            throws IOException {
        log.info("Sending " + count + " event(s) to " + uri);
        for (int i = 1; i <= count; i++) {
            if (i % 1000 == 0 && i != 0) {
                log.info(new Date() + " Number of events sent: " + i);
            }
            if (withTimestamp) {
                sendEvent(uri, testName, siddhiAppName, streamName, String.valueOf(i), String.valueOf(i));
            } else {
                sendEvent(uri, testName, siddhiAppName, streamName, String.valueOf(i), 0f);
            }
        }
    }
}


<|start_filename|>modules/integration/tests-kubernetes-integration/src/main/java/org/sp/integration/tests/core/commons/GenericYamlParser.java<|end_filename|>
/*
*  Copyright (c) 2017, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
*
*  WSO2 Inc. licenses this file to you under the Apache License,
*  Version 2.0 (the "License"); you may not use this file except
*  in compliance with the License.
*  You may obtain a copy of the License at
*
*    http://www.apache.org/licenses/LICENSE-2.0
*
*  Unless required by applicable law or agreed to in writing,
*  software distributed under the License is distributed on an
*  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
*  KIND, either express or implied.  See the License for the
*  specific language governing permissions and limitations
*  under the License.
*/

package org.sp.integration.tests.core.commons;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.yaml.snakeyaml.Yaml;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.Map;

/**
 * Universal yaml parser class - parses yaml files
 */
@SuppressWarnings("ALL")
class GenericYamlParser {

    private Log log = LogFactory.getLog(GenericYamlParser.class);

    Map<String, String> yamlInitializer(String fileName) {

        Yaml yaml = new Yaml();
        Map<String, String> map = null;
        InputStream inputStream = null;

        try {
            inputStream = new FileInputStream(new File(fileName));
            // Parse the YAML file and return the output as a series of Maps and Lists
            map = (Map<String, String>) yaml.load(inputStream);

        } catch (Exception e) {
            log.error("Error " + e.getMessage());
        } finally {
            if (inputStream != null) {
                try {
                    inputStream.close();
                } catch (IOException e) {
                    log.error(e.getMessage());
                }
            }
        }
        return map;
    }
}


<|start_filename|>modules/performance/extension/siddhi-execution-performance/component/src/main/java/org/wso2/extension/siddhi/execution/throughput/util/filewriting/ThroughputFileWriting.java<|end_filename|>
/*
 * Copyright (c) 2017, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * WSO2 Inc. licenses this file to you under the Apache License,
 * Version 2.0 (the "License"); you may not use this file except
 * in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied. See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package org.wso2.extension.siddhi.execution.throughput.util.filewriting;

import org.apache.log4j.Logger;

import java.io.IOException;
import java.io.Writer;

/**
 * Writes Throughput calculations in the csv file.
 */

public class ThroughputFileWriting implements Runnable {
    private final Logger log = Logger.getLogger(ThroughputFileWriting.class);
    private long firstTupleTime;
    private int recordWindow;
    private long currentTime;
    private long value;
    private long eventCountTotal;
    private long eventCount;
    private Writer fstream;

    public ThroughputFileWriting(long firstTupleTime, int recordWindow, long eventCountTotal, long
            eventCount, long currentTime, long value, Writer fstream) {
        this.firstTupleTime = firstTupleTime;
        this.recordWindow = recordWindow;
        this.eventCountTotal = eventCountTotal;
        this.eventCount = eventCount;
        this.currentTime = currentTime;
        this.value = value;
        this.fstream = fstream;
    }

    @Override public void run() {
        try {
            fstream.write(
                    (eventCountTotal / recordWindow) + "," + ((eventCount * 1000) / value) + ","
                            +
                            (eventCountTotal * 1000 / (currentTime - firstTupleTime)) + "," +
                            ((currentTime - firstTupleTime) / 1000f) + ","
                            + "" + eventCountTotal + "," + currentTime);
            fstream.write("\r\n");
            fstream.flush();
        } catch (IOException ex) {
            log.error("Error while writing into the file" + ex.getMessage(), ex);
        }
    }
}


<|start_filename|>modules/integration/tests-kubernetes-integration/src/test/java/org/sp/tests/ha/StateSyncIT.java<|end_filename|>
/*
 * Copyright (c) 2017, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.sp.tests.ha;

import org.apache.log4j.Logger;
import org.sp.tests.base.SPBaseTest;
import org.sp.tests.util.HTTPResponse;
import org.sp.tests.util.TestUtil;
import org.testng.Assert;
import org.testng.ITestResult;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.AfterTest;
import org.testng.annotations.BeforeTest;
import org.testng.annotations.Test;

import java.io.IOException;
import java.net.URI;

import static org.sp.tests.ha.util.SiddhiAppUtil.deployAggregateSiddhiApp;
import static org.sp.tests.ha.util.SiddhiAppUtil.sendEvent;
import static org.sp.tests.ha.util.TestCaseUtil.getExpectedEventsMessage;
import static org.sp.tests.ha.util.TestCaseUtil.getTestListener;
import static org.sp.tests.util.Constants.DEFAULT_PASSWORD;
import static org.sp.tests.util.Constants.DEFAULT_USER_NAME;
import static org.sp.tests.util.Constants.HEADER_CONTTP_JSON;
import static org.sp.tests.util.Constants.HEADER_CONTTP_TEXT;
import static org.sp.tests.util.Constants.HTTP_GET;
import static org.sp.tests.util.Constants.HTTP_POST;
import static org.sp.tests.util.Constants.HTTP_RESP_200;
import static org.sp.tests.util.Constants.HTTP_RESP_201;
import static org.sp.tests.util.Constants.HTTP_RESP_204;
import static org.sp.tests.util.Constants.SINGLE_EVENT_SIMULATION_STARTED_SUCCESSFULLY;
import static org.sp.tests.util.TestUtil.waitThread;

public class StateSyncIT extends SPBaseTest {
    private static final Logger log = Logger.getLogger(StateSyncIT.class);
    private static final String SIDDHI_APP_NAME = "TestMinimumHA";
    private static final String TEST_NAME = "cclassName:com.sp.test.StateSync";

    private URI nodeOneURI;
    private URI nodeTwoURI;
    private URI msf4jBaseURI;

    @BeforeTest
    public void testInit() {
        log.info("Starting test " + this.getClass().getCanonicalName());
        nodeOneURI = URI.create(haNodeOneURL);
        nodeTwoURI = URI.create(haNodeTwoURL);
        msf4jBaseURI = URI.create(haNodeTwoMsf4jURL);
    }

    @Test
    public void testHaInit() throws IOException {

        HTTPResponse httpResponseNodeOne = deployAggregateSiddhiApp(nodeOneURI, SIDDHI_APP_NAME);
        Assert.assertEquals(httpResponseNodeOne.getResponseCode(), HTTP_RESP_201, httpResponseNodeOne.getMessage());

        waitThread(4000); //Wait for Siddhi application to deploy

        HTTPResponse nodeOneResponse = sendEvent(nodeOneURI, TEST_NAME, SIDDHI_APP_NAME, "FooStream", "First", 20f);
        Assert.assertEquals(nodeOneResponse.getResponseCode(), HTTP_RESP_200, nodeOneResponse.getMessage());
        Assert.assertEquals(nodeOneResponse.getContentType(), HEADER_CONTTP_JSON);
        Assert.assertEquals(nodeOneResponse.getMessage(), SINGLE_EVENT_SIMULATION_STARTED_SUCCESSFULLY);
        waitThread(2000); //Wait for event to publish

        // Checking if the Msf4j Service is working properly by checking if test case is registered
        getTestListener(getExpectedEventsMessage(TEST_NAME, "First", 20), msf4jBaseURI,
                "/testresults/com.sp.test.StateSyn/?eventIndex=" + 0).waitForResults(10, 1000);
        HTTPResponse msf4jResponse = TestUtil.sendHRequest("", msf4jBaseURI, "/testresults/" +
                        "com.sp.test.StateSyn/" + "?eventIndex=" + 0, HEADER_CONTTP_TEXT, HTTP_GET,
                DEFAULT_USER_NAME, DEFAULT_PASSWORD);
        Assert.assertEquals(msf4jResponse.getMessage(), getExpectedEventsMessage(TEST_NAME, "First", 20));

        // Begin Tests
        sendEvent(nodeOneURI, TEST_NAME, SIDDHI_APP_NAME, "FooStream", "message", 100f);
        sendEvent(nodeOneURI, TEST_NAME, SIDDHI_APP_NAME, "FooStream", "message", 150f);
        sendEvent(nodeOneURI, TEST_NAME, SIDDHI_APP_NAME, "FooStream", "message", 90f);

        HTTPResponse httpResponseNodeTwo = deployAggregateSiddhiApp(nodeTwoURI, SIDDHI_APP_NAME);
        Assert.assertEquals(httpResponseNodeTwo.getResponseCode(), HTTP_RESP_201, httpResponseNodeTwo.getMessage());

        waitThread(2000); //Wait for siddhi application to deploy

        super.runBashScript("ha-scripts", "shutdown-node-1-server.sh");

        sendEvent(nodeTwoURI, TEST_NAME, SIDDHI_APP_NAME, "FooStream", "message", 100f);

        getTestListener(getExpectedEventsMessage(TEST_NAME, "First", 20), msf4jBaseURI,
                "/testresults/com.sp.test.StateSyn/?eventIndex=" + 4).waitForResults(10, 1000);
        msf4jResponse = TestUtil.sendHRequest("", msf4jBaseURI, "/testresults/" +
                        "com.sp.test.StateSyn/" + "?eventIndex=" + 4, HEADER_CONTTP_TEXT, HTTP_GET,
                DEFAULT_USER_NAME, DEFAULT_PASSWORD);
        Assert.assertEquals(msf4jResponse.getMessage(), getExpectedEventsMessage(TEST_NAME, "message", 150));

    }

    @AfterMethod
    public void afterMethod(ITestResult result) throws IOException {
        log.info("After Method Running!");
        if (result.getStatus() == ITestResult.FAILURE) {
            log.info("Found Errors. Printing Log");
            runBashScript("ha-scripts", "print-log.sh");
        }
    }

    @AfterTest
    public void tearDown() {
        log.info("Finishing test " + this.getClass().getCanonicalName());
        HTTPResponse msf4jClearResponse = null;
        try {
            msf4jClearResponse = TestUtil.sendHRequest("", msf4jBaseURI, "/testresults/clear",
                    HEADER_CONTTP_TEXT, HTTP_POST, DEFAULT_USER_NAME, DEFAULT_PASSWORD);
            Assert.assertEquals(msf4jClearResponse.getResponseCode(), HTTP_RESP_204);
        } catch (IOException e) {
            TestUtil.handleException("IOException occurred when tearing down", e);
        }
    }
}


<|start_filename|>modules/integration/tests-kubernetes-integration/src/test/java/org/sp/tests/TestXMLInput.java<|end_filename|>
/*
*  Copyright (c) 2017, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
*
*  WSO2 Inc. licenses this file to you under the Apache License,
*  Version 2.0 (the "License"); you may not use this file except
*  in compliance with the License.
*  You may obtain a copy of the License at
*
*    http://www.apache.org/licenses/LICENSE-2.0
*
*  Unless required by applicable law or agreed to in writing,
*  software distributed under the License is distributed on an
*  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
*  KIND, either express or implied.  See the License for the
*  specific language governing permissions and limitations
*  under the License.
*/
package org.sp.tests;

import org.apache.log4j.Logger;
import org.sp.tests.base.SPBaseTest;
import org.sp.tests.util.HTTPResponse;
import org.sp.tests.util.TestResults;
import org.sp.tests.util.TestUtil;
import org.testng.Assert;
import org.testng.annotations.AfterTest;
import org.testng.annotations.BeforeTest;
import org.testng.annotations.Test;

import java.io.IOException;
import java.net.URI;

import static org.sp.tests.util.Constants.DEFAULT_PASSWORD;
import static org.sp.tests.util.Constants.DEFAULT_USER_NAME;
import static org.sp.tests.util.Constants.HEADER_CONTTP_JSON;
import static org.sp.tests.util.Constants.HEADER_CONTTP_TEXT;
import static org.sp.tests.util.Constants.HTTP_GET;
import static org.sp.tests.util.Constants.HTTP_POST;
import static org.sp.tests.util.Constants.SINGLE_EVENT_SIMULATION_STARTED_SUCCESSFULLY;
import static org.sp.tests.util.TestUtil.waitThread;

/**
 * TestXMLInput class.
 */
public class TestXMLInput extends SPBaseTest {
    private static final Logger log = Logger.getLogger(TestXMLInput.class);
    private URI msf4jbaseURI;
    //private final URI appApibaseURI = URI.create(SIDDHI_APP_API);
    private URI appApibaseURI;

    @BeforeTest
    public void testInit() {
        appApibaseURI = URI.create(spURL);
        msf4jbaseURI = URI.create(msf4jURL);
    }

    @Test
    public void testXmlInputMapping() {
        //Deploy siddhi app having input receiver with xml event mapper
        // and include http sink to send event data to msf4j service
        //String pubUrl = msf4jbaseURI + "/testresults";
        String pubUrl = "http://localhost:8080/testresults";
        String appbody = "@App:name('TestSiddhiAppxml')\n" +
                "@source(type='inMemory', topic='symbol', @map(type='xml'))\n" +
                "define stream FooStream (symbol string, price float, class string);\n" +
                "@Sink(type='http', publisher.url='" + pubUrl + "', method='{{method}}',headers='{{headers}}',\n" +
                "@map(type='json'))\n" +
                "define stream BarStream (message string,method String,headers String);\n" +

                "from FooStream#log()\n" +
                "select symbol as message, 'POST' as method, class as headers\n" +
                "insert into BarStream;";
        //Deploy siddhi app

        String appApipath = "/siddhi-apps";

        log.info("Deploying Siddhi App...");
        HTTPResponse httpResponse = null;
        try {
            httpResponse = TestUtil.sendHRequest(appbody, appApibaseURI, appApipath, HEADER_CONTTP_TEXT,
                    HTTP_POST, DEFAULT_USER_NAME, DEFAULT_PASSWORD);
        } catch (IOException e) {
            TestUtil.handleException("Error creating Siddhi App ", e);
        }
        //Assert whether siddhi app deployed - RestAPI call
        Assert.assertEquals(httpResponse.getResponseCode(), 201);
        //wait untill siddhi app get deployed
        waitThread(5000);

        //Test case name(fully qualified method name)
        String testName = "com.sp.test.VerifyXML";
        String querybody = "{\n" +
                "  \"streamName\": \"FooStream\",\n" +
                "  \"siddhiAppName\": \"TestSiddhiAppxml\",\n" +
                "  \"timestamp\": null,\n" +
                "  \"data\": [\n" +
                "   \"TestData\",\n" +
                "   5.0,\n" +
                "   \"cclassName:" + testName + "\"\n" +
                "  ]\n" +
                "}";
        String simulatorpath = "/simulation/single";

        //Send single event
        log.info("Publishing a single event");
        HTTPResponse queryresponse = null;
        try {
            queryresponse = TestUtil.sendHRequest(querybody, appApibaseURI, simulatorpath,
                    HEADER_CONTTP_TEXT, HTTP_POST, DEFAULT_USER_NAME, DEFAULT_PASSWORD);
        } catch (IOException e) {
            TestUtil.handleException("Error sending event ", e);
        }
        //wait till events published
        waitThread(3000);
        //Event should published to msf4j service using above sink
        Assert.assertEquals(queryresponse.getResponseCode(), 200);
        Assert.assertEquals(queryresponse.getContentType(), HEADER_CONTTP_JSON);
        Assert.assertEquals(queryresponse.getMessage(), SINGLE_EVENT_SIMULATION_STARTED_SUCCESSFULLY);

        HTTPResponse msf4jgetrespns = new HTTPResponse(-1, "invalid", "default message");
        try {
            msf4jgetrespns = TestUtil.sendHRequest("",
                    msf4jbaseURI, "/testresults" + "/com.sp.test.VerifyXM/" + "?eventIndex=" + 0, HEADER_CONTTP_TEXT,
                    HTTP_GET, DEFAULT_USER_NAME, DEFAULT_PASSWORD);
        } catch (IOException e) {
            TestUtil.handleException("Error checking results ", e);
        }
        //expected result
        final String msgPattern = "{\"value\":0.0,\"message\":\"TestData\",\"method\":\"POST\"," +
                "\"headers\":\"cclassName:com.sp.test.VerifyXML\"}";

        HTTPResponse finalMsf4jgetrespns = msf4jgetrespns;
        TestResults testResults = new TestResults() {
            @Override
            public void waitForResults(int retryCount, long interval) {
                super.verifyResult(interval, retryCount, finalMsf4jgetrespns.getMessage());
            }

            @Override
            public boolean resultsFound(String eventMessage) {
                return msgPattern.equalsIgnoreCase(eventMessage);
            }
        };
        //verify event results
        testResults.waitForResults(10, 1000);
        Assert.assertEquals(msf4jgetrespns.getMessage(), msgPattern);
        //Delete siddhi app - optional
        HTTPResponse apiDeleteResponse = null;
        try {
            apiDeleteResponse = TestUtil.sendHRequest("", appApibaseURI, "/siddhi-apps/TestSiddhiAppxml",
                    HEADER_CONTTP_TEXT, "DELETE", DEFAULT_USER_NAME, DEFAULT_PASSWORD);
        } catch (IOException e) {
            TestUtil.handleException("Error deleting Siddhi App ", e);
        }
        //verify app deleted
        Assert.assertEquals(apiDeleteResponse.getResponseCode(), 200);
    }

    @AfterTest
    public void tearDown() {
        //clearing msf4j service
        HTTPResponse msf4jclearrespns = null;
        try {
            msf4jclearrespns = TestUtil.sendHRequest("", msf4jbaseURI, "/testresults/clear",
                    HEADER_CONTTP_TEXT, HTTP_POST, DEFAULT_USER_NAME, DEFAULT_PASSWORD);
        } catch (IOException e) {
            TestUtil.handleException("Error tearing down ", e);
        }
        Assert.assertEquals(msf4jclearrespns.getResponseCode(), 204);
    }
}


<|start_filename|>modules/samples/artifacts/MessageTracer/simulation-configs/MessageTracerSimulation.json<|end_filename|>
{"properties":{"simulationName":"MessageTracerSimulation","startTimestamp":"","endTimestamp":"","noOfEvents":"","description":"","timeInterval":"10"},"sources":[{"siddhiAppName":"message-tracer","streamName":"SpanStreamIn","timestampInterval":"1000","simulationType":"CSV_SIMULATION","fileName":"message_tracer_events.csv","delimiter":",","isOrdered":true,"indices":"0,1,2,3,4,5,6,7,8,9,10,11,12,13,14"}]}

<|start_filename|>modules/samples/sample-clients/wso2event-client/src/main/java/org/wso2/carbon/sample/wso2event/Client.java<|end_filename|>
/*
 * Copyright (c) 2017, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * WSO2 Inc. licenses this file to you under the Apache License,
 * Version 2.0 (the "License"); you may not use this file except
 * in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package org.wso2.carbon.sample.wso2event;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.wso2.carbon.databridge.agent.AgentHolder;
import org.wso2.carbon.databridge.agent.DataPublisher;
import org.wso2.carbon.databridge.commons.Event;
import org.wso2.carbon.databridge.commons.utils.DataBridgeCommonsUtils;

import java.util.Arrays;

/**
 * WSO2Event Client Publisher.
 */
public class Client {
    private static Log log = LogFactory.getLog(Client.class);
    private static final String STREAM_NAME = "SweetProductionStream";
    private static final String VERSION = "1.0.0";
    private static String agentConfigFileName = "sync.data.agent.config.yaml";

    public static void main(String[] args) {

        DataPublisherUtil.setKeyStoreParams();
        DataPublisherUtil.setTrustStoreParams();

        log.info("These are the provided configurations: " + Arrays.deepToString(args));

        String protocol = args[0];
        String host = args[1];
        String port = args[2];
        int sslPort = Integer.parseInt(port) + 100;
        String username = args[3];
        String password = args[4];
        String numberOfEventsStr = args[5];
        int numberOfEvents = Integer.parseInt(numberOfEventsStr);

        try {
            log.info("Starting WSO2 Event Client");

            AgentHolder.setConfigPath(DataPublisherUtil.getDataAgentConfigPath(agentConfigFileName));
            DataPublisher dataPublisher = new DataPublisher(protocol, "tcp://" + host + ":" + port,
                    "ssl://" + host + ":" + sslPort, username, password);
            Event event = new Event();
            event.setStreamId(DataBridgeCommonsUtils.generateStreamId(STREAM_NAME, VERSION));
            event.setMetaData(new Object[]{"127.0.0.1"});
            event.setCorrelationData(null);
            event.setPayloadData(new Object[]{"WSO2", 123.4});

            for (int i = 0; i < numberOfEvents; i++) {
                dataPublisher.publish(event);
            }

            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                log.error(e);
            }
            dataPublisher.shutdown();
            log.info("Events published successfully");

        } catch (Throwable e) {
            log.error(e);
        }
    }
}


<|start_filename|>modules/samples/artifacts/ApproximateComputingSample/Test-approximate-distinctCountEver.json<|end_filename|>
{"properties":{"simulationName":"Test-approximate-distinctCountEver","startTimestamp":"","endTimestamp":"","noOfEvents":"","description":"","timeInterval":"1000"},"sources":[{"siddhiAppName":"approximate-distinctCountEver-sample","streamName":"requestStream","timestampInterval":"1000","simulationType":"CSV_SIMULATION","fileName":"approximate-distinctCountEver.csv","delimiter":"\\n","isOrdered":true}]}

<|start_filename|>modules/integration/test-results-service/src/main/java/org/wso2/sp/tests/eventscollector/VerifyTest.java<|end_filename|>
/*
 * Copyright (c) 2017, WSO2 Inc. (http://wso2.com) All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.wso2.sp.tests.eventscollector;

import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiParam;
import io.swagger.annotations.ApiResponse;
import io.swagger.annotations.ApiResponses;
import io.swagger.annotations.Contact;
import io.swagger.annotations.Info;
import io.swagger.annotations.License;
import io.swagger.annotations.SwaggerDefinition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.wso2.msf4j.Request;
import org.wso2.sp.tests.eventscollector.exception.TestNotFoundException;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
import javax.ws.rs.Consumes;
import javax.ws.rs.GET;
import javax.ws.rs.HeaderParam;
import javax.ws.rs.POST;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
import javax.ws.rs.Produces;
import javax.ws.rs.QueryParam;
import javax.ws.rs.core.Context;
import javax.ws.rs.core.Response;


/**
 * VerifyTest class.
 */

@Api(value = "testresults")
@SwaggerDefinition(
        info = @Info(
                title = "SP Test Events Swagger Definition", version = "1.0",
                description = "Stream Processor Test Events in-memory persisting service",
                license = @License(name = "Apache 2.0", url = "http://www.apache.org/licenses/LICENSE-2.0"),
                contact = @Contact(
                        name = "<NAME>",
                        email = "<EMAIL>",
                        url = "http://wso2.com"
                ))
)

@Path("/testresults")
public class VerifyTest {

    private static final Logger log = LoggerFactory.getLogger(VerifyTest.class);
    private Map<String, ArrayList<Event>> testResults = new HashMap<>();

    @GET
    @Path("/{testCaseName}")
    @Produces({"application/json", "text/xml"})
    @ApiOperation(
            value = "Return event corresponding to the testcase name or testcase name + index",
            notes = "Returns HTTP 404 if the testcase is not found")
    @ApiResponses(value = {
            @ApiResponse(code = 200, message = "Valid event found"),
            @ApiResponse(code = 404, message = "Test case not found")})
    public Response getSingleEvent(@ApiParam(value = "testCaseName", required = true)
                                   @PathParam("testCaseName") String testCaseName,
                                   @ApiParam(value = "Event index", required = true)
                                   @QueryParam("eventIndex") String index) throws TestNotFoundException {
        if (testResults.containsKey(testCaseName)) {
            if (index == null) {
                log.info("Retrieving siddhi events by testcase: " + testCaseName);
                ArrayList<Event> events = testResults.get(testCaseName);
                return Response.ok().entity(events).build();
            } else if (index.isEmpty()) {
                log.warn("Index is invalid : " + testCaseName);
                return Response.status(400).build();
            } else {
                try {
                    int eventIndex = Integer.parseInt(index);
                    if (testResults.get(testCaseName).size() > eventIndex) {
                        log.info("Retrieving siddhi events by testcase: " + testCaseName + " and index: " + eventIndex);
                        Event result = testResults.get(testCaseName).get(eventIndex);
                        return Response.ok().entity(result).build();
                    } else {
                        log.warn("No events found for index : " + eventIndex);
                        return Response.status(404).
                                entity("{\"message\":\"No events found for the index: " + eventIndex + "\"}").build();
                    }
                } catch (NumberFormatException nfe) {
                    log.warn("Index is invalid : " + testCaseName);
                    return Response.status(400).build();
                }
            }
        }
        log.warn("Couldn't find test case : " + testCaseName);
        return Response.status(404).entity("{\"message\":\"Couldn't find test case: " + testCaseName + "\"}").build();
    }

    @GET
    @Path("/{testCaseName}/count")
    @Produces({"application/json", "text/xml"})
    @ApiOperation(
            value = "Return event count corresponding to the testcase name",
            notes = "Returns HTTP 404 if the testcase is not found")
    @ApiResponses(value = {
            @ApiResponse(code = 200, message = "Valid event/s found"),
            @ApiResponse(code = 404, message = "Events not found")})
    public Response getEventsCount(@ApiParam(value = "testCaseName", required = true)
                                   @PathParam("testCaseName") String testCaseName) throws TestNotFoundException {
        log.info("Retrieving siddhi event count by testcase: " + testCaseName);

        if (testResults.containsKey(testCaseName)) {
            Integer count = testResults.get(testCaseName).size();
            return Response.ok().entity("{\"testCase\":\"" + testCaseName + "\",\"eventCount\":" + count + "}").build();
        }
        log.warn("Not Found : " + testCaseName);
        return Response.status(404).build();
    }

    @POST
    @Path("/")
    @Consumes("application/json")
    @ApiOperation(
            value = "Add a test result",
            notes = "Add a valid method name and res")
    public void addResult(@ApiParam(value = "Events object", required = true) EventWrapper event,
                          @ApiParam(value = "className string", required = true)
                          @HeaderParam("className") String className, @Context Request request) {
        log.info("POST invoked");
        request.getHeaders().getAll().forEach(entry -> log.info(entry.getName() + "=" + entry.getValue()));

        if (testResults.containsKey(className)) {
            log.info("adding event under existing test case.");
            ArrayList<Event> eventArrayList = testResults.get(className);
            eventArrayList.add(event.event);
            testResults.put(className, eventArrayList);
        } else {
            log.info("adding event under new test case.");
            ArrayList<Event> eventList = new ArrayList<>();
            eventList.add(event.event);
            testResults.put(className, eventList);
        }
        log.info("TestCaseName: " + className);

    }

    @POST
    @Path("/clear")
    public void clearMap() {
        log.info("event map clearing...");
        testResults.clear();
    }
}


<|start_filename|>modules/samples/artifacts/ApproximateComputingSample/Test-approximate-distinctCount.json<|end_filename|>
{"properties":{"simulationName":"Test-approximate-distinctCount","startTimestamp":"","endTimestamp":"","noOfEvents":"","description":"","timeInterval":"1000"},"sources":[{"siddhiAppName":"approximate-distinctCount-sample","streamName":"sensorStream","timestampInterval":"1000","simulationType":"CSV_SIMULATION","fileName":"approximate-distinctCount.csv","delimiter":"\\n","isOrdered":true}]}

<|start_filename|>modules/samples/artifacts/ClusTreeSample/clusTreeTestFeed.json<|end_filename|>
{"properties":{"simulationName":"clusTreeTestFeed","startTimestamp":"","endTimestamp":"","noOfEvents":"","description":"","timeInterval":""},"sources":[{"siddhiAppName":"ClusTreeTestApp","streamName":"InputStream","timeInterval":"1000","simulationType":"CSV_SIMULATION","fileName":"clusTreeFileTest.csv","delimiter":",","isOrdered":true}]}

<|start_filename|>modules/integration/tests-kubernetes-integration/src/test/java/org/sp/tests/connectors/rdbms/transactions/RDBMSTransactions.java<|end_filename|>
/*
*  Copyright (c) 2017, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
*
*  WSO2 Inc. licenses this file to you under the Apache License,
*  Version 2.0 (the "License"); you may not use this file except
*  in compliance with the License.
*  You may obtain a copy of the License at
*
*    http://www.apache.org/licenses/LICENSE-2.0
*
*  Unless required by applicable law or agreed to in writing,
*  software distributed under the License is distributed on an
*  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
*  KIND, either express or implied.  See the License for the
*  specific language governing permissions and limitations
*  under the License.
*/

package org.sp.tests.connectors.rdbms.transactions;

import org.apache.commons.httpclient.HttpClient;
import org.apache.commons.httpclient.HttpStatus;
import org.apache.commons.httpclient.methods.PostMethod;
import org.apache.commons.httpclient.methods.StringRequestEntity;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.sp.tests.base.SPBaseTest;
import org.sp.tests.util.Constants;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;

import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;

import static org.testng.Assert.assertEquals;

/**
 * Tests to cover RDBMS transactions
 */
public class RDBMSTransactions extends SPBaseTest {

    private static final Log log = LogFactory.getLog(RDBMSTransactions.class);
    Connection conn = null;
    Statement stmt = null;
    Statement stmt2 = null;

    @BeforeClass(alwaysRun = true) public void initializeTables() {
        String createPersonLKTable = "CREATE TABLE personsinlanka (\n" + "    personid int NOT NULL,\n"
                + "    lastname varchar(255) NOT NULL,\n" + "    firstname varchar(255),\n"
                + "    address varchar(255),\n" + "    city varchar(255),\n" + "    PRIMARY KEY (personid)\n" + ");";
        String createPersonsUSTable = "CREATE TABLE personsinus (\n" + "    personid int NOT NULL,\n"
                + "    lastname varchar(255) NOT NULL,\n" + "    firstname varchar(255),\n"
                + "    address varchar(255),\n" + "    city varchar(255),\n" + "    PRIMARY KEY (personid)\n" + ");";

        try {
            conn = DriverManager.getConnection(
                    "jdbc:mysql://" + mysqlURL + "?" + "user=" + Constants.MYSQL_USERNAME + "&password="
                            + Constants.MYSQL_PASSWORD);

            stmt = conn.createStatement();
            stmt.executeUpdate(createPersonLKTable);
            stmt.executeUpdate(createPersonsUSTable);

        } catch (SQLException ex) {
            log.error("SQLException: " + ex.getMessage());
            log.error("SQLState: " + ex.getSQLState());
            log.error("VendorError: " + ex.getErrorCode());
        } finally {
            if (conn != null) {
                try {
                    conn.close();
                } catch (SQLException sqlEx) {
                }
            }
        }
    }

    @Test public static void insertIntoTable() {

        String serviceURL = spURL + "/persons/insert/success";
        String payload = "{  \n" + "   \"id\":1,\n" + "   \"firstname\":\"danuja\",\n" + "   \"lastname\":\"perera\",\n"
                + "   \"address\":\"260, Mahawatta Rd, Colombo 14\",\n" + "   \"city\":\"kentaky\"\n" + "}";
        try {
            StringRequestEntity requestEntity = new StringRequestEntity(payload, "application/json", "UTF-8");

            HttpClient client = new HttpClient();
            PostMethod post = new PostMethod(serviceURL);

            post.setRequestEntity(requestEntity);

            int statuscode = client.executeMethod(post);

            // Asserting the Status code. Expected 200 OK
            assertEquals(statuscode, HttpStatus.SC_OK);

            byte[] response = post.getResponseBody();

            // Asserting the Response Message.
            assertEquals(new String(response), "Data Insertion Successful");

            //TODO : Assert the actual database table values

        } catch (IOException e) {
            log.error("Error while calling the BE server : " + e.getMessage(), e);
        }
    }

    @Test public void abortONSQLError() {
        String serviceURL = spURL + "/persons/insert/sqlError";
        String payload = "{  \n" + "   \"id\":1,\n" + "   \"firstname\":\"danuja\",\n" + "   \"lastname\":\"perera\",\n"
                + "   \"address\":\"260, Mahawatta Rd, Colombo 14\",\n" + "   \"city\":\"kentaky\"\n" + "}";
        try {
            StringRequestEntity requestEntity = new StringRequestEntity(payload, "application/json", "UTF-8");

            HttpClient client = new HttpClient();
            PostMethod post = new PostMethod(serviceURL);

            post.setRequestEntity(requestEntity);

            int statuscode = client.executeMethod(post);

            // Asserting the Status code. Expected 200 OK
            assertEquals(statuscode, HttpStatus.SC_OK);

            byte[] response = post.getResponseBody();

            // Asserting the Response Message.
            assertEquals(new String(response), "Data Insertion Failed");

            //TODO : Assert the actual database table values

        } catch (IOException e) {
            log.error("Error while calling the BE server : " + e.getMessage(), e);
        }
    }

    @Test public void forceAbort() {
        String serviceURL = spURL + "/persons/insert/sqlError";
        String payload = "{  \n" + "   \"id\":1,\n" + "   \"firstname\":\"danuja\",\n" + "   \"lastname\":\"perera\",\n"
                + "   \"address\":\"260, Mahawatta Rd, Colombo 14\",\n" + "   \"city\":\"kentaky\"\n" + "}";
        try {
            StringRequestEntity requestEntity = new StringRequestEntity(payload, "application/json", "UTF-8");

            HttpClient client = new HttpClient();
            PostMethod post = new PostMethod(serviceURL);

            post.setRequestEntity(requestEntity);

            int statuscode = client.executeMethod(post);

            // Asserting the Status code. Expected 200 OK
            assertEquals(statuscode, HttpStatus.SC_OK);

            byte[] response = post.getResponseBody();

            // Asserting the Response Message.
            assertEquals(new String(response), "Data Insertion Failed");

        } catch (IOException e) {
            log.error("Error while calling the BE server : " + e.getMessage(), e);
        }
    }
}


<|start_filename|>modules/samples/sample-clients/jms-consumer/src/main/java/org/wso2/sp/sample/jms/consumer/JmsReceiver.java<|end_filename|>
/*
 * Copyright (c) 2017, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * WSO2 Inc. licenses this file to you under the Apache License,
 * Version 2.0 (the "License"); you may not use this file except
 * in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied. See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package org.wso2.sp.sample.jms.consumer;

import org.apache.log4j.Logger;
import org.wso2.siddhi.core.SiddhiAppRuntime;
import org.wso2.siddhi.core.SiddhiManager;

/**
 * Test client for Jms source.
 */
public class JmsReceiver {
    private static final Logger log = Logger.getLogger(JmsReceiver.class);

    /**
     * Main method to start the test client.
     *
     * @param args no args need to be provided
     */
    public static void main(String[] args) {
        log.info("Initialize Jms receiver.");
        SiddhiManager siddhiManager = new SiddhiManager();
        String publisherUrl = args[0];
        String destination = args[1];
        String type = args[2];
        String factoryType = args[3];
        String jndiName = args[4];
        SiddhiAppRuntime siddhiAppRuntime = siddhiManager.createSiddhiAppRuntime(
                        "@App:name('JmsReceiver')\n" +
                        "@sink(type='log')" +
                        "define stream logStream(name string, amount double);\n" +
                        "@source(type='jms',@map(type='" + type + "'),\n" +
                        "factory.initial='org.apache.activemq.jndi.ActiveMQInitialContextFactory',\n" +
                        "provider.url='" + publisherUrl + "', destination='" + destination + "', " +
                        "connection.factory.type='" + factoryType + "', \n" +
                        "connection.factory.jndi.name='" + jndiName + "')\n" +
                        "define stream jmsConsumerStream(name string, amount double);\n" +
                        "from jmsConsumerStream\n" +
                        "select * \n" +
                        "insert into logStream;");
        siddhiAppRuntime.start();
        while (true) {
        }
    }
}



<|start_filename|>modules/integration/tests-kubernetes-integration/src/test/java/org/sp/tests/util/HTTPResponse.java<|end_filename|>
/*
*  Copyright (c) 2017, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
*
*  WSO2 Inc. licenses this file to you under the Apache License,
*  Version 2.0 (the "License"); you may not use this file except
*  in compliance with the License.
*  You may obtain a copy of the License at
*
*    http://www.apache.org/licenses/LICENSE-2.0
*
*  Unless required by applicable law or agreed to in writing,
*  software distributed under the License is distributed on an
*  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
*  KIND, either express or implied.  See the License for the
*  specific language governing permissions and limitations
*  under the License.
*/package org.sp.tests.util;
/**
 *
 * TestUtil class
 */
public class HTTPResponse {

    private int responseCode;
    private String contentType;
    private String message;

    public HTTPResponse(int responseCode, String contentType, String message) {
        this.responseCode = responseCode;
        this.contentType = contentType;
        this.message = message;
    }

    public HTTPResponse() {
    }


    public int getResponseCode() {
        return responseCode;
    }

    public void setResponseCode(int responseCode) {
        this.responseCode = responseCode;
    }

    public String getContentType() {
        return contentType;
    }

    public void setContentType(String contentType) {
        this.contentType = contentType;
    }

    public String getMessage() {
        return message;
    }

    public void setMessage(String message) {
        this.message = message;
    }
}


<|start_filename|>modules/integration/tests-kubernetes-integration/src/test/java/org/sp/tests/ha/util/TestCaseUtil.java<|end_filename|>
/*
 * Copyright (c) 2017, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.sp.tests.ha.util;

import org.sp.tests.util.HTTPResponse;
import org.sp.tests.util.TestResults;
import org.sp.tests.util.TestUtil;

import java.io.IOException;
import java.net.URI;

import static org.sp.tests.util.Constants.DEFAULT_PASSWORD;
import static org.sp.tests.util.Constants.DEFAULT_USER_NAME;
import static org.sp.tests.util.Constants.HEADER_CONTTP_TEXT;
import static org.sp.tests.util.Constants.HTTP_GET;

public class TestCaseUtil {

    public static TestResults getTestListener(String expectedResponse, URI msf4jBaseURI, String path) {

        return new TestResults() {

            @Override
            public void waitForResults(int retryCount, long interval) {
                HTTPResponse msf4jResponse = null;
                try {
                    msf4jResponse = TestUtil.sendHRequest("", msf4jBaseURI, path, HEADER_CONTTP_TEXT, HTTP_GET,
                            DEFAULT_USER_NAME, DEFAULT_PASSWORD);
                    super.verifyResult(interval, retryCount, msf4jResponse.getMessage());
                } catch (IOException e) {
                    TestUtil.handleException("Error in getting event count ", e);
                }
            }

            @Override
            public boolean resultsFound(String eventMessage) {
                return expectedResponse.equalsIgnoreCase(eventMessage);
            }
        };
    }

    public static String getExpectedEventsCountMessage(String testCase, int eventCount) {
        return "{\"testCase\":\"" + testCase + "\",\"eventCount\":" + Integer.toString(eventCount) + "}";
    }

    public static String getExpectedEventsMessage(String testCase, String data, float value) {
        return "{\"value\":" + String.valueOf(value) + ",\"message\":\""
                + data + "\",\"method\":\"POST\",\"headers\":\"" + testCase + "\"}";
    }
}


<|start_filename|>modules/samples/artifacts/ApproximateComputingSample/Test-approximate-count.json<|end_filename|>
{"properties":{"simulationName":"Test-approximate-count","startTimestamp":"","endTimestamp":"","noOfEvents":"","description":"","timeInterval":"1000"},"sources":[{"siddhiAppName":"approximate-count-sample","streamName":"transactionStream","timestampInterval":"","simulationType":"CSV_SIMULATION","fileName":"approximate-count.csv","delimiter":",","timestampAttribute":"","isOrdered":true}]}

<|start_filename|>modules/integration/test-results-service/src/main/java/org/wso2/sp/tests/eventscollector/Event.java<|end_filename|>
/*
 * Copyright (c) 2017, WSO2 Inc. (http://wso2.com) All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.wso2.sp.tests.eventscollector;

import org.codehaus.jackson.map.annotate.JsonRootName;
import java.io.Serializable;
import javax.xml.bind.annotation.XmlAccessType;
import javax.xml.bind.annotation.XmlAccessorType;
import javax.xml.bind.annotation.XmlElement;
import javax.xml.bind.annotation.XmlRootElement;

/**
 * Event class.
 */
@XmlRootElement
@XmlAccessorType(XmlAccessType.FIELD)
@JsonRootName(value = "event")
public class Event implements Serializable {

    public Event() {

    }

    public Event(String message, float value, String method, String headers) {
        this.message = message;
        this.value = value;
        this.method = method;
        this.headers = headers;
    }

    @XmlElement(required = true)
    private float value;
    @XmlElement(required = true)
    private String message;
    @XmlElement(required = true)
    private String method;
    @XmlElement(required = true)
    private String headers;

    public float getValue() {
        return value;
    }

    public String getMessage() {
        return message;
    }


    public String getMethod() {
        return method;
    }


    public String getHeaders() {
        return headers;
    }

    public void setValue(float value) {
        this.value = value;
    }

    public void setMessage(String message) {
        this.message = message;
    }


    public void setMethod(String method) {
        this.method = method;
    }


    public void setHeaders(String headers) {
        this.headers = headers;
    }

}


<|start_filename|>modules/integration/tests-kubernetes-integration/src/test/java/org/sp/tests/ha/PassiveNodeIT.java<|end_filename|>
/*
 * Copyright (c) 2017, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.sp.tests.ha;

import org.apache.log4j.Logger;
import org.sp.tests.base.SPBaseTest;
import org.sp.tests.util.HTTPResponse;
import org.sp.tests.util.TestUtil;
import org.testng.Assert;
import org.testng.ITestResult;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.AfterTest;
import org.testng.annotations.BeforeTest;
import org.testng.annotations.Test;

import java.io.IOException;
import java.net.URI;

import static org.sp.tests.ha.util.SiddhiAppUtil.deployPassThroughSiddhiApp;
import static org.sp.tests.ha.util.SiddhiAppUtil.sendEvent;
import static org.sp.tests.ha.util.SiddhiAppUtil.sendEvents;
import static org.sp.tests.ha.util.TestCaseUtil.getExpectedEventsCountMessage;
import static org.sp.tests.ha.util.TestCaseUtil.getTestListener;
import static org.sp.tests.util.Constants.DEFAULT_PASSWORD;
import static org.sp.tests.util.Constants.DEFAULT_USER_NAME;
import static org.sp.tests.util.Constants.HEADER_CONTTP_JSON;
import static org.sp.tests.util.Constants.HEADER_CONTTP_TEXT;
import static org.sp.tests.util.Constants.HTTP_GET;
import static org.sp.tests.util.Constants.HTTP_POST;
import static org.sp.tests.util.Constants.HTTP_RESP_200;
import static org.sp.tests.util.Constants.HTTP_RESP_201;
import static org.sp.tests.util.Constants.HTTP_RESP_204;
import static org.sp.tests.util.Constants.SINGLE_EVENT_SIMULATION_STARTED_SUCCESSFULLY;
import static org.sp.tests.util.TestUtil.waitThread;

public class PassiveNodeIT extends SPBaseTest {
    private static final Logger log = Logger.getLogger(PassiveNodeIT.class);
    private static final String SIDDHI_APP_NAME = "TestMinimumHA";
    private static final String TEST_NAME = "cclassName:com.sp.test.TwoNodeHa";
    public static final String TEST_RESULTS_COUNT_PATH = "/testresults/com.sp.test.TwoNodeH/count";

    private URI nodeOneURI;
    private URI nodeTwoURI;
    private URI msf4jBaseURI;

    @BeforeTest
    public void testInit() {
        log.info("Starting test " + this.getClass().getCanonicalName());
        nodeOneURI = URI.create(haNodeOneURL);
        nodeTwoURI = URI.create(haNodeTwoURL);
        msf4jBaseURI = URI.create(haNodeTwoMsf4jURL);
    }

    @Test
    public void testHaInit() throws IOException {

        HTTPResponse httpResponseNodeOne = deployPassThroughSiddhiApp(nodeOneURI, SIDDHI_APP_NAME);
        Assert.assertEquals(httpResponseNodeOne.getResponseCode(), HTTP_RESP_201, httpResponseNodeOne.getMessage());

        HTTPResponse httpResponseNodeTwo = deployPassThroughSiddhiApp(nodeTwoURI, SIDDHI_APP_NAME);
        Assert.assertEquals(httpResponseNodeTwo.getResponseCode(), HTTP_RESP_201, httpResponseNodeTwo.getMessage());

        waitThread(1000); //Wait for Siddhi applications to deploy

        // Checking if both Siddhi Applications are receiving events
        HTTPResponse nodeOneResponse = sendEvent(nodeOneURI, TEST_NAME, SIDDHI_APP_NAME, "FooStream", "First", "1");
        Assert.assertEquals(nodeOneResponse.getResponseCode(), HTTP_RESP_200, nodeOneResponse.getMessage());
        Assert.assertEquals(nodeOneResponse.getContentType(), HEADER_CONTTP_JSON);
        Assert.assertEquals(nodeOneResponse.getMessage(), SINGLE_EVENT_SIMULATION_STARTED_SUCCESSFULLY);
        waitThread(2000); //Wait for event to publish

        HTTPResponse nodeTwoResponse = sendEvent(nodeTwoURI, TEST_NAME, SIDDHI_APP_NAME, "FooStream", "First", "1");
        Assert.assertEquals(nodeTwoResponse.getResponseCode(), HTTP_RESP_200, nodeTwoResponse.getMessage());
        Assert.assertEquals(nodeTwoResponse.getContentType(), HEADER_CONTTP_JSON);
        Assert.assertEquals(nodeTwoResponse.getMessage(), SINGLE_EVENT_SIMULATION_STARTED_SUCCESSFULLY);
        waitThread(2000); //Wait for event to publish


        String expectedResult = "{\"value\":0.0,\"message\":\"First\",\"method\":\"POST\"," +
                "\"headers\":\"cclassName:com.sp.test.TwoNodeHa\"}";

        // Checking if the Msf4j Service is working properly by checking if test case is registered
        getTestListener(expectedResult, msf4jBaseURI, "/testresults/com.sp.test.TwoNodeH/?eventIndex=" + 0).
                waitForResults(10, 1000);
        HTTPResponse msf4jResponse = TestUtil.sendHRequest("", msf4jBaseURI, "/testresults/" +
                        "com.sp.test.TwoNodeH/" + "?eventIndex=" + 0, HEADER_CONTTP_TEXT, HTTP_GET,
                DEFAULT_USER_NAME, DEFAULT_PASSWORD);
        Assert.assertEquals(msf4jResponse.getMessage(), expectedResult);

        // Resetting event count
        TestUtil.sendHRequest("", msf4jBaseURI, "/testresults/clear",
                HEADER_CONTTP_TEXT, HTTP_POST, DEFAULT_USER_NAME, DEFAULT_PASSWORD);

        // Begin Tests
        // Sending 10 events each to both nodes
        sendEvents(10, nodeOneURI, TEST_NAME, SIDDHI_APP_NAME, "FooStream", false);
        sendEvents(10, nodeTwoURI, TEST_NAME, SIDDHI_APP_NAME, "FooStream", false);

        getTestListener(getExpectedEventsCountMessage("com.sp.test.TwoNodeH", 10), msf4jBaseURI,
                TEST_RESULTS_COUNT_PATH).waitForResults(10, 5000);
        msf4jResponse = TestUtil.sendHRequest("", msf4jBaseURI, TEST_RESULTS_COUNT_PATH,
                HEADER_CONTTP_TEXT, HTTP_GET, DEFAULT_USER_NAME, DEFAULT_PASSWORD);
        Assert.assertEquals(msf4jResponse.getMessage(), getExpectedEventsCountMessage("com.sp.test.TwoNodeH", 10));

        // Sending 5 events to Active Node
        sendEvents(5, nodeOneURI, TEST_NAME, SIDDHI_APP_NAME, "FooStream", false);

        getTestListener(getExpectedEventsCountMessage("com.sp.test.TwoNodeH", 15), msf4jBaseURI,
                TEST_RESULTS_COUNT_PATH).waitForResults(10, 5000);
        msf4jResponse = TestUtil.sendHRequest("", msf4jBaseURI, TEST_RESULTS_COUNT_PATH,
                HEADER_CONTTP_TEXT, HTTP_GET, DEFAULT_USER_NAME, DEFAULT_PASSWORD);
        Assert.assertEquals(msf4jResponse.getMessage(), getExpectedEventsCountMessage("com.sp.test.TwoNodeH", 15));

        // Sending 5 events to Passive Node
        sendEvents(5, nodeTwoURI, TEST_NAME, SIDDHI_APP_NAME, "FooStream", false);

        waitThread(20000); // Waiting for Passive Node to Sync with Active Node
        getTestListener(getExpectedEventsCountMessage("com.sp.test.TwoNodeH", 15), msf4jBaseURI,
                TEST_RESULTS_COUNT_PATH).waitForResults(10, 5000);
        msf4jResponse = TestUtil.sendHRequest("", msf4jBaseURI, TEST_RESULTS_COUNT_PATH,
                HEADER_CONTTP_TEXT, HTTP_GET, DEFAULT_USER_NAME, DEFAULT_PASSWORD);
        Assert.assertEquals(msf4jResponse.getMessage(), getExpectedEventsCountMessage("com.sp.test.TwoNodeH", 15));

        // Shutting down Active Node. Passive Node will now change states
        super.runBashScript("ha-scripts", "undeploy-node-1.sh");
        waitThread(60000); // Waiting for node to go down

        getTestListener(getExpectedEventsCountMessage("com.sp.test.TwoNodeH", 20), msf4jBaseURI,
                TEST_RESULTS_COUNT_PATH).waitForResults(10, 5000);
        msf4jResponse = TestUtil.sendHRequest("", msf4jBaseURI, TEST_RESULTS_COUNT_PATH,
                HEADER_CONTTP_TEXT, HTTP_GET, DEFAULT_USER_NAME, DEFAULT_PASSWORD);
        Assert.assertEquals(msf4jResponse.getMessage(), getExpectedEventsCountMessage("com.sp.test.TwoNodeH", 20));
        if (msf4jResponse.getMessage().equals(getExpectedEventsCountMessage("com.sp.test.TwoNodeH", 15))) {
            log.error("Error");
        }
        // Starting a new Passive Node
        super.runBashScript("ha-scripts", "deploy-node-1.sh");
        waitThread(20000); //Waiting for node to be deployed

        httpResponseNodeOne = deployPassThroughSiddhiApp(nodeOneURI, SIDDHI_APP_NAME);
        Assert.assertEquals(httpResponseNodeOne.getResponseCode(), HTTP_RESP_201, httpResponseNodeOne.getMessage());
        waitThread(1000); //Wait for Siddhi applications to deploy

        sendEvents(10, nodeOneURI, TEST_NAME, SIDDHI_APP_NAME, "FooStream", false);
        sendEvents(5, nodeTwoURI, TEST_NAME, SIDDHI_APP_NAME, "FooStream", false);
        waitThread(500); //Make sure node one receives events after node two
        sendEvents(5, nodeOneURI, TEST_NAME, SIDDHI_APP_NAME, "FooStream", false);
        waitThread(2000); //Wait for events to publish

        getTestListener(getExpectedEventsCountMessage("com.sp.test.TwoNodeH", 25), msf4jBaseURI,
                TEST_RESULTS_COUNT_PATH).waitForResults(10, 5000);
        msf4jResponse = TestUtil.sendHRequest("", msf4jBaseURI, TEST_RESULTS_COUNT_PATH,
                HEADER_CONTTP_TEXT, HTTP_GET, DEFAULT_USER_NAME, DEFAULT_PASSWORD);
        Assert.assertEquals(msf4jResponse.getMessage(), getExpectedEventsCountMessage("com.sp.test.TwoNodeH", 25));

        log.info("Waiting for Sync");
        waitThread(20000); // Wait for Sync

        super.runBashScript("ha-scripts", "shutdown-node-2-server.sh");
        log.info("Waiting for node 2 to shutdown");
        waitThread(20000); // Waiting for node to go down

        sendEvents(2, nodeOneURI, TEST_NAME, SIDDHI_APP_NAME, "FooStream", false);

        getTestListener(getExpectedEventsCountMessage("com.sp.test.TwoNodeH", 32), msf4jBaseURI,
                TEST_RESULTS_COUNT_PATH).waitForResults(10, 5000);
        msf4jResponse = TestUtil.sendHRequest("", msf4jBaseURI, TEST_RESULTS_COUNT_PATH,
                HEADER_CONTTP_TEXT, HTTP_GET, DEFAULT_USER_NAME, DEFAULT_PASSWORD);
        Assert.assertEquals(msf4jResponse.getMessage(), getExpectedEventsCountMessage("com.sp.test.TwoNodeH", 32));

    }

    @AfterMethod
    public void afterMethod(ITestResult result) throws IOException {
        log.info("After Method Running!");
        if (result.getStatus() == ITestResult.FAILURE) {
            log.info("Found Errors. Printing Log");
            runBashScript("ha-scripts", "print-log.sh");
        }
    }

    @AfterTest
    public void tearDown() {
        log.info("Finishing test " + this.getClass().getCanonicalName());
        HTTPResponse msf4jClearResponse = null;
        try {
            msf4jClearResponse = TestUtil.sendHRequest("", msf4jBaseURI, "/testresults/clear",
                    HEADER_CONTTP_TEXT, HTTP_POST, DEFAULT_USER_NAME, DEFAULT_PASSWORD);
            Assert.assertEquals(msf4jClearResponse.getResponseCode(), HTTP_RESP_204);
        } catch (IOException e) {
            TestUtil.handleException("IOException occurred when tearing down", e);
        }
    }
}


<|start_filename|>modules/integration/tests-kubernetes-integration/src/test/java/org/sp/tests/base/SPConstants.java<|end_filename|>
/*
*  Copyright (c) 2017, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
*
*  WSO2 Inc. licenses this file to you under the Apache License,
*  Version 2.0 (the "License"); you may not use this file except
*  in compliance with the License.
*  You may obtain a copy of the License at
*
*    http://www.apache.org/licenses/LICENSE-2.0
*
*  Unless required by applicable law or agreed to in writing,
*  software distributed under the License is distributed on an
*  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
*  KIND, either express or implied.  See the License for the
*  specific language governing permissions and limitations
*  under the License.
*/
package org.sp.tests.base;


public class SPConstants {

    public static final String ENABLE_URL_REMAP = "ENABLE_REMAP";
    public static final String POD_TAG_NAME = "sp";
    public static final String SP_PORT_NAME = "servlet-http";
    public static final String MSF4J_PORT_NAME = "msf4j-http";
    public static final String HA_NODE_1_POD_NAME = "sp-1";
    public static final String HA_NODE_2_POD_NAME = "sp-2";
    public static final String MYSQL_POD_NAME = "sp-mysql-db";
    public static final String MYSQL_PORT_NAME = "mysql-port";
/**public static final String POD_TAG_NAME_MYSQL = "mysql";
 # public static final String MYSQL_PORT_NAME = "mysql-port";
 */
}


<|start_filename|>modules/integration/tests-kubernetes-integration/src/test/java/org/sp/tests/util/TestUtil.java<|end_filename|>
/*
*  Copyright (c) 2017, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
*
*  WSO2 Inc. licenses this file to you under the Apache License,
*  Version 2.0 (the "License"); you may not use this file except
*  in compliance with the License.
*  You may obtain a copy of the License at
*
*    http://www.apache.org/licenses/LICENSE-2.0
*
*  Unless required by applicable law or agreed to in writing,
*  software distributed under the License is distributed on an
*  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
*  KIND, either express or implied.  See the License for the
*  specific language governing permissions and limitations
*  under the License.
*/
package org.sp.tests.util;


import io.netty.handler.codec.http.HttpMethod;
import org.apache.log4j.Logger;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.net.HttpURLConnection;
import java.net.URI;
import java.net.URL;


/**
 * TestUtil class
 */
public class TestUtil {
    private static final Logger logger = Logger.getLogger(TestUtil.class);

    public static HTTPResponse sendHRequest(String body, URI baseURI, String path, String contentType,
                                            String methodType, String userName, String password)
            throws IOException {

        HttpURLConnection urlConn = null;
        try {
            urlConn = TestUtil.generateRequest(baseURI, path, methodType, false);

            TestUtil.setHeader(urlConn, "Authorization", "Basic " + java.util.Base64.getEncoder().
                    encodeToString((userName + ":" + password).getBytes()));
            if (contentType != null) {
                TestUtil.setHeader(urlConn, "Content-Type", contentType);
            }
            TestUtil.setHeader(urlConn, "HTTP_METHOD", methodType);
            if (methodType.equals(HttpMethod.POST.name()) || methodType.equals(HttpMethod.PUT.name())) {
                TestUtil.writeContent(urlConn, body);
            }
            assert urlConn != null;

            HTTPResponse httpResponseMessage = new HTTPResponse(urlConn.getResponseCode(),
                    urlConn.getContentType(), TestUtil.getResponseMessage(urlConn));
            urlConn.disconnect();
            return httpResponseMessage;
        } catch (IOException e) {
            throw new IOException("Error generating request to " + baseURI + path, e);
        }
    }

    @SuppressWarnings("ConstantConditions")
    private static String getResponseMessage(HttpURLConnection urlConn) {
        StringBuilder sb;
        String value = null;
        try {

            BufferedReader br = null;
            if (200 <= urlConn.getResponseCode() && urlConn.getResponseCode() <= 299) {

                if (urlConn.getContentLength() > 0) {
                    br = new BufferedReader(new InputStreamReader((urlConn.getInputStream())));
                }

            } else {
                br = new BufferedReader(new InputStreamReader((urlConn.getErrorStream())));
            }

            sb = new StringBuilder();
            String output;
            while ((output = br.readLine()) != null) {
                sb.append(output);
            }
            /*while (true) {
                final String line = br.readLine();
                if (line == null) break;
                    sb.append(line);
                            }*/
            value = sb.toString();
        } catch (IOException e) {
            TestUtil.handleException("IOException occurred while getting the response message: ", e);
        } catch (NullPointerException ne) {
            sb = new StringBuilder();
            sb.append("Resource Not Found");
        }
        return value;

    }

    private static void writeContent(HttpURLConnection urlConn, String content) throws IOException {
        OutputStreamWriter out = new OutputStreamWriter(urlConn.getOutputStream());
        out.write(content);
        out.close();
    }

    private static HttpURLConnection generateRequest(URI baseURI, String path, String method, boolean keepAlive)
            throws IOException {
        URL url = baseURI.resolve(path).toURL();
        HttpURLConnection urlConn = (HttpURLConnection) url.openConnection();
        urlConn.setRequestMethod(method);
        if (method.equals(HttpMethod.POST.name()) || method.equals(HttpMethod.PUT.name())) {
            urlConn.setDoOutput(true);
        }
        if (keepAlive) {
            urlConn.setRequestProperty("Connection", "Keep-Alive");
        }
        return urlConn;
    }

    private static void setHeader(HttpURLConnection urlConnection, String key, String value) {
        urlConnection.setRequestProperty(key, value);
    }

    public static void handleException(String msg, Exception ex) {
        logger.error(msg, ex);
    }

    public static void waitThread(long timeInMilliseconds) {
        try {
            Thread.sleep(timeInMilliseconds);
        } catch (InterruptedException e) {
            TestUtil.handleException("IO Exception when thread sleep : ", e);
        }
    }

}
