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

=str - Introduces a TCK-verified SingleElementPublisher to optimize Source.single and things depending on it

Browse source
This commit is contained in:
Viktor Klang 2015-06-06 16:07:35 +02:00
parent 8527e0347e
commit b49746b0da
3 changed files with 56 additions and 11 deletions
Download patch file Download diff file Expand all files Collapse all files

14
akka-stream/src/main/scala/akka/stream/scaladsl/Source.scala
View file

@ -192,11 +192,8 @@ object Source extends SourceApply {
* Elements are pulled out of the iterator in accordance with the demand coming
* from the downstream transformation steps.
*/
def apply[T](f: () ⇒ Iterator[T]): Source[T, Unit] = {
apply(new immutable.Iterable[T] {
override def iterator: Iterator[T] = f()
})
}
def apply[T](f: () ⇒ Iterator[T]): Source[T, Unit] =
apply(new immutable.Iterable[T] { override def iterator: Iterator[T] = f() })
/**
* A graph with the shape of a source logically is a source, this method makes
@ -214,9 +211,8 @@ object Source extends SourceApply {
* stream will see an individual flow of elements (always starting from the
* beginning) regardless of when they subscribed.
*/
def apply[T](iterable: immutable.Iterable[T]): Source[T, Unit] = {
Source.single(()).mapConcat((_: Unit) ⇒ iterable).withAttributes(DefaultAttributes.iterableSource)
}
def apply[T](iterable: immutable.Iterable[T]): Source[T, Unit] =
Source.single(iterable).mapConcat(identity).withAttributes(DefaultAttributes.iterableSource)
/**
* Start a new `Source` from the given `Future`. The stream will consist of
@ -242,7 +238,7 @@ object Source extends SourceApply {
* Every connected `Sink` of this stream will see an individual stream consisting of one element.
*/
def single[T](element: T): Source[T, Unit] =
apply(SynchronousIterablePublisher(List(element), "SingleSource")).withAttributes(DefaultAttributes.singleSource) // FIXME optimize
apply(SingleElementPublisher(element, "SingleSource")).withAttributes(DefaultAttributes.singleSource) // FIXME optimize
/**
* Create a `Source` that will continually emit the given element.