Alpakka UDP: Как я могу ответить на полученные дейтаграммы через уже связанный сокет?

Question

Я использую Alpakkas UDP.bindFlow для пересылки входящих UDP-дейтаграмм брокеру Kafka. Устаревшее приложение, которое отправляет эти дейтаграммы, требует UDP-ответ от того же порта, на который было отправлено сообщение. Я изо всех сил пытаюсь смоделировать это поведение, так как это потребует от меня подключения выхода потока к его входу.

Я пробовал это решение, но оно не работает, потому что датаграмма ответа отправляется с другого исходного порта:

import java.net.InetSocketAddress
import akka.actor.ActorSystem
import akka.kafka.ProducerSettings
import akka.kafka.scaladsl.Producer
import akka.stream.ActorMaterializer
import akka.stream.alpakka.udp.Datagram
import akka.stream.alpakka.udp.scaladsl.Udp
import akka.stream.scaladsl.{Flow, Source}
import akka.util.ByteString
import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.common.serialization.StringSerializer

object UdpInput extends App {

  implicit val system: ActorSystem = ActorSystem()
  implicit val materializer: ActorMaterializer = ActorMaterializer()

  val socket = new InetSocketAddress("0.0.0.0", 40000)
  val udpBindFlow = Udp.bindFlow(socket)
  val producerSettings = ProducerSettings(system, new StringSerializer, new StringSerializer)
  val kafkaSink = Flow[Datagram].map(toProducerRecord).to(Producer.plainSink(producerSettings))

  def toProducerRecord(datagram: Datagram) = new ProducerRecord[String, String]("udp", datagram.data.utf8String)
  def toResponseDatagram(datagram: Datagram) = Datagram(ByteString("OK"), datagram.remote)

  // Does not model the behaviour I'm looking for because
  // the response datagram is sent from a different source port
  Source.asSubscriber
    .via(udpBindFlow)
    .alsoTo(kafkaSink)
    .map(toResponseDatagram)
    .to(Udp.sendSink)
    .run
}

Answer 1

В конечном итоге я использовал GraphDSL для реализации циклического графа. Спасибо dvim за то, что указал мне правильное направление!

import java.net.InetSocketAddress
import akka.actor.ActorSystem
import akka.kafka.ProducerSettings
import akka.kafka.scaladsl.Producer
import akka.stream.alpakka.udp.Datagram
import akka.stream.alpakka.udp.scaladsl.Udp
import akka.stream.scaladsl.GraphDSL.Implicits._
import akka.stream.scaladsl.{Broadcast, Flow, GraphDSL, MergePreferred, RunnableGraph, Source}
import akka.stream.{ActorMaterializer, ClosedShape}
import akka.util.ByteString
import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.common.serialization.StringSerializer

object UdpInput extends App {

  implicit val system: ActorSystem = ActorSystem()
  implicit val materializer: ActorMaterializer = ActorMaterializer()

  val producerSettings = ProducerSettings(system, new StringSerializer, new StringSerializer)
  val socket = new InetSocketAddress("0.0.0.0", 40000)
  val udpBindFlow = Udp.bindFlow(socket)
  val udpResponseFlow = Flow[Datagram].map(toResponseDatagram)
  val kafkaSink = Flow[Datagram].map(toProducerRecord).to(Producer.plainSink(producerSettings))

  def toProducerRecord(datagram: Datagram) = new ProducerRecord[String, String]("udp", datagram.data.utf8String)
  def toResponseDatagram(datagram: Datagram) = Datagram(ByteString("OK"), datagram.remote)

  RunnableGraph.fromGraph(GraphDSL.create() { implicit b =>
    val merge = b.add(MergePreferred[Datagram](1))
    val bcast = b.add(Broadcast[Datagram](2))

    Source.asSubscriber ~> merge           ~>   udpBindFlow   ~> bcast ~> kafkaSink
                           merge.preferred <~ udpResponseFlow <~ bcast
    ClosedShape
  }).run
}