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

=str - Makes GraphStageWithMaterializedValue covariant its type parameters

Browse source
This commit is contained in:
Viktor Klang 2015-10-26 16:05:42 +01:00
parent fc0ecfebef
commit f330cfb394
3 changed files with 8 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

8
akka-stream/src/main/scala/akka/stream/impl/fusing/GraphInterpreter.scala
View file

@ -87,7 +87,7 @@ private[stream] object GraphInterpreter {
* corresponding segments of these arrays matches the exact same order of the ports in the [[Shape]].
*
*/
final class GraphAssembly(val stages: Array[GraphStageWithMaterializedValue[_, _]],
final class GraphAssembly(val stages: Array[GraphStageWithMaterializedValue[Shape, Any]],
val ins: Array[Inlet[_]],
val inOwners: Array[Int],
val outs: Array[Outlet[_]],
@ -113,7 +113,7 @@ private[stream] object GraphInterpreter {
var i = 0
while (i < stages.length) {
// Port initialization loops, these must come first
val shape = stages(i).asInstanceOf[GraphStageWithMaterializedValue[Shape, _]].shape
val shape = stages(i).shape
var idx = 0
val inletItr = shape.inlets.iterator
@ -185,7 +185,7 @@ private[stream] object GraphInterpreter {
*/
final def apply(inlets: immutable.Seq[Inlet[_]],
outlets: immutable.Seq[Outlet[_]],
stages: GraphStageWithMaterializedValue[_, _]*): GraphAssembly = {
stages: GraphStageWithMaterializedValue[Shape, _]*): GraphAssembly = {
// add the contents of an iterator to an array starting at idx
@tailrec def add[T](i: Iterator[T], a: Array[T], idx: Int): Array[T] =
if (i.hasNext) {
@ -317,7 +317,7 @@ private[stream] final class GraphInterpreter(
// Counts how many active connections a stage has. Once it reaches zero, the stage is automatically stopped.
private[this] val shutdownCounter = Array.tabulate(assembly.stages.length) { i ⇒
val shape = assembly.stages(i).shape.asInstanceOf[Shape]
val shape = assembly.stages(i).shape
shape.inlets.size + shape.outlets.size
}

5
akka-stream/src/main/scala/akka/stream/stage/GraphStage.scala
View file

@ -12,13 +12,12 @@ import scala.collection.{ immutable, mutable }
import scala.concurrent.duration.FiniteDuration
import scala.collection.mutable.ArrayBuffer
abstract class GraphStageWithMaterializedValue[S <: Shape, M] extends Graph[S, M] {
def shape: S
abstract class GraphStageWithMaterializedValue[+S <: Shape, +M] extends Graph[S, M] {
def createLogicAndMaterializedValue: (GraphStageLogic, M)
final override private[stream] lazy val module: Module =
GraphModule(
GraphAssembly(shape.inlets, shape.outlets, Array(this): _*),
GraphAssembly(shape.inlets, shape.outlets, this),
shape,
Attributes.none)