| 1 | /* |
| 2 | * Copyright (c) 2005-2010, WSO2 Inc. (http://www.wso2.org) All Rights Reserved. |
| 3 | * |
| 4 | * WSO2 Inc. licenses this file to you under the Apache License, |
| 5 | * Version 2.0 (the "License"); you may not use this file except |
| 6 | * in compliance with the License. |
| 7 | * You may obtain a copy of the License at |
| 8 | * |
| 9 | * http://www.apache.org/licenses/LICENSE-2.0 |
| 10 | * |
| 11 | * Unless required by applicable law or agreed to in writing, |
| 12 | * software distributed under the License is distributed on an |
| 13 | * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| 14 | * KIND, either express or implied. See the License for the |
| 15 | * specific language governing permissions and limitations |
| 16 | * under the License. |
| 17 | */ |
| 18 | package org.wso2.siddhi.core.executor.expression; |
| 19 | |
| 20 | import org.wso2.siddhi.core.event.AtomicEvent; |
| 21 | import org.wso2.siddhi.core.event.Event; |
| 22 | import org.wso2.siddhi.core.event.ListEvent; |
| 23 | import org.wso2.siddhi.core.event.in.StateEvent; |
| 24 | import org.wso2.siddhi.core.event.StreamEvent; |
| 25 | import org.wso2.siddhi.query.api.query.QueryEventStream; |
| 26 | import org.wso2.siddhi.query.api.definition.Attribute; |
| 27 | import org.wso2.siddhi.query.api.definition.StreamDefinition; |
| 28 | import org.wso2.siddhi.query.api.utils.SiddhiConstants; |
| 29 | |
| 30 | import java.util.List; |
| 31 | |
| 32 | public class VariableExpressionExecutor implements ExpressionExecutor { |
| 33 | Attribute.Type type; |
| 34 | int streamPosition = -1; |
| 35 | int attributePosition = -1; |
| 36 | int innerStreamPosition = -1; //Simple Event (Default) |
| 37 | |
| 38 | |
| 39 | public VariableExpressionExecutor(String streamIdOfVariable, String attributeName, int position, |
| 40 | List<QueryEventStream> queryEventStreamList, |
| 41 | String currentStreamReference) { |
| 42 | String streamReference; |
| 43 | if (streamIdOfVariable != null) { |
| 44 | streamReference = streamIdOfVariable; |
| 45 | } else { |
| 46 | streamReference = currentStreamReference; |
| 47 | } |
| 48 | StreamDefinition streamDefinition = null; |
| 49 | int queryEventStreamListSize = queryEventStreamList.size(); |
| 50 | // if (queryEventStreamListSize == 1) { |
| 51 | // streamDefinition = queryEventStreamList.get(0).getStreamDefinition(); |
| 52 | // } else { |
| 53 | for (int i = 0; i < queryEventStreamListSize; i++) { |
| 54 | QueryEventStream queryEventStream = queryEventStreamList.get(i); |
| 55 | if (queryEventStream.getReferenceStreamId().equals(streamReference)) { |
| 56 | streamDefinition = queryEventStream.getStreamDefinition(); |
| 57 | streamPosition = i; |
| 58 | if (position > -1) { //for known positions |
| 59 | innerStreamPosition = position; |
| 60 | } else if (position == SiddhiConstants.PREV) { |
| 61 | innerStreamPosition = SiddhiConstants.PREV; |
| 62 | } else if (queryEventStream.isCounterStream()) { //to get the last event |
| 63 | innerStreamPosition = SiddhiConstants.LAST; |
| 64 | } |
| 65 | } |
| 66 | |
| 67 | // } |
| 68 | } |
| 69 | if (streamDefinition == null) { |
| 70 | streamDefinition = queryEventStreamList.get(0).getStreamDefinition(); |
| 71 | } |
| 72 | type = streamDefinition.getAttributeType(attributeName); |
| 73 | attributePosition = streamDefinition.getAttributePosition(attributeName); |
| 74 | |
| 75 | } |
| 76 | |
| 77 | @Override |
| 78 | public Object execute(AtomicEvent event) { |
| 79 | try { |
| 80 | if (event instanceof Event) { |
| 81 | return ((Event) event).getData()[attributePosition]; |
| 82 | } else if (innerStreamPosition == SiddhiConstants.PREV) { |
| 83 | StreamEvent streamEvent = ((StateEvent) event).getStreamEvent(streamPosition); |
| 84 | if (streamEvent instanceof ListEvent) { |
| 85 | int prevLast = ((ListEvent) streamEvent).getActiveEvents() - 2; |
| 86 | if (prevLast > 0) { |
| 87 | streamEvent = ((ListEvent) streamEvent).getEvent(prevLast); |
| 88 | if (streamEvent != null) { |
| 89 | return ((Event) streamEvent).getData(attributePosition); |
| 90 | } |
| 91 | } |
| 92 | } |
| 93 | if (streamPosition - 1 >= 0) { |
| 94 | streamEvent = ((StateEvent) event).getStreamEvent(streamPosition - 1); |
| 95 | if (streamEvent instanceof ListEvent) { |
| 96 | int prevLast = ((ListEvent) streamEvent).getActiveEvents() - 1; |
| 97 | if (prevLast > 0) { |
| 98 | streamEvent = ((ListEvent) streamEvent).getEvent(prevLast); |
| 99 | if (streamEvent != null) { |
| 100 | return ((Event) streamEvent).getData(attributePosition); |
| 101 | } |
| 102 | } |
| 103 | } else if (streamEvent instanceof Event) { |
| 104 | return ((Event) streamEvent).getData(attributePosition); |
| 105 | } |
| 106 | } |
| 107 | if (streamPosition - 2 >= 0) { |
| 108 | streamEvent = ((StateEvent) event).getStreamEvent(streamPosition - 2); |
| 109 | if (streamEvent instanceof ListEvent) { |
| 110 | int prevLast = ((ListEvent) streamEvent).getActiveEvents() - 1; |
| 111 | if (prevLast > 0) { |
| 112 | streamEvent = ((ListEvent) streamEvent).getEvent(prevLast); |
| 113 | if (streamEvent != null) { |
| 114 | return ((Event) streamEvent).getData(attributePosition); |
| 115 | } |
| 116 | } |
| 117 | } else if (streamEvent instanceof Event) { |
| 118 | return ((Event) streamEvent).getData(attributePosition); |
| 119 | } |
| 120 | } |
| 121 | return null; |
| 122 | // |
| 123 | } else { //State Event |
| 124 | StreamEvent streamEvent = ((StateEvent) event).getStreamEvent(streamPosition); |
| 125 | if (streamEvent == null) { |
| 126 | return null; |
| 127 | } else if (streamEvent instanceof Event) { //Single Event |
| 128 | return ((Event) streamEvent).getData(attributePosition); |
| 129 | } else if (innerStreamPosition == SiddhiConstants.LAST) { //Event List Unknown Size (get the last event) |
| 130 | streamEvent = ((ListEvent) streamEvent).getEvent(((ListEvent) streamEvent).getActiveEvents() - 1); |
| 131 | if (streamEvent == null) { |
| 132 | return null; |
| 133 | } else { |
| 134 | return ((Event) streamEvent).getData(attributePosition); |
| 135 | } |
| 136 | } else { //Event List Known Size |
| 137 | streamEvent = ((ListEvent) streamEvent).getEvent(innerStreamPosition); |
| 138 | if (streamEvent == null) { |
| 139 | return null; |
| 140 | } else { |
| 141 | return ((Event) streamEvent).getData(attributePosition); |
| 142 | } |
| 143 | } |
| 144 | } |
| 145 | } catch (NullPointerException e) { |
| 146 | return null; |
| 147 | } |
| 148 | } |
| 149 | |
| 150 | public Attribute.Type getType() { |
| 151 | return type; |
| 152 | } |
| 153 | |
| 154 | |
| 155 | } |