//设置package包名称以及导入依赖的类
package nl.bigdatarepublic.streaming.embedded.adapter.kafka

import java.util.Properties

import com.typesafe.scalalogging.LazyLogging
import kafka.server.{KafkaConfig, KafkaServerStartable}
import nl.bigdatarepublic.streaming.embedded.entity.EmbeddedService

import scala.collection.JavaConverters._
import scala.reflect.io.Path
import scala.util.{Failure, Success, Try}


class EmbeddedKafka(props: Map[String, String], clearState: Boolean) extends LazyLogging with EmbeddedService {

  val kafka: KafkaServerStartable = new KafkaServerStartable(KafkaConfig(props.asJava))

  def start(): Unit = {
    // Clear out the existing kafka dir upon startup.
    if (clearState) {
      logger.info("Cleaning Kafka data dir before start...")
      kafka.serverConfig.logDirs.foreach { x =>
        Try(Path(x).deleteRecursively()) match {
          case Success(true) => logger.info("Successfully cleaned Kafka data dir...")
          case Success(false) => logger.info("Failed to clean Kafka data dir...")
          case Failure(e) => logger.warn("Failed to clean Kafka data dir", e)
        }
      }
    }
    logger.info("Starting embedded Kafka...")
    kafka.startup()
    logger.info("Successfully started embedded Kafka")

  }

  def stop(): Unit = {
    logger.info("Stopping embedded Kafka...")
    kafka.shutdown()
    logger.info("Successfully stopped embedded Kafka")

  }

}



object EmbeddedKafka {
  def apply(props: Map[String, String], clearState: Boolean): EmbeddedKafka = new EmbeddedKafka(props, clearState)
  def apply(props: Map[String, String]): EmbeddedKafka = new EmbeddedKafka(props, false)

  // Java compatibility
  def apply(props: Properties, clearState: Boolean): EmbeddedKafka = new EmbeddedKafka(props.asScala.toMap, clearState)
  def apply(props: Properties): EmbeddedKafka = new EmbeddedKafka(props.asScala.toMap, false)
} 

//设置package包名称以及导入依赖的类
package com.mbesida.synchronizer

import java.io.PrintWriter

import akka.actor.{ActorSystem, Props}
import akka.testkit.{ImplicitSender, TestKit}
import au.com.bytecode.opencsv.CSVWriter
import com.mbesida.synchronizer.ReadFileActor.{ReadResult, ReadValue}
import com.mbesida.synchronizer.ReadWriteFileActor.{FailedModification, ModifyFile, SuccessModification}
import org.junit.runner.RunWith
import org.scalatest.junit.JUnitRunner
import org.scalatest.{BeforeAndAfterAll, Matchers, WordSpecLike}

import scala.reflect.io.Path

@RunWith(classOf[JUnitRunner])
class ReadWriteFileActorSpec extends TestKit(ActorSystem("test"))
  with ImplicitSender
  with Matchers
  with WordSpecLike
  with BeforeAndAfterAll {


  override def beforeAll(): Unit = {
    val out = new CSVWriter(new PrintWriter("sampleFile.csv"))
    out.writeNext(Array("1.45", "2", "0.45"))
    out.close()
  }

  "ReadWriteFileActor" should {
    "modify empty file success scenario" in {
      val actor = system.actorOf(Props(classOf[ReadWriteFileActor], "samplef2.csv"))
      actor ! ModifyFile(0, 2.4)
      expectMsg(SuccessModification)
    }
    "modify empty file failure scenario" in {
      val actor = system.actorOf(Props(classOf[ReadWriteFileActor], "samplef2.csv"))
      actor ! ModifyFile(10, 20)
      expectMsg(FailedModification)
    }
    "modify non empty file" in {
      val actor = system.actorOf(Props(classOf[ReadWriteFileActor], "sampleFile.csv"))
      actor ! ReadValue(2)
      expectMsg(ReadResult(Right(0.45)))
      actor ! ModifyFile(1, 3.14)
      expectMsg(SuccessModification)
      actor ! ReadValue(1)
      expectMsg(ReadResult(Right(3.14)))
    }
  }


  override def afterAll(): Unit = {
    Path("samplef2.csv").deleteIfExists()
    Path("sampleFile.csv").deleteIfExists()
  }

} 

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

import testbed._

import scala.reflect.io.{File, Path}
import scala.sys.process.{ProcessLogger, Process}

class Numpy[ResultType](val pathToPython: String, numpyImportAlias: String = "np") {

  def isInstalled = {
    val testPythonCmd = Process(s"$pathToPython --version")
    val testNumpyCmd = Process(Seq(pathToPython, "-c",  "'import numpy; print(numpy.version)'"))

    def pythonInstalled = {
      var result = ""
      testPythonCmd ! ProcessLogger(line => result += line)

      result contains "Python"
    }

    def numpyInstalled = {
      var result = ""
      testNumpyCmd ! ProcessLogger(line => result += line)

      !(result contains "Error")
    }

    pythonInstalled && numpyInstalled
  }

  val NOT_INSTALLED_MSG: String = s"No numpy installation found with $pathToPython."

  val PYTHON_SRC_FILE_NAME: String = "src_buffer.py"

  def getResult(numpyExpr: String, bufferName: String, resultReader: String => ResultType) = {
    assert(isInstalled, NOT_INSTALLED_MSG)
    val absolutePathToBuffer = Path(bufferName).toAbsolute.path
    val pythonSrcFile = File(PYTHON_SRC_FILE_NAME)
    val pythonFileWriteExpr =
      s"""import numpy as $numpyImportAlias
         |import csv
         |with open('$absolutePathToBuffer', 'w') as buffer_file:
         |        csv_writer = csv.writer(buffer_file, delimiter=',')
         |        for result_line in $numpyExpr:
         |            result_line = [result_line]
         |            csv_writer.writerow(result_line)
       """.stripMargin

    pythonSrcFile.writeAll(pythonFileWriteExpr)

    Process(Seq(pathToPython, pythonSrcFile.toAbsolute.path))!

    resultReader(bufferName)
  }
}