Add Linked Data Notification Protocol and TCP Protocol

engine/src/main/java/it/unibo/arces/wot/sepa/engine/core/Engine.java

@@ -42,11 +42,15 @@
 import it.unibo.arces.wot.sepa.engine.protocol.websocket.WebsocketServer;
 import it.unibo.arces.wot.sepa.engine.protocol.http.HttpGate;
 import it.unibo.arces.wot.sepa.engine.protocol.http.HttpsGate;
+import it.unibo.arces.wot.sepa.engine.protocol.tcp.ServerPrincipaleQuery;
+import it.unibo.arces.wot.sepa.engine.protocol.tcp.ServerPrincipaleSubscribe;
 import it.unibo.arces.wot.sepa.engine.protocol.websocket.SecureWebsocketServer;
 
 import it.unibo.arces.wot.sepa.engine.scheduling.Scheduler;
 
 import it.unibo.arces.wot.sepa.engine.security.AuthorizationManager;
+import net.minidev.json.parser.JSONParser;
+
 
 /**
  * This class represents the SPARQL Subscription Engine (Core) of the Semantic
@@ -70,7 +74,7 @@ public class Engine implements EngineMBean {
 
 	// SPARQL 1.1 Protocol handler
 	private HttpGate httpGate = null;
-
+	
 	// SPARQL 1.1 SE Protocol handler
 	private WebsocketServer wsServer;
 	private SecureWebsocketServer wssServer;
@@ -236,6 +240,7 @@ public Engine(String[] args) throws SEPASecurityException, SEPAProtocolException
 			}
 		}
 
+
 		wssServer.start();
 		synchronized(wssServer) {
 			try {
@@ -244,6 +249,15 @@ public Engine(String[] args) throws SEPASecurityException, SEPAProtocolException
 				throw new SEPAProtocolException(e);
 			}
 		}
+
+		ServerPrincipaleQuery queryServer = new ServerPrincipaleQuery(scheduler); //da spostare sopra
+		ServerPrincipaleSubscribe subscribeServer = new ServerPrincipaleSubscribe(scheduler);
+
+
+		queryServer.start();
+		subscribeServer.start();
+
+
 		System.out.println("----------------------");
 
 		// Welcome message

engine/src/main/java/it/unibo/arces/wot/sepa/engine/protocol/http/HttpGate.java

@@ -6,7 +6,6 @@
 import java.util.concurrent.TimeUnit;
 
 import org.apache.http.ExceptionLogger;
-
 import org.apache.http.impl.nio.bootstrap.HttpServer;
 import org.apache.http.impl.nio.bootstrap.ServerBootstrap;
 import org.apache.http.impl.nio.reactor.IOReactorConfig;
@@ -20,6 +19,7 @@
 import it.unibo.arces.wot.sepa.engine.protocol.http.handler.QueryHandler;
 import it.unibo.arces.wot.sepa.engine.protocol.http.handler.UpdateHandler;
 import it.unibo.arces.wot.sepa.engine.protocol.http.handler.EchoHandler;
+import it.unibo.arces.wot.sepa.engine.protocol.http.handler.LinkedDataNotificationServlet;
 import it.unibo.arces.wot.sepa.engine.scheduling.Scheduler;
 
 public class HttpGate {
@@ -41,7 +41,24 @@ public HttpGate(EngineProperties properties, Scheduler scheduler) throws SEPAPro
 				.setServerInfo(serverInfo).setIOReactorConfig(config).setExceptionLogger(ExceptionLogger.STD_ERR)
 				.registerHandler(properties.getQueryPath(), new QueryHandler(scheduler))
 				.registerHandler(properties.getUpdatePath(), new UpdateHandler(scheduler))
-				.registerHandler("/echo", new EchoHandler()).create();
+				.registerHandler("/echo", new EchoHandler())
+				.registerHandler("/ldnServlet/*", new LinkedDataNotificationServlet(scheduler))
+				.registerHandler("/ldnServlet", new LinkedDataNotificationServlet(scheduler)).create(); //aggiunta
+
+
+
+
+
+		//qui mappo la servlet che dovrò scrivere..gli passo uno scheduler
+		//la mia servlet dovrà implementare implements HttpAsyncRequestHandler<HttpRequest>
+		//devo invocare i metodi dello scheduler (vedi serverSecondario per sapere quali metodi)
+		//devo creare un mio handler con un parametro: httpExchange
+		//invio lo spuid
+		//devo controllare che sia una get(vedi QueryHandler per vedere come fare)
+		//la stringa che dovrò creare sarà così composta:
+		//http://localhost/nomeCheVoglio/richiestaPresaDallUtente
+		//uso postman per fare query: metto come http: localhost:8000/stringaConCuiHoMappatoLaServlet
+		//il file engine.jar contiene l'elenco delle porte
 
 		try {
 			server.start();

engine/src/main/java/it/unibo/arces/wot/sepa/engine/protocol/http/handler/LinkedDataNotificationServlet.java

@@ -0,0 +1,97 @@
+package it.unibo.arces.wot.sepa.engine.protocol.http.handler;
+
+import java.io.IOException;
+import java.net.URLDecoder;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.StringTokenizer;
+
+import org.apache.http.HttpException;
+import org.apache.http.HttpRequest;
+import org.apache.http.HttpStatus;
+import org.apache.http.nio.protocol.BasicAsyncRequestConsumer;
+import org.apache.http.nio.protocol.HttpAsyncExchange;
+import org.apache.http.nio.protocol.HttpAsyncRequestConsumer;
+import org.apache.http.nio.protocol.HttpAsyncRequestHandler;
+import org.apache.http.protocol.HttpContext;
+
+import it.unibo.arces.wot.sepa.commons.request.QueryRequest;
+import it.unibo.arces.wot.sepa.commons.request.SubscribeRequest;
+import it.unibo.arces.wot.sepa.engine.protocol.http.HttpUtilities;
+import it.unibo.arces.wot.sepa.engine.scheduling.Scheduler;
+
+public class LinkedDataNotificationServlet implements HttpAsyncRequestHandler<HttpRequest>{
+
+	private Scheduler scheduler;
+
+	public LinkedDataNotificationServlet(Scheduler scheduler) {
+		this.scheduler = scheduler;
+	}
+
+	@Override
+	public HttpAsyncRequestConsumer<HttpRequest> processRequest(HttpRequest request, HttpContext context)
+			throws HttpException, IOException {
+		return new BasicAsyncRequestConsumer();
+	}
+
+	@Override
+	public void handle(HttpRequest data, HttpAsyncExchange httpExchange, HttpContext context)
+			throws HttpException, IOException {
+
+		String query = "";
+
+		if(httpExchange.getRequest().getRequestLine().getMethod().toUpperCase().equals("GET")) {
+			System.out.println("LINE: "  + httpExchange.getRequest().getRequestLine().toString());          
+			String url = httpExchange.getRequest().getRequestLine().getUri();
+			StringTokenizer token = new StringTokenizer(url, " ");
+			String path = token.nextToken(); ///ldnServlet?subscribe=select * where { ?a ?b ?c }
+			path = URLDecoder.decode(path, "UTF-8");
+			int i = path.indexOf('?');
+
+			query = i > 0 ? path.substring(i+1) : "" ;
+
+			Map<String, String> parseQuery = LinkedDataNotificationServlet.getQueryMap(query);
+
+			//se è una subscribe
+			if(parseQuery.containsKey("subscribe")){
+			query = parseQuery.get("subscribe"); //subscribe=select * where { ?a ?b ?c }
+
+			SubscribeRequest subscribeRequest = new SubscribeRequest(query);
+			scheduler.schedule(subscribeRequest, new ldnHandler(httpExchange));
+
+			} else { //case is not subscribe
+
+			String[] split = path.split("\\/");
+			String spuid = split[2];	//regex	
+			int numeroNotifica = split.length > 3 ? Integer.parseInt(split[3]) : -1;
+
+			QueryRequest queryRequest = new QueryRequest(query);
+			scheduler.schedule(queryRequest, new ReadLdnHandler(httpExchange, spuid, numeroNotifica));
+
+			}
+
+		} else {
+			HttpUtilities.sendFailureResponse(httpExchange, HttpStatus.SC_BAD_REQUEST, "Wrong format: " + httpExchange.getRequest().getRequestLine());
+			return;
+		}
+
+	}
+
+	public static Map<String, String> getQueryMap(String query)  
+	{  
+		Map<String, String> map = new HashMap<String, String>();
+		if(!query.isEmpty()){
+		    String[] params = query.split("&");  
+		    for (String param : params)  
+		    {  
+		        String name = param.split("=")[0]; 
+		        System.out.println("Key: " + name);
+		        String value = param.split("=")[1]; 
+		        System.out.println("Value: " + value);
+		        map.put(name, value);  
+		    }  
+		}
+		return map;
+	}
+
+}

engine/src/main/java/it/unibo/arces/wot/sepa/engine/protocol/http/handler/ReadLdnHandler.java

@@ -0,0 +1,119 @@
+package it.unibo.arces.wot.sepa.engine.protocol.http.handler;
+
+import java.io.FileReader;
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.http.HttpResponse;
+import org.apache.http.HttpVersion;
+import org.apache.http.entity.ContentType;
+import org.apache.http.message.BasicHttpResponse;
+import org.apache.http.message.BasicStatusLine;
+import org.apache.http.nio.entity.NStringEntity;
+import org.apache.http.nio.protocol.BasicAsyncResponseProducer;
+import org.apache.http.nio.protocol.HttpAsyncExchange;
+
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import com.google.gson.reflect.TypeToken;
+import com.google.gson.stream.JsonReader;
+
+import it.unibo.arces.wot.sepa.commons.response.Notification;
+import it.unibo.arces.wot.sepa.commons.response.Ping;
+import it.unibo.arces.wot.sepa.commons.response.Response;
+import it.unibo.arces.wot.sepa.engine.core.EventHandler;
+
+public class ReadLdnHandler implements EventHandler {
+
+	private HttpAsyncExchange exchange;
+	private String spuid;
+	private int numeroNotifica;
+
+	private static String FILENAME = "/Users/AlessioPazzani/Documents/Tesi/FileSpuid/";
+
+	public ReadLdnHandler(HttpAsyncExchange httpExchange, String spuid, int numeroNotifica) throws IllegalArgumentException {
+		this.exchange = httpExchange;
+		this.spuid = spuid;
+		this.numeroNotifica = numeroNotifica;
+	}
+
+	@Override
+	public void sendResponse(Response response) throws IOException {
+
+		if(numeroNotifica <0){ //localhost:8000/ldnServlet/spuid
+
+		System.out.println("E' stato richiesto di visionare un particolare file");
+
+		String fileName = FILENAME  + spuid;
+		Type listType = new TypeToken<List<Object>>(){}.getType();
+		Gson gson = new GsonBuilder().setPrettyPrinting().create();
+		JsonReader jsonReader = new JsonReader(new FileReader(fileName));
+		List<Object> jsonList = gson.fromJson(jsonReader, listType);
+
+		System.out.println(gson.toJson(jsonList)); //stampo l'intero file letto
+
+		ResponseListingOfNotification listOfNotification = new ResponseListingOfNotification();
+		List<String> contains = new ArrayList<String>();
+		System.out.println("Dimensione JSONList: " + jsonList.size());
+		for (int i=0; i<jsonList.size(); i++){
+			contains.add("http://example.org/inbox/" + spuid + "/" + i);
+		}
+
+
+
+		listOfNotification.setContext("http://w3.org/ns/ldp");
+		listOfNotification.setId("http://example.org/inbox");
+		listOfNotification.setContains(contains);
+
+		String responseJson = gson.toJson(listOfNotification);
+		//fromJson(stringa, nomeClasse.Class che rappresento) restituisce un oggetto della classe nomeClasse, a partire da una stringa (stringa) che rappresenta un json
+
+		NStringEntity entity = new NStringEntity(
+				responseJson,
+                ContentType.create("application/json-ld", "UTF-8"));
+		HttpResponse res = new BasicHttpResponse(new BasicStatusLine(HttpVersion.HTTP_1_0, 200, "OK"));
+        res.setEntity(entity);
+
+        exchange.submitResponse(new BasicAsyncResponseProducer(res));
+
+		} else { //localhost:8000/ldnServlet/spuid/numeroNotifica
+
+			String fileName = FILENAME + spuid;
+			Type listType = new TypeToken<List<Object>>(){}.getType();
+			Gson gson = new GsonBuilder().setPrettyPrinting().create();
+			JsonReader jsonReader = new JsonReader(new FileReader(fileName));
+			List<Object> jsonList = gson.fromJson(jsonReader, listType);
+
+			if(!jsonList.isEmpty()){ //se il file con nome spuid esiste
+
+				Object notifica = jsonList.get(numeroNotifica);
+				System.out.println("Notifica: \n\n" + notifica.toString());
+				String result = gson.toJson(notifica);
+				NStringEntity entity = new NStringEntity(
+						result,
+		                ContentType.create("application/json-ld", "UTF-8"));
+				HttpResponse res = new BasicHttpResponse(new BasicStatusLine(HttpVersion.HTTP_1_0, 200, "OK"));
+		        res.setEntity(entity);
+
+		        exchange.submitResponse(new BasicAsyncResponseProducer(res));
+			}
+
+		} //else
+
+	} //sendResponse
+
+	@Override
+	public void notifyEvent(Notification notify) throws IOException {
+		// TODO Auto-generated method stub
+
+	}
+
+	@Override
+	public void sendPing(Ping ping) throws IOException {
+		// TODO Auto-generated method stub
+
+	}
+
+}

engine/src/main/java/it/unibo/arces/wot/sepa/engine/protocol/http/handler/ResponseListingOfNotification.java

@@ -0,0 +1,45 @@
+package it.unibo.arces.wot.sepa.engine.protocol.http.handler;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import com.google.gson.annotations.SerializedName;
+
+public class ResponseListingOfNotification {
+
+
+	@SerializedName(value="@context")
+	private String context;
+	@SerializedName(value="@id")
+	private String id;
+	private List<String> contains;
+
+	public ResponseListingOfNotification(){
+		contains = new ArrayList<String>();
+	}
+
+	public String getContext() {
+		return context;
+	}
+
+	public void setContext(String context) {
+		this.context = context;
+	}
+
+	public String getId() {
+		return id;
+	}
+
+	public void setId(String id) {
+		this.id = id;
+	}
+
+	public List<String> getContains() {
+		return contains;
+	}
+
+	public void setContains(List<String> contains) {
+		this.contains = contains;
+	}
+
+}

engine/src/main/java/it/unibo/arces/wot/sepa/engine/protocol/http/handler/SPARQL11Handler.java

@@ -26,7 +26,7 @@
 public abstract class SPARQL11Handler implements HttpAsyncRequestHandler<HttpRequest>, SPARQL11HandlerMBean {
 	private static final Logger logger = LogManager.getLogger("SPARQL11Handler");
 
-	private Scheduler scheduler;
+	Scheduler scheduler;
 
 	protected HTTPHandlerBeans jmx = new HTTPHandlerBeans();