/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to you under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.clerezza.rdf.rdfjson.serializer;

import java.io.BufferedWriter;
import java.io.IOException;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.io.UnsupportedEncodingException;
import java.util.Arrays;
import java.util.Comparator;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.clerezza.commons.rdf.BlankNode;
import org.apache.clerezza.commons.rdf.BlankNodeOrIRI;
import org.apache.clerezza.commons.rdf.RDFTerm;
import org.apache.clerezza.commons.rdf.Triple;
import org.apache.clerezza.commons.rdf.Graph;
import org.apache.clerezza.commons.rdf.IRI;
import org.apache.clerezza.commons.rdf.Literal;
import org.apache.clerezza.rdf.core.serializedform.SerializingProvider;
import org.apache.clerezza.rdf.core.serializedform.SupportedFormat;
import org.apache.felix.scr.annotations.Component;
import org.apache.felix.scr.annotations.Service;
import org.json.simple.JSONArray;
import org.json.simple.JSONObject;

/**
 * A {@link org.apache.clerezza.rdf.core.serializedform.SerializingProvider} for
 * rdf/json.
 * 
 * This implementation is based on first sorting the triples within the parsed
 * {@link Graph} based on the {@link #SUBJECT_COMPARATOR subject}.
 * <p>
 * The serialization is done on a subject scope. Meaning that all triples for a
 * subject are serialized and instantly written to the provided
 * {@link OutputStream}.
 * <p>
 * 'UFT-8' is used as encoding to write the data.
 * 
 * @author tio, hasan, rwesten
 */
@Component(immediate = true)
@Service(SerializingProvider.class)
@SupportedFormat(SupportedFormat.RDF_JSON)
public class RdfJsonSerializingProvider implements SerializingProvider {

    @SuppressWarnings("unchecked")
    @Override
    public void serialize(OutputStream serializedGraph, Graph tc,
            String formatIdentifier) {
        if (tc.isEmpty()) { // ensure writing an empty element in case of an
                            // empty collection
            try {
                serializedGraph.write(new JSONObject().toJSONString().getBytes(
                        "UTF-8"));
            } catch (IOException e) {
                throw new IllegalStateException(
                        "Exception while writing to parsed OutputStream", e);
            }
            return;
        }
        BlankNodeManager bNodeMgr = new BlankNodeManager();
        BufferedWriter out;
        try {
            out = new BufferedWriter(new OutputStreamWriter(serializedGraph,
                    "UTF-8"));
        } catch (UnsupportedEncodingException e) {
            throw new IllegalStateException(
                    "Encoding 'UTF-8' is not supported by this System", e);
        }
        Triple[] sortedTriples = tc.toArray(new Triple[tc.size()]);
        Arrays.sort(sortedTriples, SUBJECT_COMPARATOR);
        Triple triple;
        BlankNodeOrIRI subject = null;
        String subjectStr = null;
        IRI predicate = null;
        Map<IRI, JSONArray> predicateValues = new HashMap<IRI, JSONArray>();
        JSONObject jSubject = new JSONObject();
        try {
            out.write("{"); // start the root object
            for (int i = 0; i < sortedTriples.length; i++) {
                triple = sortedTriples[i];
                boolean subjectChange = !triple.getSubject().equals(subject);
                if (subjectChange) {
                    if (subject != null) {
                        // write the predicate values
                        for (Entry<IRI, JSONArray> predicates : predicateValues
                                .entrySet()) {
                            jSubject.put(
                                    predicates.getKey().getUnicodeString(),
                                    predicates.getValue());
                        }
                        // write subject
                        out.write(JSONObject.toString(subjectStr, jSubject));
                        out.write(",");
                        jSubject.clear(); // just clear
                        predicateValues.clear();
                    }
                    // init next subject
                    subject = triple.getSubject();
                    if (subject instanceof BlankNode) {
                        subjectStr = bNodeMgr.getBlankNodeId((BlankNode) subject);
                    } else { // if (subject instanceof IRI)
                        subjectStr = ((IRI) subject).getUnicodeString();
                    }
                }
                predicate = triple.getPredicate();
                JSONArray values = predicateValues.get(predicate);
                if (values == null) {
                    values = new JSONArray();
                    predicateValues.put(predicate, values);
                }
                values.add(writeObject(bNodeMgr, triple.getObject()));
            }
            if (subjectStr != null) {
                for (Entry<IRI, JSONArray> predicates : predicateValues
                        .entrySet()) {
                    jSubject.put(predicates.getKey().getUnicodeString(),
                            predicates.getValue());
                }
                out.write(JSONObject.toString(subjectStr, jSubject));
            }
            out.write("}");// end the root object
            out.flush();
        } catch (IOException e) {
            throw new IllegalStateException(
                    "Exception while writing on the parsed OutputStream", e);
        }
    }

    private class BlankNodeManager {
        private Map<BlankNode, String> bNodeMap = new HashMap<BlankNode, String>();
        private int counter = 0;

        public String getBlankNodeId(BlankNode node) {
            String bNodeId = bNodeMap.get(node);
            if (bNodeId == null) {
                bNodeId = "_:b" + ++counter;
                bNodeMap.put((BlankNode) node, bNodeId);
            }
            return bNodeId;
        }
    }

    /**
     * Converts the {@link RDFTerm object} of an triple to JSON
     * 
     * @param bNodeMgr
     *            used to lookup {@link BlankNode} instances
     * @param object
     *            the object of the triple
     * @return the JSON representation of parsed object
     */
    @SuppressWarnings("unchecked")
    private JSONObject writeObject(BlankNodeManager bNodeMgr, RDFTerm object) {
        JSONObject jObject = new JSONObject();
        if (object instanceof Literal) {
            Literal literal = (Literal) object;
            jObject.put("value", literal.getLexicalForm());
            jObject.put("type", "literal");
            jObject.put("datatype", literal.getDataType().getUnicodeString());
            if (literal.getLanguage() != null) {
                jObject.put("lang", literal.getLanguage().toString());
            }
        } else if (object instanceof IRI) {
            IRI uriRef = (IRI) object;
            jObject.put("value", uriRef.getUnicodeString());
            jObject.put("type", "uri");
        } else if (object instanceof BlankNode) {
            String bNodeId = bNodeMgr.getBlankNodeId((BlankNode) object);
            jObject.put("value", bNodeId);
            jObject.put("type", "bnode");
        }
        return jObject;
    }

    /**
     * Compares only the subjects of the triples. If they are equals
     * <code>0</code> is returned. This will ensure that all triples with the
     * same subjects are sorted correctly. However it does not sort predicates
     * and objects!
     */
    public static final Comparator<Triple> SUBJECT_COMPARATOR = new Comparator<Triple>() {

        @Override
        public int compare(Triple a, Triple b) {
            return compare(a.getSubject(), b.getSubject());
        }

        private int compare(BlankNodeOrIRI a, BlankNodeOrIRI b) {
            int hashA = a.hashCode();
            int hashB = b.hashCode();
            if (hashA != hashB) {
                return hashA > hashB ? 1 : -1;
            }
            return a.toString().compareTo(b.toString());
        }

    };
}