[TASK] Downgrade to engine 0.9.6

pom.xml

@@ -27,7 +27,7 @@
 		<dependency>
 			<groupId>eu.larkc.csparql</groupId>
 			<artifactId>csparql-core</artifactId>
-			<version>0.9.7</version>
+			<version>0.9.6</version>
 		</dependency>
 		<dependency>
 			<groupId>it.polimi.deib</groupId>

src/main/java/lu/jpt/csparqlproject/SimulationContext.java

@@ -11,9 +11,11 @@ import eu.larkc.csparql.cep.api.RdfStream;
 import eu.larkc.csparql.core.engine.CsparqlEngine;
 import eu.larkc.csparql.core.engine.CsparqlEngineImpl;
 import eu.larkc.csparql.core.engine.CsparqlQueryResultProxy;
+import eu.larkc.csparql.core.engine.RDFStreamFormatter;
 import lu.jpt.csparqlproject.gui.FancyTextObserverWindow;
 import lu.jpt.csparqlproject.rentacar.RentACarSimulation;
 import lu.jpt.csparqlproject.util.CsparqlQueryHelper;
+import lu.jpt.csparqlproject.util.CsparqlQueryHelper.CsparqlQueryInfo;
 
 /**
  * This class encapsulates the use of the C-SPARQL Engine. Here, the RentACarSimulation,
@@ -153,23 +155,38 @@ public class SimulationContext {
 		RdfStream driverStream = simulation.getDriverStream();
 		this.engine.registerStream(driverStream);
 		this.registeredStreams.add(driverStream);
-		// Register all the queries and add result observers!
 		Collection<String> queriesToRegister = new ArrayList<String>();
-		queriesToRegister.add(RentACarSimulation.getEventsQuery());
-		queriesToRegister.add(RentACarSimulation.getAverageDataByCar());
+		// Collect the queries to use!
+		//queriesToRegister.add(RentACarSimulation.getEventsQuery());
+		queriesToRegister.add(RentACarSimulation.getAverageDataByCarAsStream());
+		queriesToRegister.add(RentACarSimulation.selectFromRegisteredStream());
+		// Now register each query appropriately!
 		for(String query : queriesToRegister) {
-			String queryName = CsparqlQueryHelper.getQueryName(query);
+			CsparqlQueryInfo queryInfo = CsparqlQueryHelper.getQueryInfo(query);
 			CsparqlQueryResultProxy resultProxy = null;
 			try {
-				resultProxy = this.engine.registerQuery(query, true);
-			} catch (Exception e) {
-				SimulationContext.logger.error(e.toString());
-				SimulationContext.logger.error("Could not register query "+queryName);
-				SimulationContext.logger.error(query);
+				resultProxy = this.engine.registerQuery(query, false);
+				// Take care of streams and queries differently.
+				if(queryInfo.isStream) {
+					// If the query is a stream, we need additional components to feed it back into the engine.
+					String streamUri = "http://example.org/stream/"+queryInfo.name;
+					RDFStreamFormatter rdfStreamFormatter = new RDFStreamFormatter(streamUri);
+					engine.registerStream(rdfStreamFormatter);
+					resultProxy.addObserver(rdfStreamFormatter);
+					Observer resultObserver = this.createResultObserverWindow(queryInfo.name);
+					resultProxy.addObserver(resultObserver);
+				} else {
+					// If it is a regular query, just attach a fitting observer
+					Observer resultObserver = this.createResultObserverWindow(queryInfo.name);
+					resultProxy.addObserver(resultObserver);
 				}
 				this.queryResultProxies.add(resultProxy);
-			Observer resultObserver = this.createResultObserverWindow(queryName);
-			resultProxy.addObserver(resultObserver);
+				SimulationContext.logger.info("Successfully registered query " + queryInfo.name + ": " + query);
+			} catch (Exception e) {
+				SimulationContext.logger.error(e.toString());
+				SimulationContext.logger.error("Could not register query "+queryInfo.name);
+				SimulationContext.logger.error(query);
+			}
 		}
 		// Setup complete, ready to run.
 		SimulationContext.logger.info("Simulation set up and ready to go!");

src/main/java/lu/jpt/csparqlproject/rentacar/RentACarSimulation.java

@@ -42,7 +42,7 @@ public class RentACarSimulation implements Runnable {
 	
 	public RentACarSimulation() {
 		this.registerOwnPrefixes();
-		int numberOfCars = 5;
+		int numberOfCars = 2;
 		int numberOfCustomers = 1;
 		// Create a car pool and drivers
 		this.carPool = new CarPool(numberOfCars);
@@ -204,20 +204,44 @@ public class RentACarSimulation implements Runnable {
 				+ "}";
 	}
 	
-	public static String getAverageDataByCar() {
-		return "REGISTER QUERY getAverageSpeedByCar AS "
+	public static String getAverageDataByCarAsStream() {
+		return "REGISTER STREAM getAverageSpeedByCar AS "
 				+ "PREFIX rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#> "
 				+ "PREFIX f: <http://larkc.eu/csparql/sparql/jena/ext#> "
 				+ "PREFIX xsd: <http://www.w3.org/2001/XMLSchema#> "
 				+ "PREFIX car: <"+RentACarSimulation.BASE_ONTOLOGY_IRI+"> "
-				+ "SELECT ?car (AVG(?speed) AS ?avgSpeed) (AVG(?rpm) AS ?avgRPM) "
+				+ "CONSTRUCT { "
+				+ " [] rdf:type car:AverageSpeedEvent . "
+				+ " [] car:relatedCar ?car . "
+				+ " [] car:averageSpeed ?avgSpeed . "
+				+ "} "
 				+ "FROM STREAM <"+RentACarSimulation.CAR_STREAM_IRI+"> [RANGE 5s STEP 1s] "
 				+ "WHERE { "
+				+ " { "
+				+ "  SELECT (AVG(?speed) AS ?avgSpeed) "
+				+ "  WHERE { "
 				+ "   ?e rdf:type car:CarStatusEvent . "
 				+ "   ?e car:relatedCar ?car . "
-				+ " ?e car:motorRPM ?rpm . "
 				+ "   ?e car:speed ?speed . "
-				+ "} "
-				+ "GROUP BY (?car) ";
+				+ "  } "
+				+ "  GROUP BY (?car) "
+				+ " }"
+				+ "} ";
+	}
+	
+	/**
+	 * IMPORTANT NOTE: STREAM gets matched everywhere, so don't call stuff "stream"!
+	 */
+	public static String selectFromRegisteredStream() {
+		return "REGISTER Query askRegisteredStrm AS "
+				+ "PREFIX rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#> "
+				+ "PREFIX f: <http://larkc.eu/csparql/sparql/jena/ext#> "
+				+ "PREFIX xsd: <http://www.w3.org/2001/XMLSchema#> "
+				+ "PREFIX car: <"+RentACarSimulation.BASE_ONTOLOGY_IRI+"> "
+				+ "SELECT ?s ?p ?o "
+				+ "FROM STREAM <http://example.org/stream/getAverageSpeedByCar> [RANGE 5s STEP 1s] "
+				+ "WHERE { "
+				+ " ?s ?p ?o . "
+				+ "} ";
 	}
 }

src/main/java/lu/jpt/csparqlproject/util/CsparqlQueryHelper.java

@@ -8,13 +8,26 @@ import java.util.StringTokenizer;
  */
 public class CsparqlQueryHelper {
 
+	/**
+	 * Small inner class containing query info
+	 */
+	public class CsparqlQueryInfo {
+		public boolean isStream;
+		public String name;
+		
+		public CsparqlQueryInfo() {
+			this.name = "[UNRESOLVED NAME]";
+			this.isStream = false;
+		}
+	}
+	
 	/**
 	 * Returns the csparql query name stated within the query.
 	 * @param query given csparql query
-	 * @return name of query
+	 * @return query info using class CsparqlQueryInfo
 	 */
-	public static String getQueryName(String query) {
-		String queryName = "[UNRESOLVED NAME]";
+	public static CsparqlQueryInfo getQueryInfo(String query) {
+		CsparqlQueryHelper.CsparqlQueryInfo result = new CsparqlQueryHelper().new CsparqlQueryInfo();
 		StringTokenizer tokenizer = new StringTokenizer(query);
 		boolean gotName = false;
 		int stateCounter = 0;
@@ -29,18 +42,19 @@ public class CsparqlQueryHelper {
 				break;
 			case 1:
 				if(trimmedLowerToken.equals("query") || trimmedLowerToken.equals("stream")) {
+					if(trimmedLowerToken.equals("stream")) result.isStream = true;
 					stateCounter++;
 				}
 				break;
 			case 2:
-				queryName = token.trim();
+				result.name = token.trim();
 				gotName = true;
 				break;
 			default:
 				break;
 			}
 		}
-		return queryName;
+		return result;
 	}
 	
 }