raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 10

=str Implement Source.never as a dedicated GraphStage.

Browse source
This commit is contained in:
He-Pin 2023-04-01 17:44:01 +08:00 • committed by kerr
parent d6ddba13d3
commit f24e3887f3
2 changed files with 17 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

15
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/GraphStages.scala
View file

@ -457,6 +457,21 @@ import pekko.stream.stage._
} }
} }
@InternalApi
private[pekko] object NeverSource extends GraphStage[SourceShape[Nothing]] {
private val out = Outlet[Nothing]("NeverSource.out")
val shape: SourceShape[Nothing] = SourceShape(out)
override def initialAttributes: Attributes = DefaultAttributes.neverSource
override def createLogic(inheritedAttributes: Attributes): GraphStageLogic with OutHandler =
new GraphStageLogic(shape) with OutHandler {
override def onPull(): Unit = ()
setHandler(out, this)
}
}
@InternalApi @InternalApi
private[pekko] object NeverSink extends GraphStageWithMaterializedValue[SinkShape[Any], Future[Done]] { private[pekko] object NeverSink extends GraphStageWithMaterializedValue[SinkShape[Any], Future[Done]] {
private val in = Inlet[Any]("NeverSink.in") private val in = Inlet[Any]("NeverSink.in")

5
stream/src/main/scala/org/apache/pekko/stream/scaladsl/Source.scala
View file

@ -293,7 +293,7 @@ object Source {
*/ */
def fromJavaStream[T, S <: java.util.stream.BaseStream[T, S]]( def fromJavaStream[T, S <: java.util.stream.BaseStream[T, S]](
stream: () => java.util.stream.BaseStream[T, S]): Source[T, NotUsed] = stream: () => java.util.stream.BaseStream[T, S]): Source[T, NotUsed] =
StreamConverters.fromJavaStream(stream); StreamConverters.fromJavaStream(stream)
/** /**
* Creates [[Source]] that will continually produce given elements in specified order. * Creates [[Source]] that will continually produce given elements in specified order.
@ -516,8 +516,7 @@ object Source {
* This stream could be useful in tests. * This stream could be useful in tests.
*/ */
def never[T]: Source[T, NotUsed] = _never def never[T]: Source[T, NotUsed] = _never
private[this] val _never: Source[Nothing, NotUsed] = private[this] val _never: Source[Nothing, NotUsed] = fromGraph(GraphStages.NeverSource)
future(Future.never).withAttributes(DefaultAttributes.neverSource)
/** /**
* Emits a single value when the given `CompletionStage` is successfully completed and then completes the stream. * Emits a single value when the given `CompletionStage` is successfully completed and then completes the stream.