text:
package org.apache.zeppelin.flink

import java.io.File
import java.net.{URL, URLClassLoader}
import java.nio.file.Files
import java.util.Properties
import java.util.concurrent.TimeUnit
import java.util.jar.JarFile

import org.apache.commons.lang3.StringUtils
import org.apache.commons.lang3.exception.ExceptionUtils
import org.apache.flink.api.common.JobExecutionResult
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.client.program.ClusterClient
import org.apache.flink.configuration._
import org.apache.flink.core.execution.{JobClient, JobListener}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.environment.{StreamExecutionEnvironmentFactory, StreamExecutionEnvironment => JStreamExecutionEnvironment}
import org.apache.flink.api.java.{ExecutionEnvironmentFactory, ExecutionEnvironment => JExecutionEnvironment}
import org.apache.flink.runtime.jobgraph.SavepointConfigOptions
import org.apache.flink.runtime.util.EnvironmentInformation
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.config.ExecutionConfigOptions
import org.apache.flink.table.api.{EnvironmentSettings, TableConfig, TableEnvironment}
import org.apache.flink.table.catalog.hive.HiveCatalog
import org.apache.flink.table.functions.{AggregateFunction, ScalarFunction, TableAggregateFunction, TableFunction}
import org.apache.flink.table.module.hive.HiveModule
import org.apache.flink.yarn.cli.FlinkYarnSessionCli
import org.apache.zeppelin.dep.DependencyResolver
import org.apache.zeppelin.flink.internal.FlinkShell._
import org.apache.zeppelin.flink.internal.FlinkILoop
import org.apache.zeppelin.interpreter.Interpreter.FormType
import org.apache.zeppelin.interpreter.thrift.InterpreterCompletion
import org.apache.zeppelin.interpreter.util.InterpreterOutputStream
import org.apache.zeppelin.interpreter.{InterpreterContext, InterpreterException, InterpreterHookRegistry, InterpreterResult}
import org.slf4j.{Logger, LoggerFactory}

import scala.collection.JavaConversions
import scala.collection.JavaConverters._
import scala.tools.nsc.Settings
import scala.tools.nsc.interpreter.{Completion, IMain, IR, JPrintWriter, Results, SimpleReader}

/**
 * It instantiate flink scala shell and create env, senv, btenv, stenv.
 *
 * @param properties
 */
abstract class FlinkScalaInterpreter(val properties: Properties,
                                     val flinkScalaClassLoader: URLClassLoader) {

  private lazy val LOGGER: Logger = LoggerFactory.getLogger(getClass)

  protected var flinkILoop: FlinkILoop = _
  private var cluster: Option[ClusterClient[_]] = _

  protected var scalaCompletion: Completion = _
  private val interpreterOutput = new InterpreterOutputStream(LOGGER)
  private var configuration: Configuration = _

  private var mode: ExecutionMode.Value = _

  private var tblEnvFactory: TableEnvFactory = _
  private var benv: ExecutionEnvironment = _
  private var senv: StreamExecutionEnvironment = _

  // TableEnvironment of blink planner
  private var btenv: TableEnvironment = _
  private var stenv: TableEnvironment = _

  // TableEnvironment of flink planner
  private var btenv_2: TableEnvironment = _
  private var stenv_2: TableEnvironment = _

  // PyFlink depends on java version of TableEnvironment,
  // so need to create java version of TableEnvironment
  // java version of blink TableEnvironment
  private var java_btenv: TableEnvironment = _
  private var java_stenv: TableEnvironment = _

  // java version of flink TableEnvironment
  private var java_btenv_2: TableEnvironment = _
  private var java_stenv_2: TableEnvironment = _

  private var z: FlinkZeppelinContext = _
  private var flinkVersion: FlinkVersion = _
  private var flinkShims: FlinkShims = _
  // used for calling jm rest api
  private var jmWebUrl: String = _
  // used for displaying on zeppelin ui
  private var displayedJMWebUrl: String = _
  private var jobManager: JobManager = _
  private var defaultParallelism = 1
  private var defaultSqlParallelism = 1

  // flink.execution.jars + flink.execution.jars + flink.udf.jars
  protected var userJars: Seq[String] = _
  // flink.udf.jars
  private var userUdfJars: Seq[String] = _

  def open(): Unit = {

    val config = initFlinkConfig()
    createFlinkILoop(config)
    createTableEnvs()
    setTableEnvConfig()

    // init ZeppelinContext
    this.z = new FlinkZeppelinContext(this, new InterpreterHookRegistry(),
      Integer.parseInt(properties.getProperty("zeppelin.flink.maxResult", "1000")))
    this.bind("z", z.getClass().getCanonicalName(), z, List("@transient"))
    this.jobManager = new JobManager(jmWebUrl, displayedJMWebUrl, properties)

    // register JobListener
    val jobListener = new FlinkJobListener()
    this.benv.registerJobListener(jobListener)
    this.senv.registerJobListener(jobListener)

    // register hive catalog
    if (properties.getProperty("zeppelin.flink.enableHive", "false").toBoolean) {
      LOGGER.info("Hive is enabled, registering hive catalog.")
      registerHiveCatalog()
    } else {
      LOGGER.info("Hive is disabled.")
    }

    // load udf jar
    this.userUdfJars.foreach(jar => loadUDFJar(jar))

    if (mode == ExecutionMode.YARN_APPLICATION) {
      // have to call senv.execute method before running any user code, otherwise yarn application mode
      // will cause ClassNotFound issue. Needs to do more investigation. TODO(zjffdu)
      val initCode =
        """
          |val data = senv.fromElements("hello world", "hello flink", "hello hadoop")
          |data.flatMap(line => line.split("\\s"))
          |  .map(w => (w, 1))
          |  .keyBy(0)
          |  .sum(1)
          |  .print
          |
          |senv.execute()
          |""".stripMargin

      interpret(initCode, InterpreterContext.get())
      InterpreterContext.get().out.clear()
    }
  }

  def createIMain(settings: Settings, out: JPrintWriter): IMain

  def createSettings(): Settings

  private def initFlinkConfig(): Config = {

    this.flinkVersion = new FlinkVersion(EnvironmentInformation.getVersion)
    LOGGER.info("Using flink: " + flinkVersion)
    this.flinkShims = FlinkShims.getInstance(flinkVersion, properties)

    var flinkHome = sys.env.getOrElse("FLINK_HOME", "")
    var flinkConfDir = sys.env.getOrElse("FLINK_CONF_DIR", "")
    val hadoopConfDir = sys.env.getOrElse("HADOOP_CONF_DIR", "")
    val yarnConfDir = sys.env.getOrElse("YARN_CONF_DIR", "")
    var hiveConfDir = sys.env.getOrElse("HIVE_CONF_DIR", "")

    mode = ExecutionMode.withName(
      properties.getProperty("flink.execution.mode", "LOCAL")
        .replace("-", "_")
        .toUpperCase)
    if (mode == ExecutionMode.YARN_APPLICATION) {
      if (flinkVersion.isFlink110) {
        throw new Exception("yarn-application mode is only supported after Flink 1.11")
      }
      // use current yarn container working directory as FLINK_HOME, FLINK_CONF_DIR and HIVE_CONF_DIR
      val workingDirectory = new File(".").getAbsolutePath
      flinkHome = workingDirectory
      flinkConfDir = workingDirectory
      hiveConfDir = workingDirectory
    }
    LOGGER.info("FLINK_HOME: " + flinkHome)
    LOGGER.info("FLINK_CONF_DIR: " + flinkConfDir)
    LOGGER.info("HADOOP_CONF_DIR: " + hadoopConfDir)
    LOGGER.info("YARN_CONF_DIR: " + yarnConfDir)
    LOGGER.info("HIVE_CONF_DIR: " + hiveConfDir)

    this.configuration = GlobalConfiguration.loadConfiguration(flinkConfDir)
    var config = Config(executionMode = mode)
    val jmMemory = properties.getProperty("jobmanager.memory.process.size",
      properties.getProperty("flink.jm.memory", "1024"))
    config = config.copy(yarnConfig =
      Some(ensureYarnConfig(config)
        .copy(jobManagerMemory = Some(jmMemory))))

    val tmMemory = properties.getProperty("taskmanager.memory.process.size",
      properties.getProperty("flink.tm.memory", "1024"))
    config = config.copy(yarnConfig =
      Some(ensureYarnConfig(config)
        .copy(taskManagerMemory = Some(tmMemory))))

    val appName = properties.getProperty("yarn.application.name",
      properties.getProperty("flink.yarn.appName", "Flink Yarn App Name"))
    config = config.copy(yarnConfig =
      Some(ensureYarnConfig(config)
        .copy(name = Some(appName))))

    val slotNum =  Integer.parseInt(properties.getProperty("taskmanager.numberOfTaskSlots",
     properties.getProperty("flink.tm.slot", "1")))
    config = config.copy(yarnConfig =
      Some(ensureYarnConfig(config)
        .copy(slots = Some(slotNum))))

    val queue = properties.getProperty("yarn.application.queue",
      properties.getProperty("flink.yarn.queue", "default"))
    config = config.copy(yarnConfig =
      Some(ensureYarnConfig(config)
        .copy(queue = Some(queue))))

    this.userUdfJars = getUserUdfJars()
    this.userJars = getUserJarsExceptUdfJars ++ this.userUdfJars
    LOGGER.info("UserJars: " + userJars.mkString(","))
    config = config.copy(externalJars = Some(userJars.toArray))
    LOGGER.info("Config: " + config)
    configuration.setString("flink.yarn.jars", userJars.mkString(":"))

    // load other configuration from interpreter properties
    properties.asScala.foreach(entry => configuration.setString(entry._1, entry._2))
    this.defaultParallelism = configuration.getInteger(CoreOptions.DEFAULT_PARALLELISM)
    this.defaultSqlParallelism = configuration.getInteger(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM)
    LOGGER.info("Default Parallelism: " + this.defaultParallelism)
    LOGGER.info("Default SQL Parallelism: " + this.defaultSqlParallelism)

    // set scala.color
    if (properties.getProperty("zeppelin.flink.scala.color", "true").toBoolean) {
      System.setProperty("scala.color", "true")
    }

    // set host/port when it is remote mode
    if (config.executionMode == ExecutionMode.REMOTE) {
      val host = properties.getProperty("flink.execution.remote.host")
      val port = properties.getProperty("flink.execution.remote.port")
      if (host == null) {
        throw new InterpreterException("flink.execution.remote.host is not " +
          "specified when using REMOTE mode")
      }
      if (port == null) {
        throw new InterpreterException("flink.execution.remote.port is not " +
          "specified when using REMOTE mode")
      }
      config = config.copy(host = Some(host))
        .copy(port = Some(Integer.parseInt(port)))
    }

    config
  }

  private def createFlinkILoop(config: Config): Unit = {
    val printReplOutput = properties.getProperty("zeppelin.flink.printREPLOutput", "true").toBoolean
    val replOut = if (printReplOutput) {
      new JPrintWriter(interpreterOutput, true)
    } else {
      new JPrintWriter(Console.out, true)
    }

    val (iLoop, cluster) = {
      // workaround of checking hadoop jars in yarn  mode
      if (mode == ExecutionMode.YARN) {
        try {
          Class.forName(classOf[FlinkYarnSessionCli].getName)
        } catch {
          case e: ClassNotFoundException =>
            throw new InterpreterException("Unable to load FlinkYarnSessionCli for yarn mode", e)
          case e: NoClassDefFoundError =>
            throw new InterpreterException("No hadoop jar found, make sure you have hadoop command in your PATH", e)
        }
      }

      val (effectiveConfig, cluster) = fetchConnectionInfo(config, configuration, flinkShims)
      this.configuration = effectiveConfig
      cluster match {
        case Some(clusterClient) =>
          // local mode or yarn
          if (mode == ExecutionMode.LOCAL) {
            LOGGER.info("Starting FlinkCluster in local mode")
            this.jmWebUrl = clusterClient.getWebInterfaceURL
            this.displayedJMWebUrl = this.jmWebUrl
          } else if (mode == ExecutionMode.YARN) {
            LOGGER.info("Starting FlinkCluster in yarn mode")
            this.jmWebUrl = clusterClient.getWebInterfaceURL
            val yarnAppId = HadoopUtils.getYarnAppId(clusterClient)
            this.displayedJMWebUrl = getDisplayedJMWebUrl(yarnAppId)
          } else {
            throw new Exception("Starting FlinkCluster in invalid mode: " + mode)
          }
        case None =>
          // yarn-application mode
          if (mode == ExecutionMode.YARN_APPLICATION) {
            // get yarnAppId from env `_APP_ID`
            val yarnAppId = System.getenv("_APP_ID")
            LOGGER.info("Use FlinkCluster in yarn application mode, appId: {}", yarnAppId)
            this.jmWebUrl = "http://localhost:" + HadoopUtils.getFlinkRestPort(yarnAppId)
            this.displayedJMWebUrl = getDisplayedJMWebUrl(yarnAppId)
          } else {
            LOGGER.info("Use FlinkCluster in remote mode")
            this.jmWebUrl = "http://" + config.host.get + ":" + config.port.get
            this.displayedJMWebUrl = getDisplayedJMWebUrl("")
          }
      }

      LOGGER.info(s"\nConnecting to Flink cluster: " + this.jmWebUrl)
      if (InterpreterContext.get() != null) {
        InterpreterContext.get().getIntpEventClient.sendWebUrlInfo(this.jmWebUrl)
      }
      LOGGER.info("externalJars: " +
        config.externalJars.getOrElse(Array.empty[String]).mkString(":"))
      val classLoader = Thread.currentThread().getContextClassLoader
      try {
        // use FlinkClassLoader to initialize FlinkILoop, otherwise TableFactoryService could not find
        // the TableFactory properly
        Thread.currentThread().setContextClassLoader(getFlinkClassLoader)
        val repl = new FlinkILoop(configuration, config.externalJars, None, replOut, mode,
          JExecutionEnvironment.getExecutionEnvironment,
          JStreamExecutionEnvironment.getExecutionEnvironment,
          this)
        (repl, cluster)
      } catch {
        case e: Exception =>
          LOGGER.error(ExceptionUtils.getStackTrace(e))
          throw e
      } finally {
        Thread.currentThread().setContextClassLoader(classLoader)
      }
    }

    this.flinkILoop = iLoop
    this.cluster = cluster

    val settings = createSettings()

    val outputDir = Files.createTempDirectory("flink-repl");
    val interpArguments = List(
      "-Yrepl-class-based",
      "-Yrepl-outdir", s"${outputDir.toFile.getAbsolutePath}"
    )
    settings.processArguments(interpArguments, true)

    flinkILoop.settings = settings
    flinkILoop.intp = createIMain(settings, replOut)
    flinkILoop.intp.beQuietDuring {
      // set execution environment
      flinkILoop.intp.bind("benv", flinkILoop.scalaBenv)
      flinkILoop.intp.bind("senv", flinkILoop.scalaSenv)

      val packageImports = Seq[String](
        "org.apache.flink.core.fs._",
        "org.apache.flink.core.fs.local._",
        "org.apache.flink.api.common.io._",
        "org.apache.flink.api.common.aggregators._",
        "org.apache.flink.api.common.accumulators._",
        "org.apache.flink.api.common.distributions._",
        "org.apache.flink.api.common.operators._",
        "org.apache.flink.api.common.operators.base.JoinOperatorBase.JoinHint",
        "org.apache.flink.api.common.functions._",
        "org.apache.flink.api.java.io._",
        "org.apache.flink.api.java.aggregation._",
        "org.apache.flink.api.java.functions._",
        "org.apache.flink.api.java.operators._",
        "org.apache.flink.api.java.sampling._",
        "org.apache.flink.api.scala._",
        "org.apache.flink.api.scala.utils._",
        "org.apache.flink.streaming.api.scala._",
        "org.apache.flink.streaming.api.windowing.time._",
        "org.apache.flink.types.Row"
      )

      flinkILoop.intp.interpret("import " + packageImports.mkString(", "))

      if (flinkVersion.isFlink110) {
        flinkILoop.intp.interpret("import org.apache.flink.table.api.scala._")
      } else {
        flinkILoop.intp.interpret("import org.apache.flink.table.api._")
        flinkILoop.intp.interpret("import org.apache.flink.table.api.bridge.scala._")
      }

      flinkILoop.intp.interpret("import org.apache.flink.table.functions.ScalarFunction")
      flinkILoop.intp.interpret("import org.apache.flink.table.functions.AggregateFunction")
      flinkILoop.intp.interpret("import org.apache.flink.table.functions.TableFunction")
      flinkILoop.intp.interpret("import org.apache.flink.table.functions.TableAggregateFunction")
    }

    val in0 = None
    val reader = in0.fold(flinkILoop.chooseReader(settings))(r =>
      SimpleReader(r, replOut, interactive = true))

    flinkILoop.in = reader
    flinkILoop.initializeSynchronous()
    flinkILoop.intp.setContextClassLoader()
    reader.postInit()
    this.scalaCompletion = reader.completion

    this.benv = flinkILoop.scalaBenv
    this.senv = flinkILoop.scalaSenv
    val timeType = properties.getProperty("flink.senv.timecharacteristic", "EventTime")
    this.senv.setStreamTimeCharacteristic(TimeCharacteristic.valueOf(timeType))
    this.benv.setParallelism(configuration.getInteger(CoreOptions.DEFAULT_PARALLELISM))
    this.senv.setParallelism(configuration.getInteger(CoreOptions.DEFAULT_PARALLELISM))

    setAsContext()
  }

  private def createTableEnvs(): Unit = {
    val originalClassLoader = Thread.currentThread().getContextClassLoader
    try {
      Thread.currentThread().setContextClassLoader(getFlinkClassLoader)
      val tableConfig = new TableConfig
      tableConfig.getConfiguration.addAll(configuration)

      this.tblEnvFactory = new TableEnvFactory(this.flinkVersion, this.flinkShims,
        this.benv, this.senv, tableConfig)

      // blink planner
      var btEnvSetting = EnvironmentSettings.newInstance().inBatchMode().useBlinkPlanner().build()
      this.btenv = tblEnvFactory.createJavaBlinkBatchTableEnvironment(btEnvSetting, getFlinkClassLoader);
      flinkILoop.bind("btenv", btenv.getClass().getCanonicalName(), btenv, List("@transient"))
      this.java_btenv = this.btenv

      var stEnvSetting =
        EnvironmentSettings.newInstance().inStreamingMode().useBlinkPlanner().build()
      this.stenv = tblEnvFactory.createScalaBlinkStreamTableEnvironment(stEnvSetting, getFlinkClassLoader)
      flinkILoop.bind("stenv", stenv.getClass().getCanonicalName(), stenv, List("@transient"))
      this.java_stenv = tblEnvFactory.createJavaBlinkStreamTableEnvironment(stEnvSetting, getFlinkClassLoader)

      // flink planner
      this.btenv_2 = tblEnvFactory.createScalaFlinkBatchTableEnvironment()
      flinkILoop.bind("btenv_2", btenv_2.getClass().getCanonicalName(), btenv_2, List("@transient"))
      stEnvSetting =
        EnvironmentSettings.newInstance().inStreamingMode().useOldPlanner().build()
      this.stenv_2 = tblEnvFactory.createScalaFlinkStreamTableEnvironment(stEnvSetting, getFlinkClassLoader)
      flinkILoop.bind("stenv_2", stenv_2.getClass().getCanonicalName(), stenv_2, List("@transient"))

      this.java_btenv_2 = tblEnvFactory.createJavaFlinkBatchTableEnvironment()
      btEnvSetting = EnvironmentSettings.newInstance.useOldPlanner.inStreamingMode.build
      this.java_stenv_2 = tblEnvFactory.createJavaFlinkStreamTableEnvironment(btEnvSetting, getFlinkClassLoader)
    } finally {
      Thread.currentThread().setContextClassLoader(originalClassLoader)
    }
  }

  private def setTableEnvConfig(): Unit = {
    this.properties.asScala.filter(e => e._1.startsWith("table.exec"))
      .foreach(e => {
        this.btenv.getConfig.getConfiguration.setString(e._1, e._2)
        this.java_btenv.getConfig.getConfiguration.setString(e._1, e._2)
        this.stenv.getConfig.getConfiguration.setString(e._1, e._2)
        this.java_stenv.getConfig.getConfiguration.setString(e._1, e._2)
      })

    // set python exec for PyFlink
    val pythonExec = properties.getProperty("zeppelin.pyflink.python", "")
    if (!StringUtils.isBlank(pythonExec)) {
      this.stenv.getConfig.getConfiguration.setString("python.exec", pythonExec)
      this.btenv.getConfig.getConfiguration.setString("python.exec", pythonExec)
      this.java_btenv.getConfig.getConfiguration.setString("python.exec", pythonExec)
      this.java_stenv.getConfig.getConfiguration.setString("python.exec", pythonExec)
    }

    if (java.lang.Boolean.parseBoolean(
      properties.getProperty("zeppelin.flink.disableSysoutLogging", "true"))) {
      flinkShims.disableSysoutLogging(this.benv.getConfig, this.senv.getConfig);
    }
  }

  private def registerHiveCatalog(): Unit = {
    val hiveConfDir =
      properties.getOrDefault("HIVE_CONF_DIR", System.getenv("HIVE_CONF_DIR")).toString
    if (hiveConfDir == null) {
      throw new InterpreterException("HIVE_CONF_DIR is not specified");
    }
    val database = properties.getProperty("zeppelin.flink.hive.database", "default")
    val hiveVersion = properties.getProperty("zeppelin.flink.hive.version", "2.3.4")
    val hiveCatalog = new HiveCatalog("hive", database, hiveConfDir, hiveVersion)
    this.btenv.registerCatalog("hive", hiveCatalog)
    this.btenv.useCatalog("hive")
    this.btenv.useDatabase(database)
    if (properties.getProperty("zeppelin.flink.module.enableHive", "false").toBoolean) {
      this.btenv.loadModule("hive", new HiveModule(hiveVersion))
    }
  }

  private def loadUDFJar(jar: String): Unit = {
    LOGGER.info("Loading UDF Jar: " + jar)
    val jarFile = new JarFile(jar)
    val entries = jarFile.entries

    val udfPackages = properties.getProperty("flink.udf.jars.packages", "").split(",").toSet
    val urls = Array(new URL("jar:file:" + jar + "!/"))
    val cl = new URLClassLoader(urls, getFlinkScalaShellLoader)

    while (entries.hasMoreElements) {
      val je = entries.nextElement
      if (!je.isDirectory && je.getName.endsWith(".class") && !je.getName.contains("$")) {
        try {
          // -6 because of .class
          var className = je.getName.substring(0, je.getName.length - 6)
          className = className.replace('/', '.')
          if (udfPackages.isEmpty || udfPackages.exists( p => className.startsWith(p))) {
            val c = cl.loadClass(className)
            val udf = c.newInstance()
            if (udf.isInstanceOf[ScalarFunction]) {
              val scalarUDF = udf.asInstanceOf[ScalarFunction]
              flinkShims.registerScalarFunction(btenv, c.getSimpleName, scalarUDF)
            } else if (udf.isInstanceOf[TableFunction[_]]) {
              val tableUDF = udf.asInstanceOf[TableFunction[_]]
              flinkShims.registerTableFunction(btenv, c.getSimpleName, tableUDF)
            } else if (udf.isInstanceOf[AggregateFunction[_, _]]) {
              val aggregateUDF = udf.asInstanceOf[AggregateFunction[_, _]]
              flinkShims.registerAggregateFunction(btenv, c.getSimpleName, aggregateUDF)
            } else if (udf.isInstanceOf[TableAggregateFunction[_, _]]) {
              val tableAggregateUDF = udf.asInstanceOf[TableAggregateFunction[_, _]]
              flinkShims.registerTableAggregateFunction(btenv, c.getSimpleName, tableAggregateUDF)
            } else {
              LOGGER.warn("No UDF definition found in class file: " + je.getName)
            }
          }
        } catch {
          case e : Throwable =>
            LOGGER.info("Fail to inspect udf class: " + je.getName, e)
        }
      }
    }
  }

  private def setAsContext(): Unit = {
    val streamFactory = flinkShims.createStreamExecutionEnvironmentFactory(this.senv.getJavaEnv)
    //StreamExecutionEnvironment
    var method = classOf[JStreamExecutionEnvironment].getDeclaredMethod("initializeContextEnvironment",
      classOf[StreamExecutionEnvironmentFactory])
    method.setAccessible(true)
    method.invoke(null, streamFactory);

    val batchFactory = new ExecutionEnvironmentFactory() {
      override def createExecutionEnvironment = benv.getJavaEnv
    }
    //StreamExecutionEnvironment
    method = classOf[JExecutionEnvironment].getDeclaredMethod("initializeContextEnvironment",
      classOf[ExecutionEnvironmentFactory])
    method.setAccessible(true)
    method.invoke(null, batchFactory);
  }

  protected def bind(name: String,
                     tpe: String,
                     value: Object,
                     modifier: List[String]): Unit = {
    flinkILoop.beQuietDuring {
      val result = flinkILoop.bind(name, tpe, value, modifier)
      if (result != IR.Success) {
        throw new Exception("Fail to bind variable: " + name)
      }
    }
  }

  protected def callMethod(obj: Object, name: String): Object = {
    callMethod(obj, name, Array.empty[Class[_]], Array.empty[Object])
  }

  protected def callMethod(obj: Object, name: String,
                           parameterTypes: Array[Class[_]],
                           parameters: Array[Object]): Object = {
    val method = obj.getClass.getMethod(name, parameterTypes: _ *)
    method.setAccessible(true)
    method.invoke(obj, parameters: _ *)
  }

  /**
   * This is just a workaround to make table api work in multiple threads.
   */
  def createPlannerAgain(): Unit = {
    val originalClassLoader = Thread.currentThread().getContextClassLoader
    try {
      Thread.currentThread().setContextClassLoader(getFlinkClassLoader)
      val stEnvSetting =
        EnvironmentSettings.newInstance().inStreamingMode().useBlinkPlanner().build()
      this.tblEnvFactory.createStreamPlanner(stEnvSetting)
    } finally {
      Thread.currentThread().setContextClassLoader(originalClassLoader)
    }
  }

  def interpret(code: String, context: InterpreterContext): InterpreterResult = {
    val originalStdOut = System.out
    val originalStdErr = System.err;
    if (context != null) {
      interpreterOutput.setInterpreterOutput(context.out)
      context.out.clear()
    }

    Console.withOut(if (context != null) context.out else Console.out) {
      System.setOut(Console.out)
      System.setErr(Console.out)
      interpreterOutput.ignoreLeadingNewLinesFromScalaReporter()
      // add print("") at the end in case the last line is comment which lead to INCOMPLETE
      val lines = code.split("\\n") ++ List("print(\"\")")
      var incompleteCode = ""
      var lastStatus: InterpreterResult.Code = null

      for ((line, i) <- lines.zipWithIndex if !line.trim.isEmpty) {
        val nextLine = if (incompleteCode != "") {
          incompleteCode + "\n" + line
        } else {
          line
        }
        if (i < (lines.length - 1) && lines(i + 1).trim.startsWith(".")) {
          incompleteCode = nextLine
        } else {
          flinkILoop.interpret(nextLine) match {
            case scala.tools.nsc.interpreter.IR.Success =>
              // continue the next line
              incompleteCode = ""
              lastStatus = InterpreterResult.Code.SUCCESS
            case error@scala.tools.nsc.interpreter.IR.Error =>
              return new InterpreterResult(InterpreterResult.Code.ERROR)
            case scala.tools.nsc.interpreter.IR.Incomplete =>
              // put this line into inCompleteCode for the next execution.
              incompleteCode = incompleteCode + "\n" + line
              lastStatus = InterpreterResult.Code.INCOMPLETE
          }
        }
      }
      // flush all output before returning result to frontend
      Console.flush()
      interpreterOutput.setInterpreterOutput(null)
      // reset the java stdout
      System.setOut(originalStdOut)
      System.setErr(originalStdErr)
      return new InterpreterResult(lastStatus)
    }
  }

  /**
   * Set execution.savepoint.path in the following order:
   *
   * 1. Use savepoint path stored in paragraph config, this is recorded by zeppelin when paragraph is canceled,
   * 2. Use checkpoint path stored in pararaph config, this is recorded by zeppelin in flink job progress poller.
   * 3. Use local property 'execution.savepoint.path' if user set it.
   * 4. Otherwise remove 'execution.savepoint.path' when user didn't specify it in %flink.conf
   *
   * @param context
   */
  def setSavepointPathIfNecessary(context: InterpreterContext): Unit = {
    val savepointPath = context.getConfig.getOrDefault(JobManager.SAVEPOINT_PATH, "").toString
    val resumeFromSavepoint = context.getBooleanLocalProperty(JobManager.RESUME_FROM_SAVEPOINT, false)
    // flink 1.12 use copied version of configuration, so in order to update configuration we have to
    // get the internal configuration of StreamExecutionEnvironment.
    val internalConfiguration = getConfigurationOfStreamExecutionEnv()
    if (!StringUtils.isBlank(savepointPath) && resumeFromSavepoint){
      LOGGER.info("Resume job from savepoint , savepointPath = {}", savepointPath)
      internalConfiguration.setString(SavepointConfigOptions.SAVEPOINT_PATH.key(), savepointPath);
      return
    }

    val checkpointPath = context.getConfig.getOrDefault(JobManager.LATEST_CHECKPOINT_PATH, "").toString
    val resumeFromLatestCheckpoint = context.getBooleanLocalProperty(JobManager.RESUME_FROM_CHECKPOINT, false)
    if (!StringUtils.isBlank(checkpointPath) && resumeFromLatestCheckpoint) {
      LOGGER.info("Resume job from checkpoint , checkpointPath = {}", checkpointPath)
      internalConfiguration.setString(SavepointConfigOptions.SAVEPOINT_PATH.key(), checkpointPath);
      return
    }

    val userSavepointPath = context.getLocalProperties.getOrDefault(
      SavepointConfigOptions.SAVEPOINT_PATH.key(), "")
    if (!StringUtils.isBlank(userSavepointPath)) {
      LOGGER.info("Resume job from user set savepoint , savepointPath = {}", userSavepointPath)
      internalConfiguration.setString(SavepointConfigOptions.SAVEPOINT_PATH.key(), checkpointPath)
      return;
    }

    val userSettingSavepointPath = properties.getProperty(SavepointConfigOptions.SAVEPOINT_PATH.key())
    if (StringUtils.isBlank(userSettingSavepointPath)) {
      // remove SAVEPOINT_PATH when user didn't set it via %flink.conf
      internalConfiguration.removeConfig(SavepointConfigOptions.SAVEPOINT_PATH)
    }
  }

  def setParallelismIfNecessary(context: InterpreterContext): Unit = {
    val parallelismStr = context.getLocalProperties.get("parallelism")
    if (!StringUtils.isBlank(parallelismStr)) {
      val parallelism = parallelismStr.toInt
      this.senv.setParallelism(parallelism)
      this.benv.setParallelism(parallelism)
      this.stenv.getConfig.getConfiguration
        .setString(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM.key(), parallelism + "")
      this.btenv.getConfig.getConfiguration
        .setString(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM.key(), parallelism + "")
    }
    val maxParallelismStr = context.getLocalProperties.get("maxParallelism")
    if (!StringUtils.isBlank(maxParallelismStr)) {
      val maxParallelism = maxParallelismStr.toInt
      senv.setParallelism(maxParallelism)
    }
  }

  def cancel(context: InterpreterContext): Unit = {
    jobManager.cancelJob(context)
  }

  def getProgress(context: InterpreterContext): Int = {
    jobManager.getJobProgress(context.getParagraphId)
  }

  def close(): Unit = {
    LOGGER.info("Closing FlinkScalaInterpreter")
    if (properties.getProperty("flink.interpreter.close.shutdown_cluster", "true").toBoolean) {
      if (cluster != null) {
        cluster match {
          case Some(clusterClient) =>
            LOGGER.info("Shutdown FlinkCluster")
            clusterClient.shutDownCluster()
            clusterClient.close()
            // delete staging dir
            if (mode == ExecutionMode.YARN) {
              HadoopUtils.cleanupStagingDirInternal(clusterClient)
            }
          case None =>
            LOGGER.info("Don't close the Remote FlinkCluster")
        }
      }
    } else {
      LOGGER.info("Keep cluster alive when closing interpreter")
    }

    if (flinkILoop != null) {
      flinkILoop.closeInterpreter()
      flinkILoop = null
    }
    if (jobManager != null) {
      jobManager.shutdown()
    }
  }

  def getExecutionEnvironment(): ExecutionEnvironment = this.benv

  def getStreamExecutionEnvironment(): StreamExecutionEnvironment = this.senv

  def getBatchTableEnvironment(planner: String = "blink"): TableEnvironment = {
    if (planner == "blink")
      this.btenv
    else
      this.btenv_2
  }

  def getStreamTableEnvironment(planner: String = "blink"): TableEnvironment = {
    if (planner == "blink")
      this.stenv
    else
      this.stenv_2
  }

  def getJavaBatchTableEnvironment(planner: String): TableEnvironment = {
    if (planner == "blink") {
      this.java_btenv
    } else {
      this.java_btenv_2
    }
  }

  def getJavaStreamTableEnvironment(planner: String): TableEnvironment = {
    if (planner == "blink") {
      this.java_stenv
    } else {
      this.java_stenv_2
    }
  }

  def getDefaultParallelism = this.defaultParallelism

  def getDefaultSqlParallelism = this.defaultSqlParallelism

  private def getUserJarsExceptUdfJars: Seq[String] = {
    val flinkJars =
      if (!StringUtils.isBlank(properties.getProperty("flink.execution.jars", ""))) {
        getOrDownloadJars(properties.getProperty("flink.execution.jars").split(",").toSeq)
      } else {
        Seq.empty[String]
      }

    val flinkPackageJars =
      if (!StringUtils.isBlank(properties.getProperty("flink.execution.packages", ""))) {
        val packages = properties.getProperty("flink.execution.packages")
        val dependencyResolver = new DependencyResolver(System.getProperty("user.home") + "/.m2/repository")
        packages.split(",")
          .flatMap(e => JavaConversions.asScalaBuffer(dependencyResolver.load(e)))
          .map(e => e.getAbsolutePath).toSeq
      } else {
        Seq.empty[String]
      }

    flinkJars ++ flinkPackageJars
  }

  private def getUserUdfJars(): Seq[String] = {
    if (!StringUtils.isBlank(properties.getProperty("flink.udf.jars", ""))) {
      getOrDownloadJars(properties.getProperty("flink.udf.jars").split(",").toSeq)
    } else {
      Seq.empty[String]
    }
  }

  private def getOrDownloadJars(jars: Seq[String]): Seq[String] = {
    jars.map(jar => {
      if (jar.contains("://")) {
        HadoopUtils.downloadJar(jar)
      } else {
        val jarFile = new File(jar)
        if (!jarFile.exists() || !jarFile.isFile) {
          throw new Exception(s"jar file: ${jar} doesn't exist")
        } else {
          jar
        }
      }
    })
  }

  private def getDisplayedJMWebUrl(yarnAppId: String): String = {
    // `zeppelin.flink.uiWebUrl` is flink jm url template, {{applicationId}} will be replaced
    // with real yarn app id.
    val flinkUIWebUrl = properties.getProperty("zeppelin.flink.uiWebUrl")
    if (StringUtils.isNotBlank(flinkUIWebUrl)) {
      flinkUIWebUrl.replace("{{applicationId}}", yarnAppId)
    } else {
      this.jmWebUrl
    }
  }

  def getJobManager = this.jobManager

  def getFlinkScalaShellLoader: ClassLoader = {
    val userCodeJarFile = this.flinkILoop.writeFilesToDisk()
    new URLClassLoader(Array(userCodeJarFile.toURL) ++ userJars.map(e => new File(e).toURL))
  }

  private def getFlinkClassLoader: ClassLoader = {
    new URLClassLoader(userJars.map(e => new File(e).toURL).toArray)
  }

  def getZeppelinContext = this.z

  def getFlinkConfiguration = this.configuration

  def getFormType() = FormType.NATIVE

  def getCluster = cluster

  def getFlinkILoop = flinkILoop

  def getFlinkShims = flinkShims

  def getFlinkVersion = flinkVersion

  class FlinkJobListener extends JobListener {

    override def onJobSubmitted(jobClient: JobClient, e: Throwable): Unit = {
      if (e != null) {
        LOGGER.warn("Fail to submit job")
      } else {
        if (InterpreterContext.get() == null) {
          LOGGER.warn("Job {} is submitted but unable to associate this job to paragraph, " +
            "as InterpreterContext is null", jobClient.getJobID)
        } else {
          LOGGER.info("Job {} is submitted for paragraph {}", Array(jobClient.getJobID,
            InterpreterContext.get().getParagraphId): _ *)
          jobManager.addJob(InterpreterContext.get(), jobClient)
          if (jmWebUrl != null) {
            jobManager.sendFlinkJobUrl(InterpreterContext.get());
          } else {
            LOGGER.error("Unable to link JobURL, because JobManager weburl is null")
          }
        }
      }
    }

    override def onJobExecuted(jobExecutionResult: JobExecutionResult, e: Throwable): Unit = {
      if (e != null) {
        LOGGER.warn("Fail to execute job")
      } else {
        LOGGER.info("Job {} is executed with time {} seconds", jobExecutionResult.getJobID,
          jobExecutionResult.getNetRuntime(TimeUnit.SECONDS))
      }
      if (InterpreterContext.get() != null) {
        jobManager.removeJob(InterpreterContext.get().getParagraphId)
      } else {
        if (e == null) {
          LOGGER.warn("Unable to remove this job {}, as InterpreterContext is null",
            jobExecutionResult.getJobID)
        }
      }
    }
  }

  def getUserJars:java.util.List[String] = JavaConversions.seqAsJavaList(userJars)

  def completion(buf: String, cursor: Int, context: InterpreterContext): java.util.List[InterpreterCompletion]

  private def getConfigurationOfStreamExecutionEnv(): Configuration = {
    val getConfigurationMethod = classOf[JStreamExecutionEnvironment].getDeclaredMethod("getConfiguration")
    getConfigurationMethod.setAccessible(true)
    getConfigurationMethod.invoke(this.senv.getJavaEnv).asInstanceOf[Configuration]
  }
}


text:
namespace g{namespace Fuse{namespace Entities{struct RenderNode;}}}
namespace g{namespace Fuse{struct DrawContext;}}
namespace g{namespace Fuse{struct HitTestContext;}}
namespace g{namespace Uno{namespace Collections{struct ObservableList;}}}
namespace g{namespace Uno{namespace Geometry{struct Box;}}}
namespace g{namespace Uno{struct Float2;}}

namespace g{
namespace Fuse{
namespace Entities{

// public class RenderNode :1920
// {
struct RenderNode_type : ::g::Fuse::Node_type
{
    void(*fp_OnDraw)(::g::Fuse::Entities::RenderNode*, ::g::Fuse::DrawContext*);
};

RenderNode_type* RenderNode_typeof();
void RenderNode__ctor_1_fn(RenderNode* __this);
void RenderNode__get_Children_fn(RenderNode* __this, uObject** __retval);
void RenderNode__Draw_fn(RenderNode* __this, ::g::Fuse::DrawContext* dc);
void RenderNode__GetSubNode_fn(RenderNode* __this, int* index, ::g::Fuse::Node** __retval);
void RenderNode__get_LocalBounds_fn(RenderNode* __this, ::g::Uno::Geometry::Box* __retval);
void RenderNode__New1_fn(RenderNode** __retval);
void RenderNode__OnArrangeMarginBox_fn(RenderNode* __this, ::g::Uno::Float2* position, ::g::Uno::Float2* availableSize, int* availSet, ::g::Uno::Float2* __retval);
void RenderNode__OnChildAdded_fn(RenderNode* __this, ::g::Fuse::Node* c);
void RenderNode__OnChildRemoved_fn(RenderNode* __this, ::g::Fuse::Node* c);
void RenderNode__OnDraw_fn(RenderNode* __this, ::g::Fuse::DrawContext* dc);
void RenderNode__OnHitTest_fn(RenderNode* __this, ::g::Fuse::HitTestContext* args);
void RenderNode__get_SubNodeCount_fn(RenderNode* __this, int* __retval);

struct RenderNode : ::g::Fuse::Node
{
    uStrong< ::g::Uno::Collections::ObservableList*> _children;

    void ctor_1();
    uObject* Children();
    void OnChildAdded(::g::Fuse::Node* c);
    void OnChildRemoved(::g::Fuse::Node* c);
    void OnDraw(::g::Fuse::DrawContext* dc) { (((RenderNode_type*)__type)->fp_OnDraw)(this, dc); }
    static RenderNode* New1();
    static void OnDraw(RenderNode* __this, ::g::Fuse::DrawContext* dc) { RenderNode__OnDraw_fn(__this, dc); }
};
// }

}}} // ::g::Fuse::Entities

redpajama_set_name: RedPajamaGithub


text:
{-# LANGUAGE RecordWildCards #-}

module Main where

import Control.Concurrent.Async
import qualified Data.ByteString.Char8 as S
import Data.Monoid
import Options.Applicative hiding (Parser, option)
import qualified Options.Applicative as O
import Options.Applicative.Types
import System.Log.Logger

import Marquise.Client
import Marquise.Server
import Package (package, version)
import Vaultaire.Program

data Options = Options
  { broker         :: String
  , debug          :: Bool
  , quiet          :: Bool
  , cacheFile      :: String
  , cacheFlushFreq :: Integer
  , origin         :: String
  , namespace      :: String }

justRead :: Read a => ReadM a
justRead = readerAsk >>= return . read

helpfulParser :: O.ParserInfo Options
helpfulParser = info (helper <*> optionsParser) fullDesc

optionsParser :: O.Parser Options
optionsParser = Options <$> parseBroker
                        <*> parseDebug
                        <*> parseQuiet
                        <*> parseCacheFile
                        <*> parseCacheFlushFreq
                        <*> parseOrigin
                        <*> parseNameSpace
  where
    parseBroker = strOption $
           long "broker"
        <> short 'b'
        <> metavar "BROKER"
        <> value "localhost"
        <> showDefault
        <> help "Vault broker host name or IP address"

    parseDebug = switch $
           long "debug"
        <> short 'd'
        <> help "Output lots of debugging information"

    parseQuiet = switch $
           long "quiet"
        <> short 'q'
        <> help "Only emit warnings or fatal messages"

    parseCacheFile = strOption $
           long "cache-file"
        <> short 'c'
        <> value ""
        <> help "Location to read/write cached SourceDicts"

    -- This doesn't mean that marquised will rigorously flush the cache
    -- every `t` seconds. To clarify: marquised consideres flushing the
    -- cache after it finishes reading every spool file. If the cache
    -- hasn't been flushed in the last `t` seconds, it will be flushed
    -- before processing the next spool file.
    parseCacheFlushFreq = O.option justRead $
           long "cache-flush-freq"
        <> short 't'
        <> help "Period of time to wait between cache writes, in seconds"
        <> value 42

    parseNameSpace = argument str (metavar "NAMESPACE")

    parseOrigin = argument str (metavar "ORIGIN")

defaultCacheLoc :: String -> String
defaultCacheLoc = (++) "/var/cache/marquise/"

main :: IO ()
main = do
    Options{..} <- execParser $ helpfulParser

    let level
          | debug     = Debug
          | quiet     = Quiet
          | otherwise = Normal

    quit <- initializeProgram (package ++ "-" ++ version) level

    cacheFile' <- return $ case cacheFile of
        "" -> defaultCacheLoc origin
        x  -> x

    a <- runMarquiseDaemon broker (Origin $ S.pack origin) namespace quit cacheFile' cacheFlushFreq

    -- wait forever
    wait a
    debugM "Main.main" "End"


text:
package org.ovirt.engine.api.restapi.resource;

import java.util.List;

import javax.ws.rs.core.Response;

import org.ovirt.engine.api.model.BaseResource;
import org.ovirt.engine.api.model.Label;
import org.ovirt.engine.api.model.Labels;
import org.ovirt.engine.api.model.Network;
import org.ovirt.engine.api.resource.LabelResource;
import org.ovirt.engine.api.resource.LabelsResource;
import org.ovirt.engine.core.common.action.LabelNetworkParameters;
import org.ovirt.engine.core.common.action.UnlabelNetworkParameters;
import org.ovirt.engine.core.common.action.VdcActionType;
import org.ovirt.engine.core.common.businessentities.network.pseudo.NetworkLabel;
import org.ovirt.engine.core.common.queries.IdQueryParameters;
import org.ovirt.engine.core.common.queries.VdcQueryType;
import org.ovirt.engine.core.compat.Guid;

public class BackendNetworkLabelsResource extends AbstractBackendCollectionResource<Label, NetworkLabel> implements LabelsResource {

    private Guid networkId;

    protected BackendNetworkLabelsResource(Guid networkId) {
        super(Label.class, NetworkLabel.class);
        this.networkId = networkId;
    }

    @Override
    public Labels list() {
        return mapCollection(getNetworkLabels(networkId));
    }

    private List<NetworkLabel> getNetworkLabels(Guid networkId) {
        return getBackendCollection(VdcQueryType.GetNetworkLabelsByNetworkId, new IdQueryParameters(networkId));
    }

    private Labels mapCollection(List<NetworkLabel> networkLabels) {
        Labels labels = new Labels();
        for (NetworkLabel networkLabel : networkLabels) {
            Label label = new Label();
            label.setId(networkLabel.getId());
            labels.getLabels().add(label);
            addLinks(label, Network.class);
        }

        return labels;
    }

    @Override
    public Response add(Label label) {
        validateParameters(label, "id");
        return performCreate(VdcActionType.LabelNetwork,
                new LabelNetworkParameters(networkId, label.getId()),
                new NetworkLabelIdResolver(networkId));
    }

    @SingleEntityResource
    @Override
    public LabelResource getLabelSubResource(String id) {
        return inject(new BackendNetworkLabelResource(id, this));
    }

    @Override
    protected Response performRemove(String id) {
        return performAction(VdcActionType.UnlabelNetwork, new UnlabelNetworkParameters(networkId));
    }

    @Override
    protected Label doPopulate(Label model, NetworkLabel entity) {
        return model;
    }

    @Override
    protected Label addParents(Label model) {
        model.setNetwork(new Network());
        model.getNetwork().setId(networkId.toString());
        return model;
    };

    @Override
    protected Label addLinks(Label model,
            Class<? extends BaseResource> suggestedParent,
            String... excludeSubCollectionMembers) {
        return super.addLinks(model, Network.class);
    }

    public Guid getNetworkId() {
        return networkId;
    }

    protected class NetworkLabelIdResolver extends EntityIdResolver<String> {

        private Guid networkId;

        NetworkLabelIdResolver() {
        }

        NetworkLabelIdResolver(Guid networkId) {
            this.networkId = networkId;
        }

        @Override
        public NetworkLabel lookupEntity(String id) throws BackendFailureException {
            List<NetworkLabel> labels = getNetworkLabels(networkId);
            if (!labels.isEmpty()) {
                return labels.get(0);
            }

            return null;
        }
    }
}


text:
package at.newmedialab.ldpath.template.model.transformers;

import at.newmedialab.ldpath.api.backend.RDFBackend;
import at.newmedialab.ldpath.api.transformers.NodeTransformer;
import at.newmedialab.ldpath.model.transformers.BooleanTransformer;
import freemarker.template.TemplateBooleanModel;
import freemarker.template.TemplateModelException;

/**
 * Transform a node into the freemarker boolean type (TemplateBooleanModel).
 * <p/>
 * Author: Sebastian Schaffert
 */
public class TemplateBooleanTransformer<Node> implements NodeTransformer<TemplateBooleanModel,Node> {

    private BooleanTransformer<Node> delegate;

    public TemplateBooleanTransformer() {
        delegate = new BooleanTransformer<Node>();
    }

    /**
     * Transform the KiWiNode node into the datatype T. In case the node cannot be transformed to
     * the respective datatype, throws an IllegalArgumentException that needs to be caught by the class
     * carrying out the transformation.
     *
     * @param node
     * @return
     */
    @Override
    public TemplateBooleanModel transform(final RDFBackend<Node> nodeRDFBackend, final Node node) throws IllegalArgumentException {
        return new TemplateBooleanModel() {
            @Override
            public boolean getAsBoolean() throws TemplateModelException {
                return delegate.transform(nodeRDFBackend,node);
            }
        };
    }
}


text:
(function (angular) {
    'use strict';

    angular
        .module('mllApp')
        .directive('mllApplication', mllApplication);

    function mllApplication() {
        return {
            restrict: 'AE',
            scope: {},
            controller: 'ApplicationController',
            controllerAs: 'ctrl',
            templateUrl: 'app.template.html'
        };
    }

})(window.angular);


text:
from django.db import models
from django_countries.fields import CountryField
from geoposition.fields import GeopositionField

# Create your models here.

class Team(models.Model):
    name = models.CharField(max_length=50, default='')
    foundation = models.DateField()
    image = models.ImageField(upload_to='team_image', blank=True)
    country = CountryField(blank_label='select country', null=True)

    def __str__(self):
        return self.name

    def __unicode__(self):
        return self.name

class Stadium(models.Model):
    name = models.CharField(max_length=50, default='')
    image = models.ImageField(upload_to='stadium_image', blank=True)
    capacity = models.IntegerField(default='')
    team = models.OneToOneField(Team, related_name='team_stadium')
    position = GeopositionField(null=True, blank=True)

    def __str__(self):
        return self.name

    def __unicode__(self):
        return self.name

class PointOfInterest(models.Model):
    name = models.CharField(max_length=100)
    position = GeopositionField()


text:
package es.uvigo.esei.daa.dao;

import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Base64;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.springframework.stereotype.Repository;

@Repository
public class UsersDAO extends DAO {
	private final static Logger LOG = Logger.getLogger("UsersDAO");

	public String checkLogin(String login, String password) throws DAOException {
		try (final Connection conn = this.getConnection()) {
			final String query = "SELECT password FROM user WHERE userName=?";

			try (PreparedStatement statement = conn.prepareStatement(query)) {
				statement.setString(1, login);

				try (ResultSet result = statement.executeQuery()) {
					if (result.next()) {
						final String dbPassword = result.getString("password");
						final String shaPassword = encodeSha256(password);

						if (shaPassword.equals(dbPassword)) {
							return encodeBase64(login + ":" + password);
						} else {
							return null;
						}
					} else {
						return null;
					}
				}
			}
		} catch (SQLException e) {
			LOG.log(Level.SEVERE, "Error checking login", e);
			throw new DAOException(e);
		}
	}

	public String checkToken(String token)
	throws DAOException, IllegalArgumentException {
		final String decodedToken = decodeBase64(token);
		final int colonIndex = decodedToken.indexOf(':');

		if (colonIndex < 0 || colonIndex == decodedToken.length()-1) {
			throw new IllegalArgumentException("Invalid token");
		}

		final String login = decodedToken.substring(0, decodedToken.indexOf(':'));
		final String password = encodeSha256(decodedToken.substring(decodedToken.indexOf(':') + 1));

		try (final Connection conn = this.getConnection()) {
			final String query = "SELECT password FROM user WHERE userName=?";

			try (PreparedStatement statement = conn.prepareStatement(query)) {
				statement.setString(1, login);

				try (ResultSet result = statement.executeQuery()) {
					if (result.next()) {
						final String dbPassword = result.getString("password");

						return password.equals(dbPassword) ? login : null;
					} else {
						return null;
					}
				}
			}
		} catch (SQLException e) {
			LOG.log(Level.SEVERE, "Error checking token", e);
			throw new DAOException(e);
		}
	}

	private final static String decodeBase64(String text) {
		return new String(Base64.getDecoder().decode(text.getBytes()));
	}

	private final static String encodeBase64(String text) {
		return Base64.getEncoder().encodeToString(text.getBytes());
	}

	private final static String encodeSha256(String text) {
		try {
			final MessageDigest digest = MessageDigest.getInstance("SHA-256");
			final byte[] digested = digest.digest(text.getBytes());

			return hexToString(digested);
		} catch (NoSuchAlgorithmException e) {
			LOG.log(Level.SEVERE, "SHA-256 not supported", e);
			throw new RuntimeException(e);
		}
	}

	private final static String hexToString(byte[] hex) {
		final StringBuilder sb = new StringBuilder();

		for (byte b : hex) {
			sb.append(String.format("%02x", b & 0xff));
		}

		return sb.toString();
	}
}


text:
"""
Example that trains an LSTM or GRU networks for sentiment analysis

Reference:
   See J.Li et al, EMNLP2015 - http://arxiv.org/pdf/1503.00185v5.pdf

$ python examples/imdb_lstm.py -e 2 -eval 1 --rlayer_type lstm

"""

from neon.backends import gen_backend
from neon.data.dataloaders import load_imdb
from neon.data.dataiterator import ArrayIterator
from neon.data.text_preprocessing import pad_data
from neon.initializers import Uniform, GlorotUniform
from neon.layers import (GeneralizedCost, LSTM, Affine, Dropout, LookupTable,
                         RecurrentSum, Recurrent, DeepBiLSTM, DeepBiRNN)
from neon.models import Model
from neon.optimizers import Adagrad
from neon.transforms import Logistic, Tanh, Softmax, CrossEntropyMulti, Accuracy
from neon.callbacks.callbacks import Callbacks
from neon.util.argparser import NeonArgparser, extract_valid_args

# parse the command line arguments
parser = NeonArgparser(__doc__)
parser.add_argument('--rlayer_type', default='lstm',
                    choices=['bilstm', 'lstm', 'birnn', 'rnn'],
                    help='type of recurrent layer to use (lstm, bilstm, rnn, birnn)')

args = parser.parse_args(gen_be=False)

# hyperparameters from the reference
args.batch_size = 128
gradient_clip_value = 15
vocab_size = 20000
sentence_length = 128
embedding_dim = 128
hidden_size = 128
reset_cells = True

# setup backend
be = gen_backend(**extract_valid_args(args, gen_backend))

# make dataset
path = load_imdb(path=args.data_dir)
(X_train, y_train), (X_test, y_test), nclass = pad_data(path,
                                                        vocab_size=vocab_size,
                                                        sentence_length=sentence_length)

print "Vocab size - ", vocab_size
print "Sentence Length - ", sentence_length
print "# of train sentences", X_train.shape[0]
print "# of test sentence", X_test.shape[0]

train_set = ArrayIterator(X_train, y_train, nclass=2)
valid_set = ArrayIterator(X_test, y_test, nclass=2)

# weight initialization
uni = Uniform(low=-0.1/embedding_dim, high=0.1/embedding_dim)
g_uni = GlorotUniform()

if args.rlayer_type == 'lstm':
    rlayer = LSTM(hidden_size, g_uni, activation=Tanh(),
                  gate_activation=Logistic(), reset_cells=True)
elif args.rlayer_type == 'bilstm':
    rlayer = DeepBiLSTM(hidden_size, g_uni, activation=Tanh(), depth=1,
                        gate_activation=Logistic(), reset_cells=True)
elif args.rlayer_type == 'rnn':
    rlayer = Recurrent(hidden_size, g_uni, activation=Tanh(), reset_cells=True)
elif args.rlayer_type == 'birnn':
    rlayer = DeepBiRNN(hidden_size, g_uni, activation=Tanh(), depth=1, reset_cells=True)

layers = [
    LookupTable(vocab_size=vocab_size, embedding_dim=embedding_dim, init=uni),
    rlayer,
    RecurrentSum(),
    Dropout(keep=0.5),
    Affine(2, g_uni, bias=g_uni, activation=Softmax())
]

model = Model(layers=layers)

cost = GeneralizedCost(costfunc=CrossEntropyMulti(usebits=True))
optimizer = Adagrad(learning_rate=0.01, gradient_clip_value=gradient_clip_value)

# configure callbacks
callbacks = Callbacks(model, eval_set=valid_set, **args.callback_args)

# train model
model.fit(train_set, optimizer=optimizer, num_epochs=args.epochs, cost=cost, callbacks=callbacks)

# eval model
print "Train Accuracy - ", 100 * model.eval(train_set, metric=Accuracy())
print "Test  Accuracy - ", 100 * model.eval(valid_set, metric=Accuracy())


text:
import fileinput
import re
import sys
import urllib2

"""
Assume two words and no empty strings as input.
"""

def get_dictionary(filename='./words.txt'):
# '/usr/share/dict/words' has a lot of clutter/gibberish that matched single characters.
# not just obvious ones like 'a' and 'I'.
# I replaced this with a smaller subset of english words I found via google.
  with open(filename, 'r') as words:
    webster = words.read().splitlines()
  return webster

def check_dictionary(word, base_url='http://www.dictionaryapi.com/api/v1/references/collegiate/xml',key='a520af22-06e0-4ef8-8267-b5c72b185466'):
# this was fun, but somewhat ineffective for similar reasons as '/usr/share/dict/words'.
# Inlcudes single letters as 'F' letter of alphabet, so every character matches.
# This prevents effective tokenizing on units of 1, since they will all match.
# We'd have to assume words are 2 or more characters, and it's still dicey then.
  url = '%s/%s?key=%s' % (base_url, word.lower(), key)
  #print url
  resp = urllib2.urlopen(url)
  resp_text = resp.read()
  #print resp_text
  match = re.search(r'hindex="\d+">%s' % word.lower(), resp_text, re.MULTILINE)
  if match:
    return True
  else:
    return None

def test_line(line, webster=None):
  words = []
  for i in xrange(len(line)):
    left = line[:i]
    right = line[i:]
    if webster:
      if left in webster and right in webster:
        words = [left, right]
    else:
      if check_dictionary(left) and check_dictionary(right):
        words = [left, right]
  return words

webster = get_dictionary()
for line in fileinput.input():
  line = line.rstrip()
  print 'our input string is "%s"' % line
  #print test_line(line) # test with dictionary.com api
  print test_line(line, webster) # test with local file dictionary


text:
<?php

namespace app\models\Base;

use app\models\Glauk\Glaukuchet;
use Yii;
use app\func\Proc;

/**
 * This is the model class for table "class_mkb".
 *
 * @property integer $id
 * @property string $name
 * @property string $code
 * @property integer $parent_id
 * @property string $parent_code
 * @property integer $node_count
 * @property string $additional_info
 *
 * @property Classmkb $parent
 * @property Classmkb[] $classmkbs
 * @property Glaukuchet[] $glaukuchets
 */
class Classmkb extends \yii\db\ActiveRecord {

    /**
     * @inheritdoc
     */
    public static function tableName() {
        return 'class_mkb';
    }

    /**
     * @inheritdoc
     */
    public function rules() {
        return [
            [['name', 'code'], 'required'],
            [['parent_id', 'node_count'], 'integer'],
            [['additional_info'], 'string'],
            [['name'], 'string', 'max' => 512],
            [['code', 'parent_code'], 'string', 'max' => 20],
            [['parent_id'], 'exist', 'skipOnError' => true, 'targetClass' => Classmkb::className(), 'targetAttribute' => ['parent_id' => 'id']],
        ];
    }

    /**
     * @inheritdoc
     */
    public function attributeLabels() {
        return [
            'id' => 'PK',
            'name' => 'Наименование диагноза',
            'code' => 'Код МКБ10',
            'parent_id' => 'Вышестоящий объект',
            'parent_code' => 'Код вышестоящего объекта',
            'node_count' => 'Количество вложенных в текущую ветку',
            'additional_info' => 'Дополнительные данные по диагнозу',
        ];
    }

    /**
     * @return \yii\db\ActiveQuery
     */
    public function getParent() {
        return $this->hasOne(Classmkb::className(), ['id' => 'parent_id'])->from(['parent' => Classmkb::tableName()]);
    }

    /**
     * @return \yii\db\ActiveQuery
     */
    public function getClassmkbs() {
        return $this->hasMany(Classmkb::className(), ['parent_id' => 'id'])->from(['classmkbs' => Classmkb::tableName()]);
    }

    /**
     * @return \yii\db\ActiveQuery
     */
    public function getGlaukuchets() {
        return $this->hasMany(Glaukuchet::className(), ['id_class_mkb' => 'id'])->from(['glaukuchets' => Glaukuchet::tableName()]);
    }

    public function selectinput($params) {
        $method = isset($params['init']) ? 'one' : 'all';

        // Меняем раскладку на английскую при вводе МКБ10
        if ($method === 'all') {
            preg_match('/^([а-яА-Я]\d)/ui', $params['q'], $match);
            if (!empty($match[1]))
                $params['q'] = Proc::switcher($params['q']);
        }

        $query = self::find()
                ->select(array_merge(isset($params['init']) ? [] : [self::primaryKey()[0] . ' AS id'], ['CONCAT_WS(" - ", code, name) AS text']))
                ->where(['node_count' => 0])
                ->andwhere(['or', ['like', isset($params['init']) ? 'id' : 'code', $params['q'], isset($params['init']) ? false : null], $method === 'all' ? ['like', 'name', $params['q']] : '1<>1'])
                ->andwhere(['or', ['like', 'code', 'H40%', false], ['like', 'code', 'Q15.0', false]])
                ->limit(10)
                ->asArray()
                ->$method();

        return $query;
    }

}


text:
package api

import (
	"encoding/json"
	"errors"
	"sync"

	"github.com/gorilla/websocket"
	"github.com/grafana/metrictank/stats"
	"github.com/raintank/raintank-apps/pkg/message"
	"github.com/raintank/raintank-apps/task-server/agent_session"
	"github.com/raintank/raintank-apps/task-server/model"
	"github.com/raintank/raintank-apps/task-server/sqlstore"
	"github.com/raintank/worldping-api/pkg/log"
)

var (
	taskServerAgentConnectionsActiveCount   = stats.NewGauge64("agent.connections.active")
	taskServerAgentConnectionsFailedCount   = stats.NewCounter64("agent.connections.failed")
	taskServerAgentConnectionsAcceptedCount = stats.NewCounter64("agent.connections.accepted")
	taskServerAgentAutoCreateSuccessCount   = stats.NewCounter64("agent.autocreate.success")
	taskServerAgentAutoCreateFailedCount    = stats.NewCounter64("agent.autocreate.failed")
)

var upgrader = websocket.Upgrader{} // use default options

type socketList struct {
	sync.RWMutex
	Sockets map[int64]*agent_session.AgentSession
}

func (s *socketList) CloseAll() {
	s.Lock()
	for _, sock := range s.Sockets {
		sock.Close()
	}
	s.Sockets = make(map[int64]*agent_session.AgentSession, 0)
	s.Unlock()
}

func (s *socketList) EmitTask(task *model.TaskDTO, event string) error {
	log.Debug("sending %s task event to connected agents.", event)
	agents, err := sqlstore.GetAgentsForTask(task)
	log.Debug("Task has %d agents. %v", len(agents), agents)
	if err != nil {
		return err
	}
	body, err := json.Marshal(task)
	if err != nil {
		return err
	}
	e := &message.Event{
		Event:   event,
		Payload: body,
	}
	sent := false
	s.Lock()

	for _, id := range agents {
		if as, ok := s.Sockets[id]; ok {
			log.Debug("sending %s event to agent %d", event, id)
			as.SocketSession.Emit(e)
			sent = true
		} else {
			log.Debug("agent %d is not connected to this server.", id)
		}
	}
	s.Unlock()
	if !sent {
		log.Debug("no connected agents for task %d.", task.Id)
	}
	return nil
}

func (s *socketList) NewSocket(a *agent_session.AgentSession) {
	s.Lock()
	existing, ok := s.Sockets[a.Agent.Id]
	if ok {
		log.Debug("new connection for agent %d - %s, closing existing session", a.Agent.Id, a.Agent.Name)
		existing.Close()
	}
	log.Debug("Agent %d is connected to this server.", a.Agent.Id)
	s.Sockets[a.Agent.Id] = a
	taskServerAgentConnectionsAcceptedCount.Inc()
	taskServerAgentConnectionsActiveCount.Inc()
	s.Unlock()
}

func (s *socketList) DeleteSocket(a *agent_session.AgentSession) {
	s.Lock()
	s.deleteSocket(a.Agent.Id)
	s.Unlock()
}

func (s *socketList) deleteSocket(id int64) {
	delete(s.Sockets, id)
	taskServerAgentConnectionsActiveCount.Dec()
}

func (s *socketList) CloseSocket(a *agent_session.AgentSession) {
	s.Lock()
	existing, ok := s.Sockets[a.Agent.Id]
	if ok {
		existing.Close()
		log.Debug("removing session for Agent %d from socketList.", a.Agent.Id)
		s.deleteSocket(a.Agent.Id)
	}
	s.Unlock()
}

func (s *socketList) CloseSocketByAgentId(id int64) {
	s.Lock()
	existing, ok := s.Sockets[id]
	if ok {
		existing.Close()
		log.Debug("CloseSocketByAgentId: removing session for Agent %d from socketList.", id)
		s.deleteSocket(id)
	}
	s.Unlock()
}

func newSocketList() *socketList {
	return &socketList{
		Sockets: make(map[int64]*agent_session.AgentSession),
	}
}

var ActiveSockets *socketList

func init() {
	ActiveSockets = newSocketList()
}

func connectedAgent(agentName string, orgID int64) (*model.AgentDTO, error) {
	if agentName == "" {
		return nil, errors.New("connectedAgent: agent name not specified")
	}
	q := model.GetAgentsQuery{
		Name:  agentName,
		OrgId: orgID,
	}
	agents, err := sqlstore.GetAgents(&q)
	if err != nil {
		return nil, err
	}
	if len(agents) < 1 {
		log.Debug("connectedAgent: Agent %s not found, attempting to create.", agentName)

		// auto add agent
		anAgent := model.AgentDTO{
			Name:    agentName,
			OrgId:   orgID,
			Enabled: true,
			Public:  true,
		}
		addAgentErr := sqlstore.AddAgent(&anAgent)
		if addAgentErr != nil {
			log.Error(3, addAgentErr.Error())
			taskServerAgentConnectionsFailedCount.Inc()
			taskServerAgentAutoCreateFailedCount.Inc()
			return nil, errors.New("connectedAgent: failed to auto create agent.")
		}
		log.Debug("connectedAgent: Agent %s created.", agentName)

		agentsAfter, err := sqlstore.GetAgents(&q)
		if err != nil {
			return nil, err
		}
		if len(agentsAfter) < 1 {
			taskServerAgentConnectionsFailedCount.Inc()
			taskServerAgentAutoCreateFailedCount.Inc()
			return nil, errors.New("connectedAgent: agent not found after autocreate.")
		}
		log.Debug("connectedAgent: Allowing new Agent %s.", agentName)
		taskServerAgentConnectionsAcceptedCount.Inc()
		taskServerAgentAutoCreateSuccessCount.Inc()

		return agentsAfter[0], nil
	}

	return agents[0], nil
}

func socket(ctx *Context) {
	agentName := ctx.Params(":agent")
	agentVer := ctx.ParamsInt64(":ver")
	//TODO: add auth
	owner := ctx.OrgId
	log.Debug("socket: agent name %s", agentName)
	log.Debug("socket: agent ver %d", agentVer)
	log.Debug("socket: agent orgid %d", owner)
	agent, err := connectedAgent(agentName, owner)
	if err != nil {
		taskServerAgentConnectionsFailedCount.Inc()
		log.Debug("socket: agent cant connect. %s", err)
		ctx.JSON(400, err.Error())
		return
	}

	c, err := upgrader.Upgrade(ctx.Resp, ctx.Req.Request, nil)
	if err != nil {
		log.Error(3, "socket: upgrade:", err)
		return
	}

	log.Debug("socket: agent %s connected.", agent.Name)

	sess := agent_session.NewSession(agent, agentVer, c)
	ActiveSockets.NewSocket(sess)
	sess.Start()
	//block until connection closes.
	<-sess.Done
	ActiveSockets.DeleteSocket(sess)
}


text:
declare const PlainArrowDirectionMixin: Mixin<{}, {}>;

export default PlainArrowDirectionMixin;