Skip to content

Commit fff596d

Browse files
author
Matthias Weidlich
committed
update query rest api
1 parent 40658a1 commit fff596d

File tree

5 files changed

+145
-59
lines changed

5 files changed

+145
-59
lines changed

‎seep-contribs/src/main/java/uk/ac/imperial/lsds/seep/contribs/esper/EsperSingleQueryOperator.java

Lines changed: 100 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -2,9 +2,12 @@
22

33
import java.util.ArrayList;
44
import java.util.Collections;
5+
import java.util.Iterator;
56
import java.util.LinkedHashMap;
7+
import java.util.LinkedList;
68
import java.util.List;
79
import java.util.Map;
10+
import java.util.Queue;
811

912
import org.slf4j.Logger;
1013
import org.slf4j.LoggerFactory;
@@ -59,14 +62,17 @@ public class EsperSingleQueryOperator implements StatelessOperator {
5962
private boolean enableLoggingOfMatches = true;
6063
private List<DataTuple> matchCache;
6164

62-
65+
private Queue<DataTuple> initCache;
66+
private boolean initialised = false;
67+
6368
public EsperSingleQueryOperator(String query, String url, String name) {
6469
this.esperQuery = query;
6570
this.esperEngineURL = url;
6671
this.name = name;
6772
if (enableLoggingOfMatches) {
6873
this.matchCache = Collections.synchronizedList(new ArrayList<DataTuple>());
6974
}
75+
this.initCache = new LinkedList<DataTuple>();
7076
}
7177

7278
public EsperSingleQueryOperator(String query, String url, String streamKey, String name, String[] typeBinding) {
@@ -80,7 +86,42 @@ public EsperSingleQueryOperator(String query, String url, String name, Map<Strin
8086
this.typesPerStream.put(stream, getTypes(typeBinding.get(stream)));
8187
}
8288

89+
public void initStatement() {
90+
91+
if (statement != null) {
92+
statement.removeAllListeners();
93+
statement.destroy();
94+
}
95+
96+
log.info("Creating ESPER query...");
97+
98+
/*
99+
* Build the ESPER statement
100+
*/
101+
statement = epService.getEPAdministrator().createEPL(this.esperQuery);
83102

103+
/*
104+
* Set a listener called when statement matches
105+
*/
106+
statement.addListener(new UpdateListener() {
107+
@Override
108+
public void update(EventBean[] newEvents, EventBean[] oldEvents) {
109+
if (newEvents == null) {
110+
// we don't care about events leaving the window (old
111+
// events)
112+
return;
113+
}
114+
for (EventBean theEvent : newEvents) {
115+
sendOutput(theEvent);
116+
}
117+
}
118+
});
119+
120+
initialised = true;
121+
log.info("Done with init: {}", this.esperQuery);
122+
}
123+
124+
84125
@Override
85126
public void setUp() {
86127
/*
@@ -112,43 +153,23 @@ public void setUp() {
112153
epService = EPServiceProviderManager.getProvider(esperEngineURL,
113154
configuration);
114155

115-
log.info("Creating ESPER query...");
116-
117156
/*
118-
* Build the ESPER statement
119-
*/
120-
statement = epService.getEPAdministrator().createEPL(this.esperQuery);
121-
122-
/*
123-
* Set a listener called when statement matches
157+
* Initialise the query statement
124158
*/
125-
statement.addListener(new UpdateListener() {
126-
@Override
127-
public void update(EventBean[] newEvents, EventBean[] oldEvents) {
128-
if (newEvents == null) {
129-
// we don't care about events leaving the window (old
130-
// events)
131-
return;
132-
}
133-
for (EventBean theEvent : newEvents) {
134-
sendOutput(theEvent);
135-
}
136-
}
137-
});
159+
initStatement();
138160

139161
/*
140162
* Register rest API handler
141163
*/
142164
NodeManager.restAPIRegistry.put("/query", new RestAPIEsperGetQueryDesc(this));
143165
NodeManager.restAPIRegistry.put("/matches", new RestAPIEsperGetMatches(this));
144-
166+
NodeManager.restAPIRegistry.put("/query_update", new RestAPIEsperPostQueryUpdate(this));
145167

146168
}
147169

148170
protected void sendOutput(EventBean out) {
171+
log.debug("Query returned a new result event: {}", out);
149172

150-
log.info("Query returned a new result event: {}", out);
151-
152173
DataTuple output = new DataTuple(api.getDataMapper(), new TuplePayload());
153174
List<Object> objects = new ArrayList<>();
154175

@@ -158,22 +179,38 @@ protected void sendOutput(EventBean out) {
158179
continue;
159180
objects.add(value);
160181
}
161-
DataTuple t = output.setValues(objects.toArray());
182+
DataTuple outTuple = output.setValues(objects.toArray());
183+
outTuple.getPayload().timestamp = System.currentTimeMillis();
162184

163-
log.info("Sending output {}", t.getPayload().attrValues);
185+
log.debug("At {}, sending output {}", outTuple.getPayload().timestamp, outTuple.getPayload().attrValues);
186+
187+
if (this.enableLoggingOfMatches) {
188+
long cutOffTime = System.currentTimeMillis() - 1000*60*20;
189+
synchronized (matchCache) {
190+
191+
matchCache.add(outTuple);
192+
193+
// Remove old items
194+
Iterator<DataTuple> iter = matchCache.iterator();
195+
boolean run = true;
196+
while (iter.hasNext() && run) {
197+
DataTuple t = iter.next();
198+
if (t.getPayload().timestamp < cutOffTime) {
199+
iter.remove();
200+
}
201+
else {
202+
run = false;
203+
}
204+
}
205+
}
206+
}
164207

165-
if (this.enableLoggingOfMatches)
166-
matchCache.add(t);
208+
log.info("Match cache size: {}", this.matchCache.size());
167209

168-
api.send(t);
210+
api.send(outTuple);
169211
}
170212

171-
@Override
172-
public void processData(DataTuple input) {
173-
174-
log.info("Received input tuple {}", input.toString());
175-
log.info("Map of received input tuple {}", input.getMap().toString());
176-
213+
public void sendData(DataTuple input) {
177214
String stream = input.getString(STREAM_IDENTIFIER);
178215

179216
Map<String, Object> item = new LinkedHashMap<String, Object>();
@@ -182,12 +219,29 @@ public void processData(DataTuple input) {
182219
for (String key : this.typesPerStream.get(stream).keySet())
183220
item.put(key, input.getValue(key));
184221

185-
log.info("Sending item {} with name '{}' to esper engine", item,
222+
log.debug("Sending item {} with name '{}' to esper engine", item,
186223
stream);
187-
224+
188225
epService.getEPRuntime().sendEvent(item, stream);
189226
}
190227

228+
@Override
229+
public void processData(DataTuple input) {
230+
231+
log.debug("Received input tuple {}", input.toString());
232+
log.debug("Map of received input tuple {}", input.getMap().toString());
233+
234+
if (!initialised) {
235+
this.initCache.add(input);
236+
}
237+
else {
238+
while (!this.initCache.isEmpty()) {
239+
sendData(this.initCache.poll());
240+
}
241+
sendData(input);
242+
}
243+
}
244+
191245
@Override
192246
public void processData(List<DataTuple> dataList) {
193247
for (DataTuple tuple : dataList)
@@ -248,6 +302,14 @@ public String getEsperQuery() {
248302
return esperQuery;
249303
}
250304

305+
public void initWithNewEsperQuery(String query) {
306+
log.info("init with new esper query: {}", query);
307+
initialised = false;
308+
this.esperQuery = query;
309+
NodeManager.restAPIRegistry.put("/query", new RestAPIEsperGetQueryDesc(this));
310+
initStatement();
311+
}
312+
251313
public String getName() {
252314
return name;
253315
}

‎seep-contribs/src/main/java/uk/ac/imperial/lsds/seep/contribs/esper/RestAPIEsperGetMatches.java

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -23,13 +23,12 @@ public Object getAnswer(MultiMap<String> reqParameters) {
2323
List<DataTuple> cache = this.operator.getMatchCache();
2424

2525
/*
26-
* Should we keep the matches?
26+
* Should we keep all matches?
2727
*/
2828
boolean keep = false;
2929
if (reqParameters.containsKey("keep"))
3030
keep = Boolean.valueOf(reqParameters.getValue("keep",0));
3131

32-
3332
/*
3433
* Do we have a time window?
3534
*/
@@ -51,6 +50,8 @@ public Object getAnswer(MultiMap<String> reqParameters) {
5150
}
5251
}
5352
}
53+
54+
5455
/*
5556
* Is there a step to consider?
5657
*/

‎seep-contribs/src/main/java/uk/ac/imperial/lsds/seep/contribs/esper/RestAPIEsperPostQueryUpdate.java

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,22 @@
1+
package uk.ac.imperial.lsds.seep.contribs.esper;
2+
3+
import org.eclipse.jetty.util.MultiMap;
4+
5+
import uk.ac.imperial.lsds.seep.infrastructure.api.RestAPIRegistryEntry;
6+
7+
public class RestAPIEsperPostQueryUpdate implements RestAPIRegistryEntry {
8+
9+
private EsperSingleQueryOperator operator;
10+
11+
public RestAPIEsperPostQueryUpdate(EsperSingleQueryOperator operator) {
12+
this.operator = operator;
13+
}
14+
15+
@Override
16+
public Object getAnswer(MultiMap<String> reqParameters) {
17+
18+
String query = reqParameters.getValue("query", 0);
19+
this.operator.initWithNewEsperQuery(query);
20+
return "Loaded query: " + query;
21+
}
22+
}

‎seep-system/src/main/java/uk/ac/imperial/lsds/seep/infrastructure/NodeManager.java

Lines changed: 3 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,6 @@
1818
import java.io.ObjectStreamClass;
1919
import java.io.PrintWriter;
2020
import java.net.InetAddress;
21-
import java.net.InetSocketAddress;
2221
import java.net.MalformedURLException;
2322
import java.net.ServerSocket;
2423
import java.net.Socket;
@@ -27,19 +26,16 @@
2726
import java.util.ArrayList;
2827
import java.util.HashMap;
2928
import java.util.Map;
30-
import java.util.logging.Level;
3129

3230
import org.eclipse.jetty.server.Server;
3331
import org.slf4j.Logger;
3432
import org.slf4j.LoggerFactory;
3533

36-
import com.fasterxml.jackson.databind.ser.std.InetAddressSerializer;
37-
3834
import uk.ac.imperial.lsds.seep.GLOBALS;
3935
import uk.ac.imperial.lsds.seep.comm.NodeManagerCommunication;
36+
import uk.ac.imperial.lsds.seep.infrastructure.api.RestAPIHandler;
4037
import uk.ac.imperial.lsds.seep.infrastructure.api.RestAPINodeDescription;
4138
import uk.ac.imperial.lsds.seep.infrastructure.api.RestAPIRegistryEntry;
42-
import uk.ac.imperial.lsds.seep.infrastructure.api.RestAPIHandler;
4339
import uk.ac.imperial.lsds.seep.infrastructure.dynamiccodedeployer.ExtendedObjectInputStream;
4440
import uk.ac.imperial.lsds.seep.infrastructure.dynamiccodedeployer.RuntimeClassLoader;
4541
import uk.ac.imperial.lsds.seep.infrastructure.master.Infrastructure;
@@ -48,7 +44,6 @@
4844
import uk.ac.imperial.lsds.seep.infrastructure.monitor.slave.MonitorSlaveFactory;
4945
import uk.ac.imperial.lsds.seep.operator.EndPoint;
5046
import uk.ac.imperial.lsds.seep.operator.Operator;
51-
import uk.ac.imperial.lsds.seep.operator.OperatorStaticInformation;
5247
import uk.ac.imperial.lsds.seep.runtimeengine.CoreRE;
5348
import uk.ac.imperial.lsds.seep.state.StateWrapper;
5449

@@ -86,9 +81,9 @@ public class NodeManager{
8681
// public static void main(String[] args) {
8782
//
8883
// NodeManager.restAPIRegistry = new HashMap<>();
89-
// NodeManager.restAPIRegistry.put("/nodedescription", new RestAPINodeDescription(new WorkerNodeDescription(null, 1111)));
84+
// NodeManager.restAPIRegistry.put("/query_update", new RestAPINodeDescription(new WorkerNodeDescription(null, 1111)));
9085
// Server restAPIServer = new Server(8011);
91-
// restAPIServer.setHandler(new RestAPIHandler());
86+
// restAPIServer.setHandler(new RestAPIHandler(restAPIRegistry));
9287
// try {
9388
// restAPIServer.start();
9489
// restAPIServer.join();

‎seep-system/src/main/java/uk/ac/imperial/lsds/seep/infrastructure/api/RestAPIHandler.java

Lines changed: 17 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -44,26 +44,32 @@ public void handle(String target, Request baseRequest,
4444

4545
response.setContentType("application/json;charset=utf-8");
4646
response.setStatus(HttpServletResponse.SC_OK);
47+
response.setHeader("Access-Control-Allow-Origin", "*");
48+
response.setHeader("Access-Control-Allow-Methods", "GET, POST, DELETE, PUT");
49+
4750

48-
if (baseRequest.getMethod().equals("GET")) {
49-
if (this.restAPIRegistry.containsKey(target)) {
50-
MultiMap<String> reqParameters = new MultiMap<>();
51+
if (!this.restAPIRegistry.containsKey(target)) {
52+
baseRequest.setHandled(true);
53+
if (callback != null)
54+
response.getWriter().println(callback + "(" + mapper.writeValueAsString(this.restAPIRegistry.keySet()) + ")");
55+
else
56+
response.getWriter().println(mapper.writeValueAsString(this.restAPIRegistry.keySet()));
57+
}
58+
else {
59+
if (baseRequest.getMethod().equals("GET")) {
5160
baseRequest.setHandled(true);
52-
baseRequest.getUri().decodeQueryTo(reqParameters);
5361
if (callback != null)
54-
response.getWriter().println(callback + "(" + mapper.writeValueAsString(this.restAPIRegistry.get(target).getAnswer(reqParameters)) + ")");
62+
response.getWriter().println(callback + "(" + mapper.writeValueAsString(this.restAPIRegistry.get(target).getAnswer(baseRequest.getParameters())) + ")");
5563
else
56-
response.getWriter().println(mapper.writeValueAsString(this.restAPIRegistry.get(target).getAnswer(reqParameters)));
64+
response.getWriter().println(mapper.writeValueAsString(this.restAPIRegistry.get(target).getAnswer(baseRequest.getParameters())));
5765
}
58-
else {
59-
// default case: answer with a list of available keys
66+
else if (baseRequest.getMethod().equals("POST")) {
6067
baseRequest.setHandled(true);
6168
if (callback != null)
62-
response.getWriter().println(callback + "(" + mapper.writeValueAsString(this.restAPIRegistry.keySet()) + ")");
69+
response.getWriter().println(callback + "(" + mapper.writeValueAsString(this.restAPIRegistry.get(target).getAnswer(baseRequest.getParameters())) + ")");
6370
else
64-
response.getWriter().println(mapper.writeValueAsString(this.restAPIRegistry.keySet()));
71+
response.getWriter().println(mapper.writeValueAsString(this.restAPIRegistry.get(target).getAnswer(baseRequest.getParameters())));
6572
}
6673
}
6774
}
68-
6975
}

0 commit comments

Comments
 (0)