使用Akka的Scala-TCP数据包帧

Question

在Akka中有没有什么方法可以像在Erlang中一样用{packet,4}来实现包帧？数据包如下所示：

4 bytes length in big endian | body...

00 00 00 05 H E L L O 0 0 0 5 W O R L D

00 00 00 05 H E L L

import java.net.InetSocketAddress

import akka.actor.{Props, Actor}
import akka.io.Tcp.Bind
import akka.io.{IO, Tcp}

class Server extends Actor{
    import context.system
    import Tcp._
    IO(Tcp) ! Bind(self, new InetSocketAddress("0.0.0.0", 1234))

    def receive ={
        case bound @ Bound(localAddr) =>
            println("Server is bound to "+localAddr.toString())
        case failed @ CommandFailed(_ : Bind) =>
            context stop self
        case connected @ Connected(remote, local) =>
            val handler = context.actorOf(Props[ClientHandler])
            val connection = sender()
            println(remote.toString + "connected to "+local.toString())

            connection ! Register(handler)
    }
}

Answer 1

据我所知，在Akka或Scala中没有用于此的库函数。Akka在字节字符串中进行读写交易，所以我把一个特性放在一起，它将完全满足您的要求。您将发送给您的演员的字节字符串传递给它。然后，它根据报头中的数据包长度分解流。它是无状态的，因此它返回一个元组，其中包含提取的数据包列表和TCP流中任何未使用的数据。您可以将新的TCP数据连接到以这个字节字符串返回的流的未使用部分，如下例所示。

trait Buffering {

  val MAX_PACKET_LEN: Short = 10000

  /**
   * Extracts complete packets of the specified length, preserving remainder
   * data. If there is no complete packet, then we return an empty list. If
   * there are multiple packets available, all packets are extracted, Any remaining data
   * is returned to the caller for later submission
   * @param data A list of the packets extracted from the raw data in order of receipt
   * @return A list of ByteStrings containing extracted packets as well as any remaining buffer data not consumed
   */
  def getPacket(data: ByteString): (List[ByteString], ByteString) = {

    val headerSize = 2

    @tailrec
    def multiPacket(packets: List[ByteString], current: ByteString): (List[ByteString], ByteString) = {
      if (current.length < headerSize) {
        (packets.reverse, current)
      } else {
        val len = current.iterator.getShort
        if (len > MAX_PACKET_LEN || len < 0) throw new RuntimeException(s"Invalid packet length: $len")
        if (current.length < len + headerSize) {
          (packets.reverse, current)
        } else {
          val rem = current drop headerSize // Pop off header
          val (front, back) = rem.splitAt(len) // Front contains a completed packet, back contains the remaining data
          // Pull of the packet and recurse to see if there is another packet available
          multiPacket(front :: packets, back)
        }
      }
    }
    multiPacket(List[ByteString](), data)
  }

actor的用法如下：

def receive = buffer(CompactByteString())

def buffer(buf: ByteString): Receive = {
  // Messages inbound from the network
  case Received(data) =>
    val (pkt, remainder) = getPacket(buf ++ data)
    // Do something with your packet
    context become buffer(remainder) 
  case Other Stuff => // Etc
}