001/*
002 * Copyright 2015 DuraSpace, Inc.
003 *
004 * Licensed under the Apache License, Version 2.0 (the "License");
005 * you may not use this file except in compliance with the License.
006 * You may obtain a copy of the License at
007 *
008 *     http://www.apache.org/licenses/LICENSE-2.0
009 *
010 * Unless required by applicable law or agreed to in writing, software
011 * distributed under the License is distributed on an "AS IS" BASIS,
012 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
013 * See the License for the specific language governing permissions and
014 * limitations under the License.
015 */
016package org.fcrepo.http.commons.responses;
017
018import static javax.ws.rs.core.Response.Status.NOT_ACCEPTABLE;
019import static org.openrdf.model.impl.ValueFactoryImpl.getInstance;
020import static org.openrdf.model.util.Literals.createLiteral;
021import static org.slf4j.LoggerFactory.getLogger;
022
023import java.io.OutputStream;
024import java.util.Map;
025import java.util.function.Function;
026import java.util.stream.Stream;
027
028import javax.ws.rs.WebApplicationException;
029import javax.ws.rs.core.MediaType;
030import javax.ws.rs.core.StreamingOutput;
031
032import org.fcrepo.kernel.api.exception.RepositoryRuntimeException;
033import org.fcrepo.kernel.api.RdfStream;
034
035import org.openrdf.model.Resource;
036import org.openrdf.model.Statement;
037import org.openrdf.model.URI;
038import org.openrdf.model.Value;
039import org.openrdf.model.ValueFactory;
040import org.openrdf.rio.RDFFormat;
041import org.openrdf.rio.RDFHandlerException;
042import org.openrdf.rio.RDFWriter;
043import org.openrdf.rio.RDFWriterRegistry;
044import org.openrdf.rio.Rio;
045import org.openrdf.rio.WriterConfig;
046import org.slf4j.Logger;
047
048import com.google.common.util.concurrent.AbstractFuture;
049import com.hp.hpl.jena.graph.Node;
050import com.hp.hpl.jena.graph.Triple;
051
052/**
053 * Serializes an {@link RdfStream}.
054 *
055 * @author ajs6f
056 * @since Oct 30, 2013
057 */
058public class RdfStreamStreamingOutput extends AbstractFuture<Void> implements
059        StreamingOutput {
060
061    private static final Logger LOGGER =
062        getLogger(RdfStreamStreamingOutput.class);
063
064    private static ValueFactory vfactory = getInstance();
065
066    /**
067     * This field is used to determine the correct {@link org.openrdf.rio.RDFWriter} created for the
068     * {@link javax.ws.rs.core.StreamingOutput}.
069     */
070    private final RDFFormat format;
071
072    /**
073     * This field is used to determine the {@link org.openrdf.rio.WriterConfig} details used by the created
074     * {@link org.openrdf.rio.RDFWriter}.
075     */
076    private final MediaType mediaType;
077
078    private final RdfStream rdfStream;
079
080    private final Map<String, String> namespaces;
081
082    /**
083     * Normal constructor
084     *
085     * @param rdfStream the rdf stream
086     * @param namespaces a namespace mapping
087     * @param mediaType the media type
088     */
089    public RdfStreamStreamingOutput(final RdfStream rdfStream, final Map<String, String> namespaces,
090            final MediaType mediaType) {
091        super();
092
093        if (LOGGER.isDebugEnabled()) {
094            for (final RDFFormat writeableFormats : RDFWriterRegistry.getInstance().getKeys()) {
095                LOGGER.debug("Discovered RDF writer writeableFormats: {} with mimeTypes: {}",
096                        writeableFormats.getName(), String.join(" ", writeableFormats.getMIMETypes()));
097            }
098        }
099        final RDFFormat format = Rio.getWriterFormatForMIMEType(mediaType.toString());
100        if (format != null) {
101            this.format = format;
102            this.mediaType = mediaType;
103            LOGGER.debug("Setting up to serialize to: {}", format);
104        } else {
105            throw new WebApplicationException(NOT_ACCEPTABLE);
106        }
107
108        this.rdfStream = rdfStream;
109        this.namespaces = namespaces;
110    }
111
112    @Override
113    public void write(final OutputStream output) {
114        LOGGER.debug("Serializing RDF stream in: {}", format);
115        try {
116            write(rdfStream.map(toStatement), output, format, mediaType, namespaces);
117        } catch (final RDFHandlerException e) {
118            setException(e);
119            LOGGER.debug("Error serializing RDF", e);
120            throw new WebApplicationException(e);
121        }
122    }
123
124    private static void write(final Stream<Statement> model,
125                       final OutputStream output,
126                       final RDFFormat dataFormat,
127                       final MediaType dataMediaType,
128                       final Map<String, String> nsPrefixes)
129            throws RDFHandlerException {
130        final WriterConfig settings = WriterConfigHelper.apply(dataMediaType);
131        final RDFWriter writer = Rio.createWriter(dataFormat, output);
132        writer.setWriterConfig(settings);
133
134        /**
135         * We exclude:
136         *  - xmlns, which Sesame helpfully serializes, but normal parsers may complain
137         *     about in some serializations (e.g. RDF/XML where xmlns:xmlns is forbidden by XML);
138         */
139        nsPrefixes.entrySet().stream().filter(e -> !e.getKey().equals("xmlns"))
140                .forEach(x -> {
141                    try {
142                        writer.handleNamespace(x.getKey(), x.getValue());
143                    } catch (final RDFHandlerException e) {
144                        throw new RepositoryRuntimeException(e);
145                    }
146                });
147
148        Rio.write((Iterable<Statement>)model::iterator, writer);
149    }
150
151    protected static final Function<? super Triple, Statement> toStatement = t -> {
152        final Resource subject = getResourceForSubject(t.getSubject());
153        final URI predicate = vfactory.createURI(t.getPredicate().getURI());
154        final Value object = getValueForObject(t.getObject());
155        return vfactory.createStatement(subject, predicate, object);
156    };
157
158    private static Resource getResourceForSubject(final Node subjectNode) {
159        return subjectNode.isBlank() ? vfactory.createBNode(subjectNode.getBlankNodeLabel())
160                : vfactory.createURI(subjectNode.getURI());
161    }
162
163    protected static Value getValueForObject(final Node object) {
164        if (object.isURI()) {
165            return vfactory.createURI(object.getURI());
166        } else if (object.isBlank()) {
167            return vfactory.createBNode(object.getBlankNodeLabel());
168        } else if (object.isLiteral()) {
169            final String literalValue = object.getLiteralLexicalForm();
170
171            final String literalDatatypeURI = object.getLiteralDatatypeURI();
172
173            if (!object.getLiteralLanguage().isEmpty()) {
174                return vfactory.createLiteral(literalValue, object.getLiteralLanguage());
175            } else if (literalDatatypeURI != null) {
176                final URI uri = vfactory.createURI(literalDatatypeURI);
177                return vfactory.createLiteral(literalValue, uri);
178            } else {
179                return createLiteral(vfactory, object.getLiteralValue());
180            }
181        }
182        throw new AssertionError("Unable to convert " + object +
183                " to a value, it is neither URI, blank, nor literal!");
184    }
185}