/*
 * 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.juneau.marshall.jena;


import org.apache.juneau.commons.bean.*;
import org.apache.juneau.commons.collections.*;
import org.apache.juneau.marshall.*;
import org.apache.juneau.marshall.parser.*;
import org.apache.juneau.marshall.xml.*;
import static org.apache.juneau.commons.utils.Shorts.*;

/**
 * Stream-based RDF parser for binary formats (RDF/THRIFT, RDF/PROTO).
 *
 * <p>
 * Extends {@link InputStreamParser} and delegates RDF metadata to an internal {@link RdfParser}.
 *
 * <h5 class='section'>Notes:</h5><ul>
 *    <li class='note'>This class is thread safe and reusable.
 * </ul>
 *
 * <h5 class='section'>See Also:</h5><ul>
 *    <li class='link'>{doc jmr.RdfDetails}
 * </ul>
 */
@SuppressWarnings({
   "java:S110", // Deep inheritance inherent to the RDF parser hierarchy.
   "java:S1192" // Duplicated literals (argument/property names) read more clearly inline than as constants
})
public class RdfStreamParser extends InputStreamParser implements RdfMetaProvider {

   /**
    * Builder class.
    */
   @SuppressWarnings({
      "java:S119" // 'SELF' (CRTP self-type) is intentional and clearer than a single-letter name.
   })
   public abstract static class Builder<SELF extends Builder<SELF>> extends InputStreamParser.Builder<SELF> {

      private static final Cache<HashKey,RdfStreamParser> CACHE = Cache.of(HashKey.class, RdfStreamParser.class).build();

      private String language;
      private RdfParser rdfParser;

      /**
       * Constructor, default settings.
       */
      protected Builder() {
         language = Constants.LANG_RDFTHRIFT;
      }

      /**
       * Copy constructor.
       *
       * @param copyFrom The builder to copy from.
       *    <br>Cannot be <jk>null</jk>.
       */
      protected Builder(Builder<?> copyFrom) {
         super(reqnn("copyFrom", copyFrom));
         language = copyFrom.language;
         rdfParser = copyFrom.rdfParser;
      }

      /**
       * Copy constructor.
       *
       * @param copyFrom The bean to copy from.
       *    <br>Cannot be <jk>null</jk>.
       */
      protected Builder(RdfStreamParser copyFrom) {
         super(reqnn("copyFrom", copyFrom));
         language = copyFrom.language;
         rdfParser = copyFrom.rdfParser;
      }

      /**
       * RDF language (e.g. {@link Constants#LANG_RDFTHRIFT}, {@link Constants#LANG_RDFPROTO}).
       *
       * @param value The RDF language.
       * @return This object.
       */
      public SELF language(String value) {
         language = value;
         rdfParser = null; // Rebuild on next get
         return self();
      }

      @Override
      public RdfStreamParser build() {
         return cache(CACHE).build(RdfStreamParser.class);
      }

      @Override
      public SELF consumes(String value) {
         super.consumes(value);
         return self();
      }

      @Override /* Overridden from Context.Builder<?> */
      public abstract SELF copy();

      @Override
      public HashKey hashKey() {
         return HashKey.of(super.hashKey(), language);
      }

      RdfParser getRdfParser() {
         if (rdfParser == null)
            rdfParser = RdfParser.create().language(language).build();
         return rdfParser;
      }
   }

   /**
    * Concrete default builder leaf for the non-subclassed {@link RdfStreamParser#create()} / {@link RdfStreamParser#copy()} path.
    */
   public static final class DefaultBuilder extends Builder<DefaultBuilder> {

      DefaultBuilder() {}

      DefaultBuilder(RdfStreamParser copyFrom) {
         super(copyFrom);
      }

      DefaultBuilder(Builder<?> copyFrom) {
         super(copyFrom);
      }

      @Override /* Overridden from Context.Builder<?> */
      public DefaultBuilder copy() {
         return new DefaultBuilder(this);
      }
   }

   /**
    * Creates a new builder.
    *
    * @return A new builder.
    */
   @SuppressWarnings({
      "java:S1452", // Builder<?> wildcard return intentional; callers chain via fluent API without needing the concrete type
      "java:S9149" // Public Juneau DSL factory; hiding parent create() is intentional.
   })
   public static Builder<?> create() {
      return new DefaultBuilder();
   }

   private final String language;
   private final RdfParser rdfParser;

   /**
    * Constructor.
    *
    * @param builder The builder for this object.
    *    <br>Cannot be <jk>null</jk>.
    */
   public RdfStreamParser(Builder<?> builder) {
      super(reqnn("builder", builder).consumes(getConsumes(builder)));
      language = builder.language;
      rdfParser = builder.getRdfParser();
   }

   private static String getConsumes(Builder<?> builder) {
      if (builder.getConsumes() != null)
         return builder.getConsumes();
      return switch (builder.language) {
         case Constants.LANG_RDFTHRIFT -> "application/vnd.apache.thrift.binary";
         case Constants.LANG_RDFPROTO -> "application/vnd.apache.protobuf";
         default -> "application/octet-stream";
      };
   }

   @Override
   public Builder<?> copy() {
      return new DefaultBuilder(this);
   }

   @Override
   public RdfStreamParserSession.Builder createSession() {
      return RdfStreamParserSession.create(this);
   }

   @Override
   public RdfBeanMeta getRdfBeanMeta(BeanMeta<?> bm) {
      return rdfParser.getRdfBeanMeta(bm);
   }

   @Override
   public RdfBeanPropertyMeta getRdfBeanPropertyMeta(BeanPropertyMeta bpm) {
      return rdfParser.getRdfBeanPropertyMeta(bpm);
   }

   @Override
   public RdfClassMeta getRdfClassMeta(ClassMeta<?> cm) {
      return rdfParser.getRdfClassMeta(cm);
   }

   @Override
   public XmlBeanMeta getXmlBeanMeta(BeanMeta<?> bm) {
      return rdfParser.getXmlBeanMeta(bm);
   }

   @Override
   public XmlBeanPropertyMeta getXmlBeanPropertyMeta(BeanPropertyMeta bpm) {
      return rdfParser.getXmlBeanPropertyMeta(bpm);
   }

   @Override
   public XmlClassMeta getXmlClassMeta(ClassMeta<?> cm) {
      return rdfParser.getXmlClassMeta(cm);
   }

   /**
    * Returns the RDF language.
    *
    * @return The RDF language.
    */
   public String getLanguage() {
      return language;
   }

   /**
    * Returns the delegated RdfParser.
    *
    * @return The delegated RdfParser.
    */
   protected RdfParser getRdfParser() {
      return rdfParser;
   }

   /** Collection format - delegates to RdfParser. */
   protected RdfCollectionFormat getCollectionFormat() {
      return rdfParser.getCollectionFormat();
   }

   /** Juneau namespace - delegates to RdfParser. */
   protected Namespace getJuneauNs() {
      return rdfParser.getJuneauNs();
   }

   /** Juneau BP namespace - delegates to RdfParser. */
   protected Namespace getJuneauBpNs() {
      return rdfParser.getJuneauBpNs();
   }

   /** Loose collections - delegates to RdfParser. */
   protected boolean isLooseCollections() {
      return rdfParser.isLooseCollections();
   }

   /** Trim whitespace - delegates to RdfParser. */
   protected boolean isTrimWhitespace() {
      return rdfParser.isTrimWhitespace();
   }
}