//设置package包名称以及导入依赖的类
package org.dama.datasynth.runtime.spark.operators

import org.apache.spark.sql.{Dataset, SparkSession}
import org.dama.datasynth.executionplan.ExecutionPlan.EdgeTable
import org.dama.datasynth.runtime.spark.SparkRuntime

import scala.util.Random


  def apply( node : EdgeTable) : Dataset[(Long,Long,Long)]= {
    val sparkSession = SparkRuntime.getSparkSession()
    import sparkSession.implicits._
    val generator = SparkRuntime.instantiateStructureGeneratorOperator( node.structure )
    val size = SparkRuntime.evalValueOperator(node.size).asInstanceOf[Long]
    val random : Random = new Random()
    val id : Int = random.nextInt()
    val path : String = s"/tmp/${id}"
    val sparkContext = sparkSession.sparkContext
    generator.run(size, sparkContext.hadoopConfiguration,"hdfs://"+path)
    val edgesRDD = sparkContext.textFile(path)
                               .map( s => s.split("\t"))
                               .map( l => (l(0).toLong, l(1).toLong))
                               .zipWithIndex().map( { case ((tail,head), id) =>  (id, tail, head)})
    sparkSession.createDataset(edgesRDD)
  }

} 

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

import breeze.linalg._
import fregata.model.classification.{LogisticRegressionModel, SoftMax, SoftMaxModel}

import scala.util.Random
import fregata.model.classification.LogisticRegression
import fregata.data.LibSvmReader
import fregata.loss.LogLoss
import fregata.metrics.classification.{Accuracy, AreaUnderRoc}
import fregata.{Vector => _, _}


object TestLogisticRegression {

  def main(args: Array[String]) {
    val (_,trainData) = LibSvmReader.read("/Volumes/takun/data/libsvm/a9a",123)
    val (_,testData) = LibSvmReader.read("/Volumes/takun/data/libsvm/a9a.t",123)
    println(" load over ...")
    val r = new Random(1L)
    val data2 = r.shuffle(trainData.toList)
    val model = new LogisticRegression()
      .run(data2,10)
    val predicted = model.classPredict(testData)
    val acc = Accuracy.of(predicted.map{
      case ((x,l),(p,c)) =>
        c -> l
    })
    val auc = AreaUnderRoc.of(predicted.map{
      case ((x,l),(p,c)) =>
        p -> l
    } )
    val loss = new LogLoss().of(predicted.map{
      case ((x,l),(p,c)) =>
        if( l == 1d ) {
          (l,c,p)
        }else{
          ( l , c , 1-p )
        }
    })
    println( s"Accuracy : ${acc} " )
    println( s"AreaUnderRoc : ${auc} " )
    println( s"logLoss : ${loss} " )
  }
} 

//设置package包名称以及导入依赖的类
package linalg.sgd
import scala.util.Random
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.optimization.GradientDescent
import org.apache.spark.mllib.optimization.SquaredL2Updater
import org.apache.spark.mllib.optimization.LogisticGradient
import org.apache.spark.SparkContext



object SparkSGD {
  def main(args: Array[String]): Unit = {
    val m = 4
    val n = 200000
    val sc = new SparkContext("local[2]", "")
    val points = sc.parallelize(0 until m, 2).mapPartitionsWithIndex { (idx, iter) =>
      val random = new Random(idx)
      iter.map(i => (1.0, Vectors.dense(Array.fill(n)(random.nextDouble()))))
    }.cache()
    val (weights, loss) = GradientDescent.runMiniBatchSGD(
      points,
      new LogisticGradient,
      new SquaredL2Updater,
      0.1,
      2,
      1.0,
      1.0,
      Vectors.dense(new Array[Double](n)))
    println("w:"  + weights(0))
    println("loss:" + loss(0))
    sc.stop()

  }
} 

//设置package包名称以及导入依赖的类
import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence
import org.eclipse.paho.client.mqttv3.{MqttClient, MqttMessage}

import scala.util.{Failure, Random, Success, Try}


object Publisher extends App {
  // Global information
  val name = "apdl-case-study-from-scratch"

  // MQTT Server info to publish
  val brokerURL = "tcp://mosquitto-mqtt:1883"
  val topicTemperature = s"temp"
  val topicLuminosity = s"light"
  val persistence = new MqttDefaultFilePersistence("tmp")

  Try {
    val client = new MqttClient(brokerURL, MqttClient.generateClientId(), persistence)
    client.connect()

    val topicTemp = client.getTopic(topicTemperature)
    val topicLight = client.getTopic(topicLuminosity)

    while (true) {

      val temp = Random.nextInt(10) + 20
      val light = Random.nextInt(40) + 640
      val msgTemp = new MqttMessage(s"$temp".getBytes)
      val msgLight = new MqttMessage(s"$light".getBytes)

      println(s"Publish $temp to $topicTemperature")
      println(s"Publish $light to $topicLight")

      topicTemp.publish(msgTemp)
      topicLight.publish(msgLight)
      Thread.sleep(1000)
    }
    while(true) {
      println("OK !")
      Thread.sleep(1000)
    }

  } match {
    case Failure(exception) => println(s"ERROR : $exception + ${exception.getCause}")
    case Success(_) => println(s"OK !")
  }
} 

//设置package包名称以及导入依赖的类
package org.dl4scala.examples.misc.embedding

import org.deeplearning4j.nn.conf.NeuralNetConfiguration
import org.deeplearning4j.nn.conf.layers.{EmbeddingLayer, GravesLSTM, RnnOutputLayer}
import org.deeplearning4j.nn.conf.preprocessor.{FeedForwardToRnnPreProcessor, RnnToFeedForwardPreProcessor}
import org.deeplearning4j.nn.multilayer.MultiLayerNetwork
import org.nd4j.linalg.activations.Activation
import org.nd4j.linalg.factory.Nd4j
import org.nd4j.linalg.lossfunctions.LossFunctions

import scala.util.Random


object RNNEmbedding {
  def main(args: Array[String]): Unit = {
    val nClassesIn = 10
    val batchSize = 3
    val timeSeriesLength = 8
    val inEmbedding = Nd4j.create(batchSize, 1, timeSeriesLength)
    val outLabels = Nd4j.create(batchSize, 4, timeSeriesLength)

    val r = new Random(12345)
    (0 until batchSize).foreach{i =>
      (0 until timeSeriesLength).foreach{j =>
        val classIdx = r.nextInt(nClassesIn)
        inEmbedding.putScalar(Array[Int](i, 0, j), classIdx)
        val labelIdx = r.nextInt(4)
        outLabels.putScalar(Array[Int](i, labelIdx, j), 1.0)
      }
    }

    val conf = new NeuralNetConfiguration.Builder()
      .activation(Activation.RELU)
      .list()
      .layer(0, new EmbeddingLayer.Builder().nIn(nClassesIn).nOut(5).build())
      .layer(1, new GravesLSTM.Builder().nIn(5).nOut(7).activation(Activation.SOFTSIGN).build())
      .layer(2, new RnnOutputLayer.Builder(LossFunctions.LossFunction.MCXENT).nIn(7).nOut(4).activation(Activation.SOFTMAX).build())
      .inputPreProcessor(0, new RnnToFeedForwardPreProcessor())
      .inputPreProcessor(1, new FeedForwardToRnnPreProcessor())
      .build()

    val net = new MultiLayerNetwork(conf)
    net.init()

    net.setInput(inEmbedding)
    net.setLabels(outLabels)

    net.computeGradientAndScore()
    System.out.println(net.score())
  }
} 

//设置package包名称以及导入依赖的类
package com.jetprobe.fastgen.generators.entity

import com.jetprobe.fastgen.generators._
import com.typesafe.config.Config

import scala.util.matching.Regex


class PersonGen(datasetConfig: Config, regexMatcher: Regex.MatchIterator)
    extends EntityGenerator(datasetConfig, regexMatcher)

object PersonGen {
  import scala.util.Random
  def apply(datasetConfig: Config,
            regexMatcher: Regex.MatchIterator): PersonGen = {
    val person = new PersonGen(datasetConfig, regexMatcher)

    // firstName
    person.addField("Person.firstName", StringType, (fieldOpt: FieldOption) => {
      val arr = person.dataset.get(fieldOpt.getName).get
      Generator.getRNG(arr.length, arr)
    })

    //lastName
    person.addField("Person.lastName", StringType, (fieldOpt: FieldOption) => {
      val arr = person.dataset.get(fieldOpt.getName).get
      Generator.getRNG(arr.length, arr)
    })

    //Support for parsing and generating age
    person.addField(
      "Person.age",
      IntType,
      (fieldOpt: FieldOption) => {
        val randomInt = fieldOpt match {
          case opt: IntOption =>
            opt.minVal + Random.nextInt((opt.maxVal - opt.minVal) + 1)
          case _ => Random.nextInt(100)
        }
        randomInt.toString
      }
    )

    person

  }

} 

//设置package包名称以及导入依赖的类
package com.init6.coders.commands

import com.init6.Constants._
import com.init6.channels.{User, UserError}

import scala.util.{Random, Try}


object RollCommand {

  lazy val rollRandom = new Random(System.currentTimeMillis)

  def apply(fromUser: User, message: String): Command = {
    val (min, max) = {
      if (message.nonEmpty) {
        val splt = message.split("-")
        if (splt.length > 2 || (splt.nonEmpty && !splt.forall(_.forall(_.isDigit)))) {
          return UserError(ROLL_FORMAT)
        }

        Try {
          if (splt.length == 1) {
            1 -> splt(0).toInt
          } else {
            splt(0).toInt -> splt(1).toInt
          }
        }.getOrElse(return UserError(ROLL_FORMAT))
      } else {
        1 -> 100
      }
    }

    if (min > max) {
      return UserError(ROLL_FORMAT)
    }

    val roll = rollRandom.nextInt(max - min + 1) + min

    RollCommandToChannel(fromUser, ROLL_INFO(fromUser.name, roll, min, max))
  }
}

case class RollCommandToChannel(override val fromUser: User, override val message: String) extends ChannelBroadcastable 

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

import java.util.concurrent.TimeUnit

import akka.actor.{ActorRef, PoisonPill}
import com.init6.channels.ChannelsActor
import com.init6.connection.websocket.WebSocketConnectionHandler
import com.init6.connection.{ConnectionHandler, IpLimitActor}
import com.init6.db.{DAO, DAOActor}
import com.init6.servers.{ServerAnnouncementActor, ServerRegistry}
import com.init6.users.{RankingActor, TopCommandActor, UsersActor}

import scala.concurrent.Await
import scala.concurrent.duration.Duration
import scala.util.Random


object Init6 extends App with Init6Component {

  DAO
  ServerRegistry()
  DAOActor()
  IpLimitActor(Config().Accounts.connectionLimit)
  UsersActor()
  ChannelsActor()
  TopCommandActor()
  RankingActor()
  ServerAnnouncementActor(args(0).toLong)

  val random = new Random(System.nanoTime())
  val delay =
    if (random.nextInt(100) < Config().Server.reconThreshold) {
      0
    } else {
      12
    }

  var connectionHandlers: Seq[ActorRef] = _

  import system.dispatcher
  system.scheduler.scheduleOnce(
    Duration(delay, TimeUnit.SECONDS)
  )({
    connectionHandlers = Config().Server.ports
      .map(port => {
        ConnectionHandler(Config().Server.host, port)
      }) :+
      WebSocketConnectionHandler()
  })

  sys.addShutdownHook({
    Option(connectionHandlers).foreach(_.foreach(_ ! PoisonPill))

    implicit val timeout = Duration(10, TimeUnit.SECONDS)
    Await.ready(system.terminate(), timeout)
    DAO.close()
  })
} 

//设置package包名称以及导入依赖的类
package akka.stream.alpakka.sqs.scaladsl

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import com.amazonaws.auth.{AWSCredentialsProvider, AWSStaticCredentialsProvider, BasicAWSCredentials}
import com.amazonaws.client.builder.AwsClientBuilder.EndpointConfiguration
import com.amazonaws.services.sqs.{AmazonSQSAsync, AmazonSQSAsyncClientBuilder}
import org.elasticmq.rest.sqs.{SQSRestServer, SQSRestServerBuilder}
import org.scalatest.{BeforeAndAfterAll, Suite, Tag}

import scala.concurrent.Await
import scala.concurrent.duration._
import scala.util.Random

trait DefaultTestContext extends BeforeAndAfterAll { this: Suite =>

  lazy val sqsServer: SQSRestServer = SQSRestServerBuilder.withDynamicPort().start()
  lazy val sqsAddress = sqsServer.waitUntilStarted().localAddress
  lazy val sqsPort = sqsAddress.getPort
  lazy val sqsEndpoint: String = {
    s"http://${sqsAddress.getHostName}:$sqsPort"
  }

  object Integration extends Tag("akka.stream.alpakka.sqs.scaladsl.Integration")

  //#init-mat
  implicit val system = ActorSystem()
  implicit val mat = ActorMaterializer()
  //#init-mat

  val credentialsProvider = new AWSStaticCredentialsProvider(new BasicAWSCredentials("x", "x"))

  implicit val sqsClient = createAsyncClient(sqsEndpoint, credentialsProvider)

  def randomQueueUrl(): String = sqsClient.createQueue(s"queue-${Random.nextInt}").getQueueUrl

  override protected def afterAll(): Unit = {
    super.afterAll()
    sqsServer.stopAndWait()
    Await.ready(system.terminate(), 5.seconds)
  }

  def createAsyncClient(sqsEndpoint: String, credentialsProvider: AWSCredentialsProvider): AmazonSQSAsync = {
    //#init-client
    val client: AmazonSQSAsync = AmazonSQSAsyncClientBuilder
      .standard()
      .withCredentials(credentialsProvider)
      .withEndpointConfiguration(new EndpointConfiguration(sqsEndpoint, "eu-central-1"))
      .build()
    //#init-client
    client
  }

} 

//设置package包名称以及导入依赖的类
import java.util.UUID

import play.api.libs.json.{Json, JsValue}
import services._

import scala.util.Random


object SomeRandom {

  val random = new Random()

  def double(): Double = {
    random.nextDouble()
  }

  def string(): String = {
    random.alphanumeric.take(10).mkString("")
  }

  def pointLocation(): PointLocation = {
    PointLocation(double(), double())
  }

  def pointLocationStash(): Stash = {
    stash(pointLocation())
  }

  def stash(location: Location): Stash = {
    Stash(uuidString(), string(), location)
  }

  def lineLocation(): LineLocation = {
    LineLocation(List(doublePair(), doublePair(), doublePair()))
  }

  def lineLocationStash(): Stash = {
    stash(lineLocation())
  }

  def polygonLocation(): PolygonLocation = {
    PolygonLocation(List(doublePair(), doublePair(), doublePair()))
  }

  def polygonLocationStash(): Stash = {
    stash(polygonLocation())
  }

  def uuidString(): String = {
    UUID.randomUUID().toString
  }

  def doublePair(): (Double, Double) = (double(), double())

  def jsonObj(): JsValue = {
    Json.obj(string() -> string())
  }
} 

//设置package包名称以及导入依赖的类
package de.htwg.se.dog.model

import scala.util.Random

trait CardGame {
  sealed trait Klasse
  case object Herz extends Klasse
  case object Karo extends Klasse
  case object Kreuz extends Klasse
  case object Pik extends Klasse

  sealed trait Farbe
  case object Schwarz extends Farbe
  case object Rot extends Farbe


  sealed abstract class Rank(val value: Int)
  case object Zwei extends Rank(2)
  case object Drei extends Rank(3)
  case object Vier extends Rank(4)
  case object Fünf extends Rank(5)
  case object Sechs extends Rank(6)
  case object Sieben extends Rank(7)
  case object Acht extends Rank(8)
  case object Neun extends Rank(9)
  case object Zehn extends Rank(10)
  case object Bube extends Rank(11)
  case object Dame extends Rank(12)
  case object König extends Rank(13)
  case object Ass extends Rank(14)
  case object Joker extends Rank(15)



  case class Card(suit: Klasse, rank: Rank, farbe: Farbe)

  case class Hand(cards: List[Card])
  case class Deck(cards: List[Card])

  case class Player(name: String, hand: Hand)
  case class Game(deck: Deck, players: List[Player])

  val shuffle: (Random, Deck) => Deck
  val dealCard: Deck => (Deck, Option[Card])
  val dealHand: Deck => (Deck, Hand)
} 

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

import scala.util.Random

abstract class Distribution {
  def draw(): Int
}

class Categorical(var probabilities: List[Float]) extends Distribution {

  def draw(): Int = {
    var cumulativeProbabilities = probabilities.scanLeft(0f)(_ + _)
    var p = new Random().nextFloat()
    return cumulativeProbabilities.indexWhere(_ > p)-1
  }
} 

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

import scala.util.Random
import scala.util.control.Breaks._
import scala.collection.mutable.ListBuffer
import mazes.math.Categorical
import scala.collection.TraversableOnce.flattenTraversableOnce
import mazes.generators.DungeonGenerator.Map
import mazes.generators.DungeonGenerator.Size
import mazes.generators.DungeonGenerator.Move
import mazes.generators.Dungeon.Direction




object DungeonGenerator{
  //type Matrix[T] = Array[Array[T]]
  type Map = Matrix[Dungeon.Tile]
  type Move = (Position,Dungeon.Direction)
  type Size=(Int,Int)
}
abstract class DungeonGenerator(var size:Size) {
  var r = new Random()  
  def generate(): Map
  
  def fillBorderWith(map:Map,e:Dungeon.Tile){
    
    for (c <- 0 to map.columns-1){
      map(0,c)=e
      map(map.rows-1,c)=e
    }
    
    for (r <- 0 to map.rows-1){
      map(r,0)=e
      map(r,map.columns-1)=e
    }
    
  }
}


// pimpmylib http://stackoverflow.com/questions/2633719/is-there-an-easy-way-to-convert-a-boolean-to-an-integer 

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

import java.nio.channels.FileChannel
import java.io.File
import java.nio.file.StandardOpenOption
import scala.util.Random
import java.nio.ByteBuffer
import com.spooky.bittorrent.u.GigaByte

object RandomFile {
  def main(args: Array[String]): Unit = {
    val channel = FileChannel.open(new File("O:\\tmp\\file.dump").toPath, StandardOpenOption.WRITE, StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING);
    val r = new Random(0)
    val buffer = ByteBuffer.allocate(1024 * 4)
    val bytes = GigaByte(50).toBytes
    println(bytes.capacity.toLong)
    for (_ <- 0l to(bytes.capacity.toLong, buffer.capacity.toLong)) {
      r.nextBytes(buffer.array())
      buffer.limit(buffer.capacity)
      buffer.position(0)
      channel.write(buffer)
    }
    channel.close()
  }
} 

//设置package包名称以及导入依赖的类
package com.spooky.bittorrent.mse

import scala.util.Random
import com.spooky.bittorrent.RawWrapper

sealed case class LocalSecret(override val raw: Array[Byte]) extends RawWrapper(raw)
object LocalSecret {
  def spooky: LocalSecret = {
    val secretRaw = Array.ofDim[Byte](96)
    val spooky = "spooky"
    for (i <- 0 until 96) {
      secretRaw(i) = spooky.charAt(i % spooky.length()).asInstanceOf[Byte]
      //        print(spooky.charAt(i % spooky.length()))
    }
    LocalSecret(secretRaw)
  }
  def ykoops: LocalSecret = {
    val secretRaw = Array.ofDim[Byte](96)
    val ykoops = "ykoops"
    for (i <- 0 until 96) {
      secretRaw(i) = ykoops.charAt(i % ykoops.length()).asInstanceOf[Byte]
      //        print(spooky.charAt(i % spooky.length()))
    }
    LocalSecret(secretRaw)
  }
  def random: LocalSecret = {
    val secretRaw = Array.ofDim[Byte](96)
    Random.nextBytes(secretRaw)
    LocalSecret(secretRaw)
  }
} 

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

import scala.util.Random
import scala.annotation.tailrec

object Deck {
  def shuffle(allCards: Seq[Card]) = Deck(allCards).shuffle
}

case class Deck(cards: Seq[Card]) {
  def empty = cards.isEmpty
  def draw: (Option[Card], Deck) = (cards.headOption, Deck(cards.drop(1)))

  def drop(cs: Seq[Card]) = Deck(cards diff cs)

  def deal(hands: Int, cardsPerHand: Int): (Seq[Hand], Deck) = {

    @tailrec
    def dealRec(currentHands: Seq[Hand] = Seq.fill(hands)(Hand()), currentDeck: Deck = this, nbCards: Int = hands * cardsPerHand): (Seq[Hand], Deck) =
      if (nbCards == 0)
        (currentHands, currentDeck)
      else {
        val (card, deck) = currentDeck.draw
        val hand = currentHands.head + card
        dealRec(currentHands.tail :+ hand, deck, nbCards - 1)
      }

    dealRec()

  }

  def shuffle = Deck(Random.shuffle(cards))
} 

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

import breeze.linalg._
import fregata.data.LibSvmReader
import fregata.loss.LogLoss
import fregata.metrics.classification.{Accuracy, AreaUnderRoc}
import fregata.model.classification.{CLR, CLRModel, LogisticRegression, LogisticRegressionModel}

import scala.util.Random


object TestCLR {

  def main(args: Array[String]) {
    val (_,t1) = LibSvmReader.read("/Volumes/takun/data/libsvm/a9a",123)
    val (_,t2) = LibSvmReader.read("/Volumes/takun/data/libsvm/a9a.t",123)
    val trainData = t1.map{
      case (x,label) => Array(x) -> label
    }
    val testData = t2.map{
      case (x,label) => Array(x) -> label
    }
    println(" load over ...")
    val r = new Random(1L)
    val data2 = r.shuffle(trainData.toList)
    val model = new CLR()
      .run(data2,Array(Array(0,0)) )
    val predicted = model.clrPredict(testData)
    val acc = Accuracy.of(predicted.map{
      case ((x,l),(p,c)) =>
        c -> l
    })
    val auc = AreaUnderRoc.of(predicted.map{
      case ((x,l),(p,c)) =>
        p -> l
    } )
    val loss = new LogLoss().of(predicted.map{
      case ((x,l),(p,c)) =>
        if( l == 1d ) {
          (l,c,p)
        }else{
          ( l , c , 1-p )
        }
    })
    println( s"Accuracy : ${acc} " )
    println( s"AreaUnderRoc : ${auc} " )
    println( s"logLoss : ${loss} " )
  }
} 

//设置package包名称以及导入依赖的类
import scala.collection.mutable.Map
import scala.util.Random


object Caching {

	val TestMap = Map[Int, String]()

	def cache(key: Int) = {

		if(TestMap.size > 1000){

			val random = new Random
			val r_val = TestMap(random.nextInt(TestMap.size)).toInt

			//var r_val = TestMap.toVector(random(TestMap.size))
			TestMap -= r_val
			TestMap += (key -> "Testing val")
		}
		else{
			TestMap += (key -> "Testing val")
		}		
	}

	def time[R](block: => R): R = {  
        val t0 = System.nanoTime()
        val result = block    // call-by-name
        val t1 = System.nanoTime()
        println("Elapsed time: " + (t1 - t0) + "ns")
        result
  	}

	def main(args: Array[String]): Unit = {

		val rand = new Random
		var tot = 0
		for (a <- 0 to 1010){
			if (a > 1005)
				time {cache(rand.nextInt(3000))}
			else
				cache(rand.nextInt(3000))
			
		}

    	val runtime = Runtime.getRuntime
    	println("** Used Memory:  " + (runtime.totalMemory - runtime.freeMemory)/1010)
	}
	
} 

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

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{Flow, Sink, Source}
import scala.util.Random

object InputCustomer {
  def random(): InputCustomer = {
    InputCustomer(s"FirstName${Random.nextInt(1000)} " +
      s"LastName${Random.nextInt(1000)}")
  }
}

case class InputCustomer(name: String)

case class OutputCustomer(firstName: String, lastName: String)

object CustomersExample extends App {
  implicit val actorSystem = ActorSystem()

  import actorSystem.dispatcher

  implicit val materializer = ActorMaterializer()

  val inputCustomers = Source((1 to 100).map(_ => InputCustomer.random()))

  val normalize = Flow[InputCustomer]
    .map(c => c.name.split(" ").toList)
    .collect {
      case firstName :: lastName :: Nil => OutputCustomer(firstName, lastName)
    }

  val writeCustomers = Sink.foreach { println} // ?-conversion

  inputCustomers.via(normalize).runWith(writeCustomers).andThen {
    case _ => actorSystem.terminate()
  }
} 

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

import java.security.KeyPairGenerator
import java.util.UUID

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{Keep, Sink, Source}
import akka.util.ByteString
import com.cryptoutility.protocol.Events._
import com.cryptoutility.protocol.EventSerializer
import org.scalatestplus.play.PlaySpec
import play.api.http.websocket.BinaryMessage

import scala.concurrent.Await
import scala.concurrent.duration._
import scala.language.postfixOps
import scala.util.Random


class EventDecoderSpec extends PlaySpec{

  implicit val system = ActorSystem("test-system")
  implicit val mat = ActorMaterializer()
  val id = () => UUID.randomUUID().toString

  def publicKey = {
    KeyPairGenerator.getInstance("RSA").generateKeyPair().getPublic
  }

  "event decode" should{
    "decode byte streams to an event" in {
      val expected = Initialized(true, UserInfo("James", "Carl", "[email protected]", publicKey, id()))
      val serialized = EventSerializer.serialize(expected)

      val f =
        Source.single(BinaryMessage(ByteString(serialized)))
          .via(new EventDecoder)
            .toMat(Sink.head)(Keep.right).run()

      val actual = Await.result(f, 500 millis)

      actual mustBe expected
    }
    "of an invalid byte stream" in {
      a [InvalidFormatException] should be thrownBy{
        val data = new Array[Byte](256)
        Random.nextBytes(data)
        EventSerializer.deserialize(data)
      }
    }
  }

}