commit 572aa24c5fab9bfde1efabeb45ba77c9ae8ed4b3 Author: Harshan Liyanage Date: Thu Dec 8 16:40:18 2016 +0530 Resolved conflicts diff --git a/pom.xml b/pom.xml new file mode 100644 index 0000000000..b1671cc446 --- /dev/null +++ b/pom.xml @@ -0,0 +1,91 @@ + + + + + + + org.wso2.carbon.devicemgt-plugins + siddhi-extensions + 3.0.3-SNAPSHOT + ../pom.xml + + + 4.0.0 + org.wso2.extension.siddhi.execution.json + bundle + WSO2 Siddhi Execution Extension - Json + http://wso2.org + + + + org.wso2.siddhi + siddhi-core + + + org.wso2.siddhi + siddhi-query-api + + + log4j + log4j + + + org.json.wso2 + json + ${analytics.json.version} + + + junit + junit + test + + + + + + + org.apache.maven.plugins + maven-compiler-plugin + + ${wso2.maven.compiler.source} + ${wso2.maven.compiler.target} + + + + org.apache.felix + maven-bundle-plugin + true + + + ${project.artifactId} + ${project.artifactId} + + org.wso2.extension.siddhi.execution.json, + org.wso2.extension.siddhi.execution.json.* + + + org.json, + org.wso2.siddhi.core.*, + org.wso2.siddhi.query.api.*, + + + + + + + \ No newline at end of file diff --git a/src/main/java/org/wso2/extension/siddhi/execution/json/getPropertyFunctionExtension.java b/src/main/java/org/wso2/extension/siddhi/execution/json/getPropertyFunctionExtension.java new file mode 100644 index 0000000000..8f68577a60 --- /dev/null +++ b/src/main/java/org/wso2/extension/siddhi/execution/json/getPropertyFunctionExtension.java @@ -0,0 +1,112 @@ +/* + * Copyright (c) 2016, 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.json; + +import org.json.JSONException; +import org.json.JSONObject; +import org.wso2.siddhi.core.config.ExecutionPlanContext; +import org.wso2.siddhi.core.exception.ExecutionPlanRuntimeException; +import org.wso2.siddhi.core.executor.ExpressionExecutor; +import org.wso2.siddhi.core.executor.function.FunctionExecutor; +import org.wso2.siddhi.query.api.definition.Attribute; +import org.wso2.siddhi.query.api.exception.ExecutionPlanValidationException; + +/** + * getProperty(json , propertyName) + * Returns the vale of the property from the given json json + * Accept Type(s): (STRING, STRING) + * Return Type(s): (STRING|INT|DOUBLE|FLOAT|OBJECT) + */ +public class getPropertyFunctionExtension extends FunctionExecutor { + + Attribute.Type returnType = Attribute.Type.STRING; + + @Override + protected void init(ExpressionExecutor[] attributeExpressionExecutors, + ExecutionPlanContext executionPlanContext) { + if (attributeExpressionExecutors.length != 2) { + throw new ExecutionPlanValidationException( + "Invalid no of arguments passed to json:getProperty() function," + " required 2, but found " + + attributeExpressionExecutors.length); + } + if (attributeExpressionExecutors[0].getReturnType() != Attribute.Type.STRING) { + throw new ExecutionPlanValidationException( + "Invalid parameter type found for the first argument of json:getProperty() function, " + "required " + + Attribute.Type.STRING + ", but found " + attributeExpressionExecutors[0].getReturnType() + .toString()); + } + if (attributeExpressionExecutors[1].getReturnType() != Attribute.Type.STRING) { + throw new ExecutionPlanValidationException( + "Invalid parameter type found for the second argument of json:getProperty() function, " + "required " + + Attribute.Type.STRING + ", but found " + attributeExpressionExecutors[1].getReturnType() + .toString()); + } + } + + @Override + protected Object execute(Object[] data) { + if (data[0] == null) { + throw new ExecutionPlanRuntimeException("Invalid input given to json:getProperty() function. First argument cannot be null"); + } + if (data[1] == null) { + throw new ExecutionPlanRuntimeException("Invalid input given to json:getProperty() function. Second argument cannot be null"); + } + String jsonString = (String) data[0]; + String property = (String) data[1]; + JSONObject jsonObject = null; + try { + jsonObject = new JSONObject(jsonString); + return jsonObject.get(property).toString(); + } catch (JSONException e) { + throw new ExecutionPlanRuntimeException("Cannot parse JSON String in json:getPeroperty() function. " + e); + } + } + + @Override + protected Object execute(Object data) { + return null; //Since the getProperty function takes in 2 parameters, this method does not get called. Hence,not implemented. + } + + @Override + public void start() { + //Nothing to start + } + + @Override + public void stop() { + //Nothing to stop + } + + @Override + public Attribute.Type getReturnType() { + return returnType; + } + + @Override + public Object[] currentState() { + return null; //No need to maintain a state. + } + + @Override + public void restoreState(Object[] state) { + //Since there's no need to maintain a state, nothing needs to be done here. + } +} + + diff --git a/src/main/resources/json.siddhiext b/src/main/resources/json.siddhiext new file mode 100644 index 0000000000..f1886dd63e --- /dev/null +++ b/src/main/resources/json.siddhiext @@ -0,0 +1,19 @@ +# +# Copyright (c) 2016, 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. +# + +getProperty=org.wso2.extension.siddhi.execution.json.getPropertyFunctionExtension diff --git a/src/test/java/org/wso2/extension/siddhi/execution/json/getPropertyFunctionTestCase.java b/src/test/java/org/wso2/extension/siddhi/execution/json/getPropertyFunctionTestCase.java new file mode 100644 index 0000000000..7bc66478c2 --- /dev/null +++ b/src/test/java/org/wso2/extension/siddhi/execution/json/getPropertyFunctionTestCase.java @@ -0,0 +1,88 @@ +/* + * Copyright (c) 2016, 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.json; + +import junit.framework.Assert; +import org.apache.log4j.Logger; +import org.junit.Before; +import org.junit.Test; +import org.wso2.siddhi.core.ExecutionPlanRuntime; +import org.wso2.siddhi.core.SiddhiManager; +import org.wso2.siddhi.core.event.Event; +import org.wso2.siddhi.core.query.output.callback.QueryCallback; +import org.wso2.siddhi.core.stream.input.InputHandler; +import org.wso2.siddhi.core.util.EventPrinter; +import org.wso2.extension.siddhi.execution.json.test.util.SiddhiTestHelper; + +import java.util.concurrent.atomic.AtomicInteger; + +public class getPropertyFunctionTestCase { + static final Logger log = Logger.getLogger(getPropertyFunctionTestCase.class); + private AtomicInteger count = new AtomicInteger(0); + private volatile boolean eventArrived; + + @Before + public void init() { + count.set(0); + eventArrived = false; + } + + @Test + public void testGetPropertyFunctionExtension() throws InterruptedException { + log.info("getPropertyFunctionExtension TestCase"); + SiddhiManager siddhiManager = new SiddhiManager(); + + String inStreamDefinition = "define stream inputStream (payload string, id string, volume long);"; + String query = ("@info(name = 'query1') from inputStream select id, json:getProperty(payload, 'latitude') " + + "as latitude insert into outputStream;"); + ExecutionPlanRuntime executionPlanRuntime = siddhiManager.createExecutionPlanRuntime(inStreamDefinition + query); + + executionPlanRuntime.addCallback("query1", new QueryCallback() { + @Override + public void receive(long timeStamp, Event[] inEvents, Event[] removeEvents) { + EventPrinter.print(timeStamp, inEvents, removeEvents); + for (Event event : inEvents) { + count.incrementAndGet(); + if (count.get() == 1) { + Assert.assertEquals("1.5", event.getData(1)); + eventArrived = true; + } + if (count.get() == 2) { + Assert.assertEquals("67.5", event.getData(1)); + eventArrived = true; + } + if (count.get() == 3) { + Assert.assertEquals("7.5", event.getData(1)); + eventArrived = true; + } + } + } + }); + + InputHandler inputHandler = executionPlanRuntime.getInputHandler("inputStream"); + executionPlanRuntime.start(); + inputHandler.send(new Object[]{"{'latitude' : 1.5, 'longitude' : 78.5}","IBM",100l}); + inputHandler.send(new Object[]{"{'latitude' : 67.5, 'longitude' : 34.9}","WSO2", 200l}); + inputHandler.send(new Object[]{"{'latitude' : 7.5, 'longitude' : 44.9}", "XYZ", 200l}); + SiddhiTestHelper.waitForEvents(100, 3, count, 60000); + Assert.assertEquals(3, count.get()); + Assert.assertTrue(eventArrived); + executionPlanRuntime.shutdown(); + } +} diff --git a/src/test/java/org/wso2/extension/siddhi/execution/json/test/util/SiddhiTestHelper.java b/src/test/java/org/wso2/extension/siddhi/execution/json/test/util/SiddhiTestHelper.java new file mode 100644 index 0000000000..9cfd6c307d --- /dev/null +++ b/src/test/java/org/wso2/extension/siddhi/execution/json/test/util/SiddhiTestHelper.java @@ -0,0 +1,32 @@ +/* + * Copyright (c) 2016, 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.json.test.util; + +import java.util.concurrent.atomic.AtomicInteger; + +public class SiddhiTestHelper { + public static void waitForEvents(long sleepTime, int expectedCount, AtomicInteger actualCount, long timeout) throws InterruptedException { + long currentWaitTime = 0; + long startTime = System.currentTimeMillis(); + while ((actualCount.get() < expectedCount) && (currentWaitTime <= timeout)) { + Thread.sleep(sleepTime); + currentWaitTime = System.currentTimeMillis() - startTime; + } + } +} diff --git a/src/test/resources/log4j.properties b/src/test/resources/log4j.properties new file mode 100644 index 0000000000..96c79e9449 --- /dev/null +++ b/src/test/resources/log4j.properties @@ -0,0 +1,36 @@ +# +# Copyright (c) 2015, 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. +# + + +# For the general syntax of property based configuration files see the +# documenation of org.apache.log4j.PropertyConfigurator. + +# The root category uses the appender called A1. Since no priority is +# specified, the root category assumes the default priority for root +# which is DEBUG in log4j. The root category is the only category that +# has a default priority. All other categories need not be assigned a +# priority in which case they inherit their priority from the +# hierarchy. + +#log4j.rootLogger=DEBUG, stdout +log4j.rootLogger=INFO, stdout + +log4j.appender.stdout=org.apache.log4j.ConsoleAppender +log4j.appender.stdout.layout=org.apache.log4j.PatternLayout +log4j.appender.stdout.layout.ConversionPattern=%m%n +#log4j.appender.stdout.layout.ConversionPattern=[%t] %-5p %c %x - %m%n