Commit ccd19408 authored by nseydoux's avatar nseydoux

Adapted to now vocabulary

parent 593adb8a
......@@ -12,11 +12,12 @@ CONSTRUCT {
?service ?serviceProperty ?serviceObject.
} WHERE {
?uri ?nodeProperty ?nodeObject;
?uri a lmu:HostNode;
?nodeProperty ?nodeObject;
ioto:hasId "{{ id }}".
?nodeProperty a edr:ParentAnnouncedProperty.
OPTIONAL {
?uri iotl:exposes ?service.
?service ?serviceProperty ?serviceObject.
}
}
}
PREFIX edr: <http://w3id.org/laas-iot/edr#>
PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>
PREFIX lmu: <http://w3id.org/laas-iot/lmu#>
INSERT {
?host ?predicate ?property.
} WHERE {
# Captures interests, AMONG OTHER PROPERTIES
?host a lmu:HostNode.
?node ?predicate ?property.
?predicate a edr:ChildrenProxiedProperty.
FILTER NOT EXISTS {
?node a lmu:HostNode.
}
}
PREFIX edr: <http://w3id.org/laas-iot/edr#>
PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>
PREFIX lmu: <http://w3id.org/laas-iot/lmu#>
SELECT ?predicate ?value
WHERE {
?node edr:isInterestedIn ?interest;
{
?node a lmu:HostNode.
} UNION {
?host a lmu:HostNode;
lmu:hasUpstreamNode ?node.
}
}
PREFIX edr: <http://w3id.org/laas-iot/edr#>
PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>
PREFIX lmu: <http://w3id.org/laas-iot/lmu#>
SELECT ?property
WHERE {
# Captures interests, AMONG OTHER PROPERTIES
?node ?predicate ?property.
?predicate a edr:ChildrenProxiedProperty.
FILTER NOT EXISTS {
?node a lmu:HostNode.
}
}
PREFIX edr: <http://w3id.org/laas-iot/edr#>
PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>
PREFIX lmu: <http://w3id.org/laas-iot/lmu#>
SELECT ?node ?property
WHERE {
# Captures interests, AMONG OTHER PROPERTIES
?node ?predicate ?property.
?predicate a edr:ChildrenProxiedProperty.
FILTER NOT EXISTS {
?node a lmu:HostNode.
}
}
PREFIX lmu: <http://w3id.org/laas-iot/lmu#>
PREFIX edr: <http://w3id.org/laas-iot/edr#>
prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#>
INSERT {
<{{ self }}> ?predicate ?object.
......@@ -7,5 +8,5 @@ INSERT {
<{{ self }}> lmu:hasDownstreamNode <{{ remote }}>.
# captures for instance "producesDataOn ?property"
<{{ remote }}> ?predicate ?object.
?predicate rdfs:subClassOf+ edr:ParentAnnouncedProperty.
?predicate rdfs:subClassOf+ edr:ParentProxiedProperty.
}
PREFIX edr: <http://w3id.org/laas-iot/edr#>
PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>
PREFIX lmu: <http://w3id.org/laas-iot/lmu#>
INSERT {
?host ?predicate ?property.
} WHERE {
# Captures interests, AMONG OTHER PROPERTIES
?host a lmu:HostNode.
?node ?predicate ?property.
?predicate a edr:ChildrenProxiedProperty.
FILTER NOT EXISTS {
?node a lmu:HostNode.
}
}
package fr.irit.melodi.edr.errors;
public class UnimplementedFeatureException extends Exception {
private static final long serialVersionUID = -5330248399909835154L;
public UnimplementedFeatureException(String message) {
super(message);
}
}
......@@ -9,6 +9,7 @@ import javax.ws.rs.core.Response;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import fr.irit.melodi.edr.errors.UnimplementedFeatureException;
import fr.irit.melodi.edr.node.Node;
@Path("/app")
......@@ -17,7 +18,7 @@ public class ApplicationEndpoint {
@POST
@Path("/rule")
public Response pushRule(String rule){
public Response pushRule(String rule) throws UnimplementedFeatureException{
LOGGER.trace("Received new application rule "+rule);
Node.getInstance().getRuleManager().addRule(rule);
return Response.status(HttpURLConnection.HTTP_OK).build();
......
......@@ -22,6 +22,7 @@ import com.fasterxml.jackson.core.JsonGenerationException;
import com.fasterxml.jackson.databind.JsonMappingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import fr.irit.melodi.edr.errors.UnimplementedFeatureException;
import fr.irit.melodi.edr.node.Node;
@Path("/topology")
......@@ -69,7 +70,7 @@ public class TopologyEndPoint {
.build();
}
private static String processAnnounce(String profile){
private static String processAnnounce(String profile) throws UnimplementedFeatureException{
Model m = ModelFactory.createDefaultModel();
InputStream stream = new ByteArrayInputStream(profile.getBytes(StandardCharsets.UTF_8));
m.read(stream, "http://example.com/ns#", "TTL");
......
......@@ -16,6 +16,7 @@ import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import fr.irit.melodi.edr.configuration.E_Centralization;
import fr.irit.melodi.edr.errors.UnimplementedFeatureException;
import fr.irit.melodi.edr.node.Node;
import fr.irit.melodi.edr.out.resources.AnnouncerClient;
import fr.irit.melodi.sparql.exceptions.IncompleteSubstitutionException;
......@@ -61,13 +62,13 @@ public class InterestManager {
}
}
public void propagateInterests(){
public void propagateInterests() throws UnimplementedFeatureException{
LOGGER.debug("Begining interest propagation for "+Node.getInstance().getName()+"...");
if(Node.getInstance().getConfiguration().getCentralization().equals(E_Centralization.CENTRAL_DIRECT_RAW) &&
Node.getInstance().getConfiguration().getRemotes().size() > 0){
LOGGER.debug("Transferring interest of the central node");
// For the root node, interest propagation is always the same
Map<String, Set<String>> interests = this.listRemoteInterestsByNodes();
Map<String, Set<String>> interests = this.listProxiedPropertiesByNodes();
LOGGER.debug(interests);
for(String remote : interests.keySet()){
for(String interest : interests.get(remote)){
......@@ -75,19 +76,32 @@ public class InterestManager {
}
}
} else {
Set<String> interests = this.listInterests();
// Assimilates the proxied properties, and announces them
this.assimilateChildrenProxiedProperties();
LOGGER.warn("There is no discrimination between proxied properties and the other regarding allchildrenannouced vs somechildrenannounced");
Set<Entry<String, String>> interests = this.listChildrenAnnouncedProperties();
LOGGER.debug("Propagating "+interests.size()+" interests: "+interests);
for(String interest : interests){
this.propagateInterest(Node.getInstance().getUri(), interest);
for(Entry<String, String> proxiedProperty : interests){
if(proxiedProperty.getKey().equals("http://w3id.org/laas-iot/edr#isInterestedIn")) {
this.propagateInterest(Node.getInstance().getUri(), proxiedProperty.getValue());
}
else {
LOGGER.error("Generic propagation unimplemented");
throw new UnimplementedFeatureException("Trying to propagate an unexpected property");
}
}
}
LOGGER.debug("interest propagation done.");
}
public void assimilateChildrenProxiedProperties() {
Node.getInstance().getKBManager().updateModel(
this.queries.getQueries().get("assimilate_children_proxied_property")
);
}
public void propagateInterest(String interested, String interestURI){
Set<String> targets;
//if(!E_Centralization.rulePropagationAllowed(Node.getInstance().getConfiguration().getCentralization())){
// targets = Node.getInstance().getProductionManager().getProducers(interestURI);
if(E_Centralization.rawDataPropagation(Node.getInstance().getConfiguration().getCentralization())){
targets = Node.getInstance().getProductionManager().getProducers(interestURI);
LOGGER.debug(targets.size()+" nodes under "+Node.getInstance().getName()+" are producers of "+interestURI);
......@@ -100,15 +114,11 @@ public class InterestManager {
targets = Node.getInstance().getProductionManager().getProducers(interestURI);
LOGGER.debug(targets.size()+" nodes under "+Node.getInstance().getName()+" are producers of "+interestURI+", propagating for a parent");
}
//} else {
// targets.addAll(Node.getInstance().getProductionManager().getPartialProducers(interestURI));
//}
for(String downstream : targets){
String downstreamEndpoint = Node.getInstance().getTopologyManager().getEndpoint(downstream);
if(downstreamEndpoint != null && !(downstreamEndpoint.equals(""))){
LOGGER.debug("Propagating interests of "+interested+" for"+interestURI+" to "+downstream+" at endpoint "+downstreamEndpoint);
LOGGER.debug("Propagating interests of "+interested+" for "+interestURI+" to "+downstream+" at endpoint "+downstreamEndpoint);
String interestedEndpoint = Node.getInstance().getTopologyManager().getEndpoint(interested);
AnnouncerClient.annouceToDownstream(
downstreamEndpoint,
......@@ -161,27 +171,25 @@ public class InterestManager {
}
}
public Set<String> listInterests(){
Set<String> result = new HashSet<>();
// Map<String, String> substitution = new HashMap<>();
// substitution.put("node", Node.getInstance().getUri());
String query = this.queries.getQueries().get("get_interests");
public Set<Entry<String, String>> listChildrenAnnouncedProperties(){
Set<Entry<String, String>> result = new HashSet<>();
String query = this.queries.getQueries().get("get_children_announced_properties");
for(Map<String, String> row : Node.getInstance().getKBManager().queryModel(query))
{
result.add(row.get("interest"));
result.add(new AbstractMap.SimpleEntry<String, String>(row.get("predicate"), row.get("value")));
}
return result;
}
public Map<String, Set<String>> listRemoteInterestsByNodes(){
public Map<String, Set<String>> listProxiedPropertiesByNodes(){
Map<String, Set<String>> result = new HashMap<>();
String query = this.queries.getQueries().get("get_remote_interests");
String query = this.queries.getQueries().get("get_proxied_properties");
for(Map<String, String> row : Node.getInstance().getKBManager().queryModel(query))
{
if(!result.containsKey(row.get("node"))){
result.put(row.get("node"), new HashSet<>());
}
result.get(row.get("node")).add(row.get("interest"));
result.get(row.get("node")).add(row.get("property"));
}
return result;
}
......
......@@ -20,6 +20,7 @@ import org.apache.logging.log4j.Marker;
import org.topbraid.shacl.util.ModelPrinter;
import fr.irit.melodi.edr.configuration.E_Centralization;
import fr.irit.melodi.edr.errors.UnimplementedFeatureException;
import fr.irit.melodi.edr.node.Node;
import fr.irit.melodi.edr.out.resources.RulePropagationClient;
import fr.irit.melodi.sparql.exceptions.IncompleteSubstitutionException;
......@@ -52,7 +53,7 @@ public class RuleManager {
}
}
public void updateRule(Rule r) {
public void updateRule(Rule r) throws UnimplementedFeatureException {
if (E_Centralization.rulePropagationAllowed(Node.getInstance().getConfiguration().getCentralization())) {
Boolean aggregates = this.isAggregator(r);
if (aggregates != null) {
......@@ -105,7 +106,7 @@ public class RuleManager {
}
public void updateRules() {
public void updateRules() throws UnimplementedFeatureException {
for (String ruleURI : this.getAllRulesRef()) {
Rule r = extractRule(ruleURI);
this.updateRule(r);
......@@ -222,8 +223,9 @@ public class RuleManager {
* @param rule
* A RDF/TTL description of the rule, representing a shacl
* CONSTRUCT
* @throws UnimplementedFeatureException
*/
public void addRule(String rule) {
public void addRule(String rule) throws UnimplementedFeatureException {
StringReader sr = new StringReader(rule);
Model m = ModelFactory.createDefaultModel();
m.read(sr, "http://example.com/ns#", "TTL");
......@@ -242,7 +244,7 @@ public class RuleManager {
// Node.getInstance().getShapesManager().deactivateNodeSensitiveElements();
}
public Model triggerReasoning(Boolean activateNodeElements, Boolean deactivateNodeElements){
public Model triggerReasoning(Boolean activateNodeElements, Boolean deactivateNodeElements) throws UnimplementedFeatureException{
LOGGER.trace("Starting a reasoning step");
Model inference = Node.getInstance().getKBManager().executeRules(activateNodeElements, deactivateNodeElements);
if(inference.size() > 0){
......@@ -257,7 +259,7 @@ public class RuleManager {
}
// Deductions made based on observations
public void propagateRules(Model inference){
public void propagateRules(Model inference) throws UnimplementedFeatureException{
Map<String, List<String>> transfers = Node.getInstance()
.getShapesManager().extractTransferNodes(inference);
Boolean activatedRule = false;
......
......@@ -22,6 +22,7 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import fr.irit.melodi.edr.configuration.E_Centralization;
import fr.irit.melodi.edr.configuration.NodeConfiguration;
import fr.irit.melodi.edr.errors.UnimplementedFeatureException;
import fr.irit.melodi.edr.in.server.RESTServer;
import fr.irit.melodi.edr.model.DataEnricher;
import fr.irit.melodi.edr.model.DeductionManager;
......@@ -208,7 +209,7 @@ public class Node {
this.rm.updateAllDownStreamNodes();
}
public synchronized void addDownstreamNode(String remoteDescription){
public synchronized void addDownstreamNode(String remoteDescription) throws UnimplementedFeatureException{
Model remoteModel = ModelFactory.createDefaultModel();
StringReader sr = new StringReader(remoteDescription);
remoteModel.read(sr, Node.BASE_URI, "TTL");
......@@ -243,7 +244,7 @@ public class Node {
this.setAnnounced(true);
}
public void subscribe(){
public void subscribe() throws UnimplementedFeatureException{
for(Map<String, String> sensor : this.config.getSensors()){
String propertyInstance = Node.BASE_URI+Node.getInstance().getName()+sensor.get("produces");
LOGGER.debug(sensor.toString());
......@@ -268,7 +269,7 @@ public class Node {
}
}
public void processNewRawData(String data){
public void processNewRawData(String data) throws UnimplementedFeatureException{
// Data is dropped while the node isn't ready
if(this.getConfigured()){
Model enrichedData = this.enricher.enrichData(data);
......@@ -276,7 +277,7 @@ public class Node {
}
}
public void processNewData(Model data){
public void processNewData(Model data) throws UnimplementedFeatureException{
logModel(data);
if(this.getDataEnricher().isObservation(data)){
// Node.logTrace("Received observation,"+this.enricher.getObservation(data)+",");
......@@ -402,7 +403,7 @@ public class Node {
DataClient.sendData(endpoint, serializedEnrichedData);
}
public void reasonOverNewData(Model data){
public void reasonOverNewData(Model data) throws UnimplementedFeatureException{
LOGGER.debug("Reasoning over new data...");
Tracer.reasoningStarts(this.enricher.getObservation(data));
Model inference = Node.getInstance().getRuleManager().triggerReasoning(false, false);
......@@ -427,7 +428,7 @@ public class Node {
}
}
public void processUpstreamMessage(Model message){
public void processUpstreamMessage(Model message) throws UnimplementedFeatureException{
this.getKBManager().addToModel(message);
this.im.propagateInterests();
}
......@@ -438,7 +439,7 @@ public class Node {
LOGFILE.info(sw.toString());
}
public static int main(String[] args) {
public static int main(String[] args) throws UnimplementedFeatureException {
LOGGER.debug("Starting node");
org.apache.logging.log4j.core.LoggerContext ctx =
(org.apache.logging.log4j.core.LoggerContext) LogManager.getContext(false);
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment