package eu.dnetlib.dhp.application

import scala.io.Source

/**
 * This is the main Interface SparkApplication
 * where all the Spark Scala class should inherit
 *
 */
trait SparkScalaApplication {
  /**
   * This is the path in the classpath of the json
   * describes all the argument needed to run
   */
  val propertyPath: String

  /**
   * Utility to parse the arguments using the
   * property json in the classpath identified from
   * the variable propertyPath
   *
   * @param args the list of arguments
   */
  def parseArguments(args: Array[String]): ArgumentApplicationParser = {
    val parser = new ArgumentApplicationParser(Source.fromInputStream(getClass.getResourceAsStream(propertyPath)).mkString)
    parser.parseArgument(args)
    parser
  }

  /**
   * Here all the spark applications runs this method
   * where the whole logic of the spark node is defined
   */
  def run(): Unit
}


import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
import org.slf4j.Logger

abstract class AbstractScalaApplication (val propertyPath:String, val args:Array[String], log:Logger) extends  SparkScalaApplication {

  var parser: ArgumentApplicationParser = null

  var spark:SparkSession = null


  def initialize():SparkScalaApplication = {
    parser = parseArguments(args)
    spark = createSparkSession()
    this
  }

  /**
   * Utility for creating a spark session starting from parser
   *
   * @return a spark Session
   */
  private def createSparkSession():SparkSession = {
    require(parser!= null)

    val conf:SparkConf = new SparkConf()
    val master = parser.get("master")
    log.info(s"Creating Spark session: Master: $master")
    SparkSession.builder().config(conf)
      .appName(getClass.getSimpleName)
      .master(master)
      .getOrCreate()
  }

}