将回调方法实现转换为 akka 流 Source答案

【问题标题】：Converting a callback-method implementation into an akka stream Source将回调方法实现转换为 akka 流 Source
【发布时间】：2015-06-08 18:52:23
【问题描述】：

我正在与我无法控制的 java 库中的数据发布者合作。发布者库使用典型的回调设置；库代码中的某处（库是 java，但为了简洁起见，我将在 scala 中描述）：

type DataType = ???

trait DataConsumer {
  def onData(data : DataType) : Unit
}

该库的用户需要编写一个实现onData 方法的类并将其传递给DataProducer，该库代码如下所示：

class DataProducer(consumer : DataConsumer) {...}

DataProducer 有自己无法控制的内部线程，以及伴随的数据缓冲区，只要有另一个 DataType 对象要使用，就会调用 onData。

所以，我的问题是：如何编写一个层来将原始库模式转换/翻译成 akka 流 Source 对象？

提前谢谢你。

【问题讨论】：

我认为与此答案有一些相似之处，因为您将构建一个动态源，当您在 onData 中调用 onData 时，该源将添加项目@ impl：stackoverflow.com/questions/29072963/…跨度>

标签： scala akka akka-stream

【解决方案1】：

回调 --> 来源

详细阐述 Endre Varga 的答案，下面是创建 DataConsumer 回调函数的代码，该函数会将消息发送到 akka 流 Source。

警告：创建一个功能性的 ActorPublish 比我在下面指出的要多得多。特别是，需要进行缓冲以处理DataProducer 调用onData 的速度快于Sink 发出信号需求的情况（请参阅此example）。下面的代码只是设置了“接线”。

import akka.actor.ActorRef
import akka.actor.Actor.noSender

import akka.stream.Actor.ActorPublisher

/**Incomplete ActorPublisher, see the example link.*/
class SourceActor extends ActorPublisher[DataType] {
  def receive : Receive = {
    case message : DataType => deliverBuf() //defined in example link
  }    
}

class ActorConsumer(sourceActor : ActorRef) extends DataConsumer {
  override def onData(data : DataType) = sourceActor.tell(data, noSender)
}

//setup the actor that will feed the stream Source
val sourceActorRef = actorFactory actorOf Props[SourceActor]

//setup the Consumer object that will feed the Actor
val actorConsumer = ActorConsumer(sourceActorRef)

//setup the akka stream Source
val source = Source(ActorPublisher[DataType](sourceActorRef))

//setup the incoming data feed from 3rd party library
val dataProducer  = DataProducer(actorConsumer)

回调 --> 整个流

最初的问题专门要求对 Source 进行回调，但如果整个流已经可用（不仅仅是 Source），则处理回调更容易处理。这是因为可以使用Source#actorRef 函数将流具体化为ActorRef。举个例子：

val overflowStrategy = akka.stream.OverflowStrategy.dropHead

val bufferSize = 100

val streamRef = 
  Source
    .actorRef[DataType](bufferSize, overflowStrategy)
    .via(someFlow)
    .to(someSink)
    .run()

val streamConsumer = new DataConsumer {
  override def onData(data : DataType) : Unit = streamRef ! data
} 

val dataProducer = DataProducer(streamConsumer)

【讨论】：

【解决方案2】：

有多种方法可以解决这个问题。一种是使用 ActorPublisher：http://doc.akka.io/docs/akka-stream-and-http-experimental/1.0-M5/scala/stream-integrations.html#Integrating_with_Actors，您可以在其中更改回调，以便它向演员发送消息。根据回调的工作方式，您也可以使用 mapAsync（将回调转换为 Future）。这只有在一个请求恰好产生一个回调调用时才有效。

【讨论】：