Keep组合的含义是什么?

zero_coding

我正在尝试Keep在akka流中进行组合,并创建以下示例:

import java.nio.file.Paths

import akka.NotUsed
import akka.actor.ActorSystem
import akka.stream.{ActorMaterializer, IOResult}
import akka.stream.scaladsl.{FileIO, Flow, Keep, Sink, Source}
import akka.util.ByteString

import scala.concurrent.Future
import scala.util.{Failure, Success}

object FileConsumer extends App {

  implicit val system = ActorSystem("reactive-tweets")
  implicit val materializer = ActorMaterializer()

  val source: Source[Int, NotUsed] = Source(1 to 100)
  val factorials = source.scan(BigInt(1))((acc, next) => acc * next)

  val result: Future[IOResult] =
    factorials.map(_.toString).runWith(lineSink("factorial2.txt"))

  implicit val ec = system.dispatcher
  result.onComplete {
    case Success(v) => println(s"Fileinfo ${ v.count }")
    case Failure(e) => println(e)
  }

  def lineSink(filename: String): Sink[String, Future[IOResult]] =
    Flow[String].map(s => ByteString(s + "\n")).toMat(FileIO.toPath(Paths.get(filename)))(Keep.right)


} 

akka stream网站上说:

产生的蓝图是a Sink[String, Future[IOResult]],这意味着它接受字符串作为输入,并且在实现时将创建类型的辅助信息Future[IOResult](当在Source或Flow上进行链接操作时,将给出该辅助信息的类型-称为“物化值”在最左边的起点;由于我们要保留接收FileIO.toPath器所提供的内容,因此需要说Keep.right)。

但是,当我想将其保留ByteString在左侧时,我尝试过:

  def lineSink2(filename: String): Sink[String, Future[ByteString]] =
Flow[String].map(s => ByteString(s + "\n")).toMat(Sink.foreach(println))(Keep.left)

但它根本不编译。

我也不明白:

由最左边的起点给出

最左边的起点是Flow

我认为,我还不了解这个想法Keep

ŁukaszGawron

Sink.foreach的定义如下:

def foreach[T](f: T ⇒ Unit): Sink[T, Future[Done]]

这意味着物化价值是未来[完成]

如果有流量,您可以:

 val value: Flow[String, ByteString, NotUsed] = Flow[String].map(s => ByteString(s + "\n"))

其物化值为NotUsed

在这种情况下:

Keep.left-未使用-源或流的物化值

Keep.right -Future [Done]-水槽的总价值

Keep.both-(未使用,将来[完成])

重要的事实是,在许多情况下,物化值不是流经流的元素的值,而是

  • 诊断信息
  • 流状态
  • 有关流的其他信息

本文收集自互联网,转载请注明来源。

如有侵权,请联系 [email protected] 删除。

编辑于
0

我来说两句

0 条评论
登录 后参与评论

相关文章