Потоковая обработка данных

import akka.stream.scaladsl.Source
val stream = Source(1 to 1000).buffer(32, akka.stream.OverflowStrategy.backpressure)

// Дополнительный контекст: Граф разветвили на две ветви с разной скоростью и разными правилами обработки ошибок.
import akka.stream.scaladsl.Source
val stream = Source(1 to 1000).buffer(32, akka.stream.OverflowStrategy.backpressure)

import akka.stream.scaladsl.Source
val stream = Source(1 to 1000).buffer(32, akka.stream.OverflowStrategy.backpressure)
// На ревью требуется устранить причину дефекта, не скрывая её приведением типа или значением по умолчанию.

import akka.stream.scaladsl.*
val graph = Flow.fromGraph(GraphDSL.create() { implicit b =>
  import GraphDSL.Implicits.*
  val bcast = b.add(Broadcast[Int](2))
  val merge = b.add(Merge[Int](2))
  bcast.out(0) ~> merge.in(0)
  bcast.out(1) ~> Flow[Int].map(_ * 10) ~> merge.in(1)
  FlowShape(bcast.in, merge.out)
})

// Дополнительный контекст: Поток перезапустили после ошибки, когда часть элементов уже дошла до внешней системы.
import akka.stream.scaladsl.*
val graph = Flow.fromGraph(GraphDSL.create() { implicit b =>
  import GraphDSL.Implicits.*
  val bcast = b.add(Broadcast[Int](2))
  val merge = b.add(Merge[Int](2))
  bcast.out(0) ~> merge.in(0)
  bcast.out(1) ~> Flow[Int].map(_ * 10) ~> merge.in(1)
  FlowShape(bcast.in, merge.out)
})

import akka.stream.scaladsl.*
val graph = Flow.fromGraph(GraphDSL.create() { implicit b =>
  import GraphDSL.Implicits.*
  val bcast = b.add(Broadcast[Int](2))
  val merge = b.add(Merge[Int](2))
  bcast.out(0) ~> merge.in(0)
  bcast.out(1) ~> Flow[Int].map(_ * 10) ~> merge.in(1)
  FlowShape(bcast.in, merge.out)
})
// На ревью требуется устранить причину дефекта, не скрывая её приведением типа или значением по умолчанию.

val safe = akka.stream.scaladsl.Flow[String].map(_.toInt).recover { case _: NumberFormatException => 0 }

// Дополнительный контекст: Поток перезапустили после ошибки, когда часть элементов уже дошла до внешней системы.
val safe = akka.stream.scaladsl.Flow[String].map(_.toInt).recover { case _: NumberFormatException => 0 }

val safe = akka.stream.scaladsl.Flow[String].map(_.toInt).recover { case _: NumberFormatException => 0 }
// На ревью требуется устранить причину дефекта, не скрывая её приведением типа или значением по умолчанию.

import akka.stream.scaladsl.*
import akka.stream.*
val runnable = Source(1 to 10).toMat(Sink.seq)(Keep.right)

// Дополнительный контекст: Производитель ускорился, потребитель замедлился, а между ними появился ограниченный буфер.
import akka.stream.scaladsl.*
import akka.stream.*
val runnable = Source(1 to 10).toMat(Sink.seq)(Keep.right)

import akka.stream.scaladsl.*
import akka.stream.*
val runnable = Source(1 to 10).toMat(Sink.seq)(Keep.right)
// На ревью требуется устранить причину дефекта, не скрывая её приведением типа или значением по умолчанию.

val measured = akka.stream.scaladsl.Flow[Int].wireTap(n => println(s"seen=$n"))

// Дополнительный контекст: Производитель ускорился, потребитель замедлился, а между ними появился ограниченный буфер.
val measured = akka.stream.scaladsl.Flow[Int].wireTap(n => println(s"seen=$n"))

val measured = akka.stream.scaladsl.Flow[Int].wireTap(n => println(s"seen=$n"))
// На ревью требуется устранить причину дефекта, не скрывая её приведением типа или значением по умолчанию.

import akka.stream.scaladsl.{RestartSource, Source}
import scala.concurrent.duration.*
val source = Source(1 to 100)
val restarted = RestartSource.withBackoff(
  minBackoff = 1.second,
  maxBackoff = 30.seconds,
  randomFactor = 0.2
)(() => source)

// Дополнительный контекст: Производитель ускорился, потребитель замедлился, а между ними появился ограниченный буфер.
import akka.stream.scaladsl.{RestartSource, Source}
import scala.concurrent.duration.*
val source = Source(1 to 100)
val restarted = RestartSource.withBackoff(
  minBackoff = 1.second,
  maxBackoff = 30.seconds,
  randomFactor = 0.2
)(() => source)

import akka.stream.scaladsl.{RestartSource, Source}
import scala.concurrent.duration.*
val source = Source(1 to 100)
val restarted = RestartSource.withBackoff(
  minBackoff = 1.second,
  maxBackoff = 30.seconds,
  randomFactor = 0.2
)(() => source)
// На ревью требуется устранить причину дефекта, не скрывая её приведением типа или значением по умолчанию.