需要一个永不拉动的Sink
,用于单元测试。有一个已经可用或我需要自己编码吗?
请注意,Sink.ignore()
无济于事,因为它总是拉动。我需要一个永远不会拉的水槽。
直接回答
你可以创建一个永远不会调用org.reactivestreams.Subscriber
的Subscription.request
:
import org.reactivestreams.Subscriber
def nonSubscriber[T] : Subscriber[T] = new Subscriber[T] {
override def onComplete() : Unit = {}
override def onError(throwable: java.lang.Throwable) : Unit = {}
//should never be called therefore definition is not implemented
override def onNext(t: T) : Unit = ???
//does not call s.request
override def onSubscribe(s: Subscription) : Unit = {}
}
然后,可以使用此订阅者实例化Sink
:
import akka.NotUsed
import akka.stream.scaladsl.Sink
def nonSubscribingSink[T] : Sink[T, NotUsed] =
Sink.fromSubscriber[T](nonSubscriber[T])
间接答案
问题的本质表明您正在将“业务逻辑”与您的akka流逻辑混合在一起。您可能需要考虑a re-design,这可能使您的问题的答案变得不必要。
结束创建我自己的实现:
// sink that does not pull
val snkStage = object : GraphStage<SinkShape<Message>>() {
val shape = SinkShape(Inlet.create<Message>("in"))
override fun shape() = shape
override fun createLogic(inheritedAttributes: Attributes): GraphStageLogic = object : GraphStageLogic(shape) {
init {
setHandler(shape.`in`()) {}
}
}
}
但后来决定使用更传统的Sink.ignore()
和Source.maybe()
组合。