//设置package包名称以及导入依赖的类
package data.processing.kafkagenerator

import java.util.Properties
import java.util.concurrent.TimeUnit

import com.typesafe.config.ConfigFactory
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import com.github.andr83.scalaconfig._
import com.yammer.metrics.core.{MetricName, MetricsRegistry}
import data.processing.avro.AvroEncoder

import scala.concurrent.forkjoin.ThreadLocalRandom


object Generator {

  val metricsRegistry = new MetricsRegistry

  val config = ConfigFactory.load()
  val props = config.getConfig("kafka-client").as[Properties]
  val topic = config.getString("kafka-client.topic")
  val numberOfUsers = config.getInt("generator.number.of.users")
  val urls = config.getStringList("generator.urls")
  val eventTypes = config.getStringList("generator.event.types")
  val throughput = config.getInt("generator.throughput")

  val avroEncoder = new AvroEncoder("/event-record.json")

  def generateEvent() = {
    val id = ThreadLocalRandom.current().nextLong()
    val ts = java.lang.System.currentTimeMillis()
    val userId = ThreadLocalRandom.current().nextInt(numberOfUsers).toHexString
    val url = urls.get(ThreadLocalRandom.current().nextInt(urls.size()))
    val eventType = eventTypes.get(ThreadLocalRandom.current().nextInt(eventTypes.size()))

    (id, avroEncoder.encode((id, ts, userId, url, eventType)))
  }

  def main(args: Array[String]): Unit = {
    val meter = metricsRegistry.newMeter(new MetricName("", "", ""), "", TimeUnit.SECONDS)
    val producer = new KafkaProducer[String, Array[Byte]](props)
    while(true) {
      if (meter.meanRate < throughput) {
        meter.mark()
        val event = generateEvent()
        producer.send(new ProducerRecord[String, Array[Byte]](topic, event._1.toString, event._2))
      }
      else {
        Thread.sleep(1)
      }
    }
    producer.flush()
    producer.close()
  }
} 

//设置package包名称以及导入依赖的类
package bbc.geoip

import scala.concurrent.duration._

import io.gatling.core.Predef._
import io.gatling.http.Predef._

trait GeoIpProtocol extends Simulation {

  val httpProtocol = http
    .baseURL("https://test-lo-geoip.tools.bbc.co.uk/")
    .shareConnections
}

class GeoIpFixedIps extends GeoIpProtocol {

  val ipFeeder = csv("geoip/fixed_ips.txt").circular

  val scn = scenario("GeoIP")
    .feed(ipFeeder)

    .exec(http("GeoIP Fixed Ips")
    .get("${ip}")
      .check(status is 200))
    
  setUp(scn.inject(
    rampUsersPerSec(10) to(3000) during(30 minutes)
  ).protocols(httpProtocol))
}

class GeoIpDynamicIps extends GeoIpProtocol {

  def createDynamicIp(): String = {
    import scala.concurrent.forkjoin.ThreadLocalRandom

    def octet = ThreadLocalRandom.current().nextInt(0, 256)
    s"$octet.$octet.$octet.$octet"
  }

  val scn = scenario("GeoIP")
    .exec(_.set("dynamicIp", createDynamicIp))
    .exec(http("GeoIP Dynamic Ips")
    .get("${dynamicIp}")
      .check(status is 200))
    
  setUp(scn.inject(
    rampUsersPerSec(10) to(3000) during(30 minutes)
  ).protocols(httpProtocol))
} 

//设置package包名称以及导入依赖的类
package org.jboss.perf

import java.util.UUID

import scala.concurrent.forkjoin.ThreadLocalRandom
import scala.util.Random


object Util {
  val random = new Random(1234); // keep fixed seed

  def randomString(length: Int, rand: Random = ThreadLocalRandom.current()): String = {
    val sb = new StringBuilder;
    for (i <- 0 until length) {
      sb.append((rand.nextInt(26) + 'a').toChar)
    }
    sb.toString()
  }

  def randomUUID(rand: Random = ThreadLocalRandom.current()): String =
    new UUID(rand.nextLong(), rand.nextLong()).toString
} 

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

import akka.actor.{Actor, ActorRef, ActorSelection, Address, Props, RelativeActorPath, RootActorPath}
import akka.actor.Actor.Receive
import akka.cluster.Cluster
import akka.cluster.ClusterEvent._
import messages._

import scala.concurrent.forkjoin.ThreadLocalRandom


object PizzaClient {
  def props(servicePath: String)(user: ActorRef) = Props(classOf[PizzaClient], servicePath, user)
}

class PizzaClient(servicePath: String, user: ActorRef) extends Actor {
  val cluster = Cluster(context.system)
  var nodes = Set.empty[Address]
  val servicePathElements = servicePath match {
    case RelativeActorPath(elements) => elements
    case _ => throw new IllegalArgumentException("servicePath [%s] is not a valid relative actor path" format servicePath)
  }

  override def preStart(): Unit = {
    cluster.subscribe(self, initialStateMode = InitialStateAsEvents, classOf[MemberEvent], classOf[ReachabilityEvent])
  }

  override def postStop(): Unit = {
    cluster.unsubscribe(self)
  }

  override def receive: Receive = {
    case order: PizzaCustomerOrder => {
      // find service actor
      val service = findServiceActor()
      service ! order
    }
    case stop: PizzaCustomerStop => {
      // find service actor
      val service = findServiceActor()
      service ! stop
    }
    case result: PizzaCustomerBaked => {
      user ! result
    }
    case PizzaCustomerPing => sender() ! PizzaCustomerPong
    case MemberUp(m) if m.hasRole("chef") => nodes += m.address
    case ReachableMember(m) if m.hasRole("chef") => nodes += m.address
    case otherMemberEvent: MemberEvent => nodes -= otherMemberEvent.member.address
    case UnreachableMember(m) => nodes -= m.address
  }

  private def findServiceActor(): ActorSelection = {
    // for load balance, select random master proxy
    val address = nodes.toIndexedSeq(ThreadLocalRandom.current.nextInt(nodes.size))
    context.actorSelection(RootActorPath(address) / servicePathElements)
  }
} 

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

import java.util.UUID
import scala.concurrent.forkjoin.ThreadLocalRandom
import scala.concurrent.duration._
import akka.actor.Actor
import akka.actor.ActorLogging
import akka.actor.ActorRef

object WorkProducer {
  case object Tick
}

class WorkProducer(frontend: ActorRef) extends Actor with ActorLogging {
  import WorkProducer._
  import context.dispatcher
  def scheduler = context.system.scheduler
  def rnd = ThreadLocalRandom.current
  def nextWorkId(): String = UUID.randomUUID().toString

  var n = 0

  override def preStart(): Unit =
    scheduler.scheduleOnce(5.seconds, self, Tick)

  // override postRestart so we don't call preStart and schedule a new Tick
  override def postRestart(reason: Throwable): Unit = ()

  def receive = {
    case Tick =>
      n += 1
      log.info("Produced work: {}", n)
      val work = Work(nextWorkId(), n)
      frontend ! work
      context.become(waitAccepted(work), discardOld = false)

  }

  def waitAccepted(work: Work): Actor.Receive = {
    case Frontend.Ok =>
      context.unbecome()
      scheduler.scheduleOnce(rnd.nextInt(3, 10).seconds, self, Tick)
    case Frontend.NotOk =>
      log.info("Work not accepted, retry after a while")
      scheduler.scheduleOnce(3.seconds, frontend, work)
  }

} 

//设置package包名称以及导入依赖的类
package de.tuberlin.cit.cost.datagen.flink

import org.apache.flink.api.scala._
import org.apache.flink.core.fs.FileSystem

import scala.concurrent.forkjoin.ThreadLocalRandom
import scala.math.pow

object DataSetGenerator {
  def main(args: Array[String]) {
    if (args.length != 3) {
      Console.err.println("Usage: DataSetGenerator <points> <dimension> <outputPath>")
      System.exit(-1)
    }

    val m = args(0).toInt
    val n = args(1).toInt
    val outputPath = args(2)

    val env = ExecutionEnvironment.getExecutionEnvironment

    env
      .generateSequence(1, m)
      .map(_ => {
        val x = ThreadLocalRandom.current().nextDouble()
        val noise = ThreadLocalRandom.current().nextGaussian()

        // generate the function value with added gaussian noise
        val label = function(x) + noise

        // generate a vandermatrix from x
        val vector = polyvander(x, n - 1)

        (vector :+ label).mkString(" ")
      })
      .writeAsText(outputPath, FileSystem.WriteMode.OVERWRITE)

    env.execute("Data Set Generator")
  }

  def polyvander(x: Double, order: Int): Array[Double] = {
    (0 to order).map(pow(x, _)).toArray
  }

  def function(x: Double): Double = {
    2 * x + 10
  }
}