//设置package包名称以及导入依赖的类
package com.newegg.eims.DataPorter.HDFS

import java.io.File

import com.newegg.eims.DataPorter.HDFS.Converts._
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path
import org.apache.hadoop.security.UserGroupInformation
import org.scalatest.{FlatSpec, Matchers}


class ConvertsSpec extends FlatSpec with Matchers {
  val currentDir = new File(".").getCanonicalPath + File.separator + "target" + File.separator

  "hdfs" should "can do with hadoop user" in {
    var test = 1
    new File(currentDir + "test.txt").doAs(UserGroupInformation.createRemoteUser("vq83"), f => test = 2)
    test should be(2)
  }

  it should "no copy file when not exists file path" in {
    new File(currentDir + "test.txt").copyToHDFS(new Path(currentDir + "test1.txt"), new Configuration())
    new File(currentDir + "test1.txt").exists() shouldBe false
  }

  it should "copy file when exists file path" in {
    val f = new File(currentDir + "test.txt")
    f.createNewFile()
    f.exists() shouldBe true
    new File(currentDir + "test.txt").copyToHDFS(new Path(currentDir + "test1.txt"), new Configuration())
    val f2 = new File(currentDir + "test1.txt")
    f2.exists() shouldBe true
    f.delete() shouldBe true
    f2.delete() shouldBe true
  }
} 

//设置package包名称以及导入依赖的类
package im.yanchen.krb5.auth.passwd

import javax.security.auth.Subject
import javax.security.auth.login.AppConfigurationEntry.LoginModuleControlFlag
import javax.security.auth.login.{AppConfigurationEntry, LoginContext}

import org.apache.hadoop.security.UserGroupInformation
import org.apache.hadoop.security.UserGroupInformation.AuthenticationMethod
import org.apache.hadoop.security.authentication.util.KerberosUtil

import scala.collection.JavaConverters._
import scala.collection.immutable.HashMap


object UserGroupInformationLoginUtil {
  def loginUserFromPasswordAndReturnUGI(user: String, password: String): UserGroupInformation = {
    this.synchronized {
      if (!UserGroupInformation.isSecurityEnabled) {
        UserGroupInformation.getCurrentUser
      } else {
        val krbOptions = HashMap[String, String](
          "doNotPrompt" -> "false",
          "useTicketCache" -> "false",
          "useKeyTab" -> "false",
          "renewTGT" -> "false"
        ).asJava
        val ace = new AppConfigurationEntry(
          KerberosUtil.getKrb5LoginModuleName,
          LoginModuleControlFlag.REQUIRED,
          krbOptions)
        val dynConf = new DynamicConfiguration(Array[AppConfigurationEntry](ace))
        val loginContext = new LoginContext(
          "hadoop-password-kerberos",
          null,
          new LoginHandler(user, password),
          dynConf)
        loginContext.login()

        val loginSubject = loginContext.getSubject
        val loginPrincipals = loginSubject.getPrincipals

        val subject = new Subject()
        subject.getPrincipals().addAll(loginPrincipals)

        val newLoginUser = UserGroupInformation.getUGIFromSubject(subject)
        setUGILogin(newLoginUser, loginContext)
        newLoginUser.setAuthenticationMethod(AuthenticationMethod.KERBEROS)
        newLoginUser
      }
    }
  }

  private def setUGILogin(loginUser: UserGroupInformation, loginContext: LoginContext): Unit = {
    val cls = classOf[UserGroupInformation]
    val mtd = cls.getDeclaredMethod("setLogin", classOf[LoginContext])
    mtd.setAccessible(true)
    mtd.invoke(loginUser, loginContext)
  }
} 

//设置package包名称以及导入依赖的类
package services

import java.security.PrivilegedAction

import models.Tables.GlobalSetting
import org.apache.hadoop.security.SaslRpcServer.AuthMethod
import org.apache.hadoop.security.UserGroupInformation
import org.apache.pig.impl.logicalLayer.FrontendException
import org.apache.pig.{ExecType, PigServer}
import play.api.Logger
import qrygraph.shared.compilation.{GraphFlatten, QueryCompiler}
import qrygraph.shared.data._

import scala.collection.JavaConversions._
import scala.concurrent.{Future, Promise}



  def generateExamples(settings: GlobalSetting, qrygraph: PigQueryGraph): Future[Map[String, List[List[String]]]] = {
    Logger.info(s"Using setting for execution of Query: $settings")

    val promise = Promise[Map[String, List[List[String]]]]()
    UserGroupInformation.createRemoteUser(settings.hadoopUser, AuthMethod.SIMPLE).doAs(new PrivilegedAction[Unit] {
      override def run(): Unit = {
        Logger.error("Starting example generation")
        // create an empty map to store the resultTypes in
        var map = Map[String, List[List[String]]]()
        // create a pig server
        val pigServer = new PigServer(ExecType.MAPREDUCE, PigProperties.generateProperties(settings))
        // add compiled Pig code to PigServer
        QueryCompiler
          .compile(qrygraph)
          .foreach(pigServer.registerQuery)
        // calc the type for each node
        for (node <- GraphFlatten.flattenDependency(qrygraph)) {
          try {
            // load data from pigServer

            val x = pigServer.openIterator(node.name).toStream.take(20).toList
            val resultFormat = x.map(_.toList.map(_.toString))
            map += node.name -> resultFormat

          } catch {
            case f: FrontendException =>
              println(f.getMessage)
              f.printStackTrace()
            case e: Throwable         =>
              println(e.getMessage)
              e.printStackTrace()
          }
        }
        pigServer.shutdown()
        promise.success(map)
      }
    })
    promise.future
  }

}