package uk.ac.abdn.foodsafety.provenance;

import java.io.InputStream;
import java.text.ParseException;
import java.time.ZonedDateTime;
import java.util.Scanner;
import java.util.function.Consumer;
import java.util.function.Function;

import com.hp.hpl.jena.rdf.model.Model;
import com.hp.hpl.jena.rdf.model.RDFNode;
import com.hp.hpl.jena.rdf.model.Statement;
import com.hp.hpl.jena.rdf.model.StmtIterator;

import uk.ac.abdn.foodsafety.common.FoodSafetyException;
import eu.larkc.csparql.cep.api.RdfQuadruple;
import eu.larkc.csparql.cep.api.RdfStream;
import eu.larkc.csparql.core.engine.CsparqlEngineImpl;

/**
 * 
 * @author nhc
 *
 * A FoodSafetyEngine is a specific CsparqlEngine with methods
 * and queries for this project.
 */
public final class FoodSafetyEngine
    extends CsparqlEngineImpl 
    implements Function<ZonedDateTime, Consumer<Model>> {
    private final RdfStream rdfStream = new RdfStream("http://foodsafety/ssn");
    
    /**
     * Initializes this engine.
     */
    public FoodSafetyEngine() {
        this.initialize();
        this.registerStream(this.rdfStream);
        this.registerQueryFromResources("/window.sparql.txt");
    }
        
    /**
     * Registers a SPARQL query read from /src/main/resources
     * The query must be encoded in UTF-8.
     * A ConsoleFormatter will be added as observer to the query.
     * @param path The path to the query, relative to FoodSafety/src/main/resources,
     * example: "/myquery.sparql.txt"
     */
    private void registerQueryFromResources(final String path) {
        //Get InputStream for the file
        final InputStream resourceAsStream = FoodSafetyEngine.class.getResourceAsStream(path);
        Scanner scanner = null;
        try {
            //Read entire file as UTF-8 into String
            scanner = new Scanner(resourceAsStream, "UTF-8");
            final String text = scanner.useDelimiter("\\A").next();
            //Register query and add observer
            this.registerQuery(text, false);//.addObserver(formatter);
        } catch (final ParseException e) {
            throw FoodSafetyException.internalError(e);
        } finally { //Close Scanner
            if (scanner != null) {
                scanner.close();
            }
        }
        
    }
    
    @Override
    public Consumer<Model> apply(final ZonedDateTime t) {
        return (m -> this.add(t, m));
    }

    private void add(final ZonedDateTime t, final Model m) {
        final long timestamp = t.toInstant().toEpochMilli();
        final StmtIterator it = m.listStatements();
        while (it.hasNext()) {
            final Statement triple = it.nextStatement();
            final RDFNode o = triple.getObject();
            if (o.isAnon()) {
                FoodSafetyException.internalError(String.format("Blank node in %s", m.toString()));
            };
            this.rdfStream.put(new RdfQuadruple(
                    triple.getSubject().getURI(),
                    triple.getPredicate().getURI(),
                    (o.isResource()) ? o.asResource().getURI() : o.asLiteral().getLexicalForm(),
                    timestamp));
        }
    }
}

@@ -129,16 +129,45 @@
 teTime;%0A
+import java.util.Observable;%0A
 import j
@@ -180,24 +180,24 @@
 il.Scanner;%0A
-
 import java.
@@ -484,24 +484,69 @@
 yException;%0A
+import uk.ac.abdn.foodsafety.common.Logging;%0A
 import eu.la
@@ -618,24 +618,111 @@
 .RdfStream;%0A
+import eu.larkc.csparql.common.RDFTable;%0Aimport eu.larkc.csparql.core.ResultFormatter;%0A
 import eu.la
@@ -1016,16 +1016,58 @@
 del%3E%3E %7B%0A
+    %0A    /** This engine's sole stream */%0A
     priv
@@ -2203,11 +2203,8 @@
 lse)
-;//
 .add
@@ -2216,17 +2216,24 @@
 ver(
-f
+new TmpF
 ormatter
 );%0A 
@@ -2228,16 +2228,18 @@
 ormatter
+()
 );%0A     
@@ -3331,24 +3331,24 @@
 imestamp));%0A
-
         %7D%0A  
@@ -3351,10 +3351,419 @@
 %7D%0A    %7D%0A
+    %0A    private static class TmpFormatter extends ResultFormatter %7B%0A        /**%0A         * Called when C-Sparql emits a window.%0A         */%0A        @Override%0A        public void update(final Observable ignored, final Object rdfTableUntyped) %7B%0A            final RDFTable rdfTable = (RDFTable) rdfTableUntyped;%0A            Logging.info(String.format(%22Emitted %25d triples%22, rdfTable.size()));%0A        %7D   %0A    %7D%0A
 %7D%0A

Add basic logging to C-SPARQL query