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

import akka.actor.ActorSystem
import akka.pattern.{Backoff, BackoffSupervisor}
import akka.stream.{ActorMaterializer, ThrottleMode}
import akka.stream.scaladsl.{Sink, Source}
import sample.WindTurbineSimulator

import scala.concurrent.duration._


object SimulateWindTurbines extends App {
  implicit val system = ActorSystem()
  implicit val materializer = ActorMaterializer()

  val endpoint = "ws://127.0.0.1:8080"
  val numberOfTurbines = 5
  Source(1 to numberOfTurbines)
    .throttle(
      elements = 100, //number of elements to be taken from bucket
      per = 1.second,
      maximumBurst = 100,  //capacity of bucket
      mode = ThrottleMode.shaping
    )
    .map { _ =>
      val id = java.util.UUID.randomUUID.toString

      val supervisor = BackoffSupervisor.props(
        Backoff.onFailure(
          WindTurbineSimulator.props(id, endpoint),
          childName = id,
          minBackoff = 1.second,
          maxBackoff = 30.seconds,
          randomFactor = 0.2
        ))

      system.actorOf(supervisor, name = s"$id-backoff-supervisor")
    }
    .runWith(Sink.ignore)
} 

//设置package包名称以及导入依赖的类
package de.choffmeister.microserviceutils

import akka.Done
import akka.actor.{ActorSystem, Props}
import akka.stream.scaladsl.{Concat, Keep, Sink, Source}
import akka.stream.{KillSwitches, ThrottleMode}
import akka.testkit.{TestKit, TestProbe}
import org.scalatest._
import org.scalatest.concurrent.Eventually
import org.scalatest.time.{Millis, Seconds, Span}

import scala.collection.mutable.ListBuffer
import scala.concurrent.Future
import scala.concurrent.duration._

class PerpetualStreamTest extends TestKit(ActorSystem()) with FlatSpecLike with Matchers with Eventually with BeforeAndAfterAll {
  implicit val defaultPatience = PatienceConfig(timeout = Span(30, Seconds), interval = Span(100, Millis))

  "PerpetualStream" should "work" in {
    val list = ListBuffer.empty[Int]
    val stream = system.actorOf(Props(new NumberGeneratingPerpetualStream(list)))
    val probe = new TestProbe(system)
    probe.watch(stream)

    eventually(require(list.length >= 20, "Must have emitted at least 20 numbers"))
    list.take(20) should be((-10 until 10).toList)

    GracefulShutdownExtension(system).triggerGracefulShutdown()
    probe.expectTerminated(stream)
  }

  override def afterAll(): Unit = TestKit.shutdownActorSystem(system)

  class NumberGeneratingPerpetualStream(list: ListBuffer[Int]) extends PerpetualStream {
    var count = 0

    override def stream = {
      val currentCount = count
      count = count + 1

      val source = currentCount match {
        case 0 => Source.combine(Source(-10 until 0), Source.failed[Int](new RuntimeException("ERROR")))(Concat(_))
        case 1 => Source(0 until 10)
        case _ => Source(10 to 100000).throttle(1000, 100.millis, 100, ThrottleMode.shaping)
      }

      source
        .via(watcher)
        .viaMat(KillSwitches.single)(Keep.right)
        .map { i => list.append(i); i }
        .to(Sink.ignore)
        .mapMaterializedValue(ks => () => {
          ks.shutdown()
          Future.successful(Done)
        })
    }
  }
} 

//设置package包名称以及导入依赖的类
import akka.NotUsed
import akka.actor.SupervisorStrategy.Restart
import akka.actor.{Actor, OneForOneStrategy, SupervisorStrategy}
import akka.stream.scaladsl.{Sink, Source}
import akka.stream.{ActorMaterializer, OverflowStrategy, ThrottleMode}

import scala.concurrent.duration.Duration
import scala.io.Source.fromFile

class StreamActor extends Actor {
  override def receive: Receive = {
    case StreamBook(filename) =>
      val materializer = ActorMaterializer.create(context)  // Materializing and running a stream always requires a Materializer to be in implicit scope.
      val sink = Source
        .actorRef(1000, OverflowStrategy.dropNew)           // If the buffer is full when a new element arrives, drops the new element.
        .throttle(1, Duration(1, "seconds"), 1, ThrottleMode.shaping)   // throttle - to slow down the stream to 1 element per second.
        .to(Sink.actorRef(sender, NotUsed))                 // Sink is a set of stream processing steps that has one open input. Can be used as a Subscriber.
        .run()(materializer)
      val lines = fromFile(filename).getLines
      lines.foreach(line => sink ! StreamLine(line))
  }

  override def supervisorStrategy: SupervisorStrategy =
    OneForOneStrategy(10, Duration(60, "seconds")) {
      case _: Exception => Restart
    }
}

case class StreamBook(fileName: String)
case class StreamLine(line: String) 

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

import akka.Done
import akka.stream.scaladsl.Source
import akka.stream.{Materializer, ThrottleMode}
import models.pokemon.PokemonName

import scala.concurrent.{ExecutionContext, Future}

trait BackPressureCalls {

  import scala.concurrent.duration._

  def backPressureRequestDetailsUsing(maxPerSecond: Int)
                                     (elements: List[PokemonName])
                                     (runForEach: PokemonName => Unit)
                                     (implicit ex: ExecutionContext, mat: Materializer): Future[Done] = {
    Source(elements)
      .throttle(maxPerSecond, 1.second, 1, ThrottleMode.Shaping)
      .runForeach(runForEach(_))
  }

} 

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

import akka.Done
import akka.actor.ActorSystem
import akka.kafka.ConsumerMessage.CommittableOffsetBatch
import akka.kafka.scaladsl.Consumer
import akka.kafka.{ConsumerSettings, Subscriptions}
import akka.stream.scaladsl.Source
import akka.stream.{ActorMaterializer, ThrottleMode}
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.common.serialization.{ByteArrayDeserializer, StringDeserializer}

import scala.concurrent.duration.Duration

package object reactive_kafka {

  def throttolableConsumerFlow(implicit system: ActorSystem, mat: ActorMaterializer): Source[Done, Consumer.Control] = {
    val c = system.settings.config.getConfig("throttolable-consumer")
    implicit val ec = system.dispatcher

    val bootstrapServers = c getString "bootstrap-servers"
    val topic = c getString "topic"
    val autoRestConfig = c getString "auto-offset-reset"
    val groupId = c getString "group-id"
    val throttle = c getInt "throttle"
    val throttlePer = Duration.fromNanos((c getDuration "throttle-per").toNanos)
    val throttleBurst = c getInt "throttle-burst"
    val logPer = c getInt "log-per"
    val offsetCommitBatchSize = c getInt "offset-commit-batch-size"
    val offsetCommitParallelism = c getInt "offset-commit-parallelism"

    val consumerSettings =
      ConsumerSettings(system, new ByteArrayDeserializer, new StringDeserializer)
        .withBootstrapServers(bootstrapServers)
        .withGroupId(groupId)
        .withProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoRestConfig)

    val source = Consumer.committableSource(consumerSettings, Subscriptions.topics(Set(topic)))

    val throttled = if (throttle > 0)
      source.throttle(throttle, throttlePer, throttleBurst, ThrottleMode.shaping)
    else source

    throttled.statefulMapConcat(() => {
      var counter = 0
      msg => {
        if (counter % logPer == 0) {
          system.log.info(s"FakeConsumer consume: $msg")
          counter = 0
        }
        counter += 1
        msg :: Nil
      }
    }).batch(max = offsetCommitBatchSize, m => CommittableOffsetBatch.empty.updated(m.committableOffset))((batch, m) => batch.updated(m.committableOffset))
      .mapAsync(offsetCommitParallelism) { batch =>
      batch.commitScaladsl()
    }
  }
} 

//设置package包名称以及导入依赖的类
package lew.bing.akka.stream


import akka.actor.{ActorSystem, Cancellable}
import akka.stream.scaladsl._
import akka.stream.{ActorMaterializer, ThrottleMode}

import scala.concurrent.duration._
import scala.concurrent.{Future, Promise}


object Demo2 {

  def main(args: Array[String]): Unit = {
    implicit val system = ActorSystem("QuickStart")
    implicit val materializer = ActorMaterializer()
    val source:Source[Int,Promise[Option[Int]]] = Source.maybe[Int]
    //????
//    val flow: Flow[Int, Int, Cancellable] = Throttler.RateInt(1)
    val cancellable = new Cancellable {
      private var cancelled = false

      override def cancel(): Boolean = {
        cancelled = true
        system.terminate()
        isCancelled
      }

      override def isCancelled: Boolean = cancelled
    }
    val flow:Flow[Int, Int, Cancellable] = Flow.fromProcessorMat(() => {
      (null, cancellable)
    }).map(o => 1).throttle(1, 1.second, 1, ThrottleMode.Shaping)

    val sink:Sink[Int,Future[Int]] = Sink.head[Int]
    val r1:RunnableGraph[Promise[Option[Int]]] = source.via(flow).to(sink)
    val r2:RunnableGraph[Cancellable] = source.viaMat(flow)(Keep.right).to(sink)
    val r3 = source.via(flow).toMat(sink)(Keep.right)

    val r4 = source.via(flow).runWith(sink)
    val r5 = flow.to(sink).runWith(source)
    val r6 = flow.runWith(source,sink)



  }

} 

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

import akka.stream.{OverflowStrategy, ThrottleMode}
import akka.stream.scaladsl._

import scala.concurrent.Future
import scala.concurrent.duration._

object ExplicitBuffer extends AkkaStreamsApp {
  override def akkaStreamsExample: Future[_] =
    Source(1 to 100000000)
      .log("passing")
      .buffer(5, OverflowStrategy.backpressure)
      .throttle(1, 1 second, 1, ThrottleMode.shaping)
      .take(20)
      .runForeach(e => log.debug("received: {}", e))
  
  runExample
}