我尝试用Akka 2.4.3将TCP流重定向/转发到另一个Sink。程序应该打开一个服务器套接字,监听传入的连接,然后使用TCP流。我们的发送方不期望/接受来自我们的回复,所以我们从不发回任何东西-我们只是使用流。在构造TCP流后,我们需要将字节转换为更有用的东西,并将其发送到Sink。
到目前为止,我尝试了以下方法,但我特别纠结于如何不将TCP包发送回发送方并正确连接接收器。
import scala.util.Failure
import scala.util.Success
import akka.actor.ActorSystem
import akka.event.Logging
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.Sink
import akka.stream.scaladsl.Tcp
import akka.stream.scaladsl.Framing
import akka.util.ByteString
import java.nio.ByteOrder
import akka.stream.scaladsl.Flow
object TcpConsumeOnlyStreamToSink {
implicit val system = ActorSystem("stream-system")
private val log = Logging(system, getClass.getName)
//The Sink
//In reality this is of course a real Sink doing some useful things :-)
//The Sink accept types of "SomethingMySinkUnderstand"
val mySink = Sink.ignore;
def main(args: Array[String]): Unit = {
//our sender is not interested in getting replies from us
//so we just want to consume the tcp stream and never send back anything to the sender
val (address, port) = ("127.0.0.1", 6000)
server(system, address, port)
}
def server(system: ActorSystem, address: String, port: Int): Unit = {
implicit val sys = system
import system.dispatcher
implicit val materializer = ActorMaterializer()
val handler = Sink.foreach[Tcp.IncomingConnection] { conn =>
println("Client connected from: " + conn.remoteAddress)
conn handleWith Flow[ByteString]
//this is neccessary since we use a self developed tcp wire protocol
.via(Framing.lengthField(4, 0, 65532, ByteOrder.BIG_ENDIAN))
//here we want to map the raw bytes into something our Sink understands
.map(msg => new SomethingMySinkUnderstand(msg.utf8String))
//here we like to connect our Sink to the Tcp Source
.to(mySink) //<------ NOT COMPILING
}
val tcpSource = Tcp().bind(address, port)
val binding = tcpSource.to(handler).run()
binding.onComplete {
case Success(b) =>
println("Server started, listening on: " + b.localAddress)
case Failure(e) =>
println(s"Server could not bind to $address:$port: ${e.getMessage}")
system.terminate()
}
}
class SomethingMySinkUnderstand(x:String) {
}
}
更新:将其添加到您的build.sbt文件中,以获得必要的deps
libraryDependencies += "com.typesafe.akka" % "akka-stream_2.11" % "2.4.3"
2条答案
按热度按时间zyfwsgd61#
handleWith
需要一个Flow
,即一个盒子有一个未连接的入口和一个未连接的出口。您实际上提供了一个Source
,因为您通过使用to
操作将Flow
与Sink
连接起来。我认为您可以执行以下操作:
kxe2p93d2#
另一种(在我看来更干净)的编码方式(AKKA 2.6.x),这也将强调一个事实,即你不做任何出站流,将: