001/*
002 * The contents of this file are subject to the license and copyright
003 * detailed in the LICENSE and NOTICE files at the root of the source
004 * tree.
005 */
006package org.fcrepo.kernel.api.utils;
007
008import java.util.stream.Stream;
009
010import org.apache.jena.graph.Graph;
011import org.apache.jena.graph.Triple;
012import org.apache.jena.rdf.model.Model;
013
014import static java.util.Spliterator.IMMUTABLE;
015import static java.util.Spliterators.spliteratorUnknownSize;
016import static java.util.stream.StreamSupport.stream;
017import static org.apache.jena.graph.Node.ANY;
018import static org.apache.jena.sparql.graph.GraphFactory.createDefaultGraph;
019
020/**
021 * A wrapping {@link Stream} that calculates two differences between a
022 * {@link Graph} A and a source Stream B. The differences are (A - (A ∩ B)) and
023 * (B - (A ∩ B)). The ordinary output of this stream is (B - (A ∩ B)), and
024 * after exhaustion, sets containing (A - (A ∩ B)) and (A ∩ B) are available.
025 *
026 * @author ajs6f
027 * @author acoburn
028 * @since Oct 24, 2013
029 */
030public class GraphDifferencer {
031
032    private final Graph notCommon;
033
034    private final Graph common;
035
036    private final Stream.Builder<Triple> source = Stream.builder();
037
038    /**
039     * Diff a Model against a stream of triples
040     *
041     * @param replacement the replacement
042     * @param original the original
043     */
044    public GraphDifferencer(final Model replacement,
045                                     final Stream<Triple> original) {
046        this(replacement.getGraph(), original);
047    }
048
049    /**
050     * Diff a graph against a stream of triples
051     *
052     * @param replacement the replacement
053     * @param original the original
054     */
055    public GraphDifferencer(final Graph replacement,
056                                     final Stream<Triple> original) {
057        notCommon = replacement;
058        common = createDefaultGraph();
059        original.forEach(x -> {
060            synchronized (this) {
061                if (notCommon.contains(x)) {
062                    notCommon.remove(x.getSubject(), x.getPredicate(), x.getObject());
063                    common.add(x);
064                } else if (!common.contains(x)) {
065                    source.accept(x);
066                }
067            }
068        });
069    }
070
071    /**
072     * This method returns the difference between the two input sources.
073     *
074     * @return The differences between the two inputs.
075     */
076    public Stream<Triple> difference() {
077        return source.build();
078    }
079
080    /**
081     * This method will return null until the source iterator is exhausted.
082     *
083     * @return The elements that turned out to be common to the two inputs.
084     */
085    public Stream<Triple> common() {
086        return stream(spliteratorUnknownSize(common.find(ANY, ANY, ANY), IMMUTABLE), false);
087    }
088
089    /**
090     * This method will return null until the source iterator is exhausted.
091     *
092     * @return The elements that turned out not to be common to the two inputs.
093     */
094    public Stream<Triple> notCommon() {
095        return stream(spliteratorUnknownSize(notCommon.find(ANY, ANY, ANY), IMMUTABLE), false);
096    }
097}