CODEWITHSUNDEEP
scalatcpakka
miho
miho

Reputation: 12085

akka.io receive line by line in 2.3

I'm using Akka 2.3 (since that's the version which comes with Play) and want to connect to some TCP socket. I'm aware of the akka.io package. However I can't see any way to process the received data as UTF-8 string line by line (against just receiving byte chunks).

Searching the web there are quite a few references to the experimental Pipeline API of Akka 2.2. However this API got removed in Akka again.

What I'm looking for is what is known as readLine in most buffer classes but for the Akka I/O Framework.

Upvotes: 2

Views: 140

Answers (2)

miho
miho

Reputation: 12085

Akka Stream seems promising, however since it's still unreleased I decided to implement it myself by simply contacting all data in a buffer and waiting for the separator chars.

  private val separatorBytes = // like CRLF
  private var buffer = ByteString.empty
  private var nextPossibleMatch = 0

  // when receiving chunks of bytes they are appended to buffer and doParseLines is executed

  private def doParseLines(parsedLinesSoFar: Vector[String] = Vector()): Vector[String] = {
    val possibleMatchPos = buffer.indexOf(separatorBytes.head, from = nextPossibleMatch)
    if (possibleMatchPos == -1) {
      parsedLinesSoFar
    } else {
      if (possibleMatchPos + separatorBytes.size > buffer.size) {
        nextPossibleMatch = possibleMatchPos
        parsedLinesSoFar
      } else {
        if (buffer.slice(possibleMatchPos, possibleMatchPos + separatorBytes.size) == separatorBytes) {
          // Found a match
          val parsedLine = buffer.slice(0, possibleMatchPos).utf8String
          buffer = buffer.drop(possibleMatchPos + separatorBytes.size)
          nextPossibleMatch -= possibleMatchPos + separatorBytes.size
          doParseLines(parsedLinesSoFar :+ parsedLine)
        } else {
          nextPossibleMatch += 1
          doParseLines(parsedLinesSoFar)
        }
      }
    }
  }

Upvotes: 1

Konrad 'ktoso' Malawski
Konrad 'ktoso' Malawski

Reputation: 13130

The replacement of the old pipelines is Akka Streams, which provide a streaming TCP implementation. It is possible to add a lines parser to such stream to easily obtain a "stream of lines".

A built in line parser will be merged very soon +str #17310: Basic framing support. Currently the line parser is a cookbook you can paste into your project: Akka Streams Cookbook: Parsing lines from ByteStrings.

I suggest you first read the Working with Streaming IO docs and then modify the samples such that it matches your use case (by using the parseLines cookbook recipe).

Please note that Akka Streams are still experimental and API may change slightly. Another important information is that they implement the www.reactive-streams.org specification, so you get back-pressure out of the box without having to manually worry about it :-)

Upvotes: 0

Related Questions