如何使用Akka Streams在分隔符字符上拆分入站流

Question

我一直在玩实验性的Akka Streams API,我有一个用例,我想看看如何实现.对于我的用例,我有一个StreamTcp基于Flow将连接的输入流绑定到我的服务器套接字的源.我拥有的Flow基于ByteString进入它的数据.进入的数据将在其中具有分隔符,这意味着我应该将分隔符之前的所有内容视为一条消息,并将所有内容作为下一条消息处理到下一个分隔符之后.所以玩一个更简单的例子,不使用套接字和静态文本,这就是我提出的:

import akka.actor.ActorSystem
import akka.stream.{ FlowMaterializer, MaterializerSettings }
import akka.stream.scaladsl.Flow
import scala.util.{ Failure, Success }
import akka.util.ByteString

object BasicTransformation {

  def main(args: Array[String]): Unit = {
    implicit val system = ActorSystem("Sys")

    val data = ByteString("Lorem Ipsum is simply.Dummy text of the printing.And typesetting industry.")

    Flow(data).
      splitWhen(c => c == '.').
      foreach{producer => 
        Flow(producer).
          filter(c => c != '.').
          fold(new StringBuilder)((sb, c) => sb.append(c.toChar)).
          map(_.toString).
          filter(!_.isEmpty).
          foreach(println(_)).
          consume(FlowMaterializer(MaterializerSettings()))
      }.
      onComplete(FlowMaterializer(MaterializerSettings())) {
        case any =>
          system.shutdown
      }
  }
}

Flow我发现完成我的目标的主要功能是splitWhen,然后产生额外的子流,每个.分隔符对应一个消息.然后,我用另一个步骤流程处理每个子流程,最后在最后打印单个消息.

这一切似乎有点冗长,以实现我认为是一个非常简单和常见的用例.所以我的问题是,是否有更清晰,更简洁的方法来做到这一点,或者这是以分隔符分割流的正确和首选方式？

Answer 1

看起来最近改进了API以包含akka.stream.scaladsl.Framing.该文档还包含如何使用它的示例.关于你的具体问题:

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{Framing, Source}
import akka.util.ByteString
import com.typesafe.config.ConfigFactory

object TcpDelimiterBasedMessaging extends App {
  object chunks {
    val first = ByteString("Lorem Ipsum is simply.Dummy text of the printing.And typesetting industry.")
    val second = ByteString("More text.delimited by.a period.")
  }

  implicit val system = ActorSystem("delimiter-based-messaging", ConfigFactory.defaultReference())
  implicit val dispatcher = system.dispatcher
  implicit val materializer = ActorMaterializer()

  Source(chunks.first :: chunks.second :: Nil)
    .via(Framing.delimiter(ByteString("."), Int.MaxValue))
    .map(_.utf8String)
    .runForeach(println)
    .onComplete(_ => system.terminate())
}

产生以下输出: Lorem Ipsum is simply Dummy text of the printing And typesetting industry More text delimited by a period

Answer 2

在 Akka 用户组上发布了同样的问题后，我从 Endre Varga 和 Viktor Klang 那里得到了一些建议（https://groups.google.com/forum/#!topic/akka-user/YsnwIAjQ3EE）。我最终采用了 Endre 的建议Transformer，然后使用了transform的方法Flow。下面包含我之前示例的稍微修改的版本：

import akka.actor.ActorSystem
import akka.stream.{ FlowMaterializer, MaterializerSettings }
import akka.stream.scaladsl.Flow
import scala.util.{ Failure, Success }
import akka.util.ByteString
import akka.stream.Transformer
import akka.util.ByteStringBuilder

object BasicTransformation {

  def main(args: Array[String]): Unit = {
    implicit val system = ActorSystem("Sys")                           
    implicit val mater = FlowMaterializer(MaterializerSettings())

    val data = List(
      ByteString("Lorem Ipsum is"), 
      ByteString(" simply.Dummy text of.The prin"), 
      ByteString("ting.And typesetting industry.")
    )
    Flow(data).transform(new PeriodDelimitedTransformer).foreach(println(_))
  }
}

其定义PeriodDelimitedTransformer如下：

class PeriodDelimitedTransformer extends Transformer[ByteString,String]{
  val buffer = new ByteStringBuilder

  def onNext(msg:ByteString) = {    
    val msgString = msg.utf8String
    val delimIndex = msgString.indexOf('.')
    if (delimIndex == -1){
      buffer.append(msg)
      List.empty
    }
    else{
      val parts = msgString.split("\\.")
      val endsWithDelim = msgString.endsWith(".")

      buffer.putBytes(parts.head.getBytes())
      val currentPiece = buffer.result.utf8String            
      val otherPieces = parts.tail.dropRight(1).toList

      buffer.clear
      val lastPart = 
      if (endsWithDelim){
          List(parts.last)
      }
      else{
          buffer.putBytes(parts.last.getBytes())
          List.empty
      }           


      val result = currentPiece :: otherPieces ::: lastPart
      result
    }

  }  
}

因此，我之前的解决方案的一些复杂性都集中到了这个中Transformer，但这似乎是最好的方法。在我最初的解决方案中，流最终被分成多个子流，这并不是我真正想要的。